diff --git a/internal/core/api.go b/internal/core/api.go index 0a31462b..edcd1d01 100644 --- a/internal/core/api.go +++ b/internal/core/api.go @@ -145,26 +145,26 @@ func loadConfPathData(ctx *gin.Context) (interface{}, error) { } type apiPathManager interface { - onAPIPathsList(req pathAPIPathsListReq) pathAPIPathsListRes + apiPathsList(req pathAPIPathsListReq) pathAPIPathsListRes } type apiRTSPServer interface { - onAPISessionsList(req rtspServerAPISessionsListReq) rtspServerAPISessionsListRes - onAPISessionsKick(req rtspServerAPISessionsKickReq) rtspServerAPISessionsKickRes + apiSessionsList(req rtspServerAPISessionsListReq) rtspServerAPISessionsListRes + apiSessionsKick(req rtspServerAPISessionsKickReq) rtspServerAPISessionsKickRes } type apiRTMPServer interface { - onAPIConnsList(req rtmpServerAPIConnsListReq) rtmpServerAPIConnsListRes - onAPIConnsKick(req rtmpServerAPIConnsKickReq) rtmpServerAPIConnsKickRes + apiConnsList(req rtmpServerAPIConnsListReq) rtmpServerAPIConnsListRes + apiConnsKick(req rtmpServerAPIConnsKickReq) rtmpServerAPIConnsKickRes } type apiHLSServer interface { - onAPIHLSMuxersList(req hlsServerAPIMuxersListReq) hlsServerAPIMuxersListRes + apiHLSMuxersList(req hlsServerAPIMuxersListReq) hlsServerAPIMuxersListRes } type apiParent interface { Log(logger.Level, string, ...interface{}) - onAPIConfigSet(conf *conf.Conf) + apiConfigSet(conf *conf.Conf) } type api struct { @@ -303,7 +303,7 @@ func (a *api) onConfigSet(ctx *gin.Context) { // since reloading the configuration can cause the shutdown of the API, // call it in a goroutine - go a.parent.onAPIConfigSet(&newConf) + go a.parent.apiConfigSet(&newConf) ctx.Status(http.StatusOK) } @@ -348,7 +348,7 @@ func (a *api) onConfigPathsAdd(ctx *gin.Context) { // since reloading the configuration can cause the shutdown of the API, // call it in a goroutine - go a.parent.onAPIConfigSet(&newConf) + go a.parent.apiConfigSet(&newConf) ctx.Status(http.StatusOK) } @@ -391,7 +391,7 @@ func (a *api) onConfigPathsEdit(ctx *gin.Context) { // since reloading the configuration can cause the shutdown of the API, // call it in a goroutine - go a.parent.onAPIConfigSet(&newConf) + go a.parent.apiConfigSet(&newConf) ctx.Status(http.StatusOK) } @@ -427,13 +427,13 @@ func (a *api) onConfigPathsDelete(ctx *gin.Context) { // since reloading the configuration can cause the shutdown of the API, // call it in a goroutine - go a.parent.onAPIConfigSet(&newConf) + go a.parent.apiConfigSet(&newConf) ctx.Status(http.StatusOK) } func (a *api) onPathsList(ctx *gin.Context) { - res := a.pathManager.onAPIPathsList(pathAPIPathsListReq{}) + res := a.pathManager.apiPathsList(pathAPIPathsListReq{}) if res.err != nil { ctx.AbortWithStatus(http.StatusInternalServerError) return @@ -443,7 +443,7 @@ func (a *api) onPathsList(ctx *gin.Context) { } func (a *api) onRTSPSessionsList(ctx *gin.Context) { - res := a.rtspServer.onAPISessionsList(rtspServerAPISessionsListReq{}) + res := a.rtspServer.apiSessionsList(rtspServerAPISessionsListReq{}) if res.err != nil { ctx.AbortWithStatus(http.StatusInternalServerError) return @@ -455,7 +455,7 @@ func (a *api) onRTSPSessionsList(ctx *gin.Context) { func (a *api) onRTSPSessionsKick(ctx *gin.Context) { id := ctx.Param("id") - res := a.rtspServer.onAPISessionsKick(rtspServerAPISessionsKickReq{id: id}) + res := a.rtspServer.apiSessionsKick(rtspServerAPISessionsKickReq{id: id}) if res.err != nil { ctx.AbortWithStatus(http.StatusNotFound) return @@ -465,7 +465,7 @@ func (a *api) onRTSPSessionsKick(ctx *gin.Context) { } func (a *api) onRTSPSSessionsList(ctx *gin.Context) { - res := a.rtspsServer.onAPISessionsList(rtspServerAPISessionsListReq{}) + res := a.rtspsServer.apiSessionsList(rtspServerAPISessionsListReq{}) if res.err != nil { ctx.AbortWithStatus(http.StatusInternalServerError) return @@ -477,7 +477,7 @@ func (a *api) onRTSPSSessionsList(ctx *gin.Context) { func (a *api) onRTSPSSessionsKick(ctx *gin.Context) { id := ctx.Param("id") - res := a.rtspsServer.onAPISessionsKick(rtspServerAPISessionsKickReq{id: id}) + res := a.rtspsServer.apiSessionsKick(rtspServerAPISessionsKickReq{id: id}) if res.err != nil { ctx.AbortWithStatus(http.StatusNotFound) return @@ -487,7 +487,7 @@ func (a *api) onRTSPSSessionsKick(ctx *gin.Context) { } func (a *api) onRTMPConnsList(ctx *gin.Context) { - res := a.rtmpServer.onAPIConnsList(rtmpServerAPIConnsListReq{}) + res := a.rtmpServer.apiConnsList(rtmpServerAPIConnsListReq{}) if res.err != nil { ctx.AbortWithStatus(http.StatusInternalServerError) return @@ -499,7 +499,7 @@ func (a *api) onRTMPConnsList(ctx *gin.Context) { func (a *api) onRTMPConnsKick(ctx *gin.Context) { id := ctx.Param("id") - res := a.rtmpServer.onAPIConnsKick(rtmpServerAPIConnsKickReq{id: id}) + res := a.rtmpServer.apiConnsKick(rtmpServerAPIConnsKickReq{id: id}) if res.err != nil { ctx.AbortWithStatus(http.StatusNotFound) return @@ -509,7 +509,7 @@ func (a *api) onRTMPConnsKick(ctx *gin.Context) { } func (a *api) onHLSMuxersList(ctx *gin.Context) { - res := a.hlsServer.onAPIHLSMuxersList(hlsServerAPIMuxersListReq{}) + res := a.hlsServer.apiHLSMuxersList(hlsServerAPIMuxersListReq{}) if res.err != nil { ctx.AbortWithStatus(http.StatusInternalServerError) return @@ -518,8 +518,8 @@ func (a *api) onHLSMuxersList(ctx *gin.Context) { ctx.JSON(http.StatusOK, res.data) } -// onConfReload is called by core. -func (a *api) onConfReload(conf *conf.Conf) { +// confReload is called by core. +func (a *api) confReload(conf *conf.Conf) { a.mutex.Lock() defer a.mutex.Unlock() a.conf = conf diff --git a/internal/core/core.go b/internal/core/core.go index 03d67dd8..9b3060d0 100644 --- a/internal/core/core.go +++ b/internal/core/core.go @@ -40,7 +40,7 @@ type Core struct { confWatcher *confwatcher.ConfWatcher // in - apiConfigSet chan *conf.Conf + chAPIConfigSet chan *conf.Conf // out done chan struct{} @@ -72,11 +72,11 @@ func New(args []string) (*Core, bool) { ctx, ctxCancel := context.WithCancel(context.Background()) p := &Core{ - ctx: ctx, - ctxCancel: ctxCancel, - confPath: *argConfPath, - apiConfigSet: make(chan *conf.Conf), - done: make(chan struct{}), + ctx: ctx, + ctxCancel: ctxCancel, + confPath: *argConfPath, + chAPIConfigSet: make(chan *conf.Conf), + done: make(chan struct{}), } var err error @@ -148,7 +148,7 @@ outer: break outer } - case newConf := <-p.apiConfigSet: + case newConf := <-p.chAPIConfigSet: p.Log(logger.Info, "reloading configuration (API request)") err := p.reloadConf(newConf, true) @@ -411,7 +411,7 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) { closeMetrics { closePathManager = true } else if !reflect.DeepEqual(newConf.Paths, p.conf.Paths) { - p.pathManager.onConfReload(newConf.Paths) + p.pathManager.confReload(newConf.Paths) } closeRTSPServer := false @@ -520,7 +520,7 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) { p.api.close() p.api = nil } else if !calledByAPI { // avoid a loop - p.api.onConfReload(newConf) + p.api.confReload(newConf) } } @@ -577,10 +577,10 @@ func (p *Core) reloadConf(newConf *conf.Conf, calledByAPI bool) error { return p.createResources(false) } -// onAPIConfigSet is called by api. -func (p *Core) onAPIConfigSet(conf *conf.Conf) { +// apiConfigSet is called by api. +func (p *Core) apiConfigSet(conf *conf.Conf) { select { - case p.apiConfigSet <- conf: + case p.chAPIConfigSet <- conf: case <-p.ctx.Done(): } } diff --git a/internal/core/core_test.go b/internal/core/core_test.go index 4eadd5e2..af501f90 100644 --- a/internal/core/core_test.go +++ b/internal/core/core_test.go @@ -219,7 +219,7 @@ func TestCorePathAutoDeletion(t *testing.T) { } }() - res := p.pathManager.onAPIPathsList(pathAPIPathsListReq{}) + res := p.pathManager.apiPathsList(pathAPIPathsListReq{}) require.NoError(t, res.err) require.Equal(t, 0, len(res.data.Items)) diff --git a/internal/core/hls_muxer.go b/internal/core/hls_muxer.go index 1ad5dcaa..75c6a268 100644 --- a/internal/core/hls_muxer.go +++ b/internal/core/hls_muxer.go @@ -102,12 +102,12 @@ type hlsMuxerRequest struct { } type hlsMuxerPathManager interface { - onReaderSetupPlay(req pathReaderSetupPlayReq) pathReaderSetupPlayRes + readerSetupPlay(req pathReaderSetupPlayReq) pathReaderSetupPlayRes } type hlsMuxerParent interface { log(logger.Level, string, ...interface{}) - onMuxerClose(*hlsMuxer) + muxerClose(*hlsMuxer) } type hlsMuxer struct { @@ -134,8 +134,8 @@ type hlsMuxer struct { requests []*hlsMuxerRequest // in - request chan *hlsMuxerRequest - hlsServerAPIMuxersList chan hlsServerAPIMuxersListSubReq + chRequest chan *hlsMuxerRequest + chAPIHLSMuxersList chan hlsServerAPIMuxersListSubReq } func newHLSMuxer( @@ -177,8 +177,8 @@ func newHLSMuxer( v := time.Now().Unix() return &v }(), - request: make(chan *hlsMuxerRequest), - hlsServerAPIMuxersList: make(chan hlsServerAPIMuxersListSubReq), + chRequest: make(chan *hlsMuxerRequest), + chAPIHLSMuxersList: make(chan hlsServerAPIMuxersListSubReq), } if req != nil { @@ -231,14 +231,14 @@ func (m *hlsMuxer) run() { <-innerErr return errors.New("terminated") - case req := <-m.request: + case req := <-m.chRequest: if isReady { req.res <- m.handleRequest(req) } else { m.requests = append(m.requests, req) } - case req := <-m.hlsServerAPIMuxersList: + case req := <-m.chAPIHLSMuxersList: req.data.Items[m.name] = hlsServerAPIMuxersListItem{ LastRequest: time.Unix(atomic.LoadInt64(m.lastRequestTime), 0).String(), } @@ -266,13 +266,13 @@ func (m *hlsMuxer) run() { } } - m.parent.onMuxerClose(m) + m.parent.muxerClose(m) m.log(logger.Info, "destroyed (%v)", err) } func (m *hlsMuxer) runInner(innerCtx context.Context, innerReady chan struct{}) error { - res := m.pathManager.onReaderSetupPlay(pathReaderSetupPlayReq{ + res := m.pathManager.readerSetupPlay(pathReaderSetupPlayReq{ author: m, pathName: m.pathName, authenticate: nil, @@ -284,7 +284,7 @@ func (m *hlsMuxer) runInner(innerCtx context.Context, innerReady chan struct{}) m.path = res.path defer func() { - m.path.onReaderRemove(pathReaderRemoveReq{author: m}) + m.path.readerRemove(pathReaderRemoveReq{author: m}) }() var videoTrack *gortsplib.TrackH264 @@ -343,7 +343,7 @@ func (m *hlsMuxer) runInner(innerCtx context.Context, innerReady chan struct{}) m.ringBuffer, _ = ringbuffer.New(uint64(m.readBufferCount)) - m.path.onReaderPlay(pathReaderPlayReq{author: m}) + m.path.readerPlay(pathReaderPlayReq{author: m}) writerDone := make(chan error) go func() { @@ -510,10 +510,10 @@ func (m *hlsMuxer) authenticate(ctx *gin.Context) error { return nil } -// onRequest is called by hlsserver.Server (forwarded from ServeHTTP). -func (m *hlsMuxer) onRequest(req *hlsMuxerRequest) { +// request is called by hlsserver.Server (forwarded from ServeHTTP). +func (m *hlsMuxer) request(req *hlsMuxerRequest) { select { - case m.request <- req: + case m.chRequest <- req: case <-m.ctx.Done(): req.res <- func() *hls.MuxerFileResponse { return &hls.MuxerFileResponse{Status: http.StatusInternalServerError} @@ -521,6 +521,17 @@ func (m *hlsMuxer) onRequest(req *hlsMuxerRequest) { } } +// apiHLSMuxersList is called by api. +func (m *hlsMuxer) apiHLSMuxersList(req hlsServerAPIMuxersListSubReq) { + req.res = make(chan struct{}) + select { + case m.chAPIHLSMuxersList <- req: + <-req.res + + case <-m.ctx.Done(): + } +} + // onReaderAccepted implements reader. func (m *hlsMuxer) onReaderAccepted() { m.log(logger.Info, "is converting into HLS") @@ -531,20 +542,9 @@ func (m *hlsMuxer) onReaderData(data *data) { m.ringBuffer.Push(data) } -// onReaderAPIDescribe implements reader. -func (m *hlsMuxer) onReaderAPIDescribe() interface{} { +// apiReaderDescribe implements reader. +func (m *hlsMuxer) apiReaderDescribe() interface{} { return struct { Type string `json:"type"` }{"hlsMuxer"} } - -// onAPIHLSMuxersList is called by api. -func (m *hlsMuxer) onAPIHLSMuxersList(req hlsServerAPIMuxersListSubReq) { - req.res = make(chan struct{}) - select { - case m.hlsServerAPIMuxersList <- req: - <-req.res - - case <-m.ctx.Done(): - } -} diff --git a/internal/core/hls_server.go b/internal/core/hls_server.go index 34b59cf2..dbda6dba 100644 --- a/internal/core/hls_server.go +++ b/internal/core/hls_server.go @@ -76,11 +76,11 @@ type hlsServer struct { muxers map[string]*hlsMuxer // in - pathSourceReady chan *path - pathSourceNotReady chan *path - request chan *hlsMuxerRequest - muxerClose chan *hlsMuxer - apiMuxersList chan hlsServerAPIMuxersListReq + chPathSourceReady chan *path + chPathSourceNotReady chan *path + request chan *hlsMuxerRequest + chMuxerClose chan *hlsMuxer + chAPIMuxerList chan hlsServerAPIMuxersListReq } func newHLSServer( @@ -142,19 +142,19 @@ func newHLSServer( ln: ln, tlsConfig: tlsConfig, muxers: make(map[string]*hlsMuxer), - pathSourceReady: make(chan *path), - pathSourceNotReady: make(chan *path), + chPathSourceReady: make(chan *path), + chPathSourceNotReady: make(chan *path), request: make(chan *hlsMuxerRequest), - muxerClose: make(chan *hlsMuxer), - apiMuxersList: make(chan hlsServerAPIMuxersListReq), + chMuxerClose: make(chan *hlsMuxer), + chAPIMuxerList: make(chan hlsServerAPIMuxersListReq), } s.log(logger.Info, "listener opened on "+address) - s.pathManager.onHLSServerSet(s) + s.pathManager.hlsServerSet(s) if s.metrics != nil { - s.metrics.onHLSServerSet(s) + s.metrics.hlsServerSet(s) } s.wg.Add(1) @@ -201,12 +201,12 @@ func (s *hlsServer) run() { outer: for { select { - case pa := <-s.pathSourceReady: + case pa := <-s.chPathSourceReady: if s.hlsAlwaysRemux { s.findOrCreateMuxer(pa.Name(), "", nil) } - case pa := <-s.pathSourceNotReady: + case pa := <-s.chPathSourceNotReady: if s.hlsAlwaysRemux { c, ok := s.muxers[pa.Name()] if ok { @@ -218,7 +218,7 @@ outer: case req := <-s.request: s.findOrCreateMuxer(req.dir, req.ctx.ClientIP(), req) - case c := <-s.muxerClose: + case c := <-s.chMuxerClose: if c2, ok := s.muxers[c.PathName()]; !ok || c2 != c { continue } @@ -228,7 +228,7 @@ outer: s.findOrCreateMuxer(c.PathName(), "", nil) } - case req := <-s.apiMuxersList: + case req := <-s.chAPIMuxerList: muxers := make(map[string]*hlsMuxer) for name, m := range s.muxers { @@ -248,10 +248,10 @@ outer: hs.Shutdown(context.Background()) - s.pathManager.onHLSServerSet(nil) + s.pathManager.hlsServerSet(nil) if s.metrics != nil { - s.metrics.onHLSServerSet(nil) + s.metrics.hlsServerSet(nil) } } @@ -359,40 +359,40 @@ func (s *hlsServer) findOrCreateMuxer(pathName string, remoteAddr string, req *h s) s.muxers[pathName] = r } else if req != nil { - r.onRequest(req) + r.request(req) } return r } -// onMuxerClose is called by hlsMuxer. -func (s *hlsServer) onMuxerClose(c *hlsMuxer) { +// muxerClose is called by hlsMuxer. +func (s *hlsServer) muxerClose(c *hlsMuxer) { select { - case s.muxerClose <- c: + case s.chMuxerClose <- c: case <-s.ctx.Done(): } } -// onPathSourceReady is called by pathManager. -func (s *hlsServer) onPathSourceReady(pa *path) { +// pathSourceReady is called by pathManager. +func (s *hlsServer) pathSourceReady(pa *path) { select { - case s.pathSourceReady <- pa: + case s.chPathSourceReady <- pa: case <-s.ctx.Done(): } } -// onPathSourceNotReady is called by pathManager. -func (s *hlsServer) onPathSourceNotReady(pa *path) { +// pathSourceNotReady is called by pathManager. +func (s *hlsServer) pathSourceNotReady(pa *path) { select { - case s.pathSourceNotReady <- pa: + case s.chPathSourceNotReady <- pa: case <-s.ctx.Done(): } } -// onAPIHLSMuxersList is called by api. -func (s *hlsServer) onAPIHLSMuxersList(req hlsServerAPIMuxersListReq) hlsServerAPIMuxersListRes { +// apiHLSMuxersList is called by api. +func (s *hlsServer) apiHLSMuxersList(req hlsServerAPIMuxersListReq) hlsServerAPIMuxersListRes { req.res = make(chan hlsServerAPIMuxersListRes) select { - case s.apiMuxersList <- req: + case s.chAPIMuxerList <- req: res := <-req.res res.data = &hlsServerAPIMuxersListData{ @@ -400,7 +400,7 @@ func (s *hlsServer) onAPIHLSMuxersList(req hlsServerAPIMuxersListReq) hlsServerA } for _, pa := range res.muxers { - pa.onAPIHLSMuxersList(hlsServerAPIMuxersListSubReq{data: res.data}) + pa.apiHLSMuxersList(hlsServerAPIMuxersListSubReq{data: res.data}) } return res diff --git a/internal/core/hls_source.go b/internal/core/hls_source.go index 32e02084..4840b3c3 100644 --- a/internal/core/hls_source.go +++ b/internal/core/hls_source.go @@ -15,8 +15,8 @@ import ( type hlsSourceParent interface { log(logger.Level, string, ...interface{}) - onSourceStaticImplSetReady(req pathSourceStaticSetReadyReq) pathSourceStaticSetReadyRes - onSourceStaticImplSetNotReady(req pathSourceStaticSetNotReadyReq) + sourceStaticImplSetReady(req pathSourceStaticSetReadyReq) pathSourceStaticSetReadyRes + sourceStaticImplSetNotReady(req pathSourceStaticSetNotReadyReq) } type hlsSource struct { @@ -51,7 +51,7 @@ func (s *hlsSource) run(ctx context.Context) error { defer func() { if stream != nil { - s.parent.onSourceStaticImplSetNotReady(pathSourceStaticSetNotReadyReq{}) + s.parent.sourceStaticImplSetNotReady(pathSourceStaticSetNotReadyReq{}) } }() @@ -78,7 +78,7 @@ func (s *hlsSource) run(ctx context.Context) error { tracks = append(tracks, audioTrack) } - res := s.parent.onSourceStaticImplSetReady(pathSourceStaticSetReadyReq{tracks: tracks}) + res := s.parent.sourceStaticImplSetReady(pathSourceStaticSetReadyReq{tracks: tracks}) if res.err != nil { return res.err } @@ -161,8 +161,8 @@ func (s *hlsSource) run(ctx context.Context) error { } } -// onSourceAPIDescribe implements sourceStaticImpl. -func (*hlsSource) onSourceAPIDescribe() interface{} { +// apiSourceDescribe implements sourceStaticImpl. +func (*hlsSource) apiSourceDescribe() interface{} { return struct { Type string `json:"type"` }{"hlsSource"} diff --git a/internal/core/metrics.go b/internal/core/metrics.go index 0ea52aaf..3968cbb9 100644 --- a/internal/core/metrics.go +++ b/internal/core/metrics.go @@ -18,19 +18,19 @@ func metric(key string, value int64) string { } type metricsPathManager interface { - onAPIPathsList(req pathAPIPathsListReq) pathAPIPathsListRes + apiPathsList(req pathAPIPathsListReq) pathAPIPathsListRes } type metricsRTSPServer interface { - onAPISessionsList(req rtspServerAPISessionsListReq) rtspServerAPISessionsListRes + apiSessionsList(req rtspServerAPISessionsListReq) rtspServerAPISessionsListRes } type metricsRTMPServer interface { - onAPIConnsList(req rtmpServerAPIConnsListReq) rtmpServerAPIConnsListRes + apiConnsList(req rtmpServerAPIConnsListReq) rtmpServerAPIConnsListRes } type metricsHLSServer interface { - onAPIHLSMuxersList(req hlsServerAPIMuxersListReq) hlsServerAPIMuxersListRes + apiHLSMuxersList(req hlsServerAPIMuxersListReq) hlsServerAPIMuxersListRes } type metricsParent interface { @@ -96,7 +96,7 @@ func (m *metrics) run() { func (m *metrics) onMetrics(ctx *gin.Context) { out := "" - res := m.pathManager.onAPIPathsList(pathAPIPathsListReq{}) + res := m.pathManager.apiPathsList(pathAPIPathsListReq{}) if res.err == nil { for name, p := range res.data.Items { if p.SourceReady { @@ -108,7 +108,7 @@ func (m *metrics) onMetrics(ctx *gin.Context) { } if !interfaceIsEmpty(m.rtspServer) { - res := m.rtspServer.onAPISessionsList(rtspServerAPISessionsListReq{}) + res := m.rtspServer.apiSessionsList(rtspServerAPISessionsListReq{}) if res.err == nil { idleCount := int64(0) readCount := int64(0) @@ -135,7 +135,7 @@ func (m *metrics) onMetrics(ctx *gin.Context) { } if !interfaceIsEmpty(m.rtspsServer) { - res := m.rtspsServer.onAPISessionsList(rtspServerAPISessionsListReq{}) + res := m.rtspsServer.apiSessionsList(rtspServerAPISessionsListReq{}) if res.err == nil { idleCount := int64(0) readCount := int64(0) @@ -162,7 +162,7 @@ func (m *metrics) onMetrics(ctx *gin.Context) { } if !interfaceIsEmpty(m.rtmpServer) { - res := m.rtmpServer.onAPIConnsList(rtmpServerAPIConnsListReq{}) + res := m.rtmpServer.apiConnsList(rtmpServerAPIConnsListReq{}) if res.err == nil { idleCount := int64(0) readCount := int64(0) @@ -189,7 +189,7 @@ func (m *metrics) onMetrics(ctx *gin.Context) { } if !interfaceIsEmpty(m.hlsServer) { - res := m.hlsServer.onAPIHLSMuxersList(hlsServerAPIMuxersListReq{}) + res := m.hlsServer.apiHLSMuxersList(hlsServerAPIMuxersListReq{}) if res.err == nil { for name := range res.data.Items { out += metric("hls_muxers{name=\""+name+"\"}", 1) @@ -201,36 +201,36 @@ func (m *metrics) onMetrics(ctx *gin.Context) { io.WriteString(ctx.Writer, out) } -// onPathManagerSet is called by pathManager. -func (m *metrics) onPathManagerSet(s metricsPathManager) { +// pathManagerSet is called by pathManager. +func (m *metrics) pathManagerSet(s metricsPathManager) { m.mutex.Lock() defer m.mutex.Unlock() m.pathManager = s } -// onRTSPServer is called by rtspServer (plain). -func (m *metrics) onRTSPServerSet(s metricsRTSPServer) { +// rtspServerSet is called by rtspServer (plain). +func (m *metrics) rtspServerSet(s metricsRTSPServer) { m.mutex.Lock() defer m.mutex.Unlock() m.rtspServer = s } -// onRTSPServer is called by rtspServer (plain). -func (m *metrics) onRTSPSServerSet(s metricsRTSPServer) { +// rtspsServerSet is called by rtspServer (tls). +func (m *metrics) rtspsServerSet(s metricsRTSPServer) { m.mutex.Lock() defer m.mutex.Unlock() m.rtspsServer = s } -// onRTMPServerSet is called by rtmpServer. -func (m *metrics) onRTMPServerSet(s metricsRTMPServer) { +// rtmpServerSet is called by rtmpServer. +func (m *metrics) rtmpServerSet(s metricsRTMPServer) { m.mutex.Lock() defer m.mutex.Unlock() m.rtmpServer = s } -// onHLSServerSet is called by hlsServer. -func (m *metrics) onHLSServerSet(s metricsHLSServer) { +// hlsServerSet is called by hlsServer. +func (m *metrics) hlsServerSet(s metricsHLSServer) { m.mutex.Lock() defer m.mutex.Unlock() m.hlsServer = s diff --git a/internal/core/path.go b/internal/core/path.go index f25364b8..3ace7af7 100644 --- a/internal/core/path.go +++ b/internal/core/path.go @@ -61,8 +61,8 @@ func (pathErrAuthCritical) Error() string { type pathParent interface { log(logger.Level, string, ...interface{}) - onPathSourceReady(*path) - onPathSourceNotReady(*path) + pathSourceReady(*path) + pathSourceNotReady(*path) onPathClose(*path) } @@ -221,7 +221,7 @@ type path struct { sourceReady bool stream *stream readers map[reader]pathReaderState - describeRequestsOnHold []pathDescribeReq + chDescribeRequestsOnHold []pathDescribeReq setupPlayRequestsOnHold []pathReaderSetupPlayReq onDemandCmd *externalcmd.Cmd onReadyCmd *externalcmd.Cmd @@ -233,18 +233,18 @@ type path struct { onDemandPublisherCloseTimer *time.Timer // in - sourceStaticSetReady chan pathSourceStaticSetReadyReq - sourceStaticSetNotReady chan pathSourceStaticSetNotReadyReq - describe chan pathDescribeReq - publisherRemove chan pathPublisherRemoveReq - publisherAnnounce chan pathPublisherAnnounceReq - publisherRecord chan pathPublisherRecordReq - publisherPause chan pathPublisherPauseReq - readerRemove chan pathReaderRemoveReq - readerSetupPlay chan pathReaderSetupPlayReq - readerPlay chan pathReaderPlayReq - readerPause chan pathReaderPauseReq - apiPathsList chan pathAPIPathsListSubReq + chSourceStaticSetReady chan pathSourceStaticSetReadyReq + chSourceStaticSetNotReady chan pathSourceStaticSetNotReadyReq + chDescribe chan pathDescribeReq + chPublisherRemove chan pathPublisherRemoveReq + chPublisherAnnounce chan pathPublisherAnnounceReq + chPublisherRecord chan pathPublisherRecordReq + chPublisherPause chan pathPublisherPauseReq + chReaderRemove chan pathReaderRemoveReq + chReaderSetupPlay chan pathReaderSetupPlayReq + chReaderPlay chan pathReaderPlayReq + chReaderPause chan pathReaderPauseReq + chAPIPathsList chan pathAPIPathsListSubReq } func newPath( @@ -282,18 +282,18 @@ func newPath( onDemandStaticSourceCloseTimer: newEmptyTimer(), onDemandPublisherReadyTimer: newEmptyTimer(), onDemandPublisherCloseTimer: newEmptyTimer(), - sourceStaticSetReady: make(chan pathSourceStaticSetReadyReq), - sourceStaticSetNotReady: make(chan pathSourceStaticSetNotReadyReq), - describe: make(chan pathDescribeReq), - publisherRemove: make(chan pathPublisherRemoveReq), - publisherAnnounce: make(chan pathPublisherAnnounceReq), - publisherRecord: make(chan pathPublisherRecordReq), - publisherPause: make(chan pathPublisherPauseReq), - readerRemove: make(chan pathReaderRemoveReq), - readerSetupPlay: make(chan pathReaderSetupPlayReq), - readerPlay: make(chan pathReaderPlayReq), - readerPause: make(chan pathReaderPauseReq), - apiPathsList: make(chan pathAPIPathsListSubReq), + chSourceStaticSetReady: make(chan pathSourceStaticSetReadyReq), + chSourceStaticSetNotReady: make(chan pathSourceStaticSetNotReadyReq), + chDescribe: make(chan pathDescribeReq), + chPublisherRemove: make(chan pathPublisherRemoveReq), + chPublisherAnnounce: make(chan pathPublisherAnnounceReq), + chPublisherRecord: make(chan pathPublisherRecordReq), + chPublisherPause: make(chan pathPublisherPauseReq), + chReaderRemove: make(chan pathReaderRemoveReq), + chReaderSetupPlay: make(chan pathReaderSetupPlayReq), + chReaderPlay: make(chan pathReaderPlayReq), + chReaderPause: make(chan pathReaderPauseReq), + chAPIPathsList: make(chan pathAPIPathsListSubReq), } pa.log(logger.Debug, "created") @@ -382,10 +382,10 @@ func (pa *path) run() { for { select { case <-pa.onDemandStaticSourceReadyTimer.C: - for _, req := range pa.describeRequestsOnHold { + for _, req := range pa.chDescribeRequestsOnHold { req.res <- pathDescribeRes{err: fmt.Errorf("source of path '%s' has timed out", pa.name)} } - pa.describeRequestsOnHold = nil + pa.chDescribeRequestsOnHold = nil for _, req := range pa.setupPlayRequestsOnHold { req.res <- pathReaderSetupPlayRes{err: fmt.Errorf("source of path '%s' has timed out", pa.name)} @@ -407,10 +407,10 @@ func (pa *path) run() { } case <-pa.onDemandPublisherReadyTimer.C: - for _, req := range pa.describeRequestsOnHold { + for _, req := range pa.chDescribeRequestsOnHold { req.res <- pathDescribeRes{err: fmt.Errorf("source of path '%s' has timed out", pa.name)} } - pa.describeRequestsOnHold = nil + pa.chDescribeRequestsOnHold = nil for _, req := range pa.setupPlayRequestsOnHold { req.res <- pathReaderSetupPlayRes{err: fmt.Errorf("source of path '%s' has timed out", pa.name)} @@ -430,7 +430,7 @@ func (pa *path) run() { return fmt.Errorf("not in use") } - case req := <-pa.sourceStaticSetReady: + case req := <-pa.chSourceStaticSetReady: pa.sourceSetReady(req.tracks) if pa.hasOnDemandStaticSource() { @@ -439,12 +439,12 @@ func (pa *path) run() { pa.onDemandStaticSourceScheduleClose() - for _, req := range pa.describeRequestsOnHold { + for _, req := range pa.chDescribeRequestsOnHold { req.res <- pathDescribeRes{ stream: pa.stream, } } - pa.describeRequestsOnHold = nil + pa.chDescribeRequestsOnHold = nil for _, req := range pa.setupPlayRequestsOnHold { pa.handleReaderSetupPlayPost(req) @@ -454,7 +454,7 @@ func (pa *path) run() { req.res <- pathSourceStaticSetReadyRes{stream: pa.stream} - case req := <-pa.sourceStaticSetNotReady: + case req := <-pa.chSourceStaticSetNotReady: pa.sourceSetNotReady() if pa.hasOnDemandStaticSource() && pa.onDemandStaticSourceState != pathOnDemandStateInitial { @@ -467,50 +467,50 @@ func (pa *path) run() { return fmt.Errorf("not in use") } - case req := <-pa.describe: + case req := <-pa.chDescribe: pa.handleDescribe(req) if pa.shouldClose() { return fmt.Errorf("not in use") } - case req := <-pa.publisherRemove: + case req := <-pa.chPublisherRemove: pa.handlePublisherRemove(req) if pa.shouldClose() { return fmt.Errorf("not in use") } - case req := <-pa.publisherAnnounce: + case req := <-pa.chPublisherAnnounce: pa.handlePublisherAnnounce(req) - case req := <-pa.publisherRecord: + case req := <-pa.chPublisherRecord: pa.handlePublisherRecord(req) - case req := <-pa.publisherPause: + case req := <-pa.chPublisherPause: pa.handlePublisherPause(req) if pa.shouldClose() { return fmt.Errorf("not in use") } - case req := <-pa.readerRemove: + case req := <-pa.chReaderRemove: pa.handleReaderRemove(req) - case req := <-pa.readerSetupPlay: + case req := <-pa.chReaderSetupPlay: pa.handleReaderSetupPlay(req) if pa.shouldClose() { return fmt.Errorf("not in use") } - case req := <-pa.readerPlay: + case req := <-pa.chReaderPlay: pa.handleReaderPlay(req) - case req := <-pa.readerPause: + case req := <-pa.chReaderPause: pa.handleReaderPause(req) - case req := <-pa.apiPathsList: + case req := <-pa.chAPIPathsList: pa.handleAPIPathsList(req) case <-pa.ctx.Done(): @@ -531,7 +531,7 @@ func (pa *path) run() { pa.log(logger.Info, "runOnInit command stopped") } - for _, req := range pa.describeRequestsOnHold { + for _, req := range pa.chDescribeRequestsOnHold { req.res <- pathDescribeRes{err: fmt.Errorf("terminated")} } @@ -565,7 +565,7 @@ func (pa *path) shouldClose() bool { return pa.conf.Regexp != nil && pa.source == nil && len(pa.readers) == 0 && - len(pa.describeRequestsOnHold) == 0 && + len(pa.chDescribeRequestsOnHold) == 0 && len(pa.setupPlayRequestsOnHold) == 0 } @@ -673,11 +673,11 @@ func (pa *path) sourceSetReady(tracks gortsplib.Tracks) { }) } - pa.parent.onPathSourceReady(pa) + pa.parent.pathSourceReady(pa) } func (pa *path) sourceSetNotReady() { - pa.parent.onPathSourceNotReady(pa) + pa.parent.pathSourceNotReady(pa) for r := range pa.readers { pa.doReaderRemove(r) @@ -739,7 +739,7 @@ func (pa *path) handleDescribe(req pathDescribeReq) { if pa.onDemandStaticSourceState == pathOnDemandStateInitial { pa.onDemandStaticSourceStart() } - pa.describeRequestsOnHold = append(pa.describeRequestsOnHold, req) + pa.chDescribeRequestsOnHold = append(pa.chDescribeRequestsOnHold, req) return } @@ -747,7 +747,7 @@ func (pa *path) handleDescribe(req pathDescribeReq) { if pa.onDemandPublisherState == pathOnDemandStateInitial { pa.onDemandPublisherStart() } - pa.describeRequestsOnHold = append(pa.describeRequestsOnHold, req) + pa.chDescribeRequestsOnHold = append(pa.chDescribeRequestsOnHold, req) return } @@ -818,12 +818,12 @@ func (pa *path) handlePublisherRecord(req pathPublisherRecordReq) { pa.onDemandPublisherScheduleClose() - for _, req := range pa.describeRequestsOnHold { + for _, req := range pa.chDescribeRequestsOnHold { req.res <- pathDescribeRes{ stream: pa.stream, } } - pa.describeRequestsOnHold = nil + pa.chDescribeRequestsOnHold = nil for _, req := range pa.setupPlayRequestsOnHold { pa.handleReaderSetupPlayPost(req) @@ -938,13 +938,13 @@ func (pa *path) handleAPIPathsList(req pathAPIPathsListSubReq) { if pa.source == nil { return nil } - return pa.source.onSourceAPIDescribe() + return pa.source.apiSourceDescribe() }(), SourceReady: pa.sourceReady, Readers: func() []interface{} { ret := []interface{}{} for r := range pa.readers { - ret = append(ret, r.onReaderAPIDescribe()) + ret = append(ret, r.apiReaderDescribe()) } return ret }(), @@ -952,10 +952,10 @@ func (pa *path) handleAPIPathsList(req pathAPIPathsListSubReq) { close(req.res) } -// onSourceStaticSetReady is called by sourceStatic. -func (pa *path) onSourceStaticSetReady(sourceStaticCtx context.Context, req pathSourceStaticSetReadyReq) { +// sourceStaticSetReady is called by sourceStatic. +func (pa *path) sourceStaticSetReady(sourceStaticCtx context.Context, req pathSourceStaticSetReadyReq) { select { - case pa.sourceStaticSetReady <- req: + case pa.chSourceStaticSetReady <- req: case <-pa.ctx.Done(): req.res <- pathSourceStaticSetReadyRes{err: fmt.Errorf("terminated")} @@ -968,10 +968,10 @@ func (pa *path) onSourceStaticSetReady(sourceStaticCtx context.Context, req path } } -// onSourceStaticSetNotReady is called by sourceStatic. -func (pa *path) onSourceStaticSetNotReady(sourceStaticCtx context.Context, req pathSourceStaticSetNotReadyReq) { +// sourceStaticSetNotReady is called by sourceStatic. +func (pa *path) sourceStaticSetNotReady(sourceStaticCtx context.Context, req pathSourceStaticSetNotReadyReq) { select { - case pa.sourceStaticSetNotReady <- req: + case pa.chSourceStaticSetNotReady <- req: case <-pa.ctx.Done(): close(req.res) @@ -984,102 +984,102 @@ func (pa *path) onSourceStaticSetNotReady(sourceStaticCtx context.Context, req p } } -// onDescribe is called by a reader or publisher through pathManager. -func (pa *path) onDescribe(req pathDescribeReq) pathDescribeRes { +// describe is called by a reader or publisher through pathManager. +func (pa *path) describe(req pathDescribeReq) pathDescribeRes { select { - case pa.describe <- req: + case pa.chDescribe <- req: return <-req.res case <-pa.ctx.Done(): return pathDescribeRes{err: fmt.Errorf("terminated")} } } -// onPublisherRemove is called by a publisher. -func (pa *path) onPublisherRemove(req pathPublisherRemoveReq) { +// publisherRemove is called by a publisher. +func (pa *path) publisherRemove(req pathPublisherRemoveReq) { req.res = make(chan struct{}) select { - case pa.publisherRemove <- req: + case pa.chPublisherRemove <- req: <-req.res case <-pa.ctx.Done(): } } -// onPublisherAnnounce is called by a publisher through pathManager. -func (pa *path) onPublisherAnnounce(req pathPublisherAnnounceReq) pathPublisherAnnounceRes { +// publisherAnnounce is called by a publisher through pathManager. +func (pa *path) publisherAnnounce(req pathPublisherAnnounceReq) pathPublisherAnnounceRes { select { - case pa.publisherAnnounce <- req: + case pa.chPublisherAnnounce <- req: return <-req.res case <-pa.ctx.Done(): return pathPublisherAnnounceRes{err: fmt.Errorf("terminated")} } } -// onPublisherRecord is called by a publisher. -func (pa *path) onPublisherRecord(req pathPublisherRecordReq) pathPublisherRecordRes { +// publisherRecord is called by a publisher. +func (pa *path) publisherRecord(req pathPublisherRecordReq) pathPublisherRecordRes { req.res = make(chan pathPublisherRecordRes) select { - case pa.publisherRecord <- req: + case pa.chPublisherRecord <- req: return <-req.res case <-pa.ctx.Done(): return pathPublisherRecordRes{err: fmt.Errorf("terminated")} } } -// onPublisherPause is called by a publisher. -func (pa *path) onPublisherPause(req pathPublisherPauseReq) { +// publisherPause is called by a publisher. +func (pa *path) publisherPause(req pathPublisherPauseReq) { req.res = make(chan struct{}) select { - case pa.publisherPause <- req: + case pa.chPublisherPause <- req: <-req.res case <-pa.ctx.Done(): } } -// onReaderRemove is called by a reader. -func (pa *path) onReaderRemove(req pathReaderRemoveReq) { +// readerRemove is called by a reader. +func (pa *path) readerRemove(req pathReaderRemoveReq) { req.res = make(chan struct{}) select { - case pa.readerRemove <- req: + case pa.chReaderRemove <- req: <-req.res case <-pa.ctx.Done(): } } -// onReaderSetupPlay is called by a reader through pathManager. -func (pa *path) onReaderSetupPlay(req pathReaderSetupPlayReq) pathReaderSetupPlayRes { +// readerSetupPlay is called by a reader through pathManager. +func (pa *path) readerSetupPlay(req pathReaderSetupPlayReq) pathReaderSetupPlayRes { select { - case pa.readerSetupPlay <- req: + case pa.chReaderSetupPlay <- req: return <-req.res case <-pa.ctx.Done(): return pathReaderSetupPlayRes{err: fmt.Errorf("terminated")} } } -// onReaderPlay is called by a reader. -func (pa *path) onReaderPlay(req pathReaderPlayReq) { +// readerPlay is called by a reader. +func (pa *path) readerPlay(req pathReaderPlayReq) { req.res = make(chan struct{}) select { - case pa.readerPlay <- req: + case pa.chReaderPlay <- req: <-req.res case <-pa.ctx.Done(): } } -// onReaderPause is called by a reader. -func (pa *path) onReaderPause(req pathReaderPauseReq) { +// readerPause is called by a reader. +func (pa *path) readerPause(req pathReaderPauseReq) { req.res = make(chan struct{}) select { - case pa.readerPause <- req: + case pa.chReaderPause <- req: <-req.res case <-pa.ctx.Done(): } } -// onAPIPathsList is called by api. -func (pa *path) onAPIPathsList(req pathAPIPathsListSubReq) { +// apiPathsList is called by api. +func (pa *path) apiPathsList(req pathAPIPathsListSubReq) { req.res = make(chan struct{}) select { - case pa.apiPathsList <- req: + case pa.chAPIPathsList <- req: <-req.res case <-pa.ctx.Done(): diff --git a/internal/core/path_manager.go b/internal/core/path_manager.go index da6003b4..70c9709a 100644 --- a/internal/core/path_manager.go +++ b/internal/core/path_manager.go @@ -11,8 +11,8 @@ import ( ) type pathManagerHLSServer interface { - onPathSourceReady(*path) - onPathSourceNotReady(*path) + pathSourceReady(*path) + pathSourceNotReady(*path) } type pathManagerParent interface { @@ -36,15 +36,15 @@ type pathManager struct { paths map[string]*path // in - confReload chan map[string]*conf.PathConf - pathClose chan *path - pathSourceReady chan *path - pathSourceNotReady chan *path - describe chan pathDescribeReq - readerSetupPlay chan pathReaderSetupPlayReq - publisherAnnounce chan pathPublisherAnnounceReq - hlsServerSet chan pathManagerHLSServer - apiPathsList chan pathAPIPathsListReq + chConfReload chan map[string]*conf.PathConf + chPathClose chan *path + chPathSourceReady chan *path + chPathSourceNotReady chan *path + chDescribe chan pathDescribeReq + chReaderSetupPlay chan pathReaderSetupPlayReq + chPublisherAnnounce chan pathPublisherAnnounceReq + chHLSServerSet chan pathManagerHLSServer + chAPIPathsList chan pathAPIPathsListReq } func newPathManager( @@ -61,26 +61,26 @@ func newPathManager( ctx, ctxCancel := context.WithCancel(parentCtx) pm := &pathManager{ - rtspAddress: rtspAddress, - readTimeout: readTimeout, - writeTimeout: writeTimeout, - readBufferCount: readBufferCount, - pathConfs: pathConfs, - externalCmdPool: externalCmdPool, - metrics: metrics, - parent: parent, - ctx: ctx, - ctxCancel: ctxCancel, - paths: make(map[string]*path), - confReload: make(chan map[string]*conf.PathConf), - pathClose: make(chan *path), - pathSourceReady: make(chan *path), - pathSourceNotReady: make(chan *path), - describe: make(chan pathDescribeReq), - readerSetupPlay: make(chan pathReaderSetupPlayReq), - publisherAnnounce: make(chan pathPublisherAnnounceReq), - hlsServerSet: make(chan pathManagerHLSServer), - apiPathsList: make(chan pathAPIPathsListReq), + rtspAddress: rtspAddress, + readTimeout: readTimeout, + writeTimeout: writeTimeout, + readBufferCount: readBufferCount, + pathConfs: pathConfs, + externalCmdPool: externalCmdPool, + metrics: metrics, + parent: parent, + ctx: ctx, + ctxCancel: ctxCancel, + paths: make(map[string]*path), + chConfReload: make(chan map[string]*conf.PathConf), + chPathClose: make(chan *path), + chPathSourceReady: make(chan *path), + chPathSourceNotReady: make(chan *path), + chDescribe: make(chan pathDescribeReq), + chReaderSetupPlay: make(chan pathReaderSetupPlayReq), + chPublisherAnnounce: make(chan pathPublisherAnnounceReq), + chHLSServerSet: make(chan pathManagerHLSServer), + chAPIPathsList: make(chan pathAPIPathsListReq), } for pathConfName, pathConf := range pm.pathConfs { @@ -90,7 +90,7 @@ func newPathManager( } if pm.metrics != nil { - pm.metrics.onPathManagerSet(pm) + pm.metrics.pathManagerSet(pm) } pm.log(logger.Debug, "path manager created") @@ -118,7 +118,7 @@ func (pm *pathManager) run() { outer: for { select { - case pathConfs := <-pm.confReload: + case pathConfs := <-pm.chConfReload: // remove confs for pathConfName := range pm.pathConfs { if _, ok := pathConfs[pathConfName]; !ok { @@ -156,24 +156,24 @@ outer: } } - case pa := <-pm.pathClose: + case pa := <-pm.chPathClose: if pmpa, ok := pm.paths[pa.Name()]; !ok || pmpa != pa { continue } delete(pm.paths, pa.Name()) pa.close() - case pa := <-pm.pathSourceReady: + case pa := <-pm.chPathSourceReady: if pm.hlsServer != nil { - pm.hlsServer.onPathSourceReady(pa) + pm.hlsServer.pathSourceReady(pa) } - case pa := <-pm.pathSourceNotReady: + case pa := <-pm.chPathSourceNotReady: if pm.hlsServer != nil { - pm.hlsServer.onPathSourceNotReady(pa) + pm.hlsServer.pathSourceNotReady(pa) } - case req := <-pm.describe: + case req := <-pm.chDescribe: pathConfName, pathConf, pathMatches, err := pm.findPathConf(req.pathName) if err != nil { req.res <- pathDescribeRes{err: err} @@ -196,7 +196,7 @@ outer: req.res <- pathDescribeRes{path: pm.paths[req.pathName]} - case req := <-pm.readerSetupPlay: + case req := <-pm.chReaderSetupPlay: pathConfName, pathConf, pathMatches, err := pm.findPathConf(req.pathName) if err != nil { req.res <- pathReaderSetupPlayRes{err: err} @@ -221,7 +221,7 @@ outer: req.res <- pathReaderSetupPlayRes{path: pm.paths[req.pathName]} - case req := <-pm.publisherAnnounce: + case req := <-pm.chPublisherAnnounce: pathConfName, pathConf, pathMatches, err := pm.findPathConf(req.pathName) if err != nil { req.res <- pathPublisherAnnounceRes{err: err} @@ -244,10 +244,10 @@ outer: req.res <- pathPublisherAnnounceRes{path: pm.paths[req.pathName]} - case s := <-pm.hlsServerSet: + case s := <-pm.chHLSServerSet: pm.hlsServer = s - case req := <-pm.apiPathsList: + case req := <-pm.chAPIPathsList: paths := make(map[string]*path) for name, pa := range pm.paths { @@ -266,7 +266,7 @@ outer: pm.ctxCancel() if pm.metrics != nil { - pm.metrics.onPathManagerSet(nil) + pm.metrics.pathManagerSet(nil) } } @@ -315,26 +315,26 @@ func (pm *pathManager) findPathConf(name string) (string, *conf.PathConf, []stri return "", nil, nil, fmt.Errorf("path '%s' is not configured", name) } -// onConfReload is called by core. -func (pm *pathManager) onConfReload(pathConfs map[string]*conf.PathConf) { +// confReload is called by core. +func (pm *pathManager) confReload(pathConfs map[string]*conf.PathConf) { select { - case pm.confReload <- pathConfs: + case pm.chConfReload <- pathConfs: case <-pm.ctx.Done(): } } -// onPathSourceReady is called by path. -func (pm *pathManager) onPathSourceReady(pa *path) { +// pathSourceReady is called by path. +func (pm *pathManager) pathSourceReady(pa *path) { select { - case pm.pathSourceReady <- pa: + case pm.chPathSourceReady <- pa: case <-pm.ctx.Done(): } } -// onPathSourceNotReady is called by path. -func (pm *pathManager) onPathSourceNotReady(pa *path) { +// pathSourceNotReady is called by path. +func (pm *pathManager) pathSourceNotReady(pa *path) { select { - case pm.pathSourceNotReady <- pa: + case pm.chPathSourceNotReady <- pa: case <-pm.ctx.Done(): } } @@ -342,75 +342,75 @@ func (pm *pathManager) onPathSourceNotReady(pa *path) { // onPathClose is called by path. func (pm *pathManager) onPathClose(pa *path) { select { - case pm.pathClose <- pa: + case pm.chPathClose <- pa: case <-pm.ctx.Done(): } } -// onDescribe is called by a reader or publisher. -func (pm *pathManager) onDescribe(req pathDescribeReq) pathDescribeRes { +// describe is called by a reader or publisher. +func (pm *pathManager) describe(req pathDescribeReq) pathDescribeRes { req.res = make(chan pathDescribeRes) select { - case pm.describe <- req: + case pm.chDescribe <- req: res := <-req.res if res.err != nil { return res } - return res.path.onDescribe(req) + return res.path.describe(req) case <-pm.ctx.Done(): return pathDescribeRes{err: fmt.Errorf("terminated")} } } -// onPublisherAnnounce is called by a publisher. -func (pm *pathManager) onPublisherAnnounce(req pathPublisherAnnounceReq) pathPublisherAnnounceRes { +// publisherAnnounce is called by a publisher. +func (pm *pathManager) publisherAnnounce(req pathPublisherAnnounceReq) pathPublisherAnnounceRes { req.res = make(chan pathPublisherAnnounceRes) select { - case pm.publisherAnnounce <- req: + case pm.chPublisherAnnounce <- req: res := <-req.res if res.err != nil { return res } - return res.path.onPublisherAnnounce(req) + return res.path.publisherAnnounce(req) case <-pm.ctx.Done(): return pathPublisherAnnounceRes{err: fmt.Errorf("terminated")} } } -// onReaderSetupPlay is called by a reader. -func (pm *pathManager) onReaderSetupPlay(req pathReaderSetupPlayReq) pathReaderSetupPlayRes { +// readerSetupPlay is called by a reader. +func (pm *pathManager) readerSetupPlay(req pathReaderSetupPlayReq) pathReaderSetupPlayRes { req.res = make(chan pathReaderSetupPlayRes) select { - case pm.readerSetupPlay <- req: + case pm.chReaderSetupPlay <- req: res := <-req.res if res.err != nil { return res } - return res.path.onReaderSetupPlay(req) + return res.path.readerSetupPlay(req) case <-pm.ctx.Done(): return pathReaderSetupPlayRes{err: fmt.Errorf("terminated")} } } -// onHLSServerSet is called by hlsServer. -func (pm *pathManager) onHLSServerSet(s pathManagerHLSServer) { +// hlsServerSet is called by hlsServer. +func (pm *pathManager) hlsServerSet(s pathManagerHLSServer) { select { - case pm.hlsServerSet <- s: + case pm.chHLSServerSet <- s: case <-pm.ctx.Done(): } } -// onAPIPathsList is called by api. -func (pm *pathManager) onAPIPathsList(req pathAPIPathsListReq) pathAPIPathsListRes { +// apiPathsList is called by api. +func (pm *pathManager) apiPathsList(req pathAPIPathsListReq) pathAPIPathsListRes { req.res = make(chan pathAPIPathsListRes) select { - case pm.apiPathsList <- req: + case pm.chAPIPathsList <- req: res := <-req.res res.data = &pathAPIPathsListData{ @@ -418,7 +418,7 @@ func (pm *pathManager) onAPIPathsList(req pathAPIPathsListReq) pathAPIPathsListR } for _, pa := range res.paths { - pa.onAPIPathsList(pathAPIPathsListSubReq{data: res.data}) + pa.apiPathsList(pathAPIPathsListSubReq{data: res.data}) } return res diff --git a/internal/core/reader.go b/internal/core/reader.go index bad5c3db..ccf037ea 100644 --- a/internal/core/reader.go +++ b/internal/core/reader.go @@ -5,5 +5,5 @@ type reader interface { close() onReaderAccepted() onReaderData(*data) - onReaderAPIDescribe() interface{} + apiReaderDescribe() interface{} } diff --git a/internal/core/rtmp_conn.go b/internal/core/rtmp_conn.go index b6f3d09f..752d4f37 100644 --- a/internal/core/rtmp_conn.go +++ b/internal/core/rtmp_conn.go @@ -47,13 +47,13 @@ const ( ) type rtmpConnPathManager interface { - onReaderSetupPlay(req pathReaderSetupPlayReq) pathReaderSetupPlayRes - onPublisherAnnounce(req pathPublisherAnnounceReq) pathPublisherAnnounceRes + readerSetupPlay(req pathReaderSetupPlayReq) pathReaderSetupPlayRes + publisherAnnounce(req pathPublisherAnnounceReq) pathPublisherAnnounceRes } type rtmpConnParent interface { log(logger.Level, string, ...interface{}) - onConnClose(*rtmpConn) + connClose(*rtmpConn) } type rtmpConn struct { @@ -199,7 +199,7 @@ func (c *rtmpConn) run() { c.ctxCancel() - c.parent.onConnClose(c) + c.parent.connClose(c) c.log(logger.Info, "closed (%v)", err) } @@ -226,7 +226,7 @@ func (c *rtmpConn) runInner(ctx context.Context) error { func (c *rtmpConn) runRead(ctx context.Context, u *url.URL) error { pathName, query, rawQuery := pathNameAndQuery(u) - res := c.pathManager.onReaderSetupPlay(pathReaderSetupPlayReq{ + res := c.pathManager.readerSetupPlay(pathReaderSetupPlayReq{ author: c, pathName: pathName, authenticate: func( @@ -250,7 +250,7 @@ func (c *rtmpConn) runRead(ctx context.Context, u *url.URL) error { c.path = res.path defer func() { - c.path.onReaderRemove(pathReaderRemoveReq{author: c}) + c.path.readerRemove(pathReaderRemoveReq{author: c}) }() c.stateMutex.Lock() @@ -307,7 +307,7 @@ func (c *rtmpConn) runRead(ctx context.Context, u *url.URL) error { c.ringBuffer.Close() }() - c.path.onReaderPlay(pathReaderPlayReq{ + c.path.readerPlay(pathReaderPlayReq{ author: c, }) @@ -519,7 +519,7 @@ func (c *rtmpConn) runPublish(ctx context.Context, u *url.URL) error { pathName, query, rawQuery := pathNameAndQuery(u) - res := c.pathManager.onPublisherAnnounce(pathPublisherAnnounceReq{ + res := c.pathManager.publisherAnnounce(pathPublisherAnnounceReq{ author: c, pathName: pathName, authenticate: func( @@ -543,7 +543,7 @@ func (c *rtmpConn) runPublish(ctx context.Context, u *url.URL) error { c.path = res.path defer func() { - c.path.onPublisherRemove(pathPublisherRemoveReq{author: c}) + c.path.publisherRemove(pathPublisherRemoveReq{author: c}) }() c.stateMutex.Lock() @@ -553,7 +553,7 @@ func (c *rtmpConn) runPublish(ctx context.Context, u *url.URL) error { // disable write deadline c.nconn.SetWriteDeadline(time.Time{}) - rres := c.path.onPublisherRecord(pathPublisherRecordReq{ + rres := c.path.publisherRecord(pathPublisherRecordReq{ author: c, tracks: tracks, }) @@ -742,16 +742,16 @@ func (c *rtmpConn) onReaderData(data *data) { c.ringBuffer.Push(data) } -// onReaderAPIDescribe implements reader. -func (c *rtmpConn) onReaderAPIDescribe() interface{} { +// apiReaderDescribe implements reader. +func (c *rtmpConn) apiReaderDescribe() interface{} { return struct { Type string `json:"type"` ID string `json:"id"` }{"rtmpConn", c.id} } -// onSourceAPIDescribe implements source. -func (c *rtmpConn) onSourceAPIDescribe() interface{} { +// apiSourceDescribe implements source. +func (c *rtmpConn) apiSourceDescribe() interface{} { return struct { Type string `json:"type"` ID string `json:"id"` diff --git a/internal/core/rtmp_server.go b/internal/core/rtmp_server.go index fe12f160..ce6741c3 100644 --- a/internal/core/rtmp_server.go +++ b/internal/core/rtmp_server.go @@ -64,9 +64,9 @@ type rtmpServer struct { conns map[*rtmpConn]struct{} // in - connClose chan *rtmpConn - apiConnsList chan rtmpServerAPIConnsListReq - apiConnsKick chan rtmpServerAPIConnsKickReq + chConnClose chan *rtmpConn + chAPIConnsList chan rtmpServerAPIConnsListReq + chAPIConnsKick chan rtmpServerAPIConnsKickReq } func newRTMPServer( @@ -107,15 +107,15 @@ func newRTMPServer( ctxCancel: ctxCancel, l: l, conns: make(map[*rtmpConn]struct{}), - connClose: make(chan *rtmpConn), - apiConnsList: make(chan rtmpServerAPIConnsListReq), - apiConnsKick: make(chan rtmpServerAPIConnsKickReq), + chConnClose: make(chan *rtmpConn), + chAPIConnsList: make(chan rtmpServerAPIConnsListReq), + chAPIConnsKick: make(chan rtmpServerAPIConnsKickReq), } s.log(logger.Info, "listener opened on %s", address) if s.metrics != nil { - s.metrics.onRTMPServerSet(s) + s.metrics.rtmpServerSet(s) } s.wg.Add(1) @@ -190,13 +190,13 @@ outer: s) s.conns[c] = struct{}{} - case c := <-s.connClose: + case c := <-s.chConnClose: if _, ok := s.conns[c]; !ok { continue } delete(s.conns, c) - case req := <-s.apiConnsList: + case req := <-s.chAPIConnsList: data := &rtmpServerAPIConnsListData{ Items: make(map[string]rtmpServerAPIConnsListItem), } @@ -219,7 +219,7 @@ outer: req.res <- rtmpServerAPIConnsListRes{data: data} - case req := <-s.apiConnsKick: + case req := <-s.chAPIConnsKick: res := func() bool { for c := range s.conns { if c.ID() == req.id { @@ -246,7 +246,7 @@ outer: s.l.Close() if s.metrics != nil { - s.metrics.onRTMPServerSet(s) + s.metrics.rtmpServerSet(s) } } @@ -278,19 +278,19 @@ func (s *rtmpServer) newConnID() (string, error) { } } -// onConnClose is called by rtmpConn. -func (s *rtmpServer) onConnClose(c *rtmpConn) { +// connClose is called by rtmpConn. +func (s *rtmpServer) connClose(c *rtmpConn) { select { - case s.connClose <- c: + case s.chConnClose <- c: case <-s.ctx.Done(): } } -// onAPIConnsList is called by api. -func (s *rtmpServer) onAPIConnsList(req rtmpServerAPIConnsListReq) rtmpServerAPIConnsListRes { +// apiConnsList is called by api. +func (s *rtmpServer) apiConnsList(req rtmpServerAPIConnsListReq) rtmpServerAPIConnsListRes { req.res = make(chan rtmpServerAPIConnsListRes) select { - case s.apiConnsList <- req: + case s.chAPIConnsList <- req: return <-req.res case <-s.ctx.Done(): @@ -298,11 +298,11 @@ func (s *rtmpServer) onAPIConnsList(req rtmpServerAPIConnsListReq) rtmpServerAPI } } -// onAPIConnsKick is called by api. -func (s *rtmpServer) onAPIConnsKick(req rtmpServerAPIConnsKickReq) rtmpServerAPIConnsKickRes { +// apiConnsKick is called by api. +func (s *rtmpServer) apiConnsKick(req rtmpServerAPIConnsKickReq) rtmpServerAPIConnsKickRes { req.res = make(chan rtmpServerAPIConnsKickRes) select { - case s.apiConnsKick <- req: + case s.chAPIConnsKick <- req: return <-req.res case <-s.ctx.Done(): diff --git a/internal/core/rtmp_source.go b/internal/core/rtmp_source.go index 2b7d2309..614ff005 100644 --- a/internal/core/rtmp_source.go +++ b/internal/core/rtmp_source.go @@ -21,8 +21,8 @@ import ( type rtmpSourceParent interface { log(logger.Level, string, ...interface{}) - onSourceStaticImplSetReady(req pathSourceStaticSetReadyReq) pathSourceStaticSetReadyRes - onSourceStaticImplSetNotReady(req pathSourceStaticSetNotReadyReq) + sourceStaticImplSetReady(req pathSourceStaticSetReadyReq) pathSourceStaticSetReadyRes + sourceStaticImplSetNotReady(req pathSourceStaticSetNotReadyReq) } type rtmpSource struct { @@ -119,7 +119,7 @@ func (s *rtmpSource) run(ctx context.Context) error { tracks = append(tracks, audioTrack) } - res := s.parent.onSourceStaticImplSetReady(pathSourceStaticSetReadyReq{tracks: tracks}) + res := s.parent.sourceStaticImplSetReady(pathSourceStaticSetReadyReq{tracks: tracks}) if res.err != nil { return res.err } @@ -127,7 +127,7 @@ func (s *rtmpSource) run(ctx context.Context) error { s.Log(logger.Info, "ready") defer func() { - s.parent.onSourceStaticImplSetNotReady(pathSourceStaticSetNotReadyReq{}) + s.parent.sourceStaticImplSetNotReady(pathSourceStaticSetNotReadyReq{}) }() for { @@ -212,8 +212,8 @@ func (s *rtmpSource) run(ctx context.Context) error { } } -// onSourceAPIDescribe implements sourceStaticImpl. -func (*rtmpSource) onSourceAPIDescribe() interface{} { +// apiSourceDescribe implements sourceStaticImpl. +func (*rtmpSource) apiSourceDescribe() interface{} { return struct { Type string `json:"type"` }{"rtmpSource"} diff --git a/internal/core/rtsp_conn.go b/internal/core/rtsp_conn.go index 92038252..c0e19384 100644 --- a/internal/core/rtsp_conn.go +++ b/internal/core/rtsp_conn.go @@ -243,7 +243,7 @@ func (c *rtspConn) OnResponse(res *base.Response) { // onDescribe is called by rtspServer. func (c *rtspConn) onDescribe(ctx *gortsplib.ServerHandlerOnDescribeCtx, ) (*base.Response, *gortsplib.ServerStream, error) { - res := c.pathManager.onDescribe(pathDescribeReq{ + res := c.pathManager.describe(pathDescribeReq{ pathName: ctx.Path, url: ctx.Request.URL, authenticate: func( diff --git a/internal/core/rtsp_server.go b/internal/core/rtsp_server.go index a072a506..22fc4594 100644 --- a/internal/core/rtsp_server.go +++ b/internal/core/rtsp_server.go @@ -167,9 +167,9 @@ func newRTSPServer( if s.metrics != nil { if !isTLS { - s.metrics.onRTSPServerSet(s) + s.metrics.rtspServerSet(s) } else { - s.metrics.onRTSPSServerSet(s) + s.metrics.rtspsServerSet(s) } } @@ -219,9 +219,9 @@ outer: if s.metrics != nil { if !s.isTLS { - s.metrics.onRTSPServerSet(nil) + s.metrics.rtspServerSet(nil) } else { - s.metrics.onRTSPSServerSet(nil) + s.metrics.rtspsServerSet(nil) } } } @@ -392,8 +392,8 @@ func (s *rtspServer) OnPacketRTP(ctx *gortsplib.ServerHandlerOnPacketRTPCtx) { se.onPacketRTP(ctx) } -// onAPISessionsList is called by api and metrics. -func (s *rtspServer) onAPISessionsList(req rtspServerAPISessionsListReq) rtspServerAPISessionsListRes { +// apiSessionsList is called by api and metrics. +func (s *rtspServer) apiSessionsList(req rtspServerAPISessionsListReq) rtspServerAPISessionsListRes { select { case <-s.ctx.Done(): return rtspServerAPISessionsListRes{err: fmt.Errorf("terminated")} @@ -428,8 +428,8 @@ func (s *rtspServer) onAPISessionsList(req rtspServerAPISessionsListReq) rtspSer return rtspServerAPISessionsListRes{data: data} } -// onAPISessionsKick is called by api. -func (s *rtspServer) onAPISessionsKick(req rtspServerAPISessionsKickReq) rtspServerAPISessionsKickRes { +// apiSessionsKick is called by api. +func (s *rtspServer) apiSessionsKick(req rtspServerAPISessionsKickReq) rtspServerAPISessionsKickRes { select { case <-s.ctx.Done(): return rtspServerAPISessionsKickRes{err: fmt.Errorf("terminated")} diff --git a/internal/core/rtsp_session.go b/internal/core/rtsp_session.go index 8b8cc999..8841078a 100644 --- a/internal/core/rtsp_session.go +++ b/internal/core/rtsp_session.go @@ -20,8 +20,8 @@ const ( ) type rtspSessionPathManager interface { - onPublisherAnnounce(req pathPublisherAnnounceReq) pathPublisherAnnounceRes - onReaderSetupPlay(req pathReaderSetupPlayReq) pathReaderSetupPlayRes + publisherAnnounce(req pathPublisherAnnounceReq) pathPublisherAnnounceRes + readerSetupPlay(req pathReaderSetupPlayReq) pathReaderSetupPlayRes } type rtspSessionParent interface { @@ -112,11 +112,11 @@ func (s *rtspSession) onClose(err error) { switch s.ss.State() { case gortsplib.ServerSessionStatePrePlay, gortsplib.ServerSessionStatePlay: - s.path.onReaderRemove(pathReaderRemoveReq{author: s}) + s.path.readerRemove(pathReaderRemoveReq{author: s}) s.path = nil case gortsplib.ServerSessionStatePreRecord, gortsplib.ServerSessionStateRecord: - s.path.onPublisherRemove(pathPublisherRemoveReq{author: s}) + s.path.publisherRemove(pathPublisherRemoveReq{author: s}) s.path = nil } @@ -125,7 +125,7 @@ func (s *rtspSession) onClose(err error) { // onAnnounce is called by rtspServer. func (s *rtspSession) onAnnounce(c *rtspConn, ctx *gortsplib.ServerHandlerOnAnnounceCtx) (*base.Response, error) { - res := s.pathManager.onPublisherAnnounce(pathPublisherAnnounceReq{ + res := s.pathManager.publisherAnnounce(pathPublisherAnnounceReq{ author: s, pathName: ctx.Path, authenticate: func( @@ -185,7 +185,7 @@ func (s *rtspSession) onSetup(c *rtspConn, ctx *gortsplib.ServerHandlerOnSetupCt switch s.ss.State() { case gortsplib.ServerSessionStateInitial, gortsplib.ServerSessionStatePrePlay: // play - res := s.pathManager.onReaderSetupPlay(pathReaderSetupPlayReq{ + res := s.pathManager.readerSetupPlay(pathReaderSetupPlayReq{ author: s, pathName: ctx.Path, authenticate: func( @@ -249,7 +249,7 @@ func (s *rtspSession) onPlay(ctx *gortsplib.ServerHandlerOnPlayCtx) (*base.Respo h := make(base.Header) if s.ss.State() == gortsplib.ServerSessionStatePrePlay { - s.path.onReaderPlay(pathReaderPlayReq{author: s}) + s.path.readerPlay(pathReaderPlayReq{author: s}) if s.path.Conf().RunOnRead != "" { s.log(logger.Info, "runOnRead command started") @@ -276,7 +276,7 @@ func (s *rtspSession) onPlay(ctx *gortsplib.ServerHandlerOnPlayCtx) (*base.Respo // onRecord is called by rtspServer. func (s *rtspSession) onRecord(ctx *gortsplib.ServerHandlerOnRecordCtx) (*base.Response, error) { - res := s.path.onPublisherRecord(pathPublisherRecordReq{ + res := s.path.publisherRecord(pathPublisherRecordReq{ author: s, tracks: s.announcedTracks, }) @@ -306,14 +306,14 @@ func (s *rtspSession) onPause(ctx *gortsplib.ServerHandlerOnPauseCtx) (*base.Res s.onReadCmd.Close() } - s.path.onReaderPause(pathReaderPauseReq{author: s}) + s.path.readerPause(pathReaderPauseReq{author: s}) s.stateMutex.Lock() s.state = gortsplib.ServerSessionStatePrePlay s.stateMutex.Unlock() case gortsplib.ServerSessionStateRecord: - s.path.onPublisherPause(pathPublisherPauseReq{author: s}) + s.path.publisherPause(pathPublisherPauseReq{author: s}) s.stateMutex.Lock() s.state = gortsplib.ServerSessionStatePreRecord @@ -346,8 +346,8 @@ func (s *rtspSession) onReaderData(data *data) { // packets are routed to the session by gortsplib.ServerStream. } -// onReaderAPIDescribe implements reader. -func (s *rtspSession) onReaderAPIDescribe() interface{} { +// apiReaderDescribe implements reader. +func (s *rtspSession) apiReaderDescribe() interface{} { var typ string if s.isTLS { typ = "rtspsSession" @@ -361,8 +361,8 @@ func (s *rtspSession) onReaderAPIDescribe() interface{} { }{typ, s.id} } -// onSourceAPIDescribe implements source. -func (s *rtspSession) onSourceAPIDescribe() interface{} { +// apiSourceDescribe implements source. +func (s *rtspSession) apiSourceDescribe() interface{} { var typ string if s.isTLS { typ = "rtspsSession" diff --git a/internal/core/rtsp_source.go b/internal/core/rtsp_source.go index c807f8d7..e8177848 100644 --- a/internal/core/rtsp_source.go +++ b/internal/core/rtsp_source.go @@ -19,8 +19,8 @@ import ( type rtspSourceParent interface { log(logger.Level, string, ...interface{}) - onSourceStaticImplSetReady(req pathSourceStaticSetReadyReq) pathSourceStaticSetReadyRes - onSourceStaticImplSetNotReady(req pathSourceStaticSetNotReadyReq) + sourceStaticImplSetReady(req pathSourceStaticSetReadyReq) pathSourceStaticSetReadyRes + sourceStaticImplSetNotReady(req pathSourceStaticSetNotReadyReq) } type rtspSource struct { @@ -125,7 +125,7 @@ func (s *rtspSource) run(ctx context.Context) error { } } - res := s.parent.onSourceStaticImplSetReady(pathSourceStaticSetReadyReq{tracks: tracks}) + res := s.parent.sourceStaticImplSetReady(pathSourceStaticSetReadyReq{tracks: tracks}) if res.err != nil { return res.err } @@ -133,7 +133,7 @@ func (s *rtspSource) run(ctx context.Context) error { s.Log(logger.Info, "ready") defer func() { - s.parent.onSourceStaticImplSetNotReady(pathSourceStaticSetNotReadyReq{}) + s.parent.sourceStaticImplSetNotReady(pathSourceStaticSetNotReadyReq{}) }() c.OnPacketRTP = func(ctx *gortsplib.ClientOnPacketRTPCtx) { @@ -174,8 +174,8 @@ func (s *rtspSource) run(ctx context.Context) error { } } -// onSourceAPIDescribe implements sourceStaticImpl. -func (*rtspSource) onSourceAPIDescribe() interface{} { +// apiSourceDescribe implements sourceStaticImpl. +func (*rtspSource) apiSourceDescribe() interface{} { return struct { Type string `json:"type"` }{"rtspSource"} diff --git a/internal/core/source.go b/internal/core/source.go index 924f8bb7..390694a9 100644 --- a/internal/core/source.go +++ b/internal/core/source.go @@ -6,5 +6,5 @@ package core // - sourceStatic // - sourceRedirect type source interface { - onSourceAPIDescribe() interface{} + apiSourceDescribe() interface{} } diff --git a/internal/core/source_redirect.go b/internal/core/source_redirect.go index b24535cd..3c20e8a0 100644 --- a/internal/core/source_redirect.go +++ b/internal/core/source_redirect.go @@ -3,8 +3,8 @@ package core // sourceRedirect is a source that redirects to another one. type sourceRedirect struct{} -// onSourceAPIDescribe implements source. -func (*sourceRedirect) onSourceAPIDescribe() interface{} { +// apiSourceDescribe implements source. +func (*sourceRedirect) apiSourceDescribe() interface{} { return struct { Type string `json:"type"` }{"redirect"} diff --git a/internal/core/source_static.go b/internal/core/source_static.go index f542b807..277fd44d 100644 --- a/internal/core/source_static.go +++ b/internal/core/source_static.go @@ -17,13 +17,13 @@ const ( type sourceStaticImpl interface { Log(logger.Level, string, ...interface{}) run(context.Context) error - onSourceAPIDescribe() interface{} + apiSourceDescribe() interface{} } type sourceStaticParent interface { log(logger.Level, string, ...interface{}) - onSourceStaticSetReady(context.Context, pathSourceStaticSetReadyReq) - onSourceStaticSetNotReady(context.Context, pathSourceStaticSetNotReadyReq) + sourceStaticSetReady(context.Context, pathSourceStaticSetReadyReq) + sourceStaticSetNotReady(context.Context, pathSourceStaticSetNotReadyReq) } // sourceStatic is a static source. @@ -42,9 +42,9 @@ type sourceStatic struct { impl sourceStaticImpl running bool - done chan struct{} - sourceStaticImplSetReady chan pathSourceStaticSetReadyReq - sourceStaticImplSetNotReady chan pathSourceStaticSetNotReadyReq + done chan struct{} + chSourceStaticImplSetReady chan pathSourceStaticSetReadyReq + chSourceStaticImplSetNotReady chan pathSourceStaticSetNotReadyReq } func newSourceStatic( @@ -58,16 +58,16 @@ func newSourceStatic( parent sourceStaticParent, ) *sourceStatic { s := &sourceStatic{ - ur: ur, - protocol: protocol, - anyPortEnable: anyPortEnable, - fingerprint: fingerprint, - readTimeout: readTimeout, - writeTimeout: writeTimeout, - readBufferCount: readBufferCount, - parent: parent, - sourceStaticImplSetReady: make(chan pathSourceStaticSetReadyReq), - sourceStaticImplSetNotReady: make(chan pathSourceStaticSetNotReadyReq), + ur: ur, + protocol: protocol, + anyPortEnable: anyPortEnable, + fingerprint: fingerprint, + readTimeout: readTimeout, + writeTimeout: writeTimeout, + readBufferCount: readBufferCount, + parent: parent, + chSourceStaticImplSetReady: make(chan pathSourceStaticSetReadyReq), + chSourceStaticImplSetNotReady: make(chan pathSourceStaticSetNotReadyReq), } switch { @@ -170,11 +170,11 @@ func (s *sourceStatic) runInner() { s.impl.Log(logger.Info, "ERR: %v", err) return - case req := <-s.sourceStaticImplSetReady: - s.parent.onSourceStaticSetReady(s.ctx, req) + case req := <-s.chSourceStaticImplSetReady: + s.parent.sourceStaticSetReady(s.ctx, req) - case req := <-s.sourceStaticImplSetNotReady: - s.parent.onSourceStaticSetNotReady(s.ctx, req) + case req := <-s.chSourceStaticImplSetNotReady: + s.parent.sourceStaticSetNotReady(s.ctx, req) case <-s.ctx.Done(): innerCtxCancel() @@ -184,27 +184,27 @@ func (s *sourceStatic) runInner() { } } -// onSourceAPIDescribe implements source. -func (s *sourceStatic) onSourceAPIDescribe() interface{} { - return s.impl.onSourceAPIDescribe() +// apiSourceDescribe implements source. +func (s *sourceStatic) apiSourceDescribe() interface{} { + return s.impl.apiSourceDescribe() } -// onSourceStaticImplSetReady is called by a sourceStaticImpl. -func (s *sourceStatic) onSourceStaticImplSetReady(req pathSourceStaticSetReadyReq) pathSourceStaticSetReadyRes { +// sourceStaticImplSetReady is called by a sourceStaticImpl. +func (s *sourceStatic) sourceStaticImplSetReady(req pathSourceStaticSetReadyReq) pathSourceStaticSetReadyRes { req.res = make(chan pathSourceStaticSetReadyRes) select { - case s.sourceStaticImplSetReady <- req: + case s.chSourceStaticImplSetReady <- req: return <-req.res case <-s.ctx.Done(): return pathSourceStaticSetReadyRes{err: fmt.Errorf("terminated")} } } -// onSourceStaticImplSetNotReady is called by a sourceStaticImpl. -func (s *sourceStatic) onSourceStaticImplSetNotReady(req pathSourceStaticSetNotReadyReq) { +// sourceStaticImplSetNotReady is called by a sourceStaticImpl. +func (s *sourceStatic) sourceStaticImplSetNotReady(req pathSourceStaticSetNotReadyReq) { req.res = make(chan struct{}) select { - case s.sourceStaticImplSetNotReady <- req: + case s.chSourceStaticImplSetNotReady <- req: <-req.res case <-s.ctx.Done(): }