diff --git a/internal/core/hls_source.go b/internal/core/hls_source.go index 5f7d23bc..11e0deb1 100644 --- a/internal/core/hls_source.go +++ b/internal/core/hls_source.go @@ -2,7 +2,6 @@ package core import ( "context" - "sync" "time" "github.com/aler9/gortsplib" @@ -14,10 +13,6 @@ import ( "github.com/aler9/rtsp-simple-server/internal/logger" ) -const ( - hlsSourceRetryPause = 5 * time.Second -) - type hlsSourceParent interface { log(logger.Level, string, ...interface{}) onSourceStaticSetReady(req pathSourceStaticSetReadyReq) pathSourceStaticSetReadyRes @@ -27,80 +22,27 @@ type hlsSourceParent interface { type hlsSource struct { ur string fingerprint string - wg *sync.WaitGroup parent hlsSourceParent - - ctx context.Context - ctxCancel func() } func newHLSSource( - parentCtx context.Context, ur string, fingerprint string, - wg *sync.WaitGroup, parent hlsSourceParent, ) *hlsSource { - ctx, ctxCancel := context.WithCancel(parentCtx) - - s := &hlsSource{ + return &hlsSource{ ur: ur, fingerprint: fingerprint, - wg: wg, parent: parent, - ctx: ctx, - ctxCancel: ctxCancel, } - - s.Log(logger.Info, "started") - - s.wg.Add(1) - go s.run() - - return s -} - -func (s *hlsSource) close() { - s.Log(logger.Info, "stopped") - s.ctxCancel() } func (s *hlsSource) Log(level logger.Level, format string, args ...interface{}) { s.parent.log(level, "[hls source] "+format, args...) } -func (s *hlsSource) run() { - defer s.wg.Done() - -outer: - for { - innerCtx, innerCtxCancel := context.WithCancel(context.Background()) - innerErr := make(chan error) - go func() { - innerErr <- s.runInner(innerCtx) - }() - - select { - case err := <-innerErr: - innerCtxCancel() - s.Log(logger.Info, "ERR: %v", err) - - case <-s.ctx.Done(): - innerCtxCancel() - <-innerErr - } - - select { - case <-time.After(hlsSourceRetryPause): - case <-s.ctx.Done(): - break outer - } - } - - s.ctxCancel() -} - -func (s *hlsSource) runInner(innerCtx context.Context) error { +// run implements sourceStaticImpl. +func (s *hlsSource) run(ctx context.Context) error { var stream *stream var videoTrackID int var audioTrackID int @@ -109,7 +51,7 @@ func (s *hlsSource) runInner(innerCtx context.Context) error { defer func() { if stream != nil { - s.parent.onSourceStaticSetNotReady(pathSourceStaticSetNotReadyReq{source: s}) + s.parent.onSourceStaticSetNotReady(pathSourceStaticSetNotReadyReq{}) } }() @@ -137,7 +79,6 @@ func (s *hlsSource) runInner(innerCtx context.Context) error { } res := s.parent.onSourceStaticSetReady(pathSourceStaticSetReadyReq{ - source: s, tracks: tracks, }) if res.err != nil { @@ -145,7 +86,6 @@ func (s *hlsSource) runInner(innerCtx context.Context) error { } s.Log(logger.Info, "ready") - stream = res.stream return nil @@ -216,14 +156,14 @@ func (s *hlsSource) runInner(innerCtx context.Context) error { case err := <-c.Wait(): return err - case <-innerCtx.Done(): + case <-ctx.Done(): c.Close() <-c.Wait() return nil } } -// onSourceAPIDescribe implements source. +// onSourceAPIDescribe implements sourceStaticImpl. func (*hlsSource) onSourceAPIDescribe() interface{} { return struct { Type string `json:"type"` diff --git a/internal/core/path.go b/internal/core/path.go index 4facd883..90c60f46 100644 --- a/internal/core/path.go +++ b/internal/core/path.go @@ -92,13 +92,13 @@ type pathSourceStaticSetReadyRes struct { } type pathSourceStaticSetReadyReq struct { - source sourceStatic + source source tracks gortsplib.Tracks res chan pathSourceStaticSetReadyRes } type pathSourceStaticSetNotReadyReq struct { - source sourceStatic + source source res chan struct{} } @@ -539,7 +539,7 @@ func (pa *path) run() { } if pa.source != nil { - if source, ok := pa.source.(sourceStatic); ok { + if source, ok := pa.source.(*sourceStatic); ok { source.close() pa.sourceStaticWg.Wait() } else if source, ok := pa.source.(publisher); ok { @@ -605,7 +605,7 @@ func (pa *path) onDemandStaticSourceStop() { pa.onDemandStaticSourceState = pathOnDemandStateInitial - pa.source.(sourceStatic).close() + pa.source.(*sourceStatic).close() pa.source = nil } @@ -696,39 +696,17 @@ func (pa *path) sourceSetNotReady() { } func (pa *path) staticSourceCreate() { - switch { - case strings.HasPrefix(pa.conf.Source, "rtsp://") || - strings.HasPrefix(pa.conf.Source, "rtsps://"): - pa.source = newRTSPSource( - pa.ctx, - pa.conf.Source, - pa.conf.SourceProtocol, - pa.conf.SourceAnyPortEnable, - pa.conf.SourceFingerprint, - pa.readTimeout, - pa.writeTimeout, - pa.readBufferCount, - &pa.sourceStaticWg, - pa) - - case strings.HasPrefix(pa.conf.Source, "rtmp://"): - pa.source = newRTMPSource( - pa.ctx, - pa.conf.Source, - pa.readTimeout, - pa.writeTimeout, - &pa.sourceStaticWg, - pa) - - case strings.HasPrefix(pa.conf.Source, "http://") || - strings.HasPrefix(pa.conf.Source, "https://"): - pa.source = newHLSSource( - pa.ctx, - pa.conf.Source, - pa.conf.SourceFingerprint, - &pa.sourceStaticWg, - pa) - } + pa.source = newSourceStatic( + pa.ctx, + pa.conf.Source, + pa.conf.SourceProtocol, + pa.conf.SourceAnyPortEnable, + pa.conf.SourceFingerprint, + pa.readTimeout, + pa.writeTimeout, + pa.readBufferCount, + &pa.sourceStaticWg, + pa) } func (pa *path) doReaderRemove(r reader) { diff --git a/internal/core/rtmp_source.go b/internal/core/rtmp_source.go index a99d3c9d..c416234d 100644 --- a/internal/core/rtmp_source.go +++ b/internal/core/rtmp_source.go @@ -5,7 +5,6 @@ import ( "fmt" "net" "net/url" - "sync" "time" "github.com/aler9/gortsplib" @@ -20,10 +19,6 @@ import ( "github.com/aler9/rtsp-simple-server/internal/rtmp/message" ) -const ( - rtmpSourceRetryPause = 5 * time.Second -) - type rtmpSourceParent interface { log(logger.Level, string, ...interface{}) onSourceStaticSetReady(req pathSourceStaticSetReadyReq) pathSourceStaticSetReadyRes @@ -34,83 +29,29 @@ type rtmpSource struct { ur string readTimeout conf.StringDuration writeTimeout conf.StringDuration - wg *sync.WaitGroup parent rtmpSourceParent - - ctx context.Context - ctxCancel func() } func newRTMPSource( - parentCtx context.Context, ur string, readTimeout conf.StringDuration, writeTimeout conf.StringDuration, - wg *sync.WaitGroup, parent rtmpSourceParent, ) *rtmpSource { - ctx, ctxCancel := context.WithCancel(parentCtx) - - s := &rtmpSource{ + return &rtmpSource{ ur: ur, readTimeout: readTimeout, writeTimeout: writeTimeout, - wg: wg, parent: parent, - ctx: ctx, - ctxCancel: ctxCancel, } - - s.Log(logger.Info, "started") - - s.wg.Add(1) - go s.run() - - return s -} - -// Close closes a Source. -func (s *rtmpSource) close() { - s.Log(logger.Info, "stopped") - s.ctxCancel() } func (s *rtmpSource) Log(level logger.Level, format string, args ...interface{}) { s.parent.log(level, "[rtmp source] "+format, args...) } -func (s *rtmpSource) run() { - defer s.wg.Done() - -outer: - for { - innerCtx, innerCtxCancel := context.WithCancel(context.Background()) - innerErr := make(chan error) - go func() { - innerErr <- s.runInner(innerCtx) - }() - - select { - case err := <-innerErr: - innerCtxCancel() - s.Log(logger.Info, "ERR: %v", err) - - case <-s.ctx.Done(): - innerCtxCancel() - <-innerErr - } - - select { - case <-time.After(rtmpSourceRetryPause): - case <-s.ctx.Done(): - break outer - } - } - - s.ctxCancel() -} - -func (s *rtmpSource) runInner(innerCtx context.Context) error { +// run implements sourceStaticImpl. +func (s *rtmpSource) run(ctx context.Context) error { s.Log(logger.Debug, "connecting") u, err := url.Parse(s.ur) @@ -124,7 +65,7 @@ func (s *rtmpSource) runInner(innerCtx context.Context) error { u.Host = net.JoinHostPort(u.Host, "1935") } - ctx2, cancel2 := context.WithTimeout(innerCtx, time.Duration(s.readTimeout)) + ctx2, cancel2 := context.WithTimeout(ctx, time.Duration(s.readTimeout)) defer cancel2() var d net.Dialer @@ -179,7 +120,6 @@ func (s *rtmpSource) runInner(innerCtx context.Context) error { } res := s.parent.onSourceStaticSetReady(pathSourceStaticSetReadyReq{ - source: s, tracks: tracks, }) if res.err != nil { @@ -189,7 +129,7 @@ func (s *rtmpSource) runInner(innerCtx context.Context) error { s.Log(logger.Info, "ready") defer func() { - s.parent.onSourceStaticSetNotReady(pathSourceStaticSetNotReadyReq{source: s}) + s.parent.onSourceStaticSetNotReady(pathSourceStaticSetNotReadyReq{}) }() for { @@ -267,14 +207,14 @@ func (s *rtmpSource) runInner(innerCtx context.Context) error { nconn.Close() return err - case <-innerCtx.Done(): + case <-ctx.Done(): nconn.Close() <-readDone return nil } } -// onSourceAPIDescribe implements source. +// onSourceAPIDescribe implements sourceStaticImpl. func (*rtmpSource) onSourceAPIDescribe() interface{} { return struct { Type string `json:"type"` diff --git a/internal/core/rtsp_source.go b/internal/core/rtsp_source.go index a2b9a98f..1da73139 100644 --- a/internal/core/rtsp_source.go +++ b/internal/core/rtsp_source.go @@ -7,7 +7,6 @@ import ( "encoding/hex" "fmt" "strings" - "sync" "time" "github.com/aler9/gortsplib" @@ -18,10 +17,6 @@ import ( "github.com/aler9/rtsp-simple-server/internal/logger" ) -const ( - rtspSourceRetryPause = 5 * time.Second -) - type rtspSourceParent interface { log(logger.Level, string, ...interface{}) onSourceStaticSetReady(req pathSourceStaticSetReadyReq) pathSourceStaticSetReadyRes @@ -36,15 +31,10 @@ type rtspSource struct { readTimeout conf.StringDuration writeTimeout conf.StringDuration readBufferCount int - wg *sync.WaitGroup parent rtspSourceParent - - ctx context.Context - ctxCancel func() } func newRTSPSource( - parentCtx context.Context, ur string, proto conf.SourceProtocol, anyPortEnable bool, @@ -52,12 +42,9 @@ func newRTSPSource( readTimeout conf.StringDuration, writeTimeout conf.StringDuration, readBufferCount int, - wg *sync.WaitGroup, parent rtspSourceParent, ) *rtspSource { - ctx, ctxCancel := context.WithCancel(parentCtx) - - s := &rtspSource{ + return &rtspSource{ ur: ur, proto: proto, anyPortEnable: anyPortEnable, @@ -65,61 +52,16 @@ func newRTSPSource( readTimeout: readTimeout, writeTimeout: writeTimeout, readBufferCount: readBufferCount, - wg: wg, parent: parent, - ctx: ctx, - ctxCancel: ctxCancel, } - - s.Log(logger.Info, "started") - - s.wg.Add(1) - go s.run() - - return s -} - -func (s *rtspSource) close() { - s.Log(logger.Info, "stopped") - s.ctxCancel() } func (s *rtspSource) Log(level logger.Level, format string, args ...interface{}) { s.parent.log(level, "[rtsp source] "+format, args...) } -func (s *rtspSource) run() { - defer s.wg.Done() - -outer: - for { - innerCtx, innerCtxCancel := context.WithCancel(context.Background()) - innerErr := make(chan error) - go func() { - innerErr <- s.runInner(innerCtx) - }() - - select { - case err := <-innerErr: - innerCtxCancel() - s.Log(logger.Info, "ERR: %v", err) - - case <-s.ctx.Done(): - innerCtxCancel() - <-innerErr - } - - select { - case <-time.After(rtspSourceRetryPause): - case <-s.ctx.Done(): - break outer - } - } - - s.ctxCancel() -} - -func (s *rtspSource) runInner(innerCtx context.Context) error { +// run implements sourceStaticImpl. +func (s *rtspSource) run(ctx context.Context) error { s.Log(logger.Debug, "connecting") var tlsConfig *tls.Config @@ -184,7 +126,6 @@ func (s *rtspSource) runInner(innerCtx context.Context) error { } res := s.parent.onSourceStaticSetReady(pathSourceStaticSetReadyReq{ - source: s, tracks: c.Tracks(), }) if res.err != nil { @@ -194,7 +135,7 @@ func (s *rtspSource) runInner(innerCtx context.Context) error { s.Log(logger.Info, "ready") defer func() { - s.parent.onSourceStaticSetNotReady(pathSourceStaticSetNotReadyReq{source: s}) + s.parent.onSourceStaticSetNotReady(pathSourceStaticSetNotReadyReq{}) }() c.OnPacketRTP = func(ctx *gortsplib.ClientOnPacketRTPCtx) { @@ -228,14 +169,14 @@ func (s *rtspSource) runInner(innerCtx context.Context) error { case err := <-readErr: return err - case <-innerCtx.Done(): + case <-ctx.Done(): c.Close() <-readErr return nil } } -// onSourceAPIDescribe implements source. +// onSourceAPIDescribe implements sourceStaticImpl. func (*rtspSource) onSourceAPIDescribe() interface{} { return struct { Type string `json:"type"` diff --git a/internal/core/source.go b/internal/core/source.go index 0549c458..924f8bb7 100644 --- a/internal/core/source.go +++ b/internal/core/source.go @@ -3,24 +3,8 @@ package core // source is an entity that can provide a stream. // it can be: // - a publisher -// - a static source -// - a redirect source +// - sourceStatic +// - sourceRedirect type source interface { onSourceAPIDescribe() interface{} } - -// sourceStatic is an entity that can provide a static stream. -type sourceStatic interface { - source - close() -} - -// sourceRedirect is a source that redirects to another one. -type sourceRedirect struct{} - -// onSourceAPIDescribe implements source. -func (*sourceRedirect) onSourceAPIDescribe() interface{} { - return struct { - Type string `json:"type"` - }{"redirect"} -} diff --git a/internal/core/source_redirect.go b/internal/core/source_redirect.go new file mode 100644 index 00000000..b24535cd --- /dev/null +++ b/internal/core/source_redirect.go @@ -0,0 +1,11 @@ +package core + +// sourceRedirect is a source that redirects to another one. +type sourceRedirect struct{} + +// onSourceAPIDescribe implements source. +func (*sourceRedirect) onSourceAPIDescribe() interface{} { + return struct { + Type string `json:"type"` + }{"redirect"} +} diff --git a/internal/core/source_static.go b/internal/core/source_static.go new file mode 100644 index 00000000..11879c2f --- /dev/null +++ b/internal/core/source_static.go @@ -0,0 +1,165 @@ +package core + +import ( + "context" + "strings" + "sync" + "time" + + "github.com/aler9/rtsp-simple-server/internal/conf" + "github.com/aler9/rtsp-simple-server/internal/logger" +) + +const ( + sourceStaticRetryPause = 5 * time.Second +) + +type sourceStaticImpl interface { + Log(logger.Level, string, ...interface{}) + run(context.Context) error + onSourceAPIDescribe() interface{} +} + +type sourceStaticParent interface { + log(logger.Level, string, ...interface{}) + onSourceStaticSetReady(req pathSourceStaticSetReadyReq) pathSourceStaticSetReadyRes + onSourceStaticSetNotReady(req pathSourceStaticSetNotReadyReq) +} + +// sourceStatic is a static source. +type sourceStatic struct { + ur string + protocol conf.SourceProtocol + anyPortEnable bool + fingerprint string + readTimeout conf.StringDuration + writeTimeout conf.StringDuration + readBufferCount int + wg *sync.WaitGroup + parent sourceStaticParent + + impl sourceStaticImpl + ctx context.Context + ctxCancel func() +} + +func newSourceStatic( + parentCtx context.Context, + ur string, + protocol conf.SourceProtocol, + anyPortEnable bool, + fingerprint string, + readTimeout conf.StringDuration, + writeTimeout conf.StringDuration, + readBufferCount int, + wg *sync.WaitGroup, + parent sourceStaticParent, +) *sourceStatic { + ctx, ctxCancel := context.WithCancel(parentCtx) + + s := &sourceStatic{ + ur: ur, + protocol: protocol, + anyPortEnable: anyPortEnable, + fingerprint: fingerprint, + readTimeout: readTimeout, + writeTimeout: writeTimeout, + readBufferCount: readBufferCount, + wg: wg, + parent: parent, + ctx: ctx, + ctxCancel: ctxCancel, + } + + switch { + case strings.HasPrefix(s.ur, "rtsp://") || + strings.HasPrefix(s.ur, "rtsps://"): + s.impl = newRTSPSource( + s.ur, + s.protocol, + s.anyPortEnable, + s.fingerprint, + s.readTimeout, + s.writeTimeout, + s.readBufferCount, + s) + + case strings.HasPrefix(s.ur, "rtmp://"): + s.impl = newRTMPSource( + s.ur, + s.readTimeout, + s.writeTimeout, + s) + + case strings.HasPrefix(s.ur, "http://") || + strings.HasPrefix(s.ur, "https://"): + s.impl = newHLSSource( + s.ur, + s.fingerprint, + s) + } + + s.impl.Log(logger.Info, "started") + + s.wg.Add(1) + go s.run() + + return s +} + +func (s *sourceStatic) close() { + s.impl.Log(logger.Info, "stopped") + s.ctxCancel() +} + +func (s *sourceStatic) log(level logger.Level, format string, args ...interface{}) { + s.parent.log(level, format, args...) +} + +func (s *sourceStatic) run() { + defer s.wg.Done() + +outer: + for { + innerCtx, innerCtxCancel := context.WithCancel(context.Background()) + innerErr := make(chan error) + go func() { + innerErr <- s.impl.run(innerCtx) + }() + + select { + case err := <-innerErr: + innerCtxCancel() + s.impl.Log(logger.Info, "ERR: %v", err) + + case <-s.ctx.Done(): + innerCtxCancel() + <-innerErr + } + + select { + case <-time.After(sourceStaticRetryPause): + case <-s.ctx.Done(): + break outer + } + } + + s.ctxCancel() +} + +// onSourceAPIDescribe implements source. +func (s *sourceStatic) onSourceAPIDescribe() interface{} { + return s.impl.onSourceAPIDescribe() +} + +// onSourceStaticSetReady is called by a sourceStaticImpl. +func (s *sourceStatic) onSourceStaticSetReady(req pathSourceStaticSetReadyReq) pathSourceStaticSetReadyRes { + req.source = s + return s.parent.onSourceStaticSetReady(req) +} + +// onSourceStaticSetNotReady is called by a sourceStaticImpl. +func (s *sourceStatic) onSourceStaticSetNotReady(req pathSourceStaticSetNotReadyReq) { + req.source = s + s.parent.onSourceStaticSetNotReady(req) +}