diff --git a/internal/defs/source.go b/internal/defs/source.go index b6be1f75..e0a42bca 100644 --- a/internal/defs/source.go +++ b/internal/defs/source.go @@ -12,9 +12,9 @@ import ( // Source is an entity that can provide a stream. // it can be: -// - publisher +// - Publisher // - staticsources.Handler -// - redirectSource +// - core.sourceRedirect type Source interface { logger.Writer APISourceDescribe() APIPathSourceOrReader diff --git a/internal/defs/static_source.go b/internal/defs/static_source.go index dd5adbac..ee98719d 100644 --- a/internal/defs/static_source.go +++ b/internal/defs/static_source.go @@ -4,23 +4,8 @@ import ( "context" "github.com/bluenviron/mediamtx/internal/conf" - "github.com/bluenviron/mediamtx/internal/logger" ) -// StaticSource is a static source. -type StaticSource interface { - logger.Writer - Run(StaticSourceRunParams) error - APISourceDescribe() APIPathSourceOrReader -} - -// StaticSourceParent is the parent of a static source. -type StaticSourceParent interface { - logger.Writer - SetReady(req PathSourceStaticSetReadyReq) PathSourceStaticSetReadyRes - SetNotReady(req PathSourceStaticSetNotReadyReq) -} - // StaticSourceRunParams is the set of params passed to Run(). type StaticSourceRunParams struct { Context context.Context diff --git a/internal/staticsources/handler.go b/internal/staticsources/handler.go index 17c1cc86..7c768de3 100644 --- a/internal/staticsources/handler.go +++ b/internal/staticsources/handler.go @@ -43,6 +43,12 @@ func resolveSource(s string, matches []string, query string) string { return s } +type staticSource interface { + logger.Writer + Run(defs.StaticSourceRunParams) error + APISourceDescribe() defs.APIPathSourceOrReader +} + type handlerPathManager interface { AddReader(req defs.PathAddReaderReq) (defs.Path, *stream.Stream, error) } @@ -67,7 +73,7 @@ type Handler struct { ctx context.Context ctxCancel func() - instance defs.StaticSource + instance staticSource running bool query string diff --git a/internal/staticsources/hls/source.go b/internal/staticsources/hls/source.go index 710db16d..93e2c47c 100644 --- a/internal/staticsources/hls/source.go +++ b/internal/staticsources/hls/source.go @@ -17,10 +17,16 @@ import ( "github.com/bluenviron/mediamtx/internal/stream" ) +type parent interface { + logger.Writer + SetReady(req defs.PathSourceStaticSetReadyReq) defs.PathSourceStaticSetReadyRes + SetNotReady(req defs.PathSourceStaticSetNotReadyReq) +} + // Source is a HLS static source. type Source struct { ReadTimeout conf.Duration - Parent defs.StaticSourceParent + Parent parent } // Log implements logger.Writer. diff --git a/internal/staticsources/hls/source_test.go b/internal/staticsources/hls/source_test.go index 5a6b7915..98fb6dba 100644 --- a/internal/staticsources/hls/source_test.go +++ b/internal/staticsources/hls/source_test.go @@ -88,16 +88,28 @@ func TestSource(t *testing.T) { go s.Serve(ln) defer s.Shutdown(context.Background()) - te := test.NewSourceTester( - func(p defs.StaticSourceParent) defs.StaticSource { - return &Source{ - Parent: p, - } - }, - "http://localhost:5780/stream.m3u8", - &conf.Path{}, - ) - defer te.Close() + p := &test.StaticSourceParent{} + p.Initialize() + defer p.Close() - <-te.Unit + so := &Source{ + Parent: p, + } + + done := make(chan struct{}) + defer func() { <-done }() + + ctx, ctxCancel := context.WithCancel(context.Background()) + defer ctxCancel() + + go func() { + so.Run(defs.StaticSourceRunParams{ //nolint:errcheck + Context: ctx, + ResolvedSource: "http://localhost:5780/stream.m3u8", + Conf: &conf.Path{}, + }) + close(done) + }() + + <-p.Unit } diff --git a/internal/staticsources/rpicamera/source.go b/internal/staticsources/rpicamera/source.go index b611a6b5..8e62d8ec 100644 --- a/internal/staticsources/rpicamera/source.go +++ b/internal/staticsources/rpicamera/source.go @@ -101,7 +101,9 @@ func (*secondaryReader) APIReaderDescribe() defs.APIPathSourceOrReader { } type parent interface { - defs.StaticSourceParent + logger.Writer + SetReady(req defs.PathSourceStaticSetReadyReq) defs.PathSourceStaticSetReadyRes + SetNotReady(req defs.PathSourceStaticSetNotReadyReq) AddReader(req defs.PathAddReaderReq) (defs.Path, *stream.Stream, error) } diff --git a/internal/staticsources/rtmp/source.go b/internal/staticsources/rtmp/source.go index 5dfc5a63..5e4cc350 100644 --- a/internal/staticsources/rtmp/source.go +++ b/internal/staticsources/rtmp/source.go @@ -18,11 +18,17 @@ import ( "github.com/bluenviron/mediamtx/internal/stream" ) +type parent interface { + logger.Writer + SetReady(req defs.PathSourceStaticSetReadyReq) defs.PathSourceStaticSetReadyRes + SetNotReady(req defs.PathSourceStaticSetNotReadyReq) +} + // Source is a RTMP static source. type Source struct { ReadTimeout conf.Duration WriteTimeout conf.Duration - Parent defs.StaticSourceParent + Parent parent } // Log implements logger.Writer. diff --git a/internal/staticsources/rtmp/source_test.go b/internal/staticsources/rtmp/source_test.go index af152c52..845ce0ac 100644 --- a/internal/staticsources/rtmp/source_test.go +++ b/internal/staticsources/rtmp/source_test.go @@ -1,6 +1,7 @@ package rtmp import ( + "context" "crypto/tls" "net" "os" @@ -104,23 +105,34 @@ func TestSource(t *testing.T) { source += "localhost/teststream" - te := test.NewSourceTester( - func(p defs.StaticSourceParent) defs.StaticSource { - return &Source{ - ReadTimeout: conf.Duration(10 * time.Second), - WriteTimeout: conf.Duration(10 * time.Second), - Parent: p, - } - }, - source, - &conf.Path{ - SourceFingerprint: "33949E05FFFB5FF3E8AA16F8213A6251B4D9363804BA53233C4DA9A46D6F2739", - }, - ) + p := &test.StaticSourceParent{} + p.Initialize() + defer p.Close() - defer te.Close() + so := &Source{ + ReadTimeout: conf.Duration(10 * time.Second), + WriteTimeout: conf.Duration(10 * time.Second), + Parent: p, + } - <-te.Unit + done := make(chan struct{}) + defer func() { <-done }() + + ctx, ctxCancel := context.WithCancel(context.Background()) + defer ctxCancel() + + go func() { + so.Run(defs.StaticSourceRunParams{ //nolint:errcheck + Context: ctx, + ResolvedSource: source, + Conf: &conf.Path{ + SourceFingerprint: "33949E05FFFB5FF3E8AA16F8213A6251B4D9363804BA53233C4DA9A46D6F2739", + }, + }) + close(done) + }() + + <-p.Unit }) } } diff --git a/internal/staticsources/rtsp/source.go b/internal/staticsources/rtsp/source.go index e635474e..2ea03dd5 100644 --- a/internal/staticsources/rtsp/source.go +++ b/internal/staticsources/rtsp/source.go @@ -61,12 +61,18 @@ func createRangeHeader(cnf *conf.Path) (*headers.Range, error) { } } +type parent interface { + logger.Writer + SetReady(req defs.PathSourceStaticSetReadyReq) defs.PathSourceStaticSetReadyRes + SetNotReady(req defs.PathSourceStaticSetNotReadyReq) +} + // Source is a RTSP static source. type Source struct { ReadTimeout conf.Duration WriteTimeout conf.Duration WriteQueueSize int - Parent defs.StaticSourceParent + Parent parent } // Log implements logger.Writer. diff --git a/internal/staticsources/rtsp/source_test.go b/internal/staticsources/rtsp/source_test.go index 394947dc..2f9f2609 100644 --- a/internal/staticsources/rtsp/source_test.go +++ b/internal/staticsources/rtsp/source_test.go @@ -1,6 +1,7 @@ package rtsp import ( + "context" "crypto/tls" "os" "testing" @@ -134,46 +135,50 @@ func TestSource(t *testing.T) { require.NoError(t, err) defer strm.Close() - var te *test.SourceTester + var ur string + var cnf *conf.Path if source != "tls" { + ur = "rtsp://testuser:testpass@localhost:8555/teststream" var sp conf.RTSPTransport sp.UnmarshalJSON([]byte(`"` + source + `"`)) //nolint:errcheck - - te = test.NewSourceTester( - func(p defs.StaticSourceParent) defs.StaticSource { - return &Source{ - ReadTimeout: conf.Duration(10 * time.Second), - WriteTimeout: conf.Duration(10 * time.Second), - WriteQueueSize: 2048, - Parent: p, - } - }, - "rtsp://testuser:testpass@localhost:8555/teststream", - &conf.Path{ - RTSPTransport: sp, - }, - ) + cnf = &conf.Path{ + RTSPTransport: sp, + } } else { - te = test.NewSourceTester( - func(p defs.StaticSourceParent) defs.StaticSource { - return &Source{ - ReadTimeout: conf.Duration(10 * time.Second), - WriteTimeout: conf.Duration(10 * time.Second), - WriteQueueSize: 2048, - Parent: p, - } - }, - "rtsps://testuser:testpass@localhost:8555/teststream", - &conf.Path{ - SourceFingerprint: "33949E05FFFB5FF3E8AA16F8213A6251B4D9363804BA53233C4DA9A46D6F2739", - }, - ) + ur = "rtsps://testuser:testpass@localhost:8555/teststream" + cnf = &conf.Path{ + SourceFingerprint: "33949E05FFFB5FF3E8AA16F8213A6251B4D9363804BA53233C4DA9A46D6F2739", + } } - defer te.Close() + p := &test.StaticSourceParent{} + p.Initialize() + defer p.Close() - <-te.Unit + so := &Source{ + ReadTimeout: conf.Duration(10 * time.Second), + WriteTimeout: conf.Duration(10 * time.Second), + WriteQueueSize: 2048, + Parent: p, + } + + done := make(chan struct{}) + defer func() { <-done }() + + ctx, ctxCancel := context.WithCancel(context.Background()) + defer ctxCancel() + + go func() { + so.Run(defs.StaticSourceRunParams{ //nolint:errcheck + Context: ctx, + ResolvedSource: ur, + Conf: cnf, + }) + close(done) + }() + + <-p.Unit }) } } @@ -248,23 +253,35 @@ func TestSourceNoPassword(t *testing.T) { var sp conf.RTSPTransport sp.UnmarshalJSON([]byte(`"tcp"`)) //nolint:errcheck - te := test.NewSourceTester( - func(p defs.StaticSourceParent) defs.StaticSource { - return &Source{ - ReadTimeout: conf.Duration(10 * time.Second), - WriteTimeout: conf.Duration(10 * time.Second), - WriteQueueSize: 2048, - Parent: p, - } - }, - "rtsp://testuser:@127.0.0.1:8555/teststream", - &conf.Path{ - RTSPTransport: sp, - }, - ) - defer te.Close() + p := &test.StaticSourceParent{} + p.Initialize() + defer p.Close() - <-te.Unit + so := &Source{ + ReadTimeout: conf.Duration(10 * time.Second), + WriteTimeout: conf.Duration(10 * time.Second), + WriteQueueSize: 2048, + Parent: p, + } + + done := make(chan struct{}) + defer func() { <-done }() + + ctx, ctxCancel := context.WithCancel(context.Background()) + defer ctxCancel() + + go func() { + so.Run(defs.StaticSourceRunParams{ //nolint:errcheck + Context: ctx, + ResolvedSource: "rtsp://testuser:@127.0.0.1:8555/teststream", + Conf: &conf.Path{ + RTSPTransport: sp, + }, + }) + close(done) + }() + + <-p.Unit } func TestSourceRange(t *testing.T) { @@ -350,21 +367,33 @@ func TestSourceRange(t *testing.T) { cnf.RTSPRangeStart = "130s" } - te := test.NewSourceTester( - func(p defs.StaticSourceParent) defs.StaticSource { - return &Source{ - ReadTimeout: conf.Duration(10 * time.Second), - WriteTimeout: conf.Duration(10 * time.Second), - WriteQueueSize: 2048, - Parent: p, - } - }, - "rtsp://127.0.0.1:8555/teststream", - cnf, - ) - defer te.Close() + p := &test.StaticSourceParent{} + p.Initialize() + defer p.Close() - <-te.Unit + so := &Source{ + ReadTimeout: conf.Duration(10 * time.Second), + WriteTimeout: conf.Duration(10 * time.Second), + WriteQueueSize: 2048, + Parent: p, + } + + done := make(chan struct{}) + defer func() { <-done }() + + ctx, ctxCancel := context.WithCancel(context.Background()) + defer ctxCancel() + + go func() { + so.Run(defs.StaticSourceRunParams{ //nolint:errcheck + Context: ctx, + ResolvedSource: "rtsp://127.0.0.1:8555/teststream", + Conf: cnf, + }) + close(done) + }() + + <-p.Unit }) } } diff --git a/internal/staticsources/srt/source.go b/internal/staticsources/srt/source.go index ff3b677a..f750aa67 100644 --- a/internal/staticsources/srt/source.go +++ b/internal/staticsources/srt/source.go @@ -15,10 +15,16 @@ import ( "github.com/bluenviron/mediamtx/internal/stream" ) +type parent interface { + logger.Writer + SetReady(req defs.PathSourceStaticSetReadyReq) defs.PathSourceStaticSetReadyRes + SetNotReady(req defs.PathSourceStaticSetNotReadyReq) +} + // Source is a SRT static source. type Source struct { ReadTimeout conf.Duration - Parent defs.StaticSourceParent + Parent parent } // Log implements logger.Writer. diff --git a/internal/staticsources/srt/source_test.go b/internal/staticsources/srt/source_test.go index 652a4b86..da943a5d 100644 --- a/internal/staticsources/srt/source_test.go +++ b/internal/staticsources/srt/source_test.go @@ -2,6 +2,7 @@ package srt import ( "bufio" + "context" "testing" "time" @@ -52,17 +53,29 @@ func TestSource(t *testing.T) { time.Sleep(500 * time.Millisecond) }() - te := test.NewSourceTester( - func(p defs.StaticSourceParent) defs.StaticSource { - return &Source{ - ReadTimeout: conf.Duration(10 * time.Second), - Parent: p, - } - }, - "srt://127.0.0.1:9002?streamid=sidname&passphrase=ttest1234567", - &conf.Path{}, - ) - defer te.Close() + p := &test.StaticSourceParent{} + p.Initialize() + defer p.Close() - <-te.Unit + so := &Source{ + ReadTimeout: conf.Duration(10 * time.Second), + Parent: p, + } + + done := make(chan struct{}) + defer func() { <-done }() + + ctx, ctxCancel := context.WithCancel(context.Background()) + defer ctxCancel() + + go func() { + so.Run(defs.StaticSourceRunParams{ //nolint:errcheck + Context: ctx, + ResolvedSource: "srt://127.0.0.1:9002?streamid=sidname&passphrase=ttest1234567", + Conf: &conf.Path{}, + }) + close(done) + }() + + <-p.Unit } diff --git a/internal/staticsources/udp/source.go b/internal/staticsources/udp/source.go index 9475ea83..b7735a02 100644 --- a/internal/staticsources/udp/source.go +++ b/internal/staticsources/udp/source.go @@ -46,10 +46,16 @@ type packetConn interface { SetReadBuffer(int) error } +type parent interface { + logger.Writer + SetReady(req defs.PathSourceStaticSetReadyReq) defs.PathSourceStaticSetReadyRes + SetNotReady(req defs.PathSourceStaticSetNotReadyReq) +} + // Source is a UDP static source. type Source struct { ReadTimeout conf.Duration - Parent defs.StaticSourceParent + Parent parent } // Log implements logger.Writer. diff --git a/internal/staticsources/udp/source_test.go b/internal/staticsources/udp/source_test.go index 5e9b368a..6df1f2cd 100644 --- a/internal/staticsources/udp/source_test.go +++ b/internal/staticsources/udp/source_test.go @@ -2,6 +2,7 @@ package udp import ( "bufio" + "context" "net" "testing" "time" @@ -52,17 +53,29 @@ func TestSource(t *testing.T) { src = "udp://127.0.0.1:9001?source=127.0.1.1" } - te := test.NewSourceTester( - func(p defs.StaticSourceParent) defs.StaticSource { - return &Source{ - ReadTimeout: conf.Duration(10 * time.Second), - Parent: p, - } - }, - src, - &conf.Path{}, - ) - defer te.Close() + p := &test.StaticSourceParent{} + p.Initialize() + defer p.Close() + + so := &Source{ + ReadTimeout: conf.Duration(10 * time.Second), + Parent: p, + } + + done := make(chan struct{}) + defer func() { <-done }() + + ctx, ctxCancel := context.WithCancel(context.Background()) + defer ctxCancel() + + go func() { + so.Run(defs.StaticSourceRunParams{ //nolint:errcheck + Context: ctx, + ResolvedSource: src, + Conf: &conf.Path{}, + }) + close(done) + }() time.Sleep(50 * time.Millisecond) @@ -117,7 +130,7 @@ func TestSource(t *testing.T) { err = bw.Flush() require.NoError(t, err) - <-te.Unit + <-p.Unit }) } } diff --git a/internal/staticsources/webrtc/source.go b/internal/staticsources/webrtc/source.go index 6d575fa5..191fb5c0 100644 --- a/internal/staticsources/webrtc/source.go +++ b/internal/staticsources/webrtc/source.go @@ -18,10 +18,16 @@ import ( "github.com/bluenviron/mediamtx/internal/stream" ) +type parent interface { + logger.Writer + SetReady(req defs.PathSourceStaticSetReadyReq) defs.PathSourceStaticSetReadyRes + SetNotReady(req defs.PathSourceStaticSetNotReadyReq) +} + // Source is a WebRTC static source. type Source struct { ReadTimeout conf.Duration - Parent defs.StaticSourceParent + Parent parent } // Log implements logger.Writer. diff --git a/internal/staticsources/webrtc/source_test.go b/internal/staticsources/webrtc/source_test.go index dc7f70e2..b0330cd7 100644 --- a/internal/staticsources/webrtc/source_test.go +++ b/internal/staticsources/webrtc/source_test.go @@ -123,17 +123,29 @@ func TestSource(t *testing.T) { go httpServ.Serve(ln) defer httpServ.Shutdown(context.Background()) - te := test.NewSourceTester( - func(p defs.StaticSourceParent) defs.StaticSource { - return &Source{ - ReadTimeout: conf.Duration(10 * time.Second), - Parent: p, - } - }, - "whep://localhost:9003/my/resource", - &conf.Path{}, - ) - defer te.Close() + p := &test.StaticSourceParent{} + p.Initialize() + defer p.Close() - <-te.Unit + so := &Source{ + ReadTimeout: conf.Duration(10 * time.Second), + Parent: p, + } + + done := make(chan struct{}) + defer func() { <-done }() + + ctx, ctxCancel := context.WithCancel(context.Background()) + defer ctxCancel() + + go func() { + so.Run(defs.StaticSourceRunParams{ //nolint:errcheck + Context: ctx, + ResolvedSource: "whep://localhost:9003/my/resource", + Conf: &conf.Path{}, + }) + close(done) + }() + + <-p.Unit } diff --git a/internal/test/auth_manager.go b/internal/test/auth_manager.go index 6612b0f9..13f4cda8 100644 --- a/internal/test/auth_manager.go +++ b/internal/test/auth_manager.go @@ -1,3 +1,4 @@ +// Package test contains test utilities. package test import "github.com/bluenviron/mediamtx/internal/auth" diff --git a/internal/test/source_tester.go b/internal/test/source_tester.go deleted file mode 100644 index b900948d..00000000 --- a/internal/test/source_tester.go +++ /dev/null @@ -1,96 +0,0 @@ -// Package test contains test utilities. -package test - -import ( - "context" - - "github.com/bluenviron/mediamtx/internal/conf" - "github.com/bluenviron/mediamtx/internal/defs" - "github.com/bluenviron/mediamtx/internal/logger" - "github.com/bluenviron/mediamtx/internal/stream" - "github.com/bluenviron/mediamtx/internal/unit" -) - -// SourceTester is a static source tester. -type SourceTester struct { - ctx context.Context - ctxCancel func() - stream *stream.Stream - reader stream.Reader - - Unit chan unit.Unit - done chan struct{} -} - -// NewSourceTester allocates a SourceTester. -func NewSourceTester( - createFunc func(defs.StaticSourceParent) defs.StaticSource, - resolvedSource string, - conf *conf.Path, -) *SourceTester { - ctx, ctxCancel := context.WithCancel(context.Background()) - - t := &SourceTester{ - ctx: ctx, - ctxCancel: ctxCancel, - Unit: make(chan unit.Unit), - done: make(chan struct{}), - } - - s := createFunc(t) - - go func() { - s.Run(defs.StaticSourceRunParams{ //nolint:errcheck - Context: ctx, - ResolvedSource: resolvedSource, - Conf: conf, - }) - close(t.done) - }() - - return t -} - -// Close closes the tester. -func (t *SourceTester) Close() { - t.ctxCancel() - t.stream.RemoveReader(t.reader) - <-t.done -} - -// Log implements StaticSourceParent. -func (t *SourceTester) Log(_ logger.Level, _ string, _ ...interface{}) { -} - -// SetReady implements StaticSourceParent. -func (t *SourceTester) SetReady(req defs.PathSourceStaticSetReadyReq) defs.PathSourceStaticSetReadyRes { - t.stream = &stream.Stream{ - WriteQueueSize: 512, - RTPMaxPayloadSize: 1450, - Desc: req.Desc, - GenerateRTPPackets: req.GenerateRTPPackets, - Parent: t, - } - err := t.stream.Initialize() - if err != nil { - panic(err) - } - - t.reader = NilLogger - - t.stream.AddReader(t.reader, req.Desc.Medias[0], req.Desc.Medias[0].Formats[0], func(u unit.Unit) error { - t.Unit <- u - close(t.Unit) - return nil - }) - - t.stream.StartReader(t.reader) - - return defs.PathSourceStaticSetReadyRes{ - Stream: t.stream, - } -} - -// SetNotReady implements StaticSourceParent. -func (t *SourceTester) SetNotReady(_ defs.PathSourceStaticSetNotReadyReq) { -} diff --git a/internal/test/static_source_parent.go b/internal/test/static_source_parent.go new file mode 100644 index 00000000..4b1ef8ed --- /dev/null +++ b/internal/test/static_source_parent.go @@ -0,0 +1,58 @@ +package test + +import ( + "github.com/bluenviron/mediamtx/internal/defs" + "github.com/bluenviron/mediamtx/internal/logger" + "github.com/bluenviron/mediamtx/internal/stream" + "github.com/bluenviron/mediamtx/internal/unit" +) + +// StaticSourceParent is a dummy static source parent. +type StaticSourceParent struct { + stream *stream.Stream + reader stream.Reader + Unit chan unit.Unit +} + +// Log implements logger.Writer. +func (*StaticSourceParent) Log(logger.Level, string, ...interface{}) {} + +// Initialize initializes StaticSourceParent. +func (p *StaticSourceParent) Initialize() { + p.Unit = make(chan unit.Unit) +} + +// Close closes StaticSourceParent. +func (p *StaticSourceParent) Close() { + p.stream.RemoveReader(p.reader) +} + +// SetReady implements parent. +func (p *StaticSourceParent) SetReady(req defs.PathSourceStaticSetReadyReq) defs.PathSourceStaticSetReadyRes { + p.stream = &stream.Stream{ + WriteQueueSize: 512, + RTPMaxPayloadSize: 1450, + Desc: req.Desc, + GenerateRTPPackets: req.GenerateRTPPackets, + Parent: p, + } + err := p.stream.Initialize() + if err != nil { + panic(err) + } + + p.reader = NilLogger + + p.stream.AddReader(p.reader, req.Desc.Medias[0], req.Desc.Medias[0].Formats[0], func(u unit.Unit) error { + p.Unit <- u + close(p.Unit) + return nil + }) + + p.stream.StartReader(p.reader) + + return defs.PathSourceStaticSetReadyRes{Stream: p.stream} +} + +// SetNotReady implements parent. +func (StaticSourceParent) SetNotReady(_ defs.PathSourceStaticSetNotReadyReq) {}