OPC/mediamtx
2
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases Wiki Activity

warn users about skipped tracks when reading or publishing (#3753)

Browse Source
This commit is contained in:
Alessandro Ros
2024-09-15 19:28:05 +02:00
committed by GitHub
parent 92a5aa5057
commit 471019f606
27 changed files with 584 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+2 -2
View File
@@ -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
go.sum
+2 -2
View File
@@ -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=
internal/protocols/hls/from_stream.go
+27 -16
View File
@@ -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
}
internal/protocols/hls/from_stream_test.go
+81
View File
@@ -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)
}
internal/protocols/hls/to_stream.go
+5 -2
View File
@@ -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
}
internal/protocols/hls/to_stream_test.go
+16
View File
@@ -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)
internal/protocols/mpegts/from_stream.go
+11 -1
View File
@@ -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)
internal/protocols/mpegts/from_stream_test.go
+71
View File
@@ -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)
}
internal/protocols/mpegts/to_stream.go
+17 -6
View File
@@ -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
internal/protocols/mpegts/to_stream_test.go
+77
View File
@@ -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)
}
internal/protocols/rtmp/from_stream.go
+15 -2
View File
@@ -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
}
internal/protocols/rtmp/from_stream_test.go
+85
View File
@@ -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)
}
internal/protocols/rtmp/to_stream.go
+11 -3
View File
@@ -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
}
internal/protocols/rtmp/to_stream_test.go
+17
View File
@@ -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)
internal/protocols/webrtc/from_stream.go
+12 -2
View File
@@ -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
internal/protocols/webrtc/from_stream_test.go
+67 -8
View File
@@ -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)
internal/protocols/webrtc/to_stream.go
+13 -1
View File
@@ -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
}
internal/protocols/webrtc/to_stream_test.go
+10 -4
View File
@@ -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)
internal/recorder/format_fmp4.go
+13
View File
@@ -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))
}
internal/recorder/format_mpegts.go
+13
View File
@@ -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)
internal/servers/rtmp/conn.go
+1 -1
View File
@@ -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
}
internal/servers/srt/conn.go
+2 -2
View File
@@ -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
}
internal/servers/webrtc/session.go
+1 -1
View File
@@ -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
}
internal/staticsources/rtmp/source.go
+5
View File
@@ -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,
internal/staticsources/srt/source.go
+1 -1
View File
@@ -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
}
internal/staticsources/udp/source.go
+1 -1
View File
@@ -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
}
internal/test/logger.go
+8
View File
@@ -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...)
}