diff --git a/go.mod b/go.mod index 851d3009..8c720340 100644 --- a/go.mod +++ b/go.mod @@ -8,9 +8,10 @@ require ( github.com/MicahParks/keyfunc/v3 v3.3.3 github.com/abema/go-mp4 v1.2.0 github.com/alecthomas/kong v1.2.1 + github.com/asticode/go-astits v1.13.0 github.com/bluenviron/gohlslib v1.4.0 github.com/bluenviron/gortsplib/v4 v4.10.4 - github.com/bluenviron/mediacommon v1.12.3 + github.com/bluenviron/mediacommon v1.12.4-0.20240909171423-2f5c038aff08 github.com/datarhei/gosrt v0.7.0 github.com/fsnotify/fsnotify v1.7.0 github.com/gin-gonic/gin v1.10.0 @@ -36,7 +37,6 @@ require ( require ( github.com/asticode/go-astikit v0.30.0 // indirect - github.com/asticode/go-astits v1.13.0 // indirect github.com/benburkert/openpgp v0.0.0-20160410205803-c2471f86866c // indirect github.com/bytedance/sonic v1.11.6 // indirect github.com/bytedance/sonic/loader v0.1.1 // indirect diff --git a/go.sum b/go.sum index 85be0ca5..9c7d4a97 100644 --- a/go.sum +++ b/go.sum @@ -24,8 +24,8 @@ github.com/bluenviron/gohlslib v1.4.0 h1:3a9W1x8eqlxJUKt1sJCunPGtti5ALIY2ik4GU0R github.com/bluenviron/gohlslib v1.4.0/go.mod h1:q5ZElzNw5GRbV1VEI45qkcPbKBco6BP58QEY5HyFsmo= github.com/bluenviron/gortsplib/v4 v4.10.4 h1:7fDGKRfb7q3Foab6ctfU45BNd8q3G/ln2wwMlhcoyz8= github.com/bluenviron/gortsplib/v4 v4.10.4/go.mod h1:BsTItHGBtHOPmj3D6AygXaAMXx4+LQlJWNfDAI5PNkg= -github.com/bluenviron/mediacommon v1.12.3 h1:a7O1CzfdWsLJmTLe365KfRoAGeVxJPB/aDaCW4Jm2+U= -github.com/bluenviron/mediacommon v1.12.3/go.mod h1:HDyW2CzjvhYJXtdxstdFPio3G0qSocPhqkhUt/qffec= +github.com/bluenviron/mediacommon v1.12.4-0.20240909171423-2f5c038aff08 h1:8eI9UTEBmq5PSSwCO5fI78kujvGtLPsHApWG9MeDhgg= +github.com/bluenviron/mediacommon v1.12.4-0.20240909171423-2f5c038aff08/go.mod h1:HDyW2CzjvhYJXtdxstdFPio3G0qSocPhqkhUt/qffec= github.com/bytedance/sonic v1.11.6 h1:oUp34TzMlL+OY1OUWxHqsdkgC/Zfc85zGqw9siXjrc0= github.com/bytedance/sonic v1.11.6/go.mod h1:LysEHSvpvDySVdC2f87zGWf6CIKJcAvqab1ZaiQtds4= github.com/bytedance/sonic/loader v0.1.1 h1:c+e5Pt1k/cy5wMveRDyk2X4B9hF4g7an8N3zCYjJFNM= diff --git a/internal/protocols/hls/from_stream.go b/internal/protocols/hls/from_stream.go index f362563f..5db46c38 100644 --- a/internal/protocols/hls/from_stream.go +++ b/internal/protocols/hls/from_stream.go @@ -16,13 +16,13 @@ import ( // ErrNoSupportedCodecs is returned by FromStream when there are no supported codecs. var ErrNoSupportedCodecs = errors.New( - "the stream doesn't contain any supported codec, which are currently H265, H264, Opus, MPEG-4 Audio") + "the stream doesn't contain any supported codec, which are currently AV1, VP9, H265, H264, Opus, MPEG-4 Audio") func setupVideoTrack( stream *stream.Stream, writer *asyncwriter.Writer, muxer *gohlslib.Muxer, -) *gohlslib.Track { +) format.Format { var videoFormatAV1 *format.AV1 videoMedia := stream.Desc().FindFormat(&videoFormatAV1) @@ -42,9 +42,10 @@ func setupVideoTrack( return nil }) - return &gohlslib.Track{ + muxer.VideoTrack = &gohlslib.Track{ Codec: &codecs.AV1{}, } + return videoFormatAV1 } var videoFormatVP9 *format.VP9 @@ -66,9 +67,10 @@ func setupVideoTrack( return nil }) - return &gohlslib.Track{ + muxer.VideoTrack = &gohlslib.Track{ Codec: &codecs.VP9{}, } + return videoFormatVP9 } var videoFormatH265 *format.H265 @@ -92,13 +94,14 @@ func setupVideoTrack( vps, sps, pps := videoFormatH265.SafeParams() - return &gohlslib.Track{ + muxer.VideoTrack = &gohlslib.Track{ Codec: &codecs.H265{ VPS: vps, SPS: sps, PPS: pps, }, } + return videoFormatH265 } var videoFormatH264 *format.H264 @@ -122,12 +125,13 @@ func setupVideoTrack( sps, pps := videoFormatH264.SafeParams() - return &gohlslib.Track{ + muxer.VideoTrack = &gohlslib.Track{ Codec: &codecs.H264{ SPS: sps, PPS: pps, }, } + return videoFormatH264 } return nil @@ -138,11 +142,11 @@ func setupAudioTrack( writer *asyncwriter.Writer, muxer *gohlslib.Muxer, l logger.Writer, -) *gohlslib.Track { +) format.Format { var audioFormatOpus *format.Opus audioMedia := stream.Desc().FindFormat(&audioFormatOpus) - if audioMedia != nil { + if audioFormatOpus != nil { stream.AddReader(writer, audioMedia, audioFormatOpus, func(u unit.Unit) error { tunit := u.(*unit.Opus) @@ -157,17 +161,18 @@ func setupAudioTrack( return nil }) - return &gohlslib.Track{ + muxer.AudioTrack = &gohlslib.Track{ Codec: &codecs.Opus{ ChannelCount: audioFormatOpus.ChannelCount, }, } + return audioFormatOpus } var audioFormatMPEG4Audio *format.MPEG4Audio audioMedia = stream.Desc().FindFormat(&audioFormatMPEG4Audio) - if audioMedia != nil { + if audioFormatMPEG4Audio != nil { co := audioFormatMPEG4Audio.GetConfig() if co == nil { l.Log(logger.Warn, "skipping MPEG-4 audio track: tracks without explicit configuration are not supported") @@ -190,11 +195,12 @@ func setupAudioTrack( return nil }) - return &gohlslib.Track{ + muxer.AudioTrack = &gohlslib.Track{ Codec: &codecs.MPEG4Audio{ Config: *co, }, } + return audioFormatMPEG4Audio } } @@ -208,25 +214,30 @@ func FromStream( muxer *gohlslib.Muxer, l logger.Writer, ) error { - videoTrack := setupVideoTrack( + videoFormat := setupVideoTrack( stream, writer, muxer, ) - audioTrack := setupAudioTrack( + audioFormat := setupAudioTrack( stream, writer, muxer, l, ) - if videoTrack == nil && audioTrack == nil { + if videoFormat == nil && audioFormat == nil { return ErrNoSupportedCodecs } - muxer.VideoTrack = videoTrack - muxer.AudioTrack = audioTrack + for _, media := range stream.Desc().Medias { + for _, forma := range media.Formats { + if forma != videoFormat && forma != audioFormat { + l.Log(logger.Warn, "skipping track with codec %s", forma.Codec()) + } + } + } return nil } diff --git a/internal/protocols/hls/from_stream_test.go b/internal/protocols/hls/from_stream_test.go new file mode 100644 index 00000000..7cc5e325 --- /dev/null +++ b/internal/protocols/hls/from_stream_test.go @@ -0,0 +1,81 @@ +package hls + +import ( + "fmt" + "testing" + + "github.com/bluenviron/gohlslib" + "github.com/bluenviron/gortsplib/v4/pkg/description" + "github.com/bluenviron/gortsplib/v4/pkg/format" + "github.com/bluenviron/mediamtx/internal/asyncwriter" + "github.com/bluenviron/mediamtx/internal/logger" + "github.com/bluenviron/mediamtx/internal/stream" + "github.com/bluenviron/mediamtx/internal/test" + "github.com/stretchr/testify/require" +) + +func TestFromStreamNoSupportedCodecs(t *testing.T) { + stream, err := stream.New( + 1460, + &description.Session{Medias: []*description.Media{{ + Type: description.MediaTypeVideo, + Formats: []format.Format{&format.VP8{}}, + }}}, + true, + test.NilLogger, + ) + require.NoError(t, err) + + writer := asyncwriter.New(0, nil) + + l := test.Logger(func(logger.Level, string, ...interface{}) { + t.Error("should not happen") + }) + + err = FromStream(stream, writer, nil, l) + require.Equal(t, ErrNoSupportedCodecs, err) +} + +func TestFromStreamSkipUnsupportedTracks(t *testing.T) { + stream, err := stream.New( + 1460, + &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{}}, + }, + }}, + true, + test.NilLogger, + ) + require.NoError(t, err) + + writer := asyncwriter.New(0, nil) + + 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 with codec VP8", fmt.Sprintf(format, args...)) + case 1: + require.Equal(t, "skipping track with codec MPEG-1/2 Audio", fmt.Sprintf(format, args...)) + } + n++ + }) + + err = FromStream(stream, writer, m, l) + require.NoError(t, err) + require.Equal(t, 2, n) +} diff --git a/internal/protocols/hls/to_stream.go b/internal/protocols/hls/to_stream.go index 4e993454..703c6e01 100644 --- a/internal/protocols/hls/to_stream.go +++ b/internal/protocols/hls/to_stream.go @@ -1,7 +1,6 @@ package hls import ( - "fmt" "time" "github.com/bluenviron/gohlslib" @@ -144,11 +143,15 @@ func ToStream( }) default: - return nil, fmt.Errorf("unsupported track: %T", track.Codec) + panic("should not happen") } medias = append(medias, medi) } + if len(medias) == 0 { + return nil, ErrNoSupportedCodecs + } + return medias, nil } diff --git a/internal/protocols/hls/to_stream_test.go b/internal/protocols/hls/to_stream_test.go new file mode 100644 index 00000000..bea6daae --- /dev/null +++ b/internal/protocols/hls/to_stream_test.go @@ -0,0 +1,16 @@ +package hls + +import ( + "testing" + + "github.com/bluenviron/gohlslib" + "github.com/stretchr/testify/require" +) + +func TestToStreamNoSupportedCodecs(t *testing.T) { + _, err := ToStream(nil, []*gohlslib.Track{}, nil) + require.Equal(t, ErrNoSupportedCodecs, err) +} + +// this is impossible to test since currently we support all gohlslib.Tracks. +// func TestToStreamSkipUnsupportedTracks(t *testing.T) diff --git a/internal/protocols/mpegts/from_stream.go b/internal/protocols/mpegts/from_stream.go index 131022d0..0d686a0a 100644 --- a/internal/protocols/mpegts/from_stream.go +++ b/internal/protocols/mpegts/from_stream.go @@ -13,6 +13,7 @@ import ( srt "github.com/datarhei/gosrt" "github.com/bluenviron/mediamtx/internal/asyncwriter" + "github.com/bluenviron/mediamtx/internal/logger" "github.com/bluenviron/mediamtx/internal/stream" "github.com/bluenviron/mediamtx/internal/unit" ) @@ -28,9 +29,11 @@ func FromStream( bw *bufio.Writer, sconn srt.Conn, writeTimeout time.Duration, + l logger.Writer, ) error { var w *mcmpegts.Writer var tracks []*mcmpegts.Track + var skippedFormats []format.Format addTrack := func(codec mcmpegts.Codec) *mcmpegts.Track { track := &mcmpegts.Track{ @@ -246,12 +249,19 @@ func FromStream( } return bw.Flush() }) + + default: + skippedFormats = append(skippedFormats, forma) } } } if len(tracks) == 0 { - return ErrNoTracks + return errNoSupportedCodecs + } + + for _, forma := range skippedFormats { + l.Log(logger.Warn, "skipping track with codec %s", forma.Codec()) } w = mcmpegts.NewWriter(bw, tracks) diff --git a/internal/protocols/mpegts/from_stream_test.go b/internal/protocols/mpegts/from_stream_test.go new file mode 100644 index 00000000..a4780335 --- /dev/null +++ b/internal/protocols/mpegts/from_stream_test.go @@ -0,0 +1,71 @@ +package mpegts + +import ( + "fmt" + "testing" + + "github.com/bluenviron/gortsplib/v4/pkg/description" + "github.com/bluenviron/gortsplib/v4/pkg/format" + "github.com/bluenviron/mediamtx/internal/asyncwriter" + "github.com/bluenviron/mediamtx/internal/logger" + "github.com/bluenviron/mediamtx/internal/stream" + "github.com/bluenviron/mediamtx/internal/test" + "github.com/stretchr/testify/require" +) + +func TestFromStreamNoSupportedCodecs(t *testing.T) { + stream, err := stream.New( + 1460, + &description.Session{Medias: []*description.Media{{ + Type: description.MediaTypeVideo, + Formats: []format.Format{&format.VP8{}}, + }}}, + true, + test.NilLogger, + ) + require.NoError(t, err) + + writer := asyncwriter.New(0, nil) + + l := test.Logger(func(logger.Level, string, ...interface{}) { + t.Error("should not happen") + }) + + err = FromStream(stream, writer, nil, nil, 0, l) + require.Equal(t, errNoSupportedCodecs, err) +} + +func TestFromStreamSkipUnsupportedTracks(t *testing.T) { + stream, err := stream.New( + 1460, + &description.Session{Medias: []*description.Media{ + { + Type: description.MediaTypeVideo, + Formats: []format.Format{&format.H265{}}, + }, + { + Type: description.MediaTypeVideo, + Formats: []format.Format{&format.VP8{}}, + }, + }}, + true, + test.NilLogger, + ) + require.NoError(t, err) + + writer := asyncwriter.New(0, nil) + + 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 with codec VP8", fmt.Sprintf(format, args...)) + } + n++ + }) + + err = FromStream(stream, writer, nil, nil, 0, l) + require.NoError(t, err) + require.Equal(t, 1, n) +} diff --git a/internal/protocols/mpegts/to_stream.go b/internal/protocols/mpegts/to_stream.go index 925b1616..aae9b32d 100644 --- a/internal/protocols/mpegts/to_stream.go +++ b/internal/protocols/mpegts/to_stream.go @@ -9,17 +9,23 @@ import ( "github.com/bluenviron/gortsplib/v4/pkg/format" "github.com/bluenviron/mediacommon/pkg/formats/mpegts" + "github.com/bluenviron/mediamtx/internal/logger" "github.com/bluenviron/mediamtx/internal/stream" "github.com/bluenviron/mediamtx/internal/unit" ) -// ErrNoTracks is returned when there are no supported tracks. -var ErrNoTracks = errors.New("no supported tracks found (supported are H265, H264," + - " MPEG-4 Video, MPEG-1/2 Video, Opus, MPEG-4 Audio, MPEG-1 Audio, AC-3") +var errNoSupportedCodecs = errors.New( + "the stream doesn't contain any supported codec, which are currently " + + "H265, H264, MPEG-4 Video, MPEG-1/2 Video, Opus, MPEG-4 Audio, MPEG-1 Audio, AC-3") // ToStream maps a MPEG-TS stream to a MediaMTX stream. -func ToStream(r *mpegts.Reader, stream **stream.Stream) ([]*description.Media, error) { +func ToStream( + r *mpegts.Reader, + stream **stream.Stream, + l logger.Writer, +) ([]*description.Media, error) { var medias []*description.Media //nolint:prealloc + var unsupportedTracks []int var td *mpegts.TimeDecoder decodeTime := func(t int64) time.Duration { @@ -29,7 +35,7 @@ func ToStream(r *mpegts.Reader, stream **stream.Stream) ([]*description.Media, e return td.Decode(t) } - for _, track := range r.Tracks() { //nolint:dupl + for i, track := range r.Tracks() { //nolint:dupl var medi *description.Media switch codec := track.Codec.(type) { @@ -190,6 +196,7 @@ func ToStream(r *mpegts.Reader, stream **stream.Stream) ([]*description.Media, e }) default: + unsupportedTracks = append(unsupportedTracks, i+1) continue } @@ -197,7 +204,11 @@ func ToStream(r *mpegts.Reader, stream **stream.Stream) ([]*description.Media, e } if len(medias) == 0 { - return nil, ErrNoTracks + return nil, errNoSupportedCodecs + } + + for _, id := range unsupportedTracks { + l.Log(logger.Warn, "skipping track %d due to unsupported codec", id) } return medias, nil diff --git a/internal/protocols/mpegts/to_stream_test.go b/internal/protocols/mpegts/to_stream_test.go new file mode 100644 index 00000000..9a097829 --- /dev/null +++ b/internal/protocols/mpegts/to_stream_test.go @@ -0,0 +1,77 @@ +package mpegts + +import ( + "bytes" + "context" + "fmt" + "testing" + + "github.com/asticode/go-astits" + "github.com/bluenviron/mediacommon/pkg/formats/mpegts" + "github.com/bluenviron/mediamtx/internal/logger" + "github.com/bluenviron/mediamtx/internal/test" + "github.com/stretchr/testify/require" +) + +func TestToStreamNoSupportedCodecs(t *testing.T) { + var buf bytes.Buffer + mux := astits.NewMuxer(context.Background(), &buf) + + err := mux.AddElementaryStream(astits.PMTElementaryStream{ + ElementaryPID: 122, + StreamType: astits.StreamTypeDTSAudio, + }) + require.NoError(t, err) + + mux.SetPCRPID(122) + + _, err = mux.WriteTables() + require.NoError(t, err) + + r, err := mpegts.NewReader(&buf) + require.NoError(t, err) + + l := test.Logger(func(logger.Level, string, ...interface{}) { + t.Error("should not happen") + }) + _, err = ToStream(r, nil, l) + require.Equal(t, errNoSupportedCodecs, err) +} + +func TestToStreamSkipUnsupportedTracks(t *testing.T) { + var buf bytes.Buffer + mux := astits.NewMuxer(context.Background(), &buf) + + err := mux.AddElementaryStream(astits.PMTElementaryStream{ + ElementaryPID: 122, + StreamType: astits.StreamTypeDTSAudio, + }) + require.NoError(t, err) + + err = mux.AddElementaryStream(astits.PMTElementaryStream{ + ElementaryPID: 123, + StreamType: astits.StreamTypeH264Video, + }) + require.NoError(t, err) + + mux.SetPCRPID(122) + + _, err = mux.WriteTables() + require.NoError(t, err) + + r, err := mpegts.NewReader(&buf) + require.NoError(t, err) + + 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 due to unsupported codec", fmt.Sprintf(format, args...)) + } + n++ + }) + + _, err = ToStream(r, nil, l) + require.NoError(t, err) +} diff --git a/internal/protocols/rtmp/from_stream.go b/internal/protocols/rtmp/from_stream.go index dcead0c3..4207cb1e 100644 --- a/internal/protocols/rtmp/from_stream.go +++ b/internal/protocols/rtmp/from_stream.go @@ -11,11 +11,12 @@ import ( "github.com/bluenviron/mediacommon/pkg/codecs/mpeg1audio" "github.com/bluenviron/mediacommon/pkg/codecs/mpeg4audio" "github.com/bluenviron/mediamtx/internal/asyncwriter" + "github.com/bluenviron/mediamtx/internal/logger" "github.com/bluenviron/mediamtx/internal/stream" "github.com/bluenviron/mediamtx/internal/unit" ) -var errNoSupportedCodecs = errors.New( +var errNoSupportedCodecsFrom = errors.New( "the stream doesn't contain any supported codec, which are currently H264, MPEG-4 Audio, MPEG-1/2 Audio") func setupVideo( @@ -171,6 +172,7 @@ func FromStream( conn *Conn, nconn net.Conn, writeTimeout time.Duration, + l logger.Writer, ) error { var w *Writer @@ -191,14 +193,25 @@ func FromStream( ) if videoFormat == nil && audioFormat == nil { - return errNoSupportedCodecs + return errNoSupportedCodecsFrom } + fmt.Println(videoFormat, audioFormat) + fmt.Println(videoFormat == nil, audioFormat == nil) + var err error w, err = NewWriter(conn, videoFormat, audioFormat) if err != nil { return err } + for _, media := range stream.Desc().Medias { + for _, forma := range media.Formats { + if forma != videoFormat && forma != audioFormat { + l.Log(logger.Warn, "skipping track with codec %s", forma.Codec()) + } + } + } + return nil } diff --git a/internal/protocols/rtmp/from_stream_test.go b/internal/protocols/rtmp/from_stream_test.go new file mode 100644 index 00000000..b25f917f --- /dev/null +++ b/internal/protocols/rtmp/from_stream_test.go @@ -0,0 +1,85 @@ +package rtmp + +import ( + "bytes" + "fmt" + "testing" + + "github.com/bluenviron/gortsplib/v4/pkg/description" + "github.com/bluenviron/gortsplib/v4/pkg/format" + "github.com/bluenviron/mediamtx/internal/asyncwriter" + "github.com/bluenviron/mediamtx/internal/logger" + "github.com/bluenviron/mediamtx/internal/protocols/rtmp/bytecounter" + "github.com/bluenviron/mediamtx/internal/protocols/rtmp/message" + "github.com/bluenviron/mediamtx/internal/stream" + "github.com/bluenviron/mediamtx/internal/test" + "github.com/stretchr/testify/require" +) + +func TestFromStreamNoSupportedCodecs(t *testing.T) { + stream, err := stream.New( + 1460, + &description.Session{Medias: []*description.Media{{ + Type: description.MediaTypeVideo, + Formats: []format.Format{&format.VP8{}}, + }}}, + true, + test.NilLogger, + ) + require.NoError(t, err) + + writer := asyncwriter.New(0, nil) + + l := test.Logger(func(logger.Level, string, ...interface{}) { + t.Error("should not happen") + }) + + err = FromStream(stream, writer, nil, nil, 0, l) + require.Equal(t, errNoSupportedCodecsFrom, err) +} + +func TestFromStreamSkipUnsupportedTracks(t *testing.T) { + stream, err := stream.New( + 1460, + &description.Session{Medias: []*description.Media{ + { + Type: description.MediaTypeVideo, + Formats: []format.Format{&format.VP8{}}, + }, + { + Type: description.MediaTypeVideo, + Formats: []format.Format{&format.H264{}}, + }, + { + Type: description.MediaTypeVideo, + Formats: []format.Format{&format.H264{}}, + }, + }}, + true, + test.NilLogger, + ) + require.NoError(t, err) + + writer := asyncwriter.New(0, nil) + + 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 with codec VP8", fmt.Sprintf(format, args...)) + case 1: + require.Equal(t, "skipping track with codec H264", fmt.Sprintf(format, args...)) + } + n++ + }) + + var buf bytes.Buffer + bc := bytecounter.NewReadWriter(&buf) + conn := &Conn{mrw: message.NewReadWriter(&buf, bc, false)} + + err = FromStream(stream, writer, conn, nil, 0, l) + require.NoError(t, err) + require.Equal(t, 2, n) +} diff --git a/internal/protocols/rtmp/to_stream.go b/internal/protocols/rtmp/to_stream.go index e0547597..1853f9ca 100644 --- a/internal/protocols/rtmp/to_stream.go +++ b/internal/protocols/rtmp/to_stream.go @@ -1,7 +1,7 @@ package rtmp import ( - "fmt" + "errors" "time" "github.com/bluenviron/gortsplib/v4/pkg/description" @@ -10,6 +10,10 @@ import ( "github.com/bluenviron/mediamtx/internal/unit" ) +var errNoSupportedCodecsTo = errors.New( + "the stream doesn't contain any supported codec, which are currently " + + "AV1, VP9, H265, H264, MPEG-4 Audio, MPEG-1/2 Audio, G711, LPCM") + // ToStream maps a RTMP stream to a MediaMTX stream. func ToStream(r *Reader, stream **stream.Stream) ([]*description.Media, error) { videoFormat, audioFormat := r.Tracks() @@ -69,7 +73,7 @@ func ToStream(r *Reader, stream **stream.Stream) ([]*description.Media, error) { }) default: - return nil, fmt.Errorf("unsupported video codec: %T", videoFormat) + panic("should not happen") } } @@ -126,9 +130,13 @@ func ToStream(r *Reader, stream **stream.Stream) ([]*description.Media, error) { }) default: - return nil, fmt.Errorf("unsupported audio codec: %T", audioFormat) + panic("should not happen") } } + if len(medias) == 0 { + return nil, errNoSupportedCodecsTo + } + return medias, nil } diff --git a/internal/protocols/rtmp/to_stream_test.go b/internal/protocols/rtmp/to_stream_test.go new file mode 100644 index 00000000..d52e3f24 --- /dev/null +++ b/internal/protocols/rtmp/to_stream_test.go @@ -0,0 +1,17 @@ +package rtmp + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestToStreamNoSupportedCodecs(t *testing.T) { + r := &Reader{} + + _, err := ToStream(r, nil) + require.Equal(t, errNoSupportedCodecsTo, err) +} + +// this is impossible to test since currently we support all RTMP tracks. +// func TestToStreamSkipUnsupportedTracks(t *testing.T) diff --git a/internal/protocols/webrtc/from_stream.go b/internal/protocols/webrtc/from_stream.go index 0b87d099..b1a7a981 100644 --- a/internal/protocols/webrtc/from_stream.go +++ b/internal/protocols/webrtc/from_stream.go @@ -14,6 +14,7 @@ import ( "github.com/bluenviron/gortsplib/v4/pkg/format/rtpvp9" "github.com/bluenviron/mediacommon/pkg/codecs/g711" "github.com/bluenviron/mediamtx/internal/asyncwriter" + "github.com/bluenviron/mediamtx/internal/logger" "github.com/bluenviron/mediamtx/internal/stream" "github.com/bluenviron/mediamtx/internal/unit" "github.com/pion/webrtc/v3" @@ -23,7 +24,7 @@ const ( webrtcPayloadMaxSize = 1188 // 1200 - 12 (RTP header) ) -var errNoSupportedCodecs = errors.New( +var errNoSupportedCodecsFrom = errors.New( "the stream doesn't contain any supported codec, which are currently AV1, VP9, VP8, H264, Opus, G722, G711, LPCM") func uint16Ptr(v uint16) *uint16 { @@ -520,6 +521,7 @@ func FromStream( stream *stream.Stream, writer *asyncwriter.Writer, pc *PeerConnection, + l logger.Writer, ) error { videoFormat, err := setupVideoTrack(stream, writer, pc) if err != nil { @@ -532,7 +534,15 @@ func FromStream( } if videoFormat == nil && audioFormat == nil { - return errNoSupportedCodecs + return errNoSupportedCodecsFrom + } + + for _, media := range stream.Desc().Medias { + for _, forma := range media.Formats { + if forma != videoFormat && forma != audioFormat { + l.Log(logger.Warn, "skipping track with codec %s", forma.Codec()) + } + } } return nil diff --git a/internal/protocols/webrtc/from_stream_test.go b/internal/protocols/webrtc/from_stream_test.go index 50a16024..ed28f478 100644 --- a/internal/protocols/webrtc/from_stream_test.go +++ b/internal/protocols/webrtc/from_stream_test.go @@ -1,31 +1,90 @@ package webrtc import ( + "fmt" "testing" "github.com/bluenviron/gortsplib/v4/pkg/description" "github.com/bluenviron/gortsplib/v4/pkg/format" "github.com/bluenviron/mediamtx/internal/asyncwriter" + "github.com/bluenviron/mediamtx/internal/logger" "github.com/bluenviron/mediamtx/internal/stream" "github.com/bluenviron/mediamtx/internal/test" "github.com/stretchr/testify/require" ) +func TestFromStreamNoSupportedCodecs(t *testing.T) { + stream, err := stream.New( + 1460, + &description.Session{Medias: []*description.Media{{ + Type: description.MediaTypeVideo, + Formats: []format.Format{&format.H265{}}, + }}}, + true, + test.NilLogger, + ) + require.NoError(t, err) + + writer := asyncwriter.New(0, nil) + + l := test.Logger(func(logger.Level, string, ...interface{}) { + t.Error("should not happen") + }) + + err = FromStream(stream, writer, nil, l) + require.Equal(t, errNoSupportedCodecsFrom, err) +} + +func TestFromStreamSkipUnsupportedTracks(t *testing.T) { + stream, err := stream.New( + 1460, + &description.Session{Medias: []*description.Media{ + { + Type: description.MediaTypeVideo, + Formats: []format.Format{&format.H264{}}, + }, + { + Type: description.MediaTypeVideo, + Formats: []format.Format{&format.H265{}}, + }, + }}, + true, + test.NilLogger, + ) + require.NoError(t, err) + + writer := asyncwriter.New(0, nil) + + 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 with codec H265", fmt.Sprintf(format, args...)) + } + n++ + }) + + pc := &PeerConnection{} + + err = FromStream(stream, writer, pc, l) + require.NoError(t, err) + require.Equal(t, 1, n) +} + func TestFromStream(t *testing.T) { for _, ca := range toFromStreamCases { if ca.in == nil { continue } t.Run(ca.name, func(t *testing.T) { - desc := &description.Session{ - Medias: []*description.Media{{ - Formats: []format.Format{ca.in}, - }}, - } - stream, err := stream.New( 1460, - desc, + &description.Session{ + Medias: []*description.Media{{ + Formats: []format.Format{ca.in}, + }}, + }, false, test.NilLogger, ) @@ -36,7 +95,7 @@ func TestFromStream(t *testing.T) { pc := &PeerConnection{} - err = FromStream(stream, writer, pc) + err = FromStream(stream, writer, pc, nil) require.NoError(t, err) require.Equal(t, ca.webrtcCaps, pc.OutgoingTracks[0].Caps) diff --git a/internal/protocols/webrtc/to_stream.go b/internal/protocols/webrtc/to_stream.go index d4ba2c7b..43d05b3a 100644 --- a/internal/protocols/webrtc/to_stream.go +++ b/internal/protocols/webrtc/to_stream.go @@ -1,6 +1,7 @@ package webrtc import ( + "errors" "fmt" "strings" "time" @@ -13,8 +14,15 @@ import ( "github.com/pion/webrtc/v3" ) +var errNoSupportedCodecsTo = errors.New( + "the stream doesn't contain any supported codec, which are currently " + + "AV1, VP9, VP8, H265, H264, Opus, G722, G711, LPCM") + // ToStream maps a WebRTC connection to a MediaMTX stream. -func ToStream(pc *PeerConnection, stream **stream.Stream) ([]*description.Media, error) { +func ToStream( + pc *PeerConnection, + stream **stream.Stream, +) ([]*description.Media, error) { var medias []*description.Media //nolint:prealloc timeDecoder := rtptime.NewGlobalDecoder() @@ -145,5 +153,9 @@ func ToStream(pc *PeerConnection, stream **stream.Stream) ([]*description.Media, medias = append(medias, medi) } + if len(medias) == 0 { + return nil, errNoSupportedCodecsTo + } + return medias, nil } diff --git a/internal/protocols/webrtc/to_stream_test.go b/internal/protocols/webrtc/to_stream_test.go index c01d6030..94a63a94 100644 --- a/internal/protocols/webrtc/to_stream_test.go +++ b/internal/protocols/webrtc/to_stream_test.go @@ -14,6 +14,16 @@ import ( "github.com/stretchr/testify/require" ) +func TestToStreamNoSupportedCodecs(t *testing.T) { + pc := &PeerConnection{} + _, err := ToStream(pc, nil) + require.Equal(t, errNoSupportedCodecsTo, err) +} + +// this is impossible to test since unsupported tracks cause an error +// as they are not included inside incomingVideoCodecs or incomingAudioCodecs +// func TestToStreamSkipUnsupportedTracks(t *testing.T) + var toFromStreamCases = []struct { name string in format.Format @@ -393,10 +403,6 @@ func TestToStream(t *testing.T) { _, err = pc2.GatherIncomingTracks(context.Background()) require.NoError(t, err) - /*exp := ca.webrtcOut - exp.RTCPFeedback = inc[0].track.Codec().RTPCodecCapability.RTCPFeedback - require.Equal(t, exp, inc[0].track.Codec().RTPCodecCapability)*/ - var stream *stream.Stream medias, err := ToStream(pc2, &stream) require.NoError(t, err) diff --git a/internal/recorder/format_fmp4.go b/internal/recorder/format_fmp4.go index 156aea32..b30cbf97 100644 --- a/internal/recorder/format_fmp4.go +++ b/internal/recorder/format_fmp4.go @@ -111,6 +111,7 @@ type formatFMP4 struct { func (f *formatFMP4) initialize() { nextID := 1 var formats []rtspformat.Format + var skippedFormats []rtspformat.Format addTrack := func(format rtspformat.Format, codec fmp4.Codec) *formatFMP4Track { initTrack := &fmp4.InitTrack{ @@ -803,10 +804,22 @@ func (f *formatFMP4) initialize() { ntp: tunit.NTP, }) }) + + default: + skippedFormats = append(skippedFormats, forma) } } } + if len(formats) == 0 { + f.ai.Log(logger.Warn, "no supported tracks found, skipping recording") + return + } + + for _, forma := range skippedFormats { + f.ai.Log(logger.Warn, "skipping track with codec %s", forma.Codec()) + } + f.ai.Log(logger.Info, "recording %s", defs.FormatsInfo(formats)) } diff --git a/internal/recorder/format_mpegts.go b/internal/recorder/format_mpegts.go index c70764e0..d55e5c44 100644 --- a/internal/recorder/format_mpegts.go +++ b/internal/recorder/format_mpegts.go @@ -52,6 +52,7 @@ type formatMPEGTS struct { func (f *formatMPEGTS) initialize() { var tracks []*mpegts.Track var formats []rtspformat.Format + var skippedFormats []rtspformat.Format addTrack := func(format rtspformat.Format, codec mpegts.Codec) *mpegts.Track { track := &mpegts.Track{ @@ -302,10 +303,22 @@ func (f *formatMPEGTS) initialize() { }, ) }) + + default: + skippedFormats = append(skippedFormats, forma) } } } + if len(tracks) == 0 { + f.ai.Log(logger.Warn, "no supported tracks found, skipping recording") + return + } + + for _, forma := range skippedFormats { + f.ai.Log(logger.Warn, "skipping track with codec %s", forma.Codec()) + } + f.dw = &dynamicWriter{} f.bw = bufio.NewWriterSize(f.dw, mpegtsMaxBufferSize) f.mw = mpegts.NewWriter(f.bw, tracks) diff --git a/internal/servers/rtmp/conn.go b/internal/servers/rtmp/conn.go index 6bf2553d..16c33c44 100644 --- a/internal/servers/rtmp/conn.go +++ b/internal/servers/rtmp/conn.go @@ -190,7 +190,7 @@ func (c *conn) runRead(conn *rtmp.Conn, u *url.URL) error { writer := asyncwriter.New(c.writeQueueSize, c) defer stream.RemoveReader(writer) - err = rtmp.FromStream(stream, writer, conn, c.nconn, time.Duration(c.writeTimeout)) + err = rtmp.FromStream(stream, writer, conn, c.nconn, time.Duration(c.writeTimeout), c) if err != nil { return err } diff --git a/internal/servers/srt/conn.go b/internal/servers/srt/conn.go index 6d41bc32..1beb849c 100644 --- a/internal/servers/srt/conn.go +++ b/internal/servers/srt/conn.go @@ -216,7 +216,7 @@ func (c *conn) runPublishReader(sconn srt.Conn, path defs.Path) error { var stream *stream.Stream - medias, err := mpegts.ToStream(r, &stream) + medias, err := mpegts.ToStream(r, &stream, c) if err != nil { return err } @@ -289,7 +289,7 @@ func (c *conn) runRead(streamID *streamID) error { bw := bufio.NewWriterSize(sconn, srtMaxPayloadSize(c.udpMaxPayloadSize)) - err = mpegts.FromStream(stream, writer, bw, sconn, time.Duration(c.writeTimeout)) + err = mpegts.FromStream(stream, writer, bw, sconn, time.Duration(c.writeTimeout), c) if err != nil { return err } diff --git a/internal/servers/webrtc/session.go b/internal/servers/webrtc/session.go index eafc4358..1b99e0af 100644 --- a/internal/servers/webrtc/session.go +++ b/internal/servers/webrtc/session.go @@ -299,7 +299,7 @@ func (s *session) runRead() (int, error) { Log: s, } - err = webrtc.FromStream(stream, writer, pc) + err = webrtc.FromStream(stream, writer, pc, s) if err != nil { return http.StatusBadRequest, err } diff --git a/internal/staticsources/rtmp/source.go b/internal/staticsources/rtmp/source.go index 7aa14d58..900308b4 100644 --- a/internal/staticsources/rtmp/source.go +++ b/internal/staticsources/rtmp/source.go @@ -4,6 +4,7 @@ package rtmp import ( "context" ctls "crypto/tls" + "fmt" "net" "net/url" "time" @@ -106,6 +107,10 @@ func (s *Source) runReader(u *url.URL, nconn net.Conn) error { return err } + if len(medias) == 0 { + return fmt.Errorf("no supported tracks found") + } + res := s.Parent.SetReady(defs.PathSourceStaticSetReadyReq{ Desc: &description.Session{Medias: medias}, GenerateRTPPackets: true, diff --git a/internal/staticsources/srt/source.go b/internal/staticsources/srt/source.go index 69e65c3f..1c6c6faf 100644 --- a/internal/staticsources/srt/source.go +++ b/internal/staticsources/srt/source.go @@ -82,7 +82,7 @@ func (s *Source) runReader(sconn srt.Conn) error { var stream *stream.Stream - medias, err := mpegts.ToStream(r, &stream) + medias, err := mpegts.ToStream(r, &stream, s) if err != nil { return err } diff --git a/internal/staticsources/udp/source.go b/internal/staticsources/udp/source.go index 0d899f32..2a7aa323 100644 --- a/internal/staticsources/udp/source.go +++ b/internal/staticsources/udp/source.go @@ -119,7 +119,7 @@ func (s *Source) runReader(pc net.PacketConn) error { var stream *stream.Stream - medias, err := mpegts.ToStream(r, &stream) + medias, err := mpegts.ToStream(r, &stream, s) if err != nil { return err } diff --git a/internal/test/logger.go b/internal/test/logger.go index 4331bd19..39df66a8 100644 --- a/internal/test/logger.go +++ b/internal/test/logger.go @@ -9,3 +9,11 @@ func (nilLogger) Log(_ logger.Level, _ string, _ ...interface{}) { // NilLogger is a logger to /dev/null var NilLogger logger.Writer = &nilLogger{} + +// Logger is a test logger. +type Logger func(logger.Level, string, ...interface{}) + +// Log implements logger.Writer. +func (l Logger) Log(level logger.Level, format string, args ...interface{}) { + l(level, format, args...) +}