diff --git a/internal/protocols/hls/from_stream.go b/internal/protocols/hls/from_stream.go index 1015e3b9..e49ffcfb 100644 --- a/internal/protocols/hls/from_stream.go +++ b/internal/protocols/hls/from_stream.go @@ -4,6 +4,7 @@ package hls import ( "errors" "fmt" + "slices" "github.com/bluenviron/gohlslib/v2" "github.com/bluenviron/gohlslib/v2/pkg/codecs" @@ -20,24 +21,22 @@ var ErrNoSupportedCodecs = errors.New( "the stream doesn't contain any supported codec, which are currently AV1, VP9, H265, H264, Opus, MPEG-4 Audio") func setupVideoTrack( - strea *stream.Stream, - reader stream.Reader, + desc *description.Session, + r *stream.Reader, muxer *gohlslib.Muxer, - setuppedFormats map[format.Format]struct{}, ) { addTrack := func( media *description.Media, forma format.Format, track *gohlslib.Track, - readFunc stream.ReadFunc, + onData stream.OnDataFunc, ) { muxer.Tracks = append(muxer.Tracks, track) - setuppedFormats[forma] = struct{}{} - strea.AddReader(reader, media, forma, readFunc) + r.OnData(media, forma, onData) } var videoFormatAV1 *format.AV1 - videoMedia := strea.Desc.FindFormat(&videoFormatAV1) + videoMedia := desc.FindFormat(&videoFormatAV1) if videoFormatAV1 != nil { track := &gohlslib.Track{ @@ -72,7 +71,7 @@ func setupVideoTrack( } var videoFormatVP9 *format.VP9 - videoMedia = strea.Desc.FindFormat(&videoFormatVP9) + videoMedia = desc.FindFormat(&videoFormatVP9) if videoFormatVP9 != nil { track := &gohlslib.Track{ @@ -107,7 +106,7 @@ func setupVideoTrack( } var videoFormatH265 *format.H265 - videoMedia = strea.Desc.FindFormat(&videoFormatH265) + videoMedia = desc.FindFormat(&videoFormatH265) if videoFormatH265 != nil { vps, sps, pps := videoFormatH265.SafeParams() @@ -147,7 +146,7 @@ func setupVideoTrack( } var videoFormatH264 *format.H264 - videoMedia = strea.Desc.FindFormat(&videoFormatH264) + videoMedia = desc.FindFormat(&videoFormatH264) if videoFormatH264 != nil { sps, pps := videoFormatH264.SafeParams() @@ -187,23 +186,21 @@ func setupVideoTrack( } func setupAudioTracks( - strea *stream.Stream, - reader stream.Reader, + desc *description.Session, + r *stream.Reader, muxer *gohlslib.Muxer, - setuppedFormats map[format.Format]struct{}, ) { addTrack := func( medi *description.Media, forma format.Format, track *gohlslib.Track, - readFunc stream.ReadFunc, + onData stream.OnDataFunc, ) { muxer.Tracks = append(muxer.Tracks, track) - setuppedFormats[forma] = struct{}{} - strea.AddReader(reader, medi, forma, readFunc) + r.OnData(medi, forma, onData) } - for _, media := range strea.Desc.Medias { + for _, media := range desc.Medias { for _, forma := range media.Formats { switch forma := forma.(type) { case *format.Opus: @@ -305,35 +302,33 @@ func setupAudioTracks( // FromStream maps a MediaMTX stream to a HLS muxer. func FromStream( - stream *stream.Stream, - reader stream.Reader, + desc *description.Session, + r *stream.Reader, muxer *gohlslib.Muxer, ) error { - setuppedFormats := make(map[format.Format]struct{}) - setupVideoTrack( - stream, - reader, + desc, + r, muxer, - setuppedFormats, ) setupAudioTracks( - stream, - reader, + desc, + r, muxer, - setuppedFormats, ) if len(muxer.Tracks) == 0 { return ErrNoSupportedCodecs } + setuppedFormats := r.Formats() + n := 1 - for _, media := range stream.Desc.Medias { + for _, media := range desc.Medias { for _, forma := range media.Formats { - if _, ok := setuppedFormats[forma]; !ok { - reader.Log(logger.Warn, "skipping track %d (%s)", n, forma.Codec()) + if !slices.Contains(setuppedFormats, forma) { + r.Parent.Log(logger.Warn, "skipping track %d (%s)", n, forma.Codec()) } n++ } diff --git a/internal/protocols/hls/from_stream_test.go b/internal/protocols/hls/from_stream_test.go index de227bc8..99c85c2d 100644 --- a/internal/protocols/hls/from_stream_test.go +++ b/internal/protocols/hls/from_stream_test.go @@ -14,71 +14,58 @@ import ( ) func TestFromStreamNoSupportedCodecs(t *testing.T) { - strm := &stream.Stream{ - WriteQueueSize: 512, - RTPMaxPayloadSize: 1450, - Desc: &description.Session{Medias: []*description.Media{{ - Type: description.MediaTypeVideo, - Formats: []format.Format{&format.VP8{}}, - }}}, - GenerateRTPPackets: true, - Parent: test.NilLogger, - } - err := strm.Initialize() - require.NoError(t, err) + desc := &description.Session{Medias: []*description.Media{{ + Type: description.MediaTypeVideo, + Formats: []format.Format{&format.VP8{}}, + }}} - l := test.Logger(func(logger.Level, string, ...interface{}) { - t.Error("should not happen") - }) + r := &stream.Reader{ + Parent: test.Logger(func(logger.Level, string, ...interface{}) { + t.Error("should not happen") + }), + } m := &gohlslib.Muxer{} - err = FromStream(strm, l, m) + err := FromStream(desc, r, m) require.Equal(t, ErrNoSupportedCodecs, err) } func TestFromStreamSkipUnsupportedTracks(t *testing.T) { - strm := &stream.Stream{ - WriteQueueSize: 512, - RTPMaxPayloadSize: 1450, - Desc: &description.Session{Medias: []*description.Media{ - { - Type: description.MediaTypeVideo, - Formats: []format.Format{&format.VP9{}}, - }, - { - Type: description.MediaTypeVideo, - Formats: []format.Format{&format.VP8{}}, - }, - { - Type: description.MediaTypeAudio, - Formats: []format.Format{&format.MPEG1Audio{}}, - }, - }}, - GenerateRTPPackets: true, - Parent: test.NilLogger, - } - err := strm.Initialize() - require.NoError(t, err) + desc := &description.Session{Medias: []*description.Media{ + { + Type: description.MediaTypeVideo, + Formats: []format.Format{&format.VP9{}}, + }, + { + Type: description.MediaTypeVideo, + Formats: []format.Format{&format.VP8{}}, + }, + { + Type: description.MediaTypeAudio, + Formats: []format.Format{&format.MPEG1Audio{}}, + }, + }} m := &gohlslib.Muxer{} n := 0 - l := test.Logger(func(l logger.Level, format string, args ...interface{}) { - require.Equal(t, logger.Warn, l) - switch n { - case 0: - require.Equal(t, "skipping track 2 (VP8)", fmt.Sprintf(format, args...)) - case 1: - require.Equal(t, "skipping track 3 (MPEG-1/2 Audio)", fmt.Sprintf(format, args...)) - } - n++ - }) + r := &stream.Reader{ + Parent: test.Logger(func(l logger.Level, format string, args ...interface{}) { + require.Equal(t, logger.Warn, l) + switch n { + case 0: + require.Equal(t, "skipping track 2 (VP8)", fmt.Sprintf(format, args...)) + case 1: + require.Equal(t, "skipping track 3 (MPEG-1/2 Audio)", fmt.Sprintf(format, args...)) + } + n++ + }), + } - err = FromStream(strm, l, m) + err := FromStream(desc, r, m) require.NoError(t, err) - defer strm.RemoveReader(l) require.Equal(t, 2, n) } diff --git a/internal/protocols/hls/to_stream_test.go b/internal/protocols/hls/to_stream_test.go index 9e9b602a..655a715d 100644 --- a/internal/protocols/hls/to_stream_test.go +++ b/internal/protocols/hls/to_stream_test.go @@ -86,7 +86,7 @@ func TestToStream(t *testing.T) { var strm *stream.Stream done := make(chan struct{}) - reader := test.NilLogger + r := &stream.Reader{Parent: test.NilLogger} var c *gohlslib.Client c = &gohlslib.Client{ @@ -112,8 +112,7 @@ func TestToStream(t *testing.T) { err2 = strm.Initialize() require.NoError(t, err2) - strm.AddReader( - reader, + r.OnData( medias[0], medias[0].Formats[0], func(u unit.Unit) error { @@ -122,7 +121,7 @@ func TestToStream(t *testing.T) { return nil }) - strm.StartReader(reader) + strm.AddReader(r) return nil }, @@ -133,6 +132,6 @@ func TestToStream(t *testing.T) { <-done - strm.RemoveReader(reader) + strm.RemoveReader(r) strm.Close() } diff --git a/internal/protocols/mpegts/from_stream.go b/internal/protocols/mpegts/from_stream.go index 80772315..0cff3b0a 100644 --- a/internal/protocols/mpegts/from_stream.go +++ b/internal/protocols/mpegts/from_stream.go @@ -3,6 +3,7 @@ package mpegts import ( "bufio" "fmt" + "slices" "time" "github.com/bluenviron/gortsplib/v5/pkg/description" @@ -27,28 +28,26 @@ func multiplyAndDivide(v, m, d int64) int64 { // FromStream maps a MediaMTX stream to a MPEG-TS writer. func FromStream( - strea *stream.Stream, - reader stream.Reader, + desc *description.Session, + r *stream.Reader, bw *bufio.Writer, sconn srt.Conn, writeTimeout time.Duration, ) error { var w *mcmpegts.Writer var tracks []*mcmpegts.Track - setuppedFormats := make(map[format.Format]struct{}) addTrack := func( media *description.Media, forma format.Format, track *mcmpegts.Track, - readFunc stream.ReadFunc, + onData stream.OnDataFunc, ) { tracks = append(tracks, track) - setuppedFormats[forma] = struct{}{} - strea.AddReader(reader, media, forma, readFunc) + r.OnData(media, forma, onData) } - for _, media := range strea.Desc.Medias { + for _, media := range desc.Medias { for _, forma := range media.Formats { clockRate := forma.ClockRate() @@ -233,6 +232,7 @@ func FromStream( } return bw.Flush() }) + case *format.KLV: track := &mcmpegts.Track{ Codec: &mcmpegts.CodecKLV{ @@ -414,11 +414,13 @@ func FromStream( return errNoSupportedCodecs } + setuppedFormats := r.Formats() + n := 1 - for _, medi := range strea.Desc.Medias { + for _, medi := range desc.Medias { for _, forma := range medi.Formats { - if _, ok := setuppedFormats[forma]; !ok { - reader.Log(logger.Warn, "skipping track %d (%s)", n, forma.Codec()) + if !slices.Contains(setuppedFormats, forma) { + r.Parent.Log(logger.Warn, "skipping track %d (%s)", n, forma.Codec()) } n++ } diff --git a/internal/protocols/mpegts/from_stream_test.go b/internal/protocols/mpegts/from_stream_test.go index e9a79a03..355c072a 100644 --- a/internal/protocols/mpegts/from_stream_test.go +++ b/internal/protocols/mpegts/from_stream_test.go @@ -13,60 +13,47 @@ import ( ) func TestFromStreamNoSupportedCodecs(t *testing.T) { - strm := &stream.Stream{ - WriteQueueSize: 512, - RTPMaxPayloadSize: 1450, - Desc: &description.Session{Medias: []*description.Media{{ - Type: description.MediaTypeVideo, - Formats: []format.Format{&format.VP8{}}, - }}}, - GenerateRTPPackets: true, - Parent: test.NilLogger, + desc := &description.Session{Medias: []*description.Media{{ + Type: description.MediaTypeVideo, + Formats: []format.Format{&format.VP8{}}, + }}} + + r := &stream.Reader{ + Parent: test.Logger(func(logger.Level, string, ...interface{}) { + t.Error("should not happen") + }), } - err := strm.Initialize() - require.NoError(t, err) - l := test.Logger(func(logger.Level, string, ...interface{}) { - t.Error("should not happen") - }) - - err = FromStream(strm, l, nil, nil, 0) + err := FromStream(desc, r, nil, nil, 0) require.Equal(t, errNoSupportedCodecs, err) } func TestFromStreamSkipUnsupportedTracks(t *testing.T) { - strm := &stream.Stream{ - WriteQueueSize: 512, - RTPMaxPayloadSize: 1450, - Desc: &description.Session{Medias: []*description.Media{ - { - Type: description.MediaTypeVideo, - Formats: []format.Format{&format.H265{}}, - }, - { - Type: description.MediaTypeVideo, - Formats: []format.Format{&format.VP8{}}, - }, - }}, - GenerateRTPPackets: true, - Parent: test.NilLogger, - } - err := strm.Initialize() - require.NoError(t, err) + desc := &description.Session{Medias: []*description.Media{ + { + Type: description.MediaTypeVideo, + Formats: []format.Format{&format.H265{}}, + }, + { + Type: description.MediaTypeVideo, + Formats: []format.Format{&format.VP8{}}, + }, + }} n := 0 - l := test.Logger(func(l logger.Level, format string, args ...interface{}) { - require.Equal(t, logger.Warn, l) - if n == 0 { - require.Equal(t, "skipping track 2 (VP8)", fmt.Sprintf(format, args...)) - } - n++ - }) + r := &stream.Reader{ + Parent: test.Logger(func(l logger.Level, format string, args ...interface{}) { + require.Equal(t, logger.Warn, l) + if n == 0 { + require.Equal(t, "skipping track 2 (VP8)", fmt.Sprintf(format, args...)) + } + n++ + }), + } - err = FromStream(strm, l, nil, nil, 0) + err := FromStream(desc, r, nil, nil, 0) require.NoError(t, err) - defer strm.RemoveReader(l) require.Equal(t, 1, n) } diff --git a/internal/protocols/rtmp/from_stream.go b/internal/protocols/rtmp/from_stream.go index 7634a498..263540d3 100644 --- a/internal/protocols/rtmp/from_stream.go +++ b/internal/protocols/rtmp/from_stream.go @@ -9,6 +9,7 @@ import ( "github.com/bluenviron/gortmplib" "github.com/bluenviron/gortmplib/pkg/message" + "github.com/bluenviron/gortsplib/v5/pkg/description" "github.com/bluenviron/gortsplib/v5/pkg/format" "github.com/bluenviron/mediacommon/v2/pkg/codecs/ac3" "github.com/bluenviron/mediacommon/v2/pkg/codecs/h264" @@ -37,8 +38,8 @@ func timestampToDuration(t int64, clockRate int) time.Duration { // FromStream maps a MediaMTX stream to a RTMP stream. func FromStream( - str *stream.Stream, - reader stream.Reader, + desc *description.Session, + r *stream.Reader, conn *gortmplib.ServerConn, nconn net.Conn, writeTimeout time.Duration, @@ -46,13 +47,12 @@ func FromStream( var tracks []format.Format var w *gortmplib.Writer - for _, media := range str.Desc.Medias { + for _, media := range desc.Medias { for _, forma := range media.Formats { switch forma := forma.(type) { case *format.AV1: if slices.Contains(conn.FourCcList, interface{}(fourCCToString(message.FourCCAV1))) { - str.AddReader( - reader, + r.OnData( media, forma, func(u unit.Unit) error { @@ -74,8 +74,7 @@ func FromStream( case *format.VP9: if slices.Contains(conn.FourCcList, interface{}(fourCCToString(message.FourCCVP9))) { - str.AddReader( - reader, + r.OnData( media, forma, func(u unit.Unit) error { @@ -99,8 +98,7 @@ func FromStream( if slices.Contains(conn.FourCcList, interface{}(fourCCToString(message.FourCCHEVC))) { var videoDTSExtractor *h265.DTSExtractor - str.AddReader( - reader, + r.OnData( media, forma, func(u unit.Unit) error { @@ -137,8 +135,7 @@ func FromStream( case *format.H264: var videoDTSExtractor *h264.DTSExtractor - str.AddReader( - reader, + r.OnData( media, forma, func(u unit.Unit) error { @@ -191,8 +188,7 @@ func FromStream( case *format.Opus: if slices.Contains(conn.FourCcList, interface{}(fourCCToString(message.FourCCOpus))) { - str.AddReader( - reader, + r.OnData( media, forma, func(u unit.Unit) error { @@ -225,8 +221,7 @@ func FromStream( } case *format.MPEG4Audio: - str.AddReader( - reader, + r.OnData( media, forma, func(u unit.Unit) error { @@ -257,8 +252,7 @@ func FromStream( case *format.MPEG4AudioLATM: if !forma.CPresent { - str.AddReader( - reader, + r.OnData( media, forma, func(u unit.Unit) error { @@ -289,8 +283,7 @@ func FromStream( case *format.MPEG1Audio: // TODO: check sample rate and layer, // unfortunately they are not available at this stage. - str.AddReader( - reader, + r.OnData( media, forma, func(u unit.Unit) error { @@ -325,8 +318,7 @@ func FromStream( case *format.AC3: if slices.Contains(conn.FourCcList, interface{}(fourCCToString(message.FourCCAC3))) { - str.AddReader( - reader, + r.OnData( media, forma, func(u unit.Unit) error { @@ -353,8 +345,7 @@ func FromStream( case *format.G711: if forma.SampleRate == 8000 { - str.AddReader( - reader, + r.OnData( media, forma, func(u unit.Unit) error { @@ -381,8 +372,7 @@ func FromStream( forma.SampleRate == 11025 || forma.SampleRate == 22050 || forma.SampleRate == 44100) { - str.AddReader( - reader, + r.OnData( media, forma, func(u unit.Unit) error { @@ -420,10 +410,10 @@ func FromStream( } n := 1 - for _, media := range str.Desc.Medias { + for _, media := range desc.Medias { for _, forma := range media.Formats { if !slices.Contains(tracks, forma) { - reader.Log(logger.Warn, "skipping track %d (%s)", n, forma.Codec()) + r.Parent.Log(logger.Warn, "skipping track %d (%s)", n, forma.Codec()) } n++ } diff --git a/internal/protocols/rtmp/from_stream_test.go b/internal/protocols/rtmp/from_stream_test.go index 01c6d480..8a615a2b 100644 --- a/internal/protocols/rtmp/from_stream_test.go +++ b/internal/protocols/rtmp/from_stream_test.go @@ -409,13 +409,13 @@ func TestFromStream(t *testing.T) { err = conn.Accept() require.NoError(t, err) - reader := test.NilLogger + r := &stream.Reader{Parent: test.NilLogger} - err = FromStream(strm, reader, conn, nconn, 10*time.Second) + err = FromStream(strm.Desc, r, conn, nconn, 10*time.Second) require.NoError(t, err) - defer strm.RemoveReader(reader) - strm.StartReader(reader) + strm.AddReader(r) + defer strm.RemoveReader(r) switch ca { case "h264 + aac": @@ -675,56 +675,44 @@ func TestFromStream(t *testing.T) { } func TestFromStreamNoSupportedCodecs(t *testing.T) { - strm := &stream.Stream{ - WriteQueueSize: 512, - RTPMaxPayloadSize: 1450, - Desc: &description.Session{Medias: []*description.Media{{ - Type: description.MediaTypeVideo, - Formats: []format.Format{&format.VP8{}}, - }}}, - GenerateRTPPackets: true, - Parent: test.NilLogger, + desc := &description.Session{Medias: []*description.Media{{ + Type: description.MediaTypeVideo, + Formats: []format.Format{&format.VP8{}}, + }}} + + r := &stream.Reader{ + Parent: test.Logger(func(logger.Level, string, ...interface{}) { + t.Error("should not happen") + }), } - err := strm.Initialize() - require.NoError(t, err) - l := test.Logger(func(logger.Level, string, ...interface{}) { - t.Error("should not happen") - }) - - err = FromStream(strm, l, nil, nil, 0) + err := FromStream(desc, r, nil, nil, 0) require.Equal(t, errNoSupportedCodecsFrom, err) } func TestFromStreamSkipUnsupportedTracks(t *testing.T) { - strm := &stream.Stream{ - WriteQueueSize: 512, - RTPMaxPayloadSize: 1450, - Desc: &description.Session{Medias: []*description.Media{ - { - Type: description.MediaTypeVideo, - Formats: []format.Format{&format.VP8{}}, - }, - { - Type: description.MediaTypeVideo, - Formats: []format.Format{&format.H264{}}, - }, - }}, - GenerateRTPPackets: true, - Parent: test.NilLogger, - } - err := strm.Initialize() - require.NoError(t, err) + desc := &description.Session{Medias: []*description.Media{ + { + Type: description.MediaTypeVideo, + Formats: []format.Format{&format.VP8{}}, + }, + { + Type: description.MediaTypeVideo, + Formats: []format.Format{&format.H264{}}, + }, + }} n := 0 - l := test.Logger(func(l logger.Level, format string, args ...interface{}) { - require.Equal(t, logger.Warn, l) - if n == 0 { - require.Equal(t, "skipping track 1 (VP8)", fmt.Sprintf(format, args...)) - } - n++ - }) + r := &stream.Reader{ + Parent: test.Logger(func(l logger.Level, format string, args ...interface{}) { + require.Equal(t, logger.Warn, l) + if n == 0 { + require.Equal(t, "skipping track 1 (VP8)", fmt.Sprintf(format, args...)) + } + n++ + }), + } ln, err := net.Listen("tcp", "127.0.0.1:9121") require.NoError(t, err) @@ -754,9 +742,8 @@ func TestFromStreamSkipUnsupportedTracks(t *testing.T) { err = conn.Accept() require.NoError(t, err) - err = FromStream(strm, l, conn, nil, 0) + err = FromStream(desc, r, conn, nil, 0) require.NoError(t, err) - defer strm.RemoveReader(l) require.Equal(t, 1, n) } diff --git a/internal/protocols/webrtc/from_stream.go b/internal/protocols/webrtc/from_stream.go index 12ac4a7e..6015b9c2 100644 --- a/internal/protocols/webrtc/from_stream.go +++ b/internal/protocols/webrtc/from_stream.go @@ -6,6 +6,7 @@ import ( "fmt" "time" + "github.com/bluenviron/gortsplib/v5/pkg/description" "github.com/bluenviron/gortsplib/v5/pkg/format" "github.com/bluenviron/gortsplib/v5/pkg/format/rtpav1" "github.com/bluenviron/gortsplib/v5/pkg/format/rtph264" @@ -52,12 +53,12 @@ func timestampToDuration(t int64, clockRate int) time.Duration { } func setupVideoTrack( - stream *stream.Stream, - reader stream.Reader, + desc *description.Session, + r *stream.Reader, pc *PeerConnection, ) (format.Format, error) { var av1Format *format.AV1 - media := stream.Desc.FindFormat(&av1Format) + media := desc.FindFormat(&av1Format) if av1Format != nil { //nolint:dupl track := &OutgoingTrack{ @@ -77,8 +78,7 @@ func setupVideoTrack( return nil, err } - stream.AddReader( - reader, + r.OnData( media, av1Format, func(u unit.Unit) error { @@ -106,7 +106,7 @@ func setupVideoTrack( } var vp9Format *format.VP9 - media = stream.Desc.FindFormat(&vp9Format) + media = desc.FindFormat(&vp9Format) if vp9Format != nil { track := &OutgoingTrack{ @@ -128,8 +128,7 @@ func setupVideoTrack( return nil, err } - stream.AddReader( - reader, + r.OnData( media, vp9Format, func(u unit.Unit) error { @@ -157,7 +156,7 @@ func setupVideoTrack( } var vp8Format *format.VP8 - media = stream.Desc.FindFormat(&vp8Format) + media = desc.FindFormat(&vp8Format) if vp8Format != nil { //nolint:dupl track := &OutgoingTrack{ @@ -177,8 +176,7 @@ func setupVideoTrack( return nil, err } - stream.AddReader( - reader, + r.OnData( media, vp8Format, func(u unit.Unit) error { @@ -206,7 +204,7 @@ func setupVideoTrack( } var h265Format *format.H265 - media = stream.Desc.FindFormat(&h265Format) + media = desc.FindFormat(&h265Format) if h265Format != nil { //nolint:dupl track := &OutgoingTrack{ @@ -230,8 +228,7 @@ func setupVideoTrack( firstReceived := false var lastPTS int64 - stream.AddReader( - reader, + r.OnData( media, h265Format, func(u unit.Unit) error { @@ -266,7 +263,7 @@ func setupVideoTrack( } var h264Format *format.H264 - media = stream.Desc.FindFormat(&h264Format) + media = desc.FindFormat(&h264Format) if h264Format != nil { //nolint:dupl track := &OutgoingTrack{ @@ -290,8 +287,7 @@ func setupVideoTrack( firstReceived := false var lastPTS int64 - stream.AddReader( - reader, + r.OnData( media, h264Format, func(u unit.Unit) error { @@ -329,12 +325,12 @@ func setupVideoTrack( } func setupAudioTrack( - stream *stream.Stream, - reader stream.Reader, + desc *description.Session, + r *stream.Reader, pc *PeerConnection, ) (format.Format, error) { var opusFormat *format.Opus - media := stream.Desc.FindFormat(&opusFormat) + media := desc.FindFormat(&opusFormat) if opusFormat != nil { var caps webrtc.RTPCodecCapability @@ -371,8 +367,7 @@ func setupAudioTrack( } pc.OutgoingTracks = append(pc.OutgoingTracks, track) - stream.AddReader( - reader, + r.OnData( media, opusFormat, func(u unit.Unit) error { @@ -388,7 +383,7 @@ func setupAudioTrack( } var g722Format *format.G722 - media = stream.Desc.FindFormat(&g722Format) + media = desc.FindFormat(&g722Format) if g722Format != nil { track := &OutgoingTrack{ @@ -399,8 +394,7 @@ func setupAudioTrack( } pc.OutgoingTracks = append(pc.OutgoingTracks, track) - stream.AddReader( - reader, + r.OnData( media, g722Format, func(u unit.Unit) error { @@ -416,7 +410,7 @@ func setupAudioTrack( } var g711Format *format.G711 - media = stream.Desc.FindFormat(&g711Format) + media = desc.FindFormat(&g711Format) if g711Format != nil { // These are the sample rates and channels supported by Chrome. @@ -479,8 +473,7 @@ func setupAudioTrack( return nil, err } - stream.AddReader( - reader, + r.OnData( media, g711Format, func(u unit.Unit) error { @@ -513,8 +506,7 @@ func setupAudioTrack( return nil, err } - stream.AddReader( - reader, + r.OnData( media, g711Format, func(u unit.Unit) error { @@ -559,7 +551,7 @@ func setupAudioTrack( } var lpcmFormat *format.LPCM - media = stream.Desc.FindFormat(&lpcmFormat) + media = desc.FindFormat(&lpcmFormat) if lpcmFormat != nil { if lpcmFormat.BitDepth != 16 { @@ -602,8 +594,7 @@ func setupAudioTrack( return nil, err } - stream.AddReader( - reader, + r.OnData( media, lpcmFormat, func(u unit.Unit) error { @@ -640,16 +631,16 @@ func setupAudioTrack( // FromStream maps a MediaMTX stream to a WebRTC connection func FromStream( - stream *stream.Stream, - reader stream.Reader, + desc *description.Session, + r *stream.Reader, pc *PeerConnection, ) error { - videoFormat, err := setupVideoTrack(stream, reader, pc) + videoFormat, err := setupVideoTrack(desc, r, pc) if err != nil { return err } - audioFormat, err := setupAudioTrack(stream, reader, pc) + audioFormat, err := setupAudioTrack(desc, r, pc) if err != nil { return err } @@ -659,10 +650,10 @@ func FromStream( } n := 1 - for _, media := range stream.Desc.Medias { + for _, media := range desc.Medias { for _, forma := range media.Formats { if forma != videoFormat && forma != audioFormat { - reader.Log(logger.Warn, "skipping track %d (%s)", n, forma.Codec()) + r.Parent.Log(logger.Warn, "skipping track %d (%s)", n, forma.Codec()) } n++ } diff --git a/internal/protocols/webrtc/from_stream_test.go b/internal/protocols/webrtc/from_stream_test.go index 6e2737d4..3ead10b3 100644 --- a/internal/protocols/webrtc/from_stream_test.go +++ b/internal/protocols/webrtc/from_stream_test.go @@ -13,62 +13,49 @@ import ( ) func TestFromStreamNoSupportedCodecs(t *testing.T) { - strm := &stream.Stream{ - WriteQueueSize: 512, - RTPMaxPayloadSize: 1450, - Desc: &description.Session{Medias: []*description.Media{{ - Type: description.MediaTypeVideo, - Formats: []format.Format{&format.MJPEG{}}, - }}}, - GenerateRTPPackets: true, - Parent: test.NilLogger, + desc := &description.Session{Medias: []*description.Media{{ + Type: description.MediaTypeVideo, + Formats: []format.Format{&format.MJPEG{}}, + }}} + + r := &stream.Reader{ + Parent: test.Logger(func(logger.Level, string, ...interface{}) { + t.Error("should not happen") + }), } - err := strm.Initialize() - require.NoError(t, err) - l := test.Logger(func(logger.Level, string, ...interface{}) { - t.Error("should not happen") - }) - - err = FromStream(strm, l, nil) + err := FromStream(desc, r, nil) require.Equal(t, errNoSupportedCodecsFrom, err) } func TestFromStreamSkipUnsupportedTracks(t *testing.T) { - strm := &stream.Stream{ - WriteQueueSize: 512, - RTPMaxPayloadSize: 1450, - Desc: &description.Session{Medias: []*description.Media{ - { - Type: description.MediaTypeVideo, - Formats: []format.Format{&format.H264{}}, - }, - { - Type: description.MediaTypeVideo, - Formats: []format.Format{&format.MJPEG{}}, - }, - }}, - GenerateRTPPackets: true, - Parent: test.NilLogger, - } - err := strm.Initialize() - require.NoError(t, err) + desc := &description.Session{Medias: []*description.Media{ + { + Type: description.MediaTypeVideo, + Formats: []format.Format{&format.H264{}}, + }, + { + Type: description.MediaTypeVideo, + Formats: []format.Format{&format.MJPEG{}}, + }, + }} n := 0 - l := test.Logger(func(l logger.Level, format string, args ...interface{}) { - require.Equal(t, logger.Warn, l) - if n == 0 { - require.Equal(t, "skipping track 2 (M-JPEG)", fmt.Sprintf(format, args...)) - } - n++ - }) + r := &stream.Reader{ + Parent: test.Logger(func(l logger.Level, format string, args ...interface{}) { + require.Equal(t, logger.Warn, l) + if n == 0 { + require.Equal(t, "skipping track 2 (M-JPEG)", fmt.Sprintf(format, args...)) + } + n++ + }), + } pc := &PeerConnection{} - err = FromStream(strm, l, pc) + err := FromStream(desc, r, pc) require.NoError(t, err) - defer strm.RemoveReader(l) require.Equal(t, 1, n) } @@ -76,26 +63,17 @@ func TestFromStreamSkipUnsupportedTracks(t *testing.T) { func TestFromStream(t *testing.T) { for _, ca := range toFromStreamCases { t.Run(ca.name, func(t *testing.T) { - strm := &stream.Stream{ - WriteQueueSize: 512, - RTPMaxPayloadSize: 1450, - Desc: &description.Session{ - Medias: []*description.Media{{ - Formats: []format.Format{ca.in}, - }}, - }, - GenerateRTPPackets: false, - Parent: test.NilLogger, + desc := &description.Session{ + Medias: []*description.Media{{ + Formats: []format.Format{ca.in}, + }}, } - err := strm.Initialize() - require.NoError(t, err) - defer strm.Close() pc := &PeerConnection{} + r := &stream.Reader{Parent: test.NilLogger} - err = FromStream(strm, nil, pc) + err := FromStream(desc, r, pc) require.NoError(t, err) - defer strm.RemoveReader(nil) require.Equal(t, ca.webrtcCaps, pc.OutgoingTracks[0].Caps) }) diff --git a/internal/recorder/format_fmp4.go b/internal/recorder/format_fmp4.go index 2bd98ac9..1d533771 100644 --- a/internal/recorder/format_fmp4.go +++ b/internal/recorder/format_fmp4.go @@ -3,6 +3,7 @@ package recorder import ( "bytes" "fmt" + "slices" "time" rtspformat "github.com/bluenviron/gortsplib/v5/pkg/format" @@ -108,8 +109,6 @@ type formatFMP4 struct { func (f *formatFMP4) initialize() bool { nextID := 1 - var setuppedFormats []rtspformat.Format - setuppedFormatsMap := make(map[rtspformat.Format]struct{}) addTrack := func(format rtspformat.Format, codec mp4.Codec) *formatFMP4Track { initTrack := &fmp4.InitTrack{ @@ -125,8 +124,6 @@ func (f *formatFMP4) initialize() bool { } f.tracks = append(f.tracks, track) - setuppedFormats = append(setuppedFormats, format) - setuppedFormatsMap[format] = struct{}{} return track } @@ -143,8 +140,7 @@ func (f *formatFMP4) initialize() bool { firstReceived := false - f.ri.stream.AddReader( - f.ri, + f.ri.reader.OnData( media, forma, func(u unit.Unit) error { @@ -206,8 +202,7 @@ func (f *formatFMP4) initialize() bool { firstReceived := false - f.ri.stream.AddReader( - f.ri, + f.ri.reader.OnData( media, forma, func(u unit.Unit) error { @@ -296,8 +291,7 @@ func (f *formatFMP4) initialize() bool { var dtsExtractor *h265.DTSExtractor - f.ri.stream.AddReader( - f.ri, + f.ri.reader.OnData( media, forma, func(u unit.Unit) error { @@ -384,8 +378,7 @@ func (f *formatFMP4) initialize() bool { var dtsExtractor *h264.DTSExtractor - f.ri.stream.AddReader( - f.ri, + f.ri.reader.OnData( media, forma, func(u unit.Unit) error { @@ -465,8 +458,7 @@ func (f *formatFMP4) initialize() bool { firstReceived := false var lastPTS int64 - f.ri.stream.AddReader( - f.ri, + f.ri.reader.OnData( media, forma, func(u unit.Unit) error { @@ -519,8 +511,7 @@ func (f *formatFMP4) initialize() bool { firstReceived := false var lastPTS int64 - f.ri.stream.AddReader( - f.ri, + f.ri.reader.OnData( media, forma, func(u unit.Unit) error { @@ -573,8 +564,7 @@ func (f *formatFMP4) initialize() bool { parsed := false - f.ri.stream.AddReader( - f.ri, + f.ri.reader.OnData( media, forma, func(u unit.Unit) error { @@ -610,8 +600,7 @@ func (f *formatFMP4) initialize() bool { } track := addTrack(forma, codec) - f.ri.stream.AddReader( - f.ri, + f.ri.reader.OnData( media, forma, func(u unit.Unit) error { @@ -647,8 +636,7 @@ func (f *formatFMP4) initialize() bool { } track := addTrack(forma, codec) - f.ri.stream.AddReader( - f.ri, + f.ri.reader.OnData( media, forma, func(u unit.Unit) error { @@ -683,8 +671,7 @@ func (f *formatFMP4) initialize() bool { } track := addTrack(forma, codec) - f.ri.stream.AddReader( - f.ri, + f.ri.reader.OnData( media, forma, func(u unit.Unit) error { @@ -720,8 +707,7 @@ func (f *formatFMP4) initialize() bool { parsed := false - f.ri.stream.AddReader( - f.ri, + f.ri.reader.OnData( media, forma, func(u unit.Unit) error { @@ -780,8 +766,7 @@ func (f *formatFMP4) initialize() bool { parsed := false - f.ri.stream.AddReader( - f.ri, + f.ri.reader.OnData( media, forma, func(u unit.Unit) error { @@ -846,8 +831,7 @@ func (f *formatFMP4) initialize() bool { } track := addTrack(forma, codec) - f.ri.stream.AddReader( - f.ri, + f.ri.reader.OnData( media, forma, func(u unit.Unit) error { @@ -886,8 +870,7 @@ func (f *formatFMP4) initialize() bool { } track := addTrack(forma, codec) - f.ri.stream.AddReader( - f.ri, + f.ri.reader.OnData( media, forma, func(u unit.Unit) error { @@ -909,15 +892,17 @@ func (f *formatFMP4) initialize() bool { } } - if len(setuppedFormats) == 0 { + if len(f.tracks) == 0 { f.ri.Log(logger.Warn, "no supported tracks found, skipping recording") return false } + setuppedFormats := f.ri.reader.Formats() + n := 1 for _, medi := range f.ri.stream.Desc.Medias { for _, forma := range medi.Formats { - if _, ok := setuppedFormatsMap[forma]; !ok { + if !slices.Contains(setuppedFormats, forma) { f.ri.Log(logger.Warn, "skipping track %d (%s)", n, forma.Codec()) } n++ diff --git a/internal/recorder/format_mpegts.go b/internal/recorder/format_mpegts.go index f016d1b8..0376220a 100644 --- a/internal/recorder/format_mpegts.go +++ b/internal/recorder/format_mpegts.go @@ -5,6 +5,7 @@ import ( "bytes" "fmt" "io" + "slices" "time" rtspformat "github.com/bluenviron/gortsplib/v5/pkg/format" @@ -64,17 +65,12 @@ type formatMPEGTS struct { func (f *formatMPEGTS) initialize() bool { var tracks []*mpegts.Track - var setuppedFormats []rtspformat.Format - setuppedFormatsMap := make(map[rtspformat.Format]struct{}) - addTrack := func(format rtspformat.Format, codec mpegts.Codec) *mpegts.Track { + addTrack := func(codec mpegts.Codec) *mpegts.Track { track := &mpegts.Track{ Codec: codec, } - tracks = append(tracks, track) - setuppedFormats = append(setuppedFormats, format) - setuppedFormatsMap[format] = struct{}{} return track } @@ -84,12 +80,11 @@ func (f *formatMPEGTS) initialize() bool { switch forma := forma.(type) { case *rtspformat.H265: //nolint:dupl - track := addTrack(forma, &mpegts.CodecH265{}) + track := addTrack(&mpegts.CodecH265{}) var dtsExtractor *h265.DTSExtractor - f.ri.stream.AddReader( - f.ri, + f.ri.reader.OnData( media, forma, func(u unit.Unit) error { @@ -130,12 +125,11 @@ func (f *formatMPEGTS) initialize() bool { }) case *rtspformat.H264: //nolint:dupl - track := addTrack(forma, &mpegts.CodecH264{}) + track := addTrack(&mpegts.CodecH264{}) var dtsExtractor *h264.DTSExtractor - f.ri.stream.AddReader( - f.ri, + f.ri.reader.OnData( media, forma, func(u unit.Unit) error { @@ -176,13 +170,12 @@ func (f *formatMPEGTS) initialize() bool { }) case *rtspformat.MPEG4Video: - track := addTrack(forma, &mpegts.CodecMPEG4Video{}) + track := addTrack(&mpegts.CodecMPEG4Video{}) firstReceived := false var lastPTS int64 - f.ri.stream.AddReader( - f.ri, + f.ri.reader.OnData( media, forma, func(u unit.Unit) error { @@ -216,13 +209,12 @@ func (f *formatMPEGTS) initialize() bool { }) case *rtspformat.MPEG1Video: - track := addTrack(forma, &mpegts.CodecMPEG1Video{}) + track := addTrack(&mpegts.CodecMPEG1Video{}) firstReceived := false var lastPTS int64 - f.ri.stream.AddReader( - f.ri, + f.ri.reader.OnData( media, forma, func(u unit.Unit) error { @@ -256,12 +248,11 @@ func (f *formatMPEGTS) initialize() bool { }) case *rtspformat.Opus: - track := addTrack(forma, &mpegts.CodecOpus{ + track := addTrack(&mpegts.CodecOpus{ ChannelCount: forma.ChannelCount, }) - f.ri.stream.AddReader( - f.ri, + f.ri.reader.OnData( media, forma, func(u unit.Unit) error { @@ -286,12 +277,11 @@ func (f *formatMPEGTS) initialize() bool { }) case *rtspformat.KLV: - track := addTrack(forma, &mpegts.CodecKLV{ + track := addTrack(&mpegts.CodecKLV{ Synchronous: true, }) - f.ri.stream.AddReader( - f.ri, + f.ri.reader.OnData( media, forma, func(u unit.Unit) error { @@ -313,12 +303,11 @@ func (f *formatMPEGTS) initialize() bool { }) case *rtspformat.MPEG4Audio: - track := addTrack(forma, &mpegts.CodecMPEG4Audio{ + track := addTrack(&mpegts.CodecMPEG4Audio{ Config: *forma.Config, }) - f.ri.stream.AddReader( - f.ri, + f.ri.reader.OnData( media, forma, func(u unit.Unit) error { @@ -344,12 +333,11 @@ func (f *formatMPEGTS) initialize() bool { case *rtspformat.MPEG4AudioLATM: if !forma.CPresent { - track := addTrack(forma, &mpegts.CodecMPEG4Audio{ + track := addTrack(&mpegts.CodecMPEG4Audio{ Config: *forma.StreamMuxConfig.Programs[0].Layers[0].AudioSpecificConfig, }) - f.ri.stream.AddReader( - f.ri, + f.ri.reader.OnData( media, forma, func(u unit.Unit) error { @@ -382,10 +370,9 @@ func (f *formatMPEGTS) initialize() bool { } case *rtspformat.MPEG1Audio: - track := addTrack(forma, &mpegts.CodecMPEG1Audio{}) + track := addTrack(&mpegts.CodecMPEG1Audio{}) - f.ri.stream.AddReader( - f.ri, + f.ri.reader.OnData( media, forma, func(u unit.Unit) error { @@ -410,10 +397,9 @@ func (f *formatMPEGTS) initialize() bool { }) case *rtspformat.AC3: - track := addTrack(forma, &mpegts.CodecAC3{}) + track := addTrack(&mpegts.CodecAC3{}) - f.ri.stream.AddReader( - f.ri, + f.ri.reader.OnData( media, forma, func(u unit.Unit) error { @@ -449,15 +435,17 @@ func (f *formatMPEGTS) initialize() bool { } } - if len(setuppedFormats) == 0 { + if len(tracks) == 0 { f.ri.Log(logger.Warn, "no supported tracks found, skipping recording") return false } + setuppedFormats := f.ri.reader.Formats() + n := 1 for _, medi := range f.ri.stream.Desc.Medias { for _, forma := range medi.Formats { - if _, ok := setuppedFormatsMap[forma]; !ok { + if !slices.Contains(setuppedFormats, forma) { f.ri.Log(logger.Warn, "skipping track %d (%s)", n, forma.Codec()) } n++ diff --git a/internal/recorder/recorder_instance.go b/internal/recorder/recorder_instance.go index fb388058..7d9ba8e4 100644 --- a/internal/recorder/recorder_instance.go +++ b/internal/recorder/recorder_instance.go @@ -33,6 +33,7 @@ type recorderInstance struct { pathFormat2 string format2 format skip bool + reader *stream.Reader terminate chan struct{} done chan struct{} @@ -45,11 +46,14 @@ func (ri *recorderInstance) Log(level logger.Level, format string, args ...inter func (ri *recorderInstance) initialize() { ri.pathFormat2 = ri.pathFormat - ri.pathFormat2 = recordstore.PathAddExtension( strings.ReplaceAll(ri.pathFormat2, "%path", ri.pathName), ri.format, ) + ri.reader = &stream.Reader{ + SkipBytesSent: true, + Parent: ri, + } ri.terminate = make(chan struct{}) ri.done = make(chan struct{}) @@ -71,7 +75,7 @@ func (ri *recorderInstance) initialize() { } if !ri.skip { - ri.stream.StartReader(ri) + ri.stream.AddReader(ri.reader) } go ri.run() @@ -87,13 +91,13 @@ func (ri *recorderInstance) run() { if !ri.skip { select { - case err := <-ri.stream.ReaderError(ri): + case err := <-ri.reader.Error(): ri.Log(logger.Error, err.Error()) case <-ri.terminate: } - ri.stream.RemoveReader(ri) + ri.stream.RemoveReader(ri.reader) } else { <-ri.terminate } diff --git a/internal/servers/hls/muxer_instance.go b/internal/servers/hls/muxer_instance.go index 14f8f057..ff6e80c4 100644 --- a/internal/servers/hls/muxer_instance.go +++ b/internal/servers/hls/muxer_instance.go @@ -27,6 +27,7 @@ type muxerInstance struct { parent logger.Writer hmuxer *gohlslib.Muxer + reader *stream.Reader } func (mi *muxerInstance) initialize() error { @@ -48,21 +49,25 @@ func (mi *muxerInstance) initialize() error { }, } - err := hls.FromStream(mi.stream, mi, mi.hmuxer) + mi.reader = &stream.Reader{ + SkipBytesSent: true, + Parent: mi, + } + + err := hls.FromStream(mi.stream.Desc, mi.reader, mi.hmuxer) if err != nil { return err } err = mi.hmuxer.Start() if err != nil { - mi.stream.RemoveReader(mi) return err } mi.Log(logger.Info, "is converting into HLS, %s", - defs.FormatsInfo(mi.stream.ReaderFormats(mi))) + defs.FormatsInfo(mi.reader.Formats())) - mi.stream.StartReader(mi) + mi.stream.AddReader(mi.reader) return nil } @@ -73,7 +78,7 @@ func (mi *muxerInstance) Log(level logger.Level, format string, args ...interfac } func (mi *muxerInstance) close() { - mi.stream.RemoveReader(mi) + mi.stream.RemoveReader(mi.reader) mi.hmuxer.Close() if mi.hmuxer.Directory != "" { os.Remove(mi.hmuxer.Directory) @@ -81,7 +86,7 @@ func (mi *muxerInstance) close() { } func (mi *muxerInstance) errorChan() chan error { - return mi.stream.ReaderError(mi) + return mi.reader.Error() } func (mi *muxerInstance) handleRequest(ctx *gin.Context) { diff --git a/internal/servers/hls/server_test.go b/internal/servers/hls/server_test.go index 75bd0e72..f4022e89 100644 --- a/internal/servers/hls/server_test.go +++ b/internal/servers/hls/server_test.go @@ -324,7 +324,7 @@ func TestServerRead(t *testing.T) { s.PathReady(&dummyPath{}) - strm.WaitRunningReader() + time.Sleep(500 * time.Millisecond) for i := range 4 { strm.WriteUnit(test.MediaH264, test.FormatH264, &unit.H264{ diff --git a/internal/servers/rtmp/conn.go b/internal/servers/rtmp/conn.go index 9d511d76..c1083fa4 100644 --- a/internal/servers/rtmp/conn.go +++ b/internal/servers/rtmp/conn.go @@ -151,7 +151,7 @@ func (c *conn) runRead() error { pathName := strings.TrimLeft(c.rconn.URL.Path, "/") query := c.rconn.URL.Query() - path, stream, err := c.pathManager.AddReader(defs.PathAddReaderReq{ + path, strm, err := c.pathManager.AddReader(defs.PathAddReaderReq{ Author: c, AccessRequest: defs.PathAccessRequest{ Name: pathName, @@ -183,13 +183,15 @@ func (c *conn) runRead() error { c.query = c.rconn.URL.RawQuery c.mutex.Unlock() - err = rtmp.FromStream(stream, c, c.rconn, c.nconn, time.Duration(c.writeTimeout)) + r := &stream.Reader{Parent: c} + + err = rtmp.FromStream(strm.Desc, r, c.rconn, c.nconn, time.Duration(c.writeTimeout)) if err != nil { return err } c.Log(logger.Info, "is reading from path '%s', %s", - path.Name(), defs.FormatsInfo(stream.ReaderFormats(c))) + path.Name(), defs.FormatsInfo(r.Formats())) onUnreadHook := hooks.OnRead(hooks.OnReadParams{ Logger: c, @@ -204,14 +206,14 @@ func (c *conn) runRead() error { // disable read deadline c.nconn.SetReadDeadline(time.Time{}) - stream.StartReader(c) - defer stream.RemoveReader(c) + strm.AddReader(r) + defer strm.RemoveReader(r) select { case <-c.ctx.Done(): return fmt.Errorf("terminated") - case err = <-stream.ReaderError(c): + case err = <-r.Error(): return err } } diff --git a/internal/servers/rtmp/server_test.go b/internal/servers/rtmp/server_test.go index f685a02b..d85f231f 100644 --- a/internal/servers/rtmp/server_test.go +++ b/internal/servers/rtmp/server_test.go @@ -145,10 +145,9 @@ func TestServerPublish(t *testing.T) { recv := make(chan struct{}) - reader := test.NilLogger + r := &stream.Reader{Parent: test.NilLogger} - strm.AddReader( - reader, + r.OnData( strm.Desc.Medias[0], strm.Desc.Medias[0].Formats[0], func(u unit.Unit) error { @@ -161,8 +160,8 @@ func TestServerPublish(t *testing.T) { return nil }) - strm.StartReader(reader) - defer strm.RemoveReader(reader) + strm.AddReader(r) + defer strm.RemoveReader(r) err = w.WriteH264( test.FormatH264, @@ -259,7 +258,7 @@ func TestServerRead(t *testing.T) { defer conn.Close() go func() { - strm.WaitRunningReader() + time.Sleep(500 * time.Millisecond) strm.WriteUnit(desc.Medias[0], desc.Medias[0].Formats[0], &unit.H264{ Base: unit.Base{ diff --git a/internal/servers/rtsp/server_test.go b/internal/servers/rtsp/server_test.go index fded66bc..c6f6ab34 100644 --- a/internal/servers/rtsp/server_test.go +++ b/internal/servers/rtsp/server_test.go @@ -133,12 +133,11 @@ func TestServerPublish(t *testing.T) { <-streamCreated - reader := test.NilLogger + r := &stream.Reader{Parent: test.NilLogger} recv := make(chan struct{}) - strm.AddReader( - reader, + r.OnData( strm.Desc.Medias[0], strm.Desc.Medias[0].Formats[0], func(u unit.Unit) error { @@ -151,8 +150,8 @@ func TestServerPublish(t *testing.T) { return nil }) - strm.StartReader(reader) - defer strm.RemoveReader(reader) + strm.AddReader(r) + defer strm.RemoveReader(r) err = source.WritePacketRTP(media0, &rtp.Packet{ Header: rtp.Header{ diff --git a/internal/servers/srt/conn.go b/internal/servers/srt/conn.go index a5498eba..50f36a5f 100644 --- a/internal/servers/srt/conn.go +++ b/internal/servers/srt/conn.go @@ -255,7 +255,7 @@ func (c *conn) runPublishReader(sconn srt.Conn, streamID *streamID, pathConf *co } func (c *conn) runRead(streamID *streamID) error { - path, stream, err := c.pathManager.AddReader(defs.PathAddReaderReq{ + path, strm, err := c.pathManager.AddReader(defs.PathAddReaderReq{ Author: c, AccessRequest: defs.PathAccessRequest{ Name: streamID.path, @@ -297,7 +297,9 @@ func (c *conn) runRead(streamID *streamID) error { bw := bufio.NewWriterSize(sconn, srtMaxPayloadSize(c.udpMaxPayloadSize)) - err = mpegts.FromStream(stream, c, bw, sconn, time.Duration(c.writeTimeout)) + r := &stream.Reader{Parent: c} + + err = mpegts.FromStream(strm.Desc, r, bw, sconn, time.Duration(c.writeTimeout)) if err != nil { return err } @@ -310,7 +312,7 @@ func (c *conn) runRead(streamID *streamID) error { c.mutex.Unlock() c.Log(logger.Info, "is reading from path '%s', %s", - path.Name(), defs.FormatsInfo(stream.ReaderFormats(c))) + path.Name(), defs.FormatsInfo(r.Formats())) onUnreadHook := hooks.OnRead(hooks.OnReadParams{ Logger: c, @@ -325,14 +327,14 @@ func (c *conn) runRead(streamID *streamID) error { // disable read deadline sconn.SetReadDeadline(time.Time{}) - stream.StartReader(c) - defer stream.RemoveReader(c) + strm.AddReader(r) + defer strm.RemoveReader(r) select { case <-c.ctx.Done(): return fmt.Errorf("terminated") - case err = <-stream.ReaderError(c): + case err = <-r.Error(): return err } } diff --git a/internal/servers/srt/server_test.go b/internal/servers/srt/server_test.go index d8ec9dbf..302686a3 100644 --- a/internal/servers/srt/server_test.go +++ b/internal/servers/srt/server_test.go @@ -125,12 +125,11 @@ func TestServerPublish(t *testing.T) { <-streamCreated - reader := test.NilLogger + r := &stream.Reader{Parent: test.NilLogger} recv := make(chan struct{}) - strm.AddReader( - reader, + r.OnData( strm.Desc.Medias[0], strm.Desc.Medias[0].Formats[0], func(u unit.Unit) error { @@ -143,8 +142,8 @@ func TestServerPublish(t *testing.T) { return nil }) - strm.StartReader(reader) - defer strm.RemoveReader(reader) + strm.AddReader(r) + defer strm.RemoveReader(r) err = w.WriteH264(track, 0, 0, [][]byte{ {5, 2}, @@ -214,7 +213,7 @@ func TestServerRead(t *testing.T) { require.NoError(t, err) defer reader.Close() - strm.WaitRunningReader() + time.Sleep(500 * time.Millisecond) strm.WriteUnit(desc.Medias[0], desc.Medias[0].Formats[0], &unit.H264{ Base: unit.Base{ diff --git a/internal/servers/webrtc/server_test.go b/internal/servers/webrtc/server_test.go index ed59bdd1..a0b617eb 100644 --- a/internal/servers/webrtc/server_test.go +++ b/internal/servers/webrtc/server_test.go @@ -288,12 +288,11 @@ func TestServerPublish(t *testing.T) { <-streamCreated - reader := test.NilLogger + r := &stream.Reader{Parent: test.NilLogger} recv := make(chan struct{}) - strm.AddReader( - reader, + r.OnData( strm.Desc.Medias[0], strm.Desc.Medias[0].Formats[0], func(u unit.Unit) error { @@ -311,8 +310,8 @@ func TestServerPublish(t *testing.T) { return nil }) - strm.StartReader(reader) - defer strm.RemoveReader(reader) + strm.AddReader(r) + defer strm.RemoveReader(r) err = track.WriteRTP(&rtp.Packet{ Header: rtp.Header{ @@ -547,7 +546,7 @@ func TestServerRead(t *testing.T) { go func() { defer close(writerDone) - strm.WaitRunningReader() + time.Sleep(500 * time.Millisecond) r := reflect.New(reflect.TypeOf(ca.unit).Elem()) r.Elem().Set(reflect.ValueOf(ca.unit).Elem()) diff --git a/internal/servers/webrtc/session.go b/internal/servers/webrtc/session.go index 151e0170..2d8ada23 100644 --- a/internal/servers/webrtc/session.go +++ b/internal/servers/webrtc/session.go @@ -281,7 +281,7 @@ func (s *session) runRead() (int, error) { IP: net.ParseIP(ip), } - path, stream, err := s.pathManager.AddReader(defs.PathAddReaderReq{ + path, strm, err := s.pathManager.AddReader(defs.PathAddReaderReq{ Author: s, AccessRequest: req, }) @@ -316,14 +316,15 @@ func (s *session) runRead() (int, error) { Log: s, } - err = webrtc.FromStream(stream, s, pc) + r := &stream.Reader{Parent: s} + + err = webrtc.FromStream(strm.Desc, r, pc) if err != nil { return http.StatusBadRequest, err } err = pc.Start() if err != nil { - stream.RemoveReader(s) return http.StatusBadRequest, err } @@ -346,7 +347,6 @@ func (s *session) runRead() (int, error) { answer, err := pc.CreateFullAnswer(offer) if err != nil { - stream.RemoveReader(s) return http.StatusBadRequest, err } @@ -356,7 +356,6 @@ func (s *session) runRead() (int, error) { err = pc.WaitUntilConnected() if err != nil { - stream.RemoveReader(s) return 0, err } @@ -365,7 +364,7 @@ func (s *session) runRead() (int, error) { s.mutex.Unlock() s.Log(logger.Info, "is reading from path '%s', %s", - path.Name(), defs.FormatsInfo(stream.ReaderFormats(s))) + path.Name(), defs.FormatsInfo(r.Formats())) onUnreadHook := hooks.OnRead(hooks.OnReadParams{ Logger: s, @@ -377,14 +376,14 @@ func (s *session) runRead() (int, error) { }) defer onUnreadHook() - stream.StartReader(s) - defer stream.RemoveReader(s) + strm.AddReader(r) + defer strm.RemoveReader(r) select { case <-pc.Failed(): return 0, fmt.Errorf("peer connection closed") - case err = <-stream.ReaderError(s): + case err = <-r.Error(): return 0, err case <-s.ctx.Done(): diff --git a/internal/staticsources/rpicamera/source.go b/internal/staticsources/rpicamera/source.go index f3b53c7a..d665a362 100644 --- a/internal/staticsources/rpicamera/source.go +++ b/internal/staticsources/rpicamera/source.go @@ -282,8 +282,9 @@ func (s *Source) runSecondary(params defs.StaticSourceRunParams) error { return res.Err } - origStream.AddReader( - s, + rdr := &stream.Reader{Parent: s} + + rdr.OnData( origStream.Desc.Medias[1], origStream.Desc.Medias[1].Formats[0], func(u unit.Unit) error { @@ -299,11 +300,11 @@ func (s *Source) runSecondary(params defs.StaticSourceRunParams) error { return nil }) - origStream.StartReader(s) - defer origStream.RemoveReader(s) + origStream.AddReader(rdr) + defer origStream.RemoveReader(rdr) select { - case err = <-origStream.ReaderError(s): + case err = <-rdr.Error(): return err case <-r.ctx.Done(): diff --git a/internal/stream/reader.go b/internal/stream/reader.go new file mode 100644 index 00000000..68577241 --- /dev/null +++ b/internal/stream/reader.go @@ -0,0 +1,113 @@ +package stream + +import ( + "fmt" + + "github.com/bluenviron/gortsplib/v5/pkg/description" + "github.com/bluenviron/gortsplib/v5/pkg/format" + "github.com/bluenviron/gortsplib/v5/pkg/ringbuffer" + "github.com/bluenviron/mediamtx/internal/counterdumper" + "github.com/bluenviron/mediamtx/internal/logger" + "github.com/bluenviron/mediamtx/internal/unit" +) + +// OnDataFunc is the callback passed to OnData(). +type OnDataFunc func(unit.Unit) error + +// Reader is a stream reader. +type Reader struct { + SkipBytesSent bool + Parent logger.Writer + + onDatas map[*description.Media]map[format.Format]OnDataFunc + queueSize int + buffer *ringbuffer.RingBuffer + discardedFrames *counterdumper.CounterDumper + + // out + err chan error +} + +// OnData registers a callback that is called when data from given format is available. +func (r *Reader) OnData(medi *description.Media, forma format.Format, cb OnDataFunc) { + if r.onDatas == nil { + r.onDatas = make(map[*description.Media]map[format.Format]OnDataFunc) + } + if r.onDatas[medi] == nil { + r.onDatas[medi] = make(map[format.Format]OnDataFunc) + } + r.onDatas[medi][forma] = cb +} + +// Formats returns all formats for which the reader has registered a OnData callback. +func (r *Reader) Formats() []format.Format { + var out []format.Format + + for _, formats := range r.onDatas { + for forma := range formats { + out = append(out, forma) + } + } + + return out +} + +// error returns whenever there's an error. +// It can be called only after stream.AddReader(). +func (r *Reader) Error() chan error { + return r.err +} + +func (r *Reader) start() { + buffer, _ := ringbuffer.New(uint64(r.queueSize)) + r.buffer = buffer + r.err = make(chan error) + + r.discardedFrames = &counterdumper.CounterDumper{ + OnReport: func(val uint64) { + r.Parent.Log(logger.Warn, "reader is too slow, discarding %d %s", + val, + func() string { + if val == 1 { + return "frame" + } + return "frames" + }()) + }, + } + r.discardedFrames.Start() + + go r.run() +} + +func (r *Reader) stop() { + r.buffer.Close() + r.discardedFrames.Stop() + <-r.err +} + +func (r *Reader) run() { + r.err <- r.runInner() + close(r.err) +} + +func (r *Reader) runInner() error { + for { + cb, ok := r.buffer.Pull() + if !ok { + return fmt.Errorf("terminated") + } + + err := cb.(func() error)() + if err != nil { + return err + } + } +} + +func (r *Reader) push(cb func() error) { + ok := r.buffer.Push(cb) + if !ok { + r.discardedFrames.Increase() + } +} diff --git a/internal/stream/stream.go b/internal/stream/stream.go index 4eeeccb2..b82824be 100644 --- a/internal/stream/stream.go +++ b/internal/stream/stream.go @@ -16,14 +16,6 @@ import ( "github.com/bluenviron/mediamtx/internal/unit" ) -// Reader is a stream reader. -type Reader interface { - logger.Writer -} - -// ReadFunc is the callback passed to AddReader(). -type ReadFunc func(unit.Unit) error - // Stream is a media stream. // It stores tracks, readers and allows to write data to readers, converting it when needed. type Stream struct { @@ -35,23 +27,20 @@ type Stream struct { bytesReceived *uint64 bytesSent *uint64 - streamMedias map[*description.Media]*streamMedia + medias map[*description.Media]*streamMedia mutex sync.RWMutex rtspStream *gortsplib.ServerStream rtspsStream *gortsplib.ServerStream - streamReaders map[Reader]*streamReader + readers map[*Reader]struct{} processingErrors *counterdumper.CounterDumper - - readerRunning chan struct{} } // Initialize initializes a Stream. func (s *Stream) Initialize() error { s.bytesReceived = new(uint64) s.bytesSent = new(uint64) - s.streamMedias = make(map[*description.Media]*streamMedia) - s.streamReaders = make(map[Reader]*streamReader) - s.readerRunning = make(chan struct{}) + s.medias = make(map[*description.Media]*streamMedia) + s.readers = make(map[*Reader]struct{}) s.processingErrors = &counterdumper.CounterDumper{ OnReport: func(val uint64) { @@ -68,14 +57,14 @@ func (s *Stream) Initialize() error { s.processingErrors.Start() for _, media := range s.Desc.Medias { - s.streamMedias[media] = &streamMedia{ + s.medias[media] = &streamMedia{ rtpMaxPayloadSize: s.RTPMaxPayloadSize, media: media, generateRTPPackets: s.GenerateRTPPackets, processingErrors: s.processingErrors, parent: s.Parent, } - err := s.streamMedias[media].initialize() + err := s.medias[media].initialize() if err != nil { return err } @@ -103,10 +92,11 @@ func (s *Stream) BytesReceived() uint64 { // BytesSent returns sent bytes. func (s *Stream) BytesSent() uint64 { + bytesSent := atomic.LoadUint64(s.bytesSent) + s.mutex.RLock() defer s.mutex.RUnlock() - bytesSent := atomic.LoadUint64(s.bytesSent) if s.rtspStream != nil { stats := s.rtspStream.Stats() bytesSent += stats.BytesSent @@ -115,6 +105,7 @@ func (s *Stream) BytesSent() uint64 { stats := s.rtspsStream.Stats() bytesSent += stats.BytesSent } + return bytesSent } @@ -156,103 +147,48 @@ func (s *Stream) RTSPSStream(server *gortsplib.Server) *gortsplib.ServerStream { // AddReader adds a reader. // Used by all protocols except RTSP. -func (s *Stream) AddReader(reader Reader, medi *description.Media, forma format.Format, cb ReadFunc) { +func (s *Stream) AddReader(r *Reader) { s.mutex.Lock() defer s.mutex.Unlock() - sr, ok := s.streamReaders[reader] - if !ok { - sr = &streamReader{ - queueSize: s.WriteQueueSize, - parent: reader, - } - sr.initialize() + s.readers[r] = struct{}{} - s.streamReaders[reader] = sr + for medi, formats := range r.onDatas { + sm := s.medias[medi] + + for forma, onData := range formats { + sf := sm.formats[forma] + sf.onDatas[r] = onData + } } - sm := s.streamMedias[medi] - sf := sm.formats[forma] - sf.addReader(sr, cb) + r.queueSize = s.WriteQueueSize + r.start() } // RemoveReader removes a reader. // Used by all protocols except RTSP. -func (s *Stream) RemoveReader(reader Reader) { +func (s *Stream) RemoveReader(r *Reader) { s.mutex.Lock() defer s.mutex.Unlock() - sr := s.streamReaders[reader] + r.stop() - for _, sm := range s.streamMedias { - for _, sf := range sm.formats { - sf.removeReader(sr) + for medi, formats := range r.onDatas { + sm := s.medias[medi] + + for forma := range formats { + sf := sm.formats[forma] + delete(sf.onDatas, r) } } - delete(s.streamReaders, reader) - - sr.stop() -} - -// StartReader starts a reader. -// Used by all protocols except RTSP. -func (s *Stream) StartReader(reader Reader) { - s.mutex.Lock() - defer s.mutex.Unlock() - - sr := s.streamReaders[reader] - - sr.start() - - for _, sm := range s.streamMedias { - for _, sf := range sm.formats { - sf.startReader(sr) - } - } - - select { - case <-s.readerRunning: - default: - close(s.readerRunning) - } -} - -// ReaderError returns whenever there's an error. -func (s *Stream) ReaderError(reader Reader) chan error { - sr := s.streamReaders[reader] - return sr.error() -} - -// ReaderFormats returns all formats that a reader is reading. -func (s *Stream) ReaderFormats(reader Reader) []format.Format { - s.mutex.RLock() - defer s.mutex.RUnlock() - - sr := s.streamReaders[reader] - var formats []format.Format - - for _, sm := range s.streamMedias { - for forma, sf := range sm.formats { - if _, ok := sf.pausedReaders[sr]; ok { - formats = append(formats, forma) - } else if _, ok = sf.runningReaders[sr]; ok { - formats = append(formats, forma) - } - } - } - - return formats -} - -// WaitRunningReader waits for a running reader. -func (s *Stream) WaitRunningReader() { - <-s.readerRunning + delete(s.readers, r) } // WriteUnit writes a Unit. func (s *Stream) WriteUnit(medi *description.Media, forma format.Format, u unit.Unit) { - sm := s.streamMedias[medi] + sm := s.medias[medi] sf := sm.formats[forma] s.mutex.RLock() @@ -269,7 +205,7 @@ func (s *Stream) WriteRTPPacket( ntp time.Time, pts int64, ) { - sm := s.streamMedias[medi] + sm := s.medias[medi] sf := sm.formats[forma] s.mutex.RLock() diff --git a/internal/stream/stream_format.go b/internal/stream/stream_format.go index 2a3f396f..90c50c13 100644 --- a/internal/stream/stream_format.go +++ b/internal/stream/stream_format.go @@ -29,14 +29,12 @@ type streamFormat struct { processingErrors *counterdumper.CounterDumper parent logger.Writer - proc codecprocessor.Processor - pausedReaders map[*streamReader]ReadFunc - runningReaders map[*streamReader]ReadFunc + proc codecprocessor.Processor + onDatas map[*Reader]OnDataFunc } func (sf *streamFormat) initialize() error { - sf.pausedReaders = make(map[*streamReader]ReadFunc) - sf.runningReaders = make(map[*streamReader]ReadFunc) + sf.onDatas = make(map[*Reader]OnDataFunc) var err error sf.proc, err = codecprocessor.New(sf.rtpMaxPayloadSize, sf.format, sf.generateRTPPackets, sf.parent) @@ -47,22 +45,6 @@ func (sf *streamFormat) initialize() error { return nil } -func (sf *streamFormat) addReader(sr *streamReader, cb ReadFunc) { - sf.pausedReaders[sr] = cb -} - -func (sf *streamFormat) removeReader(sr *streamReader) { - delete(sf.pausedReaders, sr) - delete(sf.runningReaders, sr) -} - -func (sf *streamFormat) startReader(sr *streamReader) { - if cb, ok := sf.pausedReaders[sr]; ok { - delete(sf.pausedReaders, sr) - sf.runningReaders[sr] = cb - } -} - func (sf *streamFormat) writeUnit(s *Stream, medi *description.Media, u unit.Unit) { err := sf.proc.ProcessUnit(u) if err != nil { @@ -80,7 +62,7 @@ func (sf *streamFormat) writeRTPPacket( ntp time.Time, pts int64, ) { - hasNonRTSPReaders := len(sf.pausedReaders) > 0 || len(sf.runningReaders) > 0 + hasNonRTSPReaders := len(sf.onDatas) > 0 u, err := sf.proc.ProcessRTPPacket(pkt, ntp, pts, hasNonRTSPReaders) if err != nil { @@ -108,11 +90,14 @@ func (sf *streamFormat) writeUnitInner(s *Stream, medi *description.Media, u uni } } - for sr, cb := range sf.runningReaders { - ccb := cb + for sr, onData := range sf.onDatas { + csr := sr + cOnData := onData sr.push(func() error { - atomic.AddUint64(s.bytesSent, size) - return ccb(u) + if !csr.SkipBytesSent { + atomic.AddUint64(s.bytesSent, size) + } + return cOnData(u) }) } } diff --git a/internal/stream/stream_reader.go b/internal/stream/stream_reader.go deleted file mode 100644 index 42b536be..00000000 --- a/internal/stream/stream_reader.go +++ /dev/null @@ -1,86 +0,0 @@ -package stream - -import ( - "fmt" - - "github.com/bluenviron/gortsplib/v5/pkg/ringbuffer" - "github.com/bluenviron/mediamtx/internal/counterdumper" - "github.com/bluenviron/mediamtx/internal/logger" -) - -type streamReader struct { - queueSize int - parent logger.Writer - - buffer *ringbuffer.RingBuffer - started bool - discardedFrames *counterdumper.CounterDumper - - // out - err chan error -} - -func (w *streamReader) initialize() { - buffer, _ := ringbuffer.New(uint64(w.queueSize)) - w.buffer = buffer - w.err = make(chan error) -} - -func (w *streamReader) start() { - w.started = true - - w.discardedFrames = &counterdumper.CounterDumper{ - OnReport: func(val uint64) { - w.parent.Log(logger.Warn, "reader is too slow, discarding %d %s", - val, - func() string { - if val == 1 { - return "frame" - } - return "frames" - }()) - }, - } - w.discardedFrames.Start() - - go w.run() -} - -func (w *streamReader) stop() { - w.buffer.Close() - - if w.started { - w.discardedFrames.Stop() - <-w.err - } -} - -func (w *streamReader) error() chan error { - return w.err -} - -func (w *streamReader) run() { - w.err <- w.runInner() - close(w.err) -} - -func (w *streamReader) runInner() error { - for { - cb, ok := w.buffer.Pull() - if !ok { - return fmt.Errorf("terminated") - } - - err := cb.(func() error)() - if err != nil { - return err - } - } -} - -func (w *streamReader) push(cb func() error) { - ok := w.buffer.Push(cb) - if !ok { - w.discardedFrames.Increase() - } -} diff --git a/internal/stream/stream_test.go b/internal/stream/stream_test.go new file mode 100644 index 00000000..64a161ab --- /dev/null +++ b/internal/stream/stream_test.go @@ -0,0 +1,110 @@ +package stream + +import ( + "testing" + + "github.com/bluenviron/gortsplib/v5/pkg/description" + "github.com/bluenviron/gortsplib/v5/pkg/format" + "github.com/bluenviron/mediamtx/internal/unit" + "github.com/stretchr/testify/require" +) + +func TestStream(t *testing.T) { + desc := &description.Session{Medias: []*description.Media{ + { + Type: description.MediaTypeVideo, + Formats: []format.Format{&format.H264{}}, + }, + { + Type: description.MediaTypeVideo, + Formats: []format.Format{&format.VP8{}}, + }, + }} + + strm := &Stream{ + WriteQueueSize: 512, + RTPMaxPayloadSize: 1450, + Desc: desc, + GenerateRTPPackets: true, + } + err := strm.Initialize() + require.NoError(t, err) + defer strm.Close() + + r := &Reader{} + + recv := make(chan struct{}) + + r.OnData(desc.Medias[0], desc.Medias[0].Formats[0], func(_ unit.Unit) error { + close(recv) + return nil + }) + + strm.AddReader(r) + defer strm.RemoveReader(r) + + strm.WriteUnit(desc.Medias[0], desc.Medias[0].Formats[0], &unit.H264{ + Base: unit.Base{ + PTS: 30000 * 2, + }, + AU: [][]byte{ + {5, 2}, // IDR + }, + }) + + <-recv + + require.Equal(t, uint64(14), strm.BytesReceived()) + require.Equal(t, uint64(14), strm.BytesSent()) +} + +func TestStreamSkipBytesSent(t *testing.T) { + desc := &description.Session{Medias: []*description.Media{ + { + Type: description.MediaTypeVideo, + Formats: []format.Format{&format.H264{}}, + }, + { + Type: description.MediaTypeVideo, + Formats: []format.Format{&format.VP8{}}, + }, + }} + + strm := &Stream{ + WriteQueueSize: 512, + RTPMaxPayloadSize: 1450, + Desc: desc, + GenerateRTPPackets: true, + } + err := strm.Initialize() + require.NoError(t, err) + defer strm.Close() + + r := &Reader{ + SkipBytesSent: true, + } + + recv := make(chan struct{}) + + r.OnData(desc.Medias[0], desc.Medias[0].Formats[0], func(_ unit.Unit) error { + close(recv) + return nil + }) + + strm.AddReader(r) + defer strm.RemoveReader(r) + + strm.WriteUnit(desc.Medias[0], desc.Medias[0].Formats[0], &unit.H264{ + Base: unit.Base{ + PTS: 30000 * 2, + }, + AU: [][]byte{ + {5, 2}, // IDR + }, + }) + + <-recv + + require.Equal(t, uint64(14), strm.BytesReceived()) + require.Equal(t, uint64(0), strm.BytesSent()) +} diff --git a/internal/test/static_source_parent.go b/internal/test/static_source_parent.go index 4b1ef8ed..a4cd59cd 100644 --- a/internal/test/static_source_parent.go +++ b/internal/test/static_source_parent.go @@ -10,7 +10,7 @@ import ( // StaticSourceParent is a dummy static source parent. type StaticSourceParent struct { stream *stream.Stream - reader stream.Reader + reader *stream.Reader Unit chan unit.Unit } @@ -41,15 +41,18 @@ func (p *StaticSourceParent) SetReady(req defs.PathSourceStaticSetReadyReq) defs panic(err) } - p.reader = NilLogger + p.reader = &stream.Reader{Parent: 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.reader.OnData( + 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) + p.stream.AddReader(p.reader) return defs.PathSourceStaticSetReadyRes{Stream: p.stream} }