diff --git a/docs/2-usage/3-read.md b/docs/2-usage/3-read.md index 9fe39847..f022f24d 100644 --- a/docs/2-usage/3-read.md +++ b/docs/2-usage/3-read.md @@ -9,7 +9,7 @@ Live streams can be read from the server with the following protocols and codecs | [SRT](#srt) | | H265, H264, MPEG-4 Video (H263, Xvid), MPEG-1/2 Video | Opus, MPEG-4 Audio (AAC), MPEG-1/2 Audio (MP3), AC-3 | | [WebRTC](#webrtc) | WHEP | AV1, VP9, VP8, H265, H264 | Opus, G722, G711 (PCMA, PCMU) | | [RTSP](#rtsp) | UDP, UDP-Multicast, TCP, RTSPS | AV1, VP9, VP8, H265, H264, MPEG-4 Video (H263, Xvid), MPEG-1/2 Video, M-JPEG and any RTP-compatible codec | Opus, MPEG-4 Audio (AAC), MPEG-1/2 Audio (MP3), AC-3, G726, G722, G711 (PCMA, PCMU), LPCM and any RTP-compatible codec | -| [RTMP](#rtmp) | RTMP, RTMPS, Enhanced RTMP | H264 | MPEG-4 Audio (AAC), MPEG-1/2 Audio (MP3) | +| [RTMP](#rtmp) | RTMP, RTMPS, Enhanced RTMP | AV1, VP9, H265, H264 | Opus, MPEG-4 Audio (AAC), MPEG-1/2 Audio (MP3), AC-3, G711 (PCMA, PCMU), LPCM | | [HLS](#hls) | Low-Latency HLS, MP4-based HLS, legacy HLS | AV1, VP9, H265, H264 | Opus, MPEG-4 Audio (AAC) | We provide instructions for reading with the following software: @@ -182,6 +182,18 @@ The RTSP protocol supports several underlying transport protocols, each with its ffmpeg -rtsp_transport tcp -i rtsp://localhost:8554/mystream -c copy output.mp4 ``` +FFmpeg can also read a stream with RTMP: + +```sh +ffmpeg -i rtmp://localhost/mystream -c copy output.mp4 +``` + +In order to read AV1, VP9, H265, Opus, AC3 tracks and in order to read multiple video or audio tracks, the `-rtmp_enhanced_codecs` flag must be present: + +```sh +ffmpeg -rtmp_enhanced_codecs ac-3,av01,avc1,ec-3,fLaC,hvc1,.mp3,mp4a,Opus,vp09 -i rtmp://localhost/mystream -c copy output.mp4 +``` + ### GStreamer GStreamer can read a stream from the server in several ways (RTSP, RTMP, HLS, WebRTC with WHEP, SRT). The recommended one consists in reading with [RTSP](#rtsp): diff --git a/go.mod b/go.mod index b486cb97..fc0cdb24 100644 --- a/go.mod +++ b/go.mod @@ -11,7 +11,7 @@ require ( github.com/asticode/go-astits v1.13.0 github.com/bluenviron/gohlslib/v2 v2.2.2 github.com/bluenviron/gortsplib/v4 v4.16.2 - github.com/bluenviron/mediacommon/v2 v2.4.1 + github.com/bluenviron/mediacommon/v2 v2.4.2-0.20250909112826-017d0bbe41db github.com/datarhei/gosrt v0.9.0 github.com/fsnotify/fsnotify v1.9.0 github.com/gin-contrib/pprof v1.5.3 diff --git a/go.sum b/go.sum index b41188f6..14f0328e 100644 --- a/go.sum +++ b/go.sum @@ -35,8 +35,8 @@ github.com/bluenviron/gohlslib/v2 v2.2.2 h1:Q86VloPjwONKF8pu6jSEh9ENm4UzdMl5SzYv github.com/bluenviron/gohlslib/v2 v2.2.2/go.mod h1:3Lby/VMDD/cN0B3uJPd3bEEiJZ34LqXs71FEvN/fq2k= github.com/bluenviron/gortsplib/v4 v4.16.2 h1:10HaMsorjW13gscLp3R7Oj41ck2i1EHIUYCNWD2wpkI= github.com/bluenviron/gortsplib/v4 v4.16.2/go.mod h1:Vm07yUMys9XKnuZJLfTT8zluAN2n9ZOtz40Xb8RKh+8= -github.com/bluenviron/mediacommon/v2 v2.4.1 h1:PsKrO/c7hDjXxiOGRUBsYtMGNb4lKWIFea6zcOchoVs= -github.com/bluenviron/mediacommon/v2 v2.4.1/go.mod h1:a6MbPmXtYda9mKibKVMZlW20GYLLrX2R7ZkUE+1pwV0= +github.com/bluenviron/mediacommon/v2 v2.4.2-0.20250909112826-017d0bbe41db h1:yBxx462HsYC14/vKr5BF/Hlpso6WmyHzjwoE/W0td5s= +github.com/bluenviron/mediacommon/v2 v2.4.2-0.20250909112826-017d0bbe41db/go.mod h1:zy1fODPuS/kBd93ftgJS1Jhvjq7LFWfAo32KP7By9AE= github.com/bytedance/sonic v1.13.2 h1:8/H1FempDZqC4VqjptGo14QQlJx8VdZJegxs6wwfqpQ= github.com/bytedance/sonic v1.13.2/go.mod h1:o68xyaF9u2gvVBuGHPlUVCy+ZfmNNO5ETf1+KgkJhz4= github.com/bytedance/sonic/loader v0.1.1/go.mod h1:ncP89zfokxS5LZrJxl5z0UJcsk4M4yY2JpfqGeCtNLU= diff --git a/internal/core/api_test.go b/internal/core/api_test.go index 45161a5f..c19b3618 100644 --- a/internal/core/api_test.go +++ b/internal/core/api_test.go @@ -16,6 +16,7 @@ import ( "github.com/bluenviron/gortsplib/v4" "github.com/bluenviron/gortsplib/v4/pkg/description" + "github.com/bluenviron/gortsplib/v4/pkg/format" "github.com/bluenviron/mediacommon/v2/pkg/formats/mpegts" srt "github.com/datarhei/gosrt" "github.com/google/uuid" @@ -441,13 +442,13 @@ func TestAPIProtocolListGet(t *testing.T) { defer conn.Close() w := &rtmp.Writer{ - Conn: conn, - VideoTrack: test.FormatH264, + Conn: conn, + Tracks: []format.Format{test.FormatH264}, } err = w.Initialize() require.NoError(t, err) - err = w.WriteH264(2*time.Second, 2*time.Second, [][]byte{{5, 2, 3, 4}}) + err = w.WriteH264(test.FormatH264, 2*time.Second, 2*time.Second, [][]byte{{5, 2, 3, 4}}) require.NoError(t, err) time.Sleep(500 * time.Millisecond) @@ -1034,13 +1035,13 @@ func TestAPIProtocolKick(t *testing.T) { defer conn.Close() w := &rtmp.Writer{ - Conn: conn, - VideoTrack: test.FormatH264, + Conn: conn, + Tracks: []format.Format{test.FormatH264}, } err = w.Initialize() require.NoError(t, err) - err = w.WriteH264(2*time.Second, 2*time.Second, [][]byte{{5, 2, 3, 4}}) + err = w.WriteH264(test.FormatH264, 2*time.Second, 2*time.Second, [][]byte{{5, 2, 3, 4}}) require.NoError(t, err) case "webrtc": diff --git a/internal/core/metrics_test.go b/internal/core/metrics_test.go index 9d048bd9..6c65955d 100644 --- a/internal/core/metrics_test.go +++ b/internal/core/metrics_test.go @@ -14,6 +14,7 @@ import ( "github.com/bluenviron/gortsplib/v4" "github.com/bluenviron/gortsplib/v4/pkg/description" + "github.com/bluenviron/gortsplib/v4/pkg/format" "github.com/bluenviron/mediacommon/v2/pkg/formats/mpegts" srt "github.com/datarhei/gosrt" "github.com/pion/rtp" @@ -217,13 +218,13 @@ webrtc_sessions_rtcp_packets_sent 0 defer conn.Close() w := &rtmp.Writer{ - Conn: conn, - VideoTrack: test.FormatH264, + Conn: conn, + Tracks: []format.Format{test.FormatH264}, } err2 = w.Initialize() require.NoError(t, err2) - err2 = w.WriteH264(2*time.Second, 2*time.Second, [][]byte{{5, 2, 3, 4}}) + err2 = w.WriteH264(test.FormatH264, 2*time.Second, 2*time.Second, [][]byte{{5, 2, 3, 4}}) require.NoError(t, err2) <-terminate @@ -245,13 +246,13 @@ webrtc_sessions_rtcp_packets_sent 0 defer conn.Close() w := &rtmp.Writer{ - Conn: conn, - VideoTrack: test.FormatH264, + Conn: conn, + Tracks: []format.Format{test.FormatH264}, } err2 = w.Initialize() require.NoError(t, err2) - err2 = w.WriteH264(2*time.Second, 2*time.Second, [][]byte{{5, 2, 3, 4}}) + err2 = w.WriteH264(test.FormatH264, 2*time.Second, 2*time.Second, [][]byte{{5, 2, 3, 4}}) require.NoError(t, err2) <-terminate diff --git a/internal/formatprocessor/av1.go b/internal/formatprocessor/av1.go index 111ee126..b903d4a9 100644 --- a/internal/formatprocessor/av1.go +++ b/internal/formatprocessor/av1.go @@ -7,6 +7,7 @@ import ( "github.com/bluenviron/gortsplib/v4/pkg/format" "github.com/bluenviron/gortsplib/v4/pkg/format/rtpav1" + mcav1 "github.com/bluenviron/mediacommon/v2/pkg/codecs/av1" "github.com/pion/rtp" "github.com/bluenviron/mediamtx/internal/logger" @@ -48,9 +49,44 @@ func (t *av1) createEncoder() error { return t.encoder.Init() } +func (t *av1) remuxTemporalUnit(tu [][]byte) [][]byte { + n := 0 + + for _, obu := range tu { + typ := mcav1.OBUType((obu[0] >> 3) & 0b1111) + + if typ == mcav1.OBUTypeTemporalDelimiter { + continue + } + n++ + } + + if n == 0 { + return nil + } + + filteredTU := make([][]byte, n) + i := 0 + + for _, obu := range tu { + typ := mcav1.OBUType((obu[0] >> 3) & 0b1111) + + if typ == mcav1.OBUTypeTemporalDelimiter { + continue + } + + filteredTU[i] = obu + i++ + } + + return filteredTU +} + func (t *av1) ProcessUnit(uu unit.Unit) error { //nolint:dupl u := uu.(*unit.AV1) + u.TU = t.remuxTemporalUnit(u.TU) + pkts, err := t.encoder.Encode(u.TU) if err != nil { return err @@ -106,7 +142,7 @@ func (t *av1) ProcessRTPPacket( //nolint:dupl return nil, err } - u.TU = tu + u.TU = t.remuxTemporalUnit(tu) } // route packet as is diff --git a/internal/formatprocessor/av1_test.go b/internal/formatprocessor/av1_test.go new file mode 100644 index 00000000..23d9ea60 --- /dev/null +++ b/internal/formatprocessor/av1_test.go @@ -0,0 +1,34 @@ +package formatprocessor + +import ( + "testing" + + "github.com/bluenviron/gortsplib/v4/pkg/format" + mcav1 "github.com/bluenviron/mediacommon/v2/pkg/codecs/av1" + "github.com/bluenviron/mediamtx/internal/unit" + "github.com/stretchr/testify/require" +) + +func TestAV1RemoveTUD(t *testing.T) { + forma := &format.AV1{} + + p, err := New(1450, forma, true, nil) + require.NoError(t, err) + + u := &unit.AV1{ + Base: unit.Base{ + PTS: 30000, + }, + TU: [][]byte{ + {byte(mcav1.OBUTypeTemporalDelimiter) << 3}, + {5}, + }, + } + + err = p.ProcessUnit(u) + require.NoError(t, err) + + require.Equal(t, [][]byte{ + {5}, + }, u.TU) +} diff --git a/internal/formatprocessor/h264.go b/internal/formatprocessor/h264.go index 7021bda7..2f478bd7 100644 --- a/internal/formatprocessor/h264.go +++ b/internal/formatprocessor/h264.go @@ -174,13 +174,13 @@ func (t *h264) remuxAccessUnit(au [][]byte) [][]byte { typ := mch264.NALUType(nalu[0] & 0x1F) switch typ { - case mch264.NALUTypeSPS, mch264.NALUTypePPS: // parameters: remove + case mch264.NALUTypeSPS, mch264.NALUTypePPS: continue - case mch264.NALUTypeAccessUnitDelimiter: // AUD: remove + case mch264.NALUTypeAccessUnitDelimiter: continue - case mch264.NALUTypeIDR: // key frame + case mch264.NALUTypeIDR: if !isKeyFrame { isKeyFrame = true @@ -197,12 +197,12 @@ func (t *h264) remuxAccessUnit(au [][]byte) [][]byte { return nil } - filteredNALUs := make([][]byte, n) + filteredAU := make([][]byte, n) i := 0 if isKeyFrame && t.Format.SPS != nil && t.Format.PPS != nil { - filteredNALUs[0] = t.Format.SPS - filteredNALUs[1] = t.Format.PPS + filteredAU[0] = t.Format.SPS + filteredAU[1] = t.Format.PPS i = 2 } @@ -217,11 +217,11 @@ func (t *h264) remuxAccessUnit(au [][]byte) [][]byte { continue } - filteredNALUs[i] = nalu + filteredAU[i] = nalu i++ } - return filteredNALUs + return filteredAU } func (t *h264) ProcessUnit(uu unit.Unit) error { diff --git a/internal/formatprocessor/h264_test.go b/internal/formatprocessor/h264_test.go index 09436ef0..6ee097ea 100644 --- a/internal/formatprocessor/h264_test.go +++ b/internal/formatprocessor/h264_test.go @@ -28,7 +28,31 @@ func Logger(cb func(logger.Level, string, ...interface{})) logger.Writer { return &testLogger{cb: cb} } -func TestH264ProcessUnit(t *testing.T) { +func TestH264RemoveAUD(t *testing.T) { + forma := &format.H264{} + + p, err := New(1450, forma, true, nil) + require.NoError(t, err) + + u := &unit.H264{ + Base: unit.Base{ + PTS: 30000, + }, + AU: [][]byte{ + {9, 24}, // AUD + {5, 1}, // IDR + }, + } + + err = p.ProcessUnit(u) + require.NoError(t, err) + + require.Equal(t, [][]byte{ + {5, 1}, // IDR + }, u.AU) +} + +func TestH264AddParams(t *testing.T) { forma := &format.H264{} p, err := New(1450, forma, true, nil) @@ -77,11 +101,11 @@ func TestH264ProcessUnit(t *testing.T) { {5, 2}, // IDR }, u2.AU) - // test that timestamp had increased + // test that timestamp has increased require.Equal(t, u1.RTPPackets[0].Timestamp+30000, u2.RTPPackets[0].Timestamp) } -func TestH264ProcessUnitEmpty(t *testing.T) { +func TestH264ProcessEmptyUnit(t *testing.T) { forma := &format.H264{ PayloadTyp: 96, PacketizationMode: 1, @@ -104,7 +128,7 @@ func TestH264ProcessUnitEmpty(t *testing.T) { require.Equal(t, []*rtp.Packet(nil), unit.RTPPackets) } -func TestH264ProcessRTPPacketUpdateParams(t *testing.T) { +func TestH264RTPExtractParams(t *testing.T) { for _, ca := range []string{"standard", "aggregated"} { t.Run(ca, func(t *testing.T) { forma := &format.H264{ @@ -169,7 +193,7 @@ func TestH264ProcessRTPPacketUpdateParams(t *testing.T) { } } -func TestH264ProcessRTPPacketOversized(t *testing.T) { +func TestH264RTPOversized(t *testing.T) { forma := &format.H264{ PayloadTyp: 96, SPS: []byte{0x01, 0x02, 0x03, 0x04}, diff --git a/internal/formatprocessor/h265.go b/internal/formatprocessor/h265.go index dd3a6fa8..fe374f4b 100644 --- a/internal/formatprocessor/h265.go +++ b/internal/formatprocessor/h265.go @@ -205,13 +205,13 @@ func (t *h265) remuxAccessUnit(au [][]byte) [][]byte { typ := mch265.NALUType((nalu[0] >> 1) & 0b111111) switch typ { - case mch265.NALUType_VPS_NUT, mch265.NALUType_SPS_NUT, mch265.NALUType_PPS_NUT: // parameters: remove + case mch265.NALUType_VPS_NUT, mch265.NALUType_SPS_NUT, mch265.NALUType_PPS_NUT: continue - case mch265.NALUType_AUD_NUT: // AUD: remove + case mch265.NALUType_AUD_NUT: continue - case mch265.NALUType_IDR_W_RADL, mch265.NALUType_IDR_N_LP, mch265.NALUType_CRA_NUT: // key frame + case mch265.NALUType_IDR_W_RADL, mch265.NALUType_IDR_N_LP, mch265.NALUType_CRA_NUT: if !isKeyFrame { isKeyFrame = true @@ -228,13 +228,13 @@ func (t *h265) remuxAccessUnit(au [][]byte) [][]byte { return nil } - filteredNALUs := make([][]byte, n) + filteredAU := make([][]byte, n) i := 0 if isKeyFrame && t.Format.VPS != nil && t.Format.SPS != nil && t.Format.PPS != nil { - filteredNALUs[0] = t.Format.VPS - filteredNALUs[1] = t.Format.SPS - filteredNALUs[2] = t.Format.PPS + filteredAU[0] = t.Format.VPS + filteredAU[1] = t.Format.SPS + filteredAU[2] = t.Format.PPS i = 3 } @@ -249,11 +249,11 @@ func (t *h265) remuxAccessUnit(au [][]byte) [][]byte { continue } - filteredNALUs[i] = nalu + filteredAU[i] = nalu i++ } - return filteredNALUs + return filteredAU } func (t *h265) ProcessUnit(uu unit.Unit) error { //nolint:dupl diff --git a/internal/formatprocessor/h265_test.go b/internal/formatprocessor/h265_test.go index 65eb52cc..a21330d7 100644 --- a/internal/formatprocessor/h265_test.go +++ b/internal/formatprocessor/h265_test.go @@ -15,7 +15,31 @@ import ( "github.com/bluenviron/mediamtx/internal/unit" ) -func TestH265ProcessUnit(t *testing.T) { +func TestH265RemoveAUD(t *testing.T) { + forma := &format.H265{} + + p, err := New(1450, forma, true, nil) + require.NoError(t, err) + + u := &unit.H265{ + Base: unit.Base{ + PTS: 30000, + }, + AU: [][]byte{ + {byte(mch265.NALUType_AUD_NUT) << 1, 0}, + {byte(mch265.NALUType_CRA_NUT) << 1, 0}, + }, + } + + err = p.ProcessUnit(u) + require.NoError(t, err) + + require.Equal(t, [][]byte{ + {byte(mch265.NALUType_CRA_NUT) << 1, 0}, + }, u.AU) +} + +func TestH265AddParams(t *testing.T) { forma := &format.H265{} p, err := New(1450, forma, true, nil) @@ -68,11 +92,11 @@ func TestH265ProcessUnit(t *testing.T) { {byte(mch265.NALUType_CRA_NUT) << 1, 1}, }, u2.AU) - // test that timestamp had increased + // test that timestamp has increased require.Equal(t, u1.RTPPackets[0].Timestamp+30000, u2.RTPPackets[0].Timestamp) } -func TestH265ProcessUnitEmpty(t *testing.T) { +func TestH265ProcessEmptyUnit(t *testing.T) { forma := &format.H265{ PayloadTyp: 96, } @@ -95,7 +119,7 @@ func TestH265ProcessUnitEmpty(t *testing.T) { require.Equal(t, []*rtp.Packet(nil), unit.RTPPackets) } -func TestH265ProcessRTPPacketUpdateParams(t *testing.T) { +func TestH265RTPExtractParams(t *testing.T) { for _, ca := range []string{"standard", "aggregated"} { t.Run(ca, func(t *testing.T) { forma := &format.H265{ @@ -168,7 +192,7 @@ func TestH265ProcessRTPPacketUpdateParams(t *testing.T) { } } -func TestH265ProcessRTPPacketOversized(t *testing.T) { +func TestH265RTPOversized(t *testing.T) { forma := &format.H265{ PayloadTyp: 96, VPS: []byte{byte(mch265.NALUType_VPS_NUT) << 1, 10, 11, 12}, diff --git a/internal/formatprocessor/mpeg4_video_test.go b/internal/formatprocessor/mpeg4_video_test.go index f14d04c9..082fd794 100644 --- a/internal/formatprocessor/mpeg4_video_test.go +++ b/internal/formatprocessor/mpeg4_video_test.go @@ -66,6 +66,6 @@ func TestMPEG4VideoProcessUnit(t *testing.T) { 0, 0, 1, 0xF1, }, u2.Frame) - // test that timestamp had increased + // test that timestamp has increased require.Equal(t, u1.RTPPackets[0].Timestamp+30000, u2.RTPPackets[0].Timestamp) } diff --git a/internal/protocols/mpegts/from_stream.go b/internal/protocols/mpegts/from_stream.go index 3a81f03e..aa58b837 100644 --- a/internal/protocols/mpegts/from_stream.go +++ b/internal/protocols/mpegts/from_stream.go @@ -69,10 +69,8 @@ func FromStream( return nil } - randomAccess := h265.IsRandomAccess(tunit.AU) - if dtsExtractor == nil { - if !randomAccess { + if !h265.IsRandomAccess(tunit.AU) { return nil } dtsExtractor = &h265.DTSExtractor{} diff --git a/internal/protocols/rtmp/client.go b/internal/protocols/rtmp/client.go index b1c28f34..321fe4b5 100644 --- a/internal/protocols/rtmp/client.go +++ b/internal/protocols/rtmp/client.go @@ -17,8 +17,25 @@ import ( "github.com/google/uuid" ) +// RTMP 1.0 spec, section 7.2.1.1 +const ( + supportSndNone = 0x0001 + supportSndMP3 = 0x0004 + supportSndG711A = 0x0080 + supportSndG711U = 0x0100 + supportSndAAV = 0x0400 + + supportVidH264 = 0x0080 + + encodingAMF0 = 0 +) + var errAuth = errors.New("auth") +func fourCCToString(c message.FourCC) string { + return string([]byte{byte(c >> 24), byte(c >> 16), byte(c >> 8), byte(c)}) +} + func resultIsOK1(res *message.CommandAMF0) bool { if len(res.Arguments) < 2 { return false @@ -252,15 +269,45 @@ func (c *Client) initialize3() error { {Key: "app", Value: app}, {Key: "flashVer", Value: "LNX 9,0,124,2"}, {Key: "tcUrl", Value: tcURL}, + {Key: "objectEncoding", Value: float64(encodingAMF0)}, } if !c.Publish { connectArg = append(connectArg, - amf0.ObjectEntry{Key: "fpad", Value: false}, - amf0.ObjectEntry{Key: "capabilities", Value: float64(15)}, - amf0.ObjectEntry{Key: "audioCodecs", Value: float64(4071)}, - amf0.ObjectEntry{Key: "videoCodecs", Value: float64(252)}, - amf0.ObjectEntry{Key: "videoFunction", Value: float64(1)}, + amf0.ObjectEntry{ + Key: "fpad", + Value: false, + }, + amf0.ObjectEntry{ + Key: "capabilities", + Value: float64(15), + }, + amf0.ObjectEntry{ + Key: "audioCodecs", + Value: float64( + supportSndNone | supportSndMP3 | supportSndG711A | supportSndG711U | supportSndAAV), + }, + amf0.ObjectEntry{ + Key: "videoCodecs", + Value: float64(supportVidH264), + }, + amf0.ObjectEntry{ + Key: "videoFunction", + Value: float64(0), + }, + amf0.ObjectEntry{ + Key: "fourCcList", + Value: amf0.StrictArray{ + fourCCToString(message.FourCCAV1), + fourCCToString(message.FourCCVP9), + fourCCToString(message.FourCCHEVC), + fourCCToString(message.FourCCAVC), + fourCCToString(message.FourCCOpus), + fourCCToString(message.FourCCAC3), + fourCCToString(message.FourCCMP4A), + fourCCToString(message.FourCCMP3), + }, + }, ) } diff --git a/internal/protocols/rtmp/client_test.go b/internal/protocols/rtmp/client_test.go index f3380c35..47f9bb18 100644 --- a/internal/protocols/rtmp/client_test.go +++ b/internal/protocols/rtmp/client_test.go @@ -66,8 +66,8 @@ func TestClient(t *testing.T) { require.NoError(t, err2) switch authState { - case 0: - require.Equal(t, &message.CommandAMF0{ + case 0: //nolint:dupl + require.Equal(t, &message.CommandAMF0{ //nolint:dupl ChunkStreamID: 3, Name: "connect", CommandID: 1, @@ -76,17 +76,28 @@ func TestClient(t *testing.T) { {Key: "app", Value: "stream"}, {Key: "flashVer", Value: "LNX 9,0,124,2"}, {Key: "tcUrl", Value: "rtmp://127.0.0.1:9121/stream"}, + {Key: "objectEncoding", Value: float64(0)}, {Key: "fpad", Value: false}, {Key: "capabilities", Value: float64(15)}, - {Key: "audioCodecs", Value: float64(4071)}, - {Key: "videoCodecs", Value: float64(252)}, - {Key: "videoFunction", Value: float64(1)}, + {Key: "audioCodecs", Value: float64(1413)}, + {Key: "videoCodecs", Value: float64(128)}, + {Key: "videoFunction", Value: float64(0)}, + {Key: "fourCcList", Value: amf0.StrictArray{ + "av01", + "vp09", + "hvc1", + "avc1", + "Opus", + "ac-3", + "mp4a", + ".mp3", + }}, }, }, }, msg) - case 1: - require.Equal(t, &message.CommandAMF0{ + case 1: //nolint:dupl + require.Equal(t, &message.CommandAMF0{ //nolint:dupl ChunkStreamID: 3, Name: "connect", CommandID: 1, @@ -95,11 +106,22 @@ func TestClient(t *testing.T) { {Key: "app", Value: "stream?authmod=adobe&user=myuser"}, {Key: "flashVer", Value: "LNX 9,0,124,2"}, {Key: "tcUrl", Value: "rtmp://127.0.0.1:9121/stream?authmod=adobe&user=myuser"}, + {Key: "objectEncoding", Value: float64(0)}, {Key: "fpad", Value: false}, {Key: "capabilities", Value: float64(15)}, - {Key: "audioCodecs", Value: float64(4071)}, - {Key: "videoCodecs", Value: float64(252)}, - {Key: "videoFunction", Value: float64(1)}, + {Key: "audioCodecs", Value: float64(1413)}, + {Key: "videoCodecs", Value: float64(128)}, + {Key: "videoFunction", Value: float64(0)}, + {Key: "fourCcList", Value: amf0.StrictArray{ + "av01", + "vp09", + "hvc1", + "avc1", + "Opus", + "ac-3", + "mp4a", + ".mp3", + }}, }, }, }, msg) @@ -127,11 +149,22 @@ func TestClient(t *testing.T) { Value: "rtmp://127.0.0.1:9121/stream?authmod=adobe&user=myuser&challenge=" + clientChallenge + "&response=" + response, }, + {Key: "objectEncoding", Value: float64(0)}, {Key: "fpad", Value: false}, {Key: "capabilities", Value: float64(15)}, - {Key: "audioCodecs", Value: float64(4071)}, - {Key: "videoCodecs", Value: float64(252)}, - {Key: "videoFunction", Value: float64(1)}, + {Key: "audioCodecs", Value: float64(1413)}, + {Key: "videoCodecs", Value: float64(128)}, + {Key: "videoFunction", Value: float64(0)}, + {Key: "fourCcList", Value: amf0.StrictArray{ + "av01", + "vp09", + "hvc1", + "avc1", + "Opus", + "ac-3", + "mp4a", + ".mp3", + }}, }, }, }, msg) @@ -140,7 +173,7 @@ func TestClient(t *testing.T) { case "read", "read nginx rtmp": msg, err2 = mrw.Read() require.NoError(t, err2) - require.Equal(t, &message.CommandAMF0{ + require.Equal(t, &message.CommandAMF0{ //nolint:dupl ChunkStreamID: 3, Name: "connect", CommandID: 1, @@ -149,11 +182,22 @@ func TestClient(t *testing.T) { {Key: "app", Value: "stream"}, {Key: "flashVer", Value: "LNX 9,0,124,2"}, {Key: "tcUrl", Value: "rtmp://127.0.0.1:9121/stream"}, + {Key: "objectEncoding", Value: float64(0)}, {Key: "fpad", Value: false}, {Key: "capabilities", Value: float64(15)}, - {Key: "audioCodecs", Value: float64(4071)}, - {Key: "videoCodecs", Value: float64(252)}, - {Key: "videoFunction", Value: float64(1)}, + {Key: "audioCodecs", Value: float64(1413)}, + {Key: "videoCodecs", Value: float64(128)}, + {Key: "videoFunction", Value: float64(0)}, + {Key: "fourCcList", Value: amf0.StrictArray{ + "av01", + "vp09", + "hvc1", + "avc1", + "Opus", + "ac-3", + "mp4a", + ".mp3", + }}, }, }, }, msg) @@ -170,6 +214,7 @@ func TestClient(t *testing.T) { {Key: "app", Value: "stream"}, {Key: "flashVer", Value: "LNX 9,0,124,2"}, {Key: "tcUrl", Value: "rtmp://127.0.0.1:9121/stream"}, + {Key: "objectEncoding", Value: float64(0)}, }, }, }, msg) @@ -408,11 +453,11 @@ func TestClient(t *testing.T) { switch ca { case "read", "read nginx rtmp": require.Equal(t, uint64(3421), conn.BytesReceived()) - require.Equal(t, uint64(3409), conn.BytesSent()) + require.Equal(t, uint64(0xdb3), conn.BytesSent()) case "publish": require.Equal(t, uint64(3427), conn.BytesReceived()) - require.Equal(t, uint64(0xd27), conn.BytesSent()) + require.Equal(t, uint64(0xd40), conn.BytesSent()) } <-done diff --git a/internal/protocols/rtmp/conn.go b/internal/protocols/rtmp/conn.go index 3098cb6a..d4683e0e 100644 --- a/internal/protocols/rtmp/conn.go +++ b/internal/protocols/rtmp/conn.go @@ -1,9 +1,6 @@ package rtmp import ( - "io" - - "github.com/bluenviron/mediamtx/internal/protocols/rtmp/bytecounter" "github.com/bluenviron/mediamtx/internal/protocols/rtmp/message" ) @@ -14,33 +11,3 @@ type Conn interface { Read() (message.Message, error) Write(msg message.Message) error } - -type dummyConn struct { - rw io.ReadWriter - - bc *bytecounter.ReadWriter - mrw *message.ReadWriter -} - -func (c *dummyConn) initialize() { - c.bc = bytecounter.NewReadWriter(c.rw) - c.mrw = message.NewReadWriter(c.bc, c.bc, false) -} - -// BytesReceived returns the number of bytes received. -func (c *dummyConn) BytesReceived() uint64 { - return c.bc.Reader.Count() -} - -// BytesSent returns the number of bytes sent. -func (c *dummyConn) BytesSent() uint64 { - return c.bc.Writer.Count() -} - -func (c *dummyConn) Read() (message.Message, error) { - return c.mrw.Read() -} - -func (c *dummyConn) Write(msg message.Message) error { - return c.mrw.Write(msg) -} diff --git a/internal/protocols/rtmp/from_stream.go b/internal/protocols/rtmp/from_stream.go index 7aa2dd1e..be105d21 100644 --- a/internal/protocols/rtmp/from_stream.go +++ b/internal/protocols/rtmp/from_stream.go @@ -4,19 +4,25 @@ import ( "errors" "fmt" "net" + "slices" "time" "github.com/bluenviron/gortsplib/v4/pkg/format" + "github.com/bluenviron/mediacommon/v2/pkg/codecs/ac3" "github.com/bluenviron/mediacommon/v2/pkg/codecs/h264" + "github.com/bluenviron/mediacommon/v2/pkg/codecs/h265" "github.com/bluenviron/mediacommon/v2/pkg/codecs/mpeg1audio" "github.com/bluenviron/mediacommon/v2/pkg/codecs/mpeg4audio" + "github.com/bluenviron/mediacommon/v2/pkg/codecs/opus" "github.com/bluenviron/mediamtx/internal/logger" + "github.com/bluenviron/mediamtx/internal/protocols/rtmp/message" "github.com/bluenviron/mediamtx/internal/stream" "github.com/bluenviron/mediamtx/internal/unit" ) var errNoSupportedCodecsFrom = errors.New( - "the stream doesn't contain any supported codec, which are currently H264, MPEG-4 Audio, MPEG-1/2 Audio") + "the stream doesn't contain any supported codec, which are currently " + + "AV1, VP9, H265, H264, Opus, MPEG-4 Audio, MPEG-1/2 Audio, AC-3, G711, LPCM") func multiplyAndDivide2(v, m, d time.Duration) time.Duration { secs := v / d @@ -28,227 +34,391 @@ func timestampToDuration(t int64, clockRate int) time.Duration { return multiplyAndDivide2(time.Duration(t), time.Second, time.Duration(clockRate)) } -func setupVideo( - str *stream.Stream, - reader stream.Reader, - w **Writer, - nconn net.Conn, - writeTimeout time.Duration, -) format.Format { - var videoFormatH264 *format.H264 - videoMedia := str.Desc.FindFormat(&videoFormatH264) - - if videoFormatH264 != nil { - var videoDTSExtractor *h264.DTSExtractor - - str.AddReader( - reader, - videoMedia, - videoFormatH264, - func(u unit.Unit) error { - tunit := u.(*unit.H264) - - if tunit.AU == nil { - return nil - } - - idrPresent := false - nonIDRPresent := false - - for _, nalu := range tunit.AU { - typ := h264.NALUType(nalu[0] & 0x1F) - switch typ { - case h264.NALUTypeIDR: - idrPresent = true - - case h264.NALUTypeNonIDR: - nonIDRPresent = true - } - } - - // wait until we receive an IDR - if videoDTSExtractor == nil { - if !idrPresent { - return nil - } - - videoDTSExtractor = &h264.DTSExtractor{} - videoDTSExtractor.Initialize() - } else if !idrPresent && !nonIDRPresent { - return nil - } - - dts, err := videoDTSExtractor.Extract(tunit.AU, tunit.PTS) - if err != nil { - return err - } - - nconn.SetWriteDeadline(time.Now().Add(writeTimeout)) - return (*w).WriteH264( - timestampToDuration(tunit.PTS, videoFormatH264.ClockRate()), - timestampToDuration(dts, videoFormatH264.ClockRate()), - tunit.AU) - }) - - return videoFormatH264 - } - - return nil -} - -func setupAudio( - str *stream.Stream, - reader stream.Reader, - w **Writer, - nconn net.Conn, - writeTimeout time.Duration, -) format.Format { - var audioFormatMPEG4Audio *format.MPEG4Audio - audioMedia := str.Desc.FindFormat(&audioFormatMPEG4Audio) - - if audioMedia != nil { - str.AddReader( - reader, - audioMedia, - audioFormatMPEG4Audio, - func(u unit.Unit) error { - tunit := u.(*unit.MPEG4Audio) - - if tunit.AUs == nil { - return nil - } - - for i, au := range tunit.AUs { - pts := tunit.PTS + int64(i)*mpeg4audio.SamplesPerAccessUnit - - nconn.SetWriteDeadline(time.Now().Add(writeTimeout)) - err := (*w).WriteMPEG4Audio( - timestampToDuration(pts, audioFormatMPEG4Audio.ClockRate()), - au, - ) - if err != nil { - return err - } - } - - return nil - }) - - return audioFormatMPEG4Audio - } - - var audioFormatMPEG4AudioLATM *format.MPEG4AudioLATM - audioMedia = str.Desc.FindFormat(&audioFormatMPEG4AudioLATM) - - if audioMedia != nil && !audioFormatMPEG4AudioLATM.CPresent { - str.AddReader( - reader, - audioMedia, - audioFormatMPEG4AudioLATM, - func(u unit.Unit) error { - tunit := u.(*unit.MPEG4AudioLATM) - - if tunit.Element == nil { - return nil - } - - var ame mpeg4audio.AudioMuxElement - ame.StreamMuxConfig = audioFormatMPEG4AudioLATM.StreamMuxConfig - err := ame.Unmarshal(tunit.Element) - if err != nil { - return err - } - - nconn.SetWriteDeadline(time.Now().Add(writeTimeout)) - return (*w).WriteMPEG4Audio( - timestampToDuration(tunit.PTS, audioFormatMPEG4AudioLATM.ClockRate()), - ame.Payloads[0][0][0], - ) - }) - - return audioFormatMPEG4AudioLATM - } - - var audioFormatMPEG1 *format.MPEG1Audio - audioMedia = str.Desc.FindFormat(&audioFormatMPEG1) - - if audioMedia != nil { - str.AddReader( - reader, - audioMedia, - audioFormatMPEG1, - func(u unit.Unit) error { - tunit := u.(*unit.MPEG1Audio) - - pts := tunit.PTS - - for _, frame := range tunit.Frames { - var h mpeg1audio.FrameHeader - err := h.Unmarshal(frame) - if err != nil { - return err - } - - if h.MPEG2 || h.Layer != 3 { - return fmt.Errorf("RTMP only supports MPEG-1 layer 3 audio") - } - - nconn.SetWriteDeadline(time.Now().Add(writeTimeout)) - err = (*w).WriteMPEG1Audio( - timestampToDuration(pts, audioFormatMPEG1.ClockRate()), - &h, - frame) - if err != nil { - return err - } - - pts += int64(h.SampleCount()) * - int64(audioFormatMPEG1.ClockRate()) / int64(h.SampleRate) - } - - return nil - }) - - return audioFormatMPEG1 - } - - return nil -} - // FromStream maps a MediaMTX stream to a RTMP stream. func FromStream( str *stream.Stream, reader stream.Reader, - conn Conn, + conn *ServerConn, nconn net.Conn, writeTimeout time.Duration, ) error { + var tracks []format.Format var w *Writer - videoFormat := setupVideo( - str, - reader, - &w, - nconn, - writeTimeout, - ) + for _, media := range str.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, + media, + forma, + func(u unit.Unit) error { + tunit := u.(*unit.AV1) - audioFormat := setupAudio( - str, - reader, - &w, - nconn, - writeTimeout, - ) + if tunit.TU == nil { + return nil + } - if videoFormat == nil && audioFormat == nil { + nconn.SetWriteDeadline(time.Now().Add(writeTimeout)) + return (*w).WriteAV1( + forma, + timestampToDuration(tunit.PTS, forma.ClockRate()), + timestampToDuration(tunit.PTS, forma.ClockRate()), + tunit.TU) + }) + + tracks = append(tracks, forma) + } + + case *format.VP9: + if slices.Contains(conn.FourCcList, interface{}(fourCCToString(message.FourCCVP9))) { + str.AddReader( + reader, + media, + forma, + func(u unit.Unit) error { + tunit := u.(*unit.VP9) + + if tunit.Frame == nil { + return nil + } + + nconn.SetWriteDeadline(time.Now().Add(writeTimeout)) + return (*w).WriteVP9( + forma, + timestampToDuration(tunit.PTS, forma.ClockRate()), + timestampToDuration(tunit.PTS, forma.ClockRate()), + tunit.Frame) + }) + + tracks = append(tracks, forma) + } + + case *format.H265: + if slices.Contains(conn.FourCcList, interface{}(fourCCToString(message.FourCCHEVC))) { + var videoDTSExtractor *h265.DTSExtractor + + str.AddReader( + reader, + media, + forma, + func(u unit.Unit) error { + tunit := u.(*unit.H265) + + if tunit.AU == nil { + return nil + } + + if videoDTSExtractor == nil { + if !h265.IsRandomAccess(tunit.AU) { + return nil + } + videoDTSExtractor = &h265.DTSExtractor{} + videoDTSExtractor.Initialize() + } + + dts, err := videoDTSExtractor.Extract(tunit.AU, tunit.PTS) + if err != nil { + return err + } + + nconn.SetWriteDeadline(time.Now().Add(writeTimeout)) + return (*w).WriteH265( + forma, + timestampToDuration(tunit.PTS, forma.ClockRate()), + timestampToDuration(dts, forma.ClockRate()), + tunit.AU) + }) + + tracks = append(tracks, forma) + } + + case *format.H264: + var videoDTSExtractor *h264.DTSExtractor + + str.AddReader( + reader, + media, + forma, + func(u unit.Unit) error { + tunit := u.(*unit.H264) + + if tunit.AU == nil { + return nil + } + + idrPresent := false + nonIDRPresent := false + + for _, nalu := range tunit.AU { + typ := h264.NALUType(nalu[0] & 0x1F) + switch typ { + case h264.NALUTypeIDR: + idrPresent = true + + case h264.NALUTypeNonIDR: + nonIDRPresent = true + } + } + + // wait until we receive an IDR + if videoDTSExtractor == nil { + if !idrPresent { + return nil + } + + videoDTSExtractor = &h264.DTSExtractor{} + videoDTSExtractor.Initialize() + } else if !idrPresent && !nonIDRPresent { + return nil + } + + dts, err := videoDTSExtractor.Extract(tunit.AU, tunit.PTS) + if err != nil { + return err + } + + nconn.SetWriteDeadline(time.Now().Add(writeTimeout)) + return (*w).WriteH264( + forma, + timestampToDuration(tunit.PTS, forma.ClockRate()), + timestampToDuration(dts, forma.ClockRate()), + tunit.AU) + }) + + tracks = append(tracks, forma) + + case *format.Opus: + if slices.Contains(conn.FourCcList, interface{}(fourCCToString(message.FourCCOpus))) { + str.AddReader( + reader, + media, + forma, + func(u unit.Unit) error { + tunit := u.(*unit.Opus) + + if tunit.Packets == nil { + return nil + } + + pts := tunit.PTS + + for _, pkt := range tunit.Packets { + nconn.SetWriteDeadline(time.Now().Add(writeTimeout)) + err := (*w).WriteOpus( + forma, + timestampToDuration(pts, forma.ClockRate()), + pkt, + ) + if err != nil { + return err + } + + pts += opus.PacketDuration2(pkt) + } + + return nil + }) + + tracks = append(tracks, forma) + } + + case *format.MPEG4Audio: + str.AddReader( + reader, + media, + forma, + func(u unit.Unit) error { + tunit := u.(*unit.MPEG4Audio) + + if tunit.AUs == nil { + return nil + } + + for i, au := range tunit.AUs { + pts := tunit.PTS + int64(i)*mpeg4audio.SamplesPerAccessUnit + + nconn.SetWriteDeadline(time.Now().Add(writeTimeout)) + err := (*w).WriteMPEG4Audio( + forma, + timestampToDuration(pts, forma.ClockRate()), + au, + ) + if err != nil { + return err + } + } + + return nil + }) + + tracks = append(tracks, forma) + + case *format.MPEG4AudioLATM: + if !forma.CPresent { + str.AddReader( + reader, + media, + forma, + func(u unit.Unit) error { + tunit := u.(*unit.MPEG4AudioLATM) + + if tunit.Element == nil { + return nil + } + + var ame mpeg4audio.AudioMuxElement + ame.StreamMuxConfig = forma.StreamMuxConfig + err := ame.Unmarshal(tunit.Element) + if err != nil { + return err + } + + nconn.SetWriteDeadline(time.Now().Add(writeTimeout)) + return (*w).WriteMPEG4Audio( + forma, + timestampToDuration(tunit.PTS, forma.ClockRate()), + ame.Payloads[0][0][0], + ) + }) + + tracks = append(tracks, forma) + } + + case *format.MPEG1Audio: + // TODO: check sample rate and layer, + // unfortunately they are not available at this stage. + str.AddReader( + reader, + media, + forma, + func(u unit.Unit) error { + tunit := u.(*unit.MPEG1Audio) + + pts := tunit.PTS + + for _, frame := range tunit.Frames { + var h mpeg1audio.FrameHeader + err := h.Unmarshal(frame) + if err != nil { + return err + } + + if h.MPEG2 || h.Layer != 3 { + return fmt.Errorf("RTMP only supports MPEG-1 layer 3 audio") + } + + nconn.SetWriteDeadline(time.Now().Add(writeTimeout)) + err = (*w).WriteMPEG1Audio( + forma, + timestampToDuration(pts, forma.ClockRate()), + &h, + frame) + if err != nil { + return err + } + + pts += int64(h.SampleCount()) * + int64(forma.ClockRate()) / int64(h.SampleRate) + } + + return nil + }) + + tracks = append(tracks, forma) + + case *format.AC3: + if slices.Contains(conn.FourCcList, interface{}(fourCCToString(message.FourCCAC3))) { + str.AddReader( + reader, + media, + forma, + func(u unit.Unit) error { + tunit := u.(*unit.AC3) + + for i, frame := range tunit.Frames { + pts := tunit.PTS + int64(i)*ac3.SamplesPerFrame + + nconn.SetWriteDeadline(time.Now().Add(writeTimeout)) + err := (*w).WriteAC3( + forma, + timestampToDuration(pts, forma.ClockRate()), + frame) + if err != nil { + return err + } + } + + return nil + }) + + tracks = append(tracks, forma) + } + + case *format.G711: + if forma.SampleRate == 8000 { + str.AddReader( + reader, + media, + forma, + func(u unit.Unit) error { + tunit := u.(*unit.G711) + + if tunit.Samples == nil { + return nil + } + + nconn.SetWriteDeadline(time.Now().Add(writeTimeout)) + return (*w).WriteG711( + forma, + timestampToDuration(tunit.PTS, forma.ClockRate()), + tunit.Samples, + ) + }) + + tracks = append(tracks, forma) + } + + case *format.LPCM: + if (forma.ChannelCount == 1 || forma.ChannelCount == 2) && + (forma.SampleRate == 5512 || + forma.SampleRate == 11025 || + forma.SampleRate == 22050 || + forma.SampleRate == 44100) { + str.AddReader( + reader, + media, + forma, + func(u unit.Unit) error { + tunit := u.(*unit.LPCM) + + if tunit.Samples == nil { + return nil + } + + nconn.SetWriteDeadline(time.Now().Add(writeTimeout)) + return (*w).WriteLPCM( + forma, + timestampToDuration(tunit.PTS, forma.ClockRate()), + tunit.Samples, + ) + }) + + tracks = append(tracks, forma) + } + } + } + } + + if len(tracks) == 0 { return errNoSupportedCodecsFrom } w = &Writer{ - Conn: conn, - VideoTrack: videoFormat, - AudioTrack: audioFormat, + Conn: conn, + Tracks: tracks, } err := w.Initialize() if err != nil { @@ -258,7 +428,7 @@ func FromStream( n := 1 for _, media := range str.Desc.Medias { for _, forma := range media.Formats { - if forma != videoFormat && forma != audioFormat { + if !slices.Contains(tracks, forma) { reader.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 ec097d03..4f46136e 100644 --- a/internal/protocols/rtmp/from_stream_test.go +++ b/internal/protocols/rtmp/from_stream_test.go @@ -1,18 +1,628 @@ package rtmp import ( - "bytes" + "context" "fmt" + "net" + "net/url" "testing" + "time" "github.com/bluenviron/gortsplib/v4/pkg/description" "github.com/bluenviron/gortsplib/v4/pkg/format" + "github.com/bluenviron/mediamtx/internal/formatprocessor" "github.com/bluenviron/mediamtx/internal/logger" "github.com/bluenviron/mediamtx/internal/stream" "github.com/bluenviron/mediamtx/internal/test" + "github.com/bluenviron/mediamtx/internal/unit" "github.com/stretchr/testify/require" ) +func TestFromStream(t *testing.T) { + for _, ca := range []string{ + "h264 + aac", + "av1", + "vp9", + "h265", + "h264", + "opus", + "aac", + "mp3", + "ac-3", + "pcma", + "pcmu", + "lpcm", + "h265 + h264 + vp9 + av1 + opus + aac", + } { + t.Run(ca, func(t *testing.T) { + var medias []*description.Media + + switch ca { + case "h264 + aac": + medias = []*description.Media{ + { + Formats: []format.Format{test.FormatH264}, + }, + { + Formats: []format.Format{test.FormatMPEG4Audio}, + }, + } + + case "av1": + medias = []*description.Media{ + { + Formats: []format.Format{&format.AV1{ + PayloadTyp: 96, + }}, + }, + } + + case "vp9": + medias = []*description.Media{ + { + Formats: []format.Format{&format.VP9{ + PayloadTyp: 96, + }}, + }, + } + + case "h265": + medias = []*description.Media{ + { + Formats: []format.Format{test.FormatH265}, + }, + } + + case "h264": + medias = []*description.Media{ + { + Formats: []format.Format{test.FormatH264}, + }, + } + + case "opus": + medias = []*description.Media{ + { + Formats: []format.Format{&format.Opus{ + PayloadTyp: 96, + ChannelCount: 2, + }}, + }, + } + + case "aac": + medias = []*description.Media{ + { + Formats: []format.Format{test.FormatMPEG4Audio}, + }, + } + + case "mp3": + medias = []*description.Media{ + { + Formats: []format.Format{&format.MPEG1Audio{}}, + }, + } + + case "ac-3": + medias = []*description.Media{ + { + Formats: []format.Format{&format.AC3{ + SampleRate: 44100, + ChannelCount: 2, + }}, + }, + } + + case "pcma": + medias = []*description.Media{ + { + Formats: []format.Format{&format.G711{ + MULaw: false, + SampleRate: 8000, + ChannelCount: 1, + }}, + }, + } + + case "pcmu": + medias = []*description.Media{ + { + Formats: []format.Format{&format.G711{ + MULaw: true, + SampleRate: 8000, + ChannelCount: 1, + }}, + }, + } + + case "lpcm": + medias = []*description.Media{ + { + Formats: []format.Format{&format.LPCM{ + BitDepth: 16, + SampleRate: 44100, + ChannelCount: 2, + }}, + }, + } + + case "h265 + h264 + vp9 + av1 + opus + aac": + medias = []*description.Media{ + { + Formats: []format.Format{&format.H265{}}, + }, + { + Formats: []format.Format{&format.H264{}}, + }, + { + Formats: []format.Format{&format.VP9{}}, + }, + { + Formats: []format.Format{&format.AV1{}}, + }, + { + Formats: []format.Format{&format.Opus{ + PayloadTyp: 96, + ChannelCount: 2, + }}, + }, + { + Formats: []format.Format{&format.MPEG4Audio{ + PayloadTyp: 96, + Config: test.FormatMPEG4Audio.Config, + SizeLength: 13, + IndexLength: 3, + IndexDeltaLength: 3, + }}, + }, + } + } + + strm := &stream.Stream{ + WriteQueueSize: 512, + RTPMaxPayloadSize: 1450, + Desc: &description.Session{Medias: medias}, + GenerateRTPPackets: true, + Parent: test.NilLogger, + } + err := strm.Initialize() + require.NoError(t, err) + + ln, err := net.Listen("tcp", "127.0.0.1:9121") + require.NoError(t, err) + defer ln.Close() + + done := make(chan struct{}) + + go func() { + u, err2 := url.Parse("rtmp://127.0.0.1:9121/stream") + require.NoError(t, err2) + + c := &Client{ + URL: u, + } + err2 = c.Initialize(context.Background()) + require.NoError(t, err2) + + r := &Reader{ + Conn: c, + } + err2 = r.Initialize() + require.NoError(t, err2) + + switch ca { + case "h264 + aac": + require.Equal(t, []format.Format{ + &format.H264{ + SPS: test.FormatH264.SPS, + PPS: test.FormatH264.PPS, + PacketizationMode: 1, + PayloadTyp: 96, + }, + &format.MPEG4Audio{ + PayloadTyp: 96, + Config: test.FormatMPEG4Audio.Config, + SizeLength: 13, + IndexLength: 3, + IndexDeltaLength: 3, + }, + }, r.Tracks()) + + case "av1": + require.Equal(t, []format.Format{ + &format.AV1{ + PayloadTyp: 96, + }, + }, r.Tracks()) + + case "vp9": + require.Equal(t, []format.Format{ + &format.VP9{ + PayloadTyp: 96, + }, + }, r.Tracks()) + + case "h265": + require.Equal(t, []format.Format{ + &format.H265{ + VPS: test.FormatH265.VPS, + SPS: test.FormatH265.SPS, + PPS: test.FormatH265.PPS, + PayloadTyp: 96, + }, + }, r.Tracks()) + + case "h264": + require.Equal(t, []format.Format{ + &format.H264{ + SPS: test.FormatH264.SPS, + PPS: test.FormatH264.PPS, + PacketizationMode: 1, + PayloadTyp: 96, + }, + }, r.Tracks()) + + case "opus": + require.Equal(t, []format.Format{ + &format.Opus{ + PayloadTyp: 96, + ChannelCount: 2, + }, + }, r.Tracks()) + + case "aac": + require.Equal(t, []format.Format{ + &format.MPEG4Audio{ + PayloadTyp: 96, + Config: test.FormatMPEG4Audio.Config, + SizeLength: 13, + IndexLength: 3, + IndexDeltaLength: 3, + }, + }, r.Tracks()) + + case "mp3": + require.Equal(t, []format.Format{ + &format.MPEG1Audio{}, + }, r.Tracks()) + + case "ac-3": + require.Equal(t, []format.Format{ + &format.AC3{ + PayloadTyp: 96, + SampleRate: 48000, + ChannelCount: 1, + }, + }, r.Tracks()) + + case "pcma": + require.Equal(t, []format.Format{ + &format.G711{ + PayloadTyp: 8, + MULaw: false, + ChannelCount: 1, + SampleRate: 8000, + }, + }, r.Tracks()) + + case "pcmu": + require.Equal(t, []format.Format{ + &format.G711{ + MULaw: true, + ChannelCount: 1, + SampleRate: 8000, + }, + }, r.Tracks()) + + case "lpcm": + require.Equal(t, []format.Format{ + &format.LPCM{ + PayloadTyp: 96, + BitDepth: 16, + SampleRate: 44100, + ChannelCount: 2, + }, + }, r.Tracks()) + + case "h265 + h264 + vp9 + av1 + opus + aac": + require.Equal(t, []format.Format{ + &format.H265{ + PayloadTyp: 96, + VPS: formatprocessor.H265DefaultVPS, + SPS: formatprocessor.H265DefaultSPS, + PPS: formatprocessor.H265DefaultPPS, + }, + &format.H264{ + PayloadTyp: 96, + PacketizationMode: 1, + SPS: formatprocessor.H264DefaultSPS, + PPS: formatprocessor.H264DefaultPPS, + }, + &format.VP9{ + PayloadTyp: 96, + }, + &format.AV1{ + PayloadTyp: 96, + }, + &format.Opus{ + PayloadTyp: 96, + ChannelCount: 2, + }, + &format.MPEG4Audio{ + PayloadTyp: 96, + Config: test.FormatMPEG4Audio.Config, + SizeLength: 13, + IndexLength: 3, + IndexDeltaLength: 3, + }, + }, r.Tracks()) + } + + close(done) + }() + + nconn, err := ln.Accept() + require.NoError(t, err) + defer nconn.Close() + + conn := &ServerConn{ + RW: nconn, + } + err = conn.Initialize() + require.NoError(t, err) + + err = conn.Accept() + require.NoError(t, err) + + reader := test.NilLogger + + err = FromStream(strm, reader, conn, nconn, 10*time.Second) + require.NoError(t, err) + defer strm.RemoveReader(reader) + + strm.StartReader(reader) + + switch ca { + case "h264 + aac": + strm.WriteUnit(medias[0], medias[0].Formats[0], &unit.H264{ + Base: unit.Base{ + PTS: 0, + }, + AU: [][]byte{ + {5, 2}, // IDR + }, + }) + + strm.WriteUnit(medias[1], medias[1].Formats[0], &unit.MPEG4Audio{ + Base: unit.Base{ + PTS: 90000 * 5, + }, + AUs: [][]byte{ + {3, 4}, + }, + }) + + case "av1": + for i := range 2 { + strm.WriteUnit(medias[0], medias[0].Formats[0], &unit.AV1{ + Base: unit.Base{ + PTS: 90000 * 2 * int64(i), + }, + TU: [][]byte{{ + 0x0a, 0x0e, 0x00, 0x00, 0x00, 0x4a, 0xab, 0xbf, + 0xc3, 0x77, 0x6b, 0xe4, 0x40, 0x40, 0x40, 0x41, + }}, + }) + } + + case "vp9": + for i := range 2 { + strm.WriteUnit(medias[0], medias[0].Formats[0], &unit.VP9{ + Base: unit.Base{ + PTS: 90000 * 2 * int64(i), + }, + Frame: []byte{1, 2}, + }) + } + + case "h265": + for i := range 2 { + strm.WriteUnit(medias[0], medias[0].Formats[0], &unit.H265{ + Base: unit.Base{ + PTS: 90000 * 2 * int64(i), + }, + AU: [][]byte{ + {0x26, 0x1, 0xaf, 0x8, 0x42, 0x23, 0x48, 0x8a, 0x43, 0xe2}, + }, + }) + } + + case "h264": + for i := range 2 { + strm.WriteUnit(medias[0], medias[0].Formats[0], &unit.H264{ + Base: unit.Base{ + PTS: 90000 * 2 * int64(i), + }, + AU: [][]byte{ + {5, 2}, // IDR + }, + }) + } + + case "opus": + for i := range 2 { + strm.WriteUnit(medias[0], medias[0].Formats[0], &unit.Opus{ + Base: unit.Base{ + PTS: 90000 * 5 * int64(i), + }, + Packets: [][]byte{ + {3, 4}, + }, + }) + } + + case "aac": + for i := range 2 { + strm.WriteUnit(medias[0], medias[0].Formats[0], &unit.MPEG4Audio{ + Base: unit.Base{ + PTS: 90000 * 5 * int64(i), + }, + AUs: [][]byte{ + {3, 4}, + }, + }) + } + + case "mp3": + for i := range 2 { + strm.WriteUnit(medias[0], medias[0].Formats[0], &unit.MPEG1Audio{ + Base: unit.Base{ + PTS: 90000 * 5 * int64(i), + }, + Frames: [][]byte{ + { + 0xff, 0xfa, 0x52, 0x04, 0x00, + }, + }, + }) + } + + case "ac-3": + for i := range 2 { + strm.WriteUnit(medias[0], medias[0].Formats[0], &unit.AC3{ + Base: unit.Base{ + PTS: 90000 * 5 * int64(i), + }, + Frames: [][]byte{ + { + 0x0b, 0x77, 0x47, 0x11, 0x0c, 0x40, 0x2f, 0x84, + 0x2b, 0xc1, 0x07, 0x7a, 0xb0, 0xfa, 0xbb, 0xea, + 0xef, 0x9f, 0x57, 0x7c, 0xf9, 0xf3, 0xf7, 0xcf, + 0x9f, 0x3e, 0x32, 0xfe, 0xd5, 0xc1, 0x50, 0xde, + 0xc5, 0x1e, 0x73, 0xd2, 0x6c, 0xa6, 0x94, 0x46, + 0x4e, 0x92, 0x8c, 0x0f, 0xb9, 0xcf, 0xad, 0x07, + 0x54, 0x4a, 0x2e, 0xf3, 0x7d, 0x07, 0x2e, 0xa4, + 0x2f, 0xba, 0xbf, 0x39, 0xb5, 0xc9, 0x92, 0xa6, + 0xe1, 0xb4, 0x70, 0xc5, 0xc4, 0xb5, 0xe6, 0x5d, + 0x0f, 0xa8, 0x71, 0xa4, 0xcc, 0xc5, 0xbc, 0x75, + 0x67, 0x92, 0x52, 0x4f, 0x7e, 0x62, 0x1c, 0xa9, + 0xd9, 0xb5, 0x19, 0x6a, 0xd7, 0xb0, 0x44, 0x92, + 0x30, 0x3b, 0xf7, 0x61, 0xd6, 0x49, 0x96, 0x66, + 0x98, 0x28, 0x1a, 0x95, 0xa9, 0x42, 0xad, 0xb7, + 0x50, 0x90, 0xad, 0x1c, 0x34, 0x80, 0xe2, 0xef, + 0xcd, 0x41, 0x0b, 0xf0, 0x9d, 0x57, 0x62, 0x78, + 0xfd, 0xc6, 0xc2, 0x19, 0x9e, 0x26, 0x31, 0xca, + 0x1e, 0x75, 0xb1, 0x7a, 0x8e, 0xb5, 0x51, 0x3a, + 0xfe, 0xe4, 0xf1, 0x0b, 0x4f, 0x14, 0x90, 0xdb, + 0x9f, 0x44, 0x50, 0xbb, 0xef, 0x74, 0x00, 0x8c, + 0x1f, 0x97, 0xa1, 0xa2, 0xfa, 0x72, 0x16, 0x47, + 0xc6, 0xc0, 0xe5, 0xfe, 0x67, 0x03, 0x9c, 0xfe, + 0x62, 0x01, 0xa1, 0x00, 0x5d, 0xff, 0xa5, 0x03, + 0x59, 0xfa, 0xa8, 0x25, 0x5f, 0x6b, 0x83, 0x51, + 0xf2, 0xc0, 0x44, 0xff, 0x2d, 0x05, 0x4b, 0xee, + 0xe0, 0x54, 0x9e, 0xae, 0x86, 0x45, 0xf3, 0xbd, + 0x0e, 0x42, 0xf2, 0xbf, 0x0f, 0x7f, 0xc6, 0x09, + 0x07, 0xdc, 0x22, 0x11, 0x77, 0xbe, 0x31, 0x27, + 0x5b, 0xa4, 0x13, 0x47, 0x07, 0x32, 0x9f, 0x1f, + 0xcb, 0xb0, 0xdf, 0x3e, 0x7d, 0x0d, 0xf3, 0xe7, + 0xcf, 0x9f, 0x3e, 0xae, 0xf9, 0xf3, 0xe7, 0xcf, + 0x9f, 0x3e, 0x85, 0x5d, 0xf3, 0xe7, 0xcf, 0x9f, + 0x3e, 0x7c, 0xf9, 0xf3, 0xe7, 0xcf, 0x9f, 0x3f, + 0x53, 0x5d, 0xf3, 0xe7, 0xcf, 0x9f, 0x3e, 0x7c, + 0xf9, 0xf3, 0xe7, 0xcf, 0x9f, 0x3e, 0x7c, 0xf9, + 0xf3, 0xe7, 0xcf, 0x9f, 0x3e, 0x7c, 0xf9, 0xf3, + 0xe7, 0xcf, 0x9f, 0x3e, 0x00, 0x46, 0x28, 0x26, + 0x20, 0x4a, 0x5a, 0xc0, 0x8a, 0xc5, 0xae, 0xa0, + 0x55, 0x78, 0x82, 0x7a, 0x38, 0x10, 0x09, 0xc9, + 0xb8, 0x0c, 0xfa, 0x5b, 0xc9, 0xd2, 0xec, 0x44, + 0x25, 0xf8, 0x20, 0xf2, 0xc8, 0x8a, 0xe9, 0x40, + 0x18, 0x06, 0xc6, 0x2b, 0xc8, 0xed, 0x8f, 0x33, + 0x09, 0x92, 0x28, 0x1e, 0xc4, 0x24, 0xd8, 0x33, + 0xa5, 0x00, 0xf5, 0xea, 0x18, 0xfa, 0x90, 0x97, + 0x97, 0xe8, 0x39, 0x6a, 0xcf, 0xf1, 0xdd, 0xff, + 0x9e, 0x8e, 0x04, 0x02, 0xae, 0x65, 0x87, 0x5c, + 0x4e, 0x72, 0xfd, 0x3c, 0x01, 0x86, 0xfe, 0x56, + 0x59, 0x74, 0x44, 0x3a, 0x40, 0x00, 0xec, 0xfc, + }, + }, + }) + } + + case "pcma", "pcmu": + for i := range 2 { + strm.WriteUnit(medias[0], medias[0].Formats[0], &unit.G711{ + Base: unit.Base{ + PTS: 90000 * 5 * int64(i), + }, + Samples: []byte{ + 3, 4, + }, + }) + } + + case "lpcm": + for i := range 2 { + strm.WriteUnit(medias[0], medias[0].Formats[0], &unit.LPCM{ + Base: unit.Base{ + PTS: 90000 * 5 * int64(i), + }, + Samples: []byte{ + 3, 4, 5, 6, + }, + }) + } + + case "h265 + h264 + vp9 + av1 + opus + aac": + strm.WriteUnit(medias[0], medias[0].Formats[0], &unit.H265{ + AU: [][]byte{ + formatprocessor.H265DefaultVPS, + formatprocessor.H265DefaultSPS, + formatprocessor.H265DefaultPPS, + {0x26, 0x1, 0xaf, 0x8, 0x42, 0x23, 0x48, 0x8a, 0x43, 0xe2}, + }, + }) + + strm.WriteUnit(medias[1], medias[1].Formats[0], &unit.H264{ + AU: [][]byte{ + formatprocessor.H264DefaultSPS, + formatprocessor.H264DefaultPPS, + {5, 2}, // IDR + }, + }) + + strm.WriteUnit(medias[2], medias[2].Formats[0], &unit.VP9{ + Frame: []byte{1, 2}, + }) + + strm.WriteUnit(medias[3], medias[3].Formats[0], &unit.AV1{ + TU: [][]byte{{ + 0x0a, 0x0e, 0x00, 0x00, 0x00, 0x4a, 0xab, 0xbf, + 0xc3, 0x77, 0x6b, 0xe4, 0x40, 0x40, 0x40, 0x41, + }}, + }) + + strm.WriteUnit(medias[4], medias[4].Formats[0], &unit.Opus{ + Packets: [][]byte{ + {3, 4}, + }, + }) + + strm.WriteUnit(medias[5], medias[5].Formats[0], &unit.MPEG4Audio{ + Base: unit.Base{ + PTS: 90000 * 5, + }, + AUs: [][]byte{ + {3, 4}, + }, + }) + } + + <-done + }) + } +} + func TestFromStreamNoSupportedCodecs(t *testing.T) { strm := &stream.Stream{ WriteQueueSize: 512, @@ -48,10 +658,6 @@ func TestFromStreamSkipUnsupportedTracks(t *testing.T) { Type: description.MediaTypeVideo, Formats: []format.Format{&format.H264{}}, }, - { - Type: description.MediaTypeVideo, - Formats: []format.Format{&format.H264{}}, - }, }}, GenerateRTPPackets: true, Parent: test.NilLogger, @@ -63,24 +669,43 @@ func TestFromStreamSkipUnsupportedTracks(t *testing.T) { l := test.Logger(func(l logger.Level, format string, args ...interface{}) { require.Equal(t, logger.Warn, l) - switch n { - case 0: + if n == 0 { require.Equal(t, "skipping track 1 (VP8)", fmt.Sprintf(format, args...)) - case 1: - require.Equal(t, "skipping track 3 (H264)", fmt.Sprintf(format, args...)) } n++ }) - var buf bytes.Buffer - c := &dummyConn{ - rw: &buf, - } - c.initialize() + ln, err := net.Listen("tcp", "127.0.0.1:9121") + require.NoError(t, err) + defer ln.Close() - err = FromStream(strm, l, c, nil, 0) + go func() { + u, err2 := url.Parse("rtmp://127.0.0.1:9121/stream") + require.NoError(t, err2) + + c := &Client{ + URL: u, + } + err2 = c.Initialize(context.Background()) + require.NoError(t, err2) + }() + + nconn, err := ln.Accept() + require.NoError(t, err) + defer nconn.Close() + + conn := &ServerConn{ + RW: nconn, + } + err = conn.Initialize() + require.NoError(t, err) + + err = conn.Accept() + require.NoError(t, err) + + err = FromStream(strm, l, conn, nil, 0) require.NoError(t, err) defer strm.RemoveReader(l) - require.Equal(t, 2, n) + require.Equal(t, 1, n) } diff --git a/internal/protocols/rtmp/message/msg_audio.go b/internal/protocols/rtmp/message/msg_audio.go index 3adaafb5..f36dff69 100644 --- a/internal/protocols/rtmp/message/msg_audio.go +++ b/internal/protocols/rtmp/message/msg_audio.go @@ -97,13 +97,10 @@ func (m *Audio) unmarshal(raw *rawmessage.Message) error { } func (m Audio) marshalBodySize() int { - var l int - if m.Codec == CodecMPEG1Audio { - l = 1 + len(m.Payload) - } else { - l = 2 + len(m.Payload) + if m.Codec == CodecMPEG4Audio { + return 2 + len(m.Payload) } - return l + return 1 + len(m.Payload) } func (m Audio) marshal() (*rawmessage.Message, error) { diff --git a/internal/protocols/rtmp/rawmessage/reader_test.go b/internal/protocols/rtmp/rawmessage/reader_test.go index 1cd2779c..8247d369 100644 --- a/internal/protocols/rtmp/rawmessage/reader_test.go +++ b/internal/protocols/rtmp/rawmessage/reader_test.go @@ -200,6 +200,67 @@ var cases = []struct { }, }, }, + { + "decreasing timestamp", + []*Message{ + { + ChunkStreamID: 27, + Timestamp: 16 * time.Second, + Type: 6, + MessageStreamID: 3123, + Body: []byte{1, 2}, + }, + { + ChunkStreamID: 27, + Timestamp: 17 * time.Second, + Type: 6, + MessageStreamID: 3123, + Body: []byte{3, 4}, + }, + { + ChunkStreamID: 27, + Timestamp: 16 * time.Second, + Type: 6, + MessageStreamID: 3123, + Body: []byte{5, 6}, + }, + { + ChunkStreamID: 27, + Timestamp: 17 * time.Second, + Type: 6, + MessageStreamID: 3123, + Body: []byte{7, 8}, + }, + }, + []chunk.Chunk{ + &chunk.Chunk0{ + ChunkStreamID: 27, + Timestamp: 16000, + Type: 6, + MessageStreamID: 3123, + BodyLen: 2, + Body: []byte{1, 2}, + }, + &chunk.Chunk2{ + ChunkStreamID: 27, + TimestampDelta: 1000, + Body: []byte{3, 4}, + }, + &chunk.Chunk0{ + ChunkStreamID: 27, + Timestamp: 16000, + Type: 6, + MessageStreamID: 3123, + BodyLen: 2, + Body: []byte{5, 6}, + }, + &chunk.Chunk2{ + ChunkStreamID: 27, + TimestampDelta: 1000, + Body: []byte{7, 8}, + }, + }, + }, } func TestReader(t *testing.T) { diff --git a/internal/protocols/rtmp/rawmessage/writer.go b/internal/protocols/rtmp/rawmessage/writer.go index ae3f466a..f9c88844 100644 --- a/internal/protocols/rtmp/rawmessage/writer.go +++ b/internal/protocols/rtmp/rawmessage/writer.go @@ -119,20 +119,6 @@ func (wc *writerChunkStream) writeMessage(msg *Message) error { return err } } - - v1 := msg.MessageStreamID - wc.lastMessageStreamID = &v1 - v2 := msg.Type - wc.lastType = &v2 - v3 := bodyLen - wc.lastBodyLen = &v3 - v4 := timestamp - wc.lastTimestamp = &v4 - - if timestampDelta != nil { - v5 := *timestampDelta - wc.lastTimestampDelta = &v5 - } } else { err := wc.writeChunk(&chunk.Chunk3{ ChunkStreamID: msg.ChunkStreamID, @@ -146,9 +132,21 @@ func (wc *writerChunkStream) writeMessage(msg *Message) error { pos += chunkBodyLen if (bodyLen - pos) == 0 { - return wc.mw.bw.Flush() + break } } + + v1 := msg.MessageStreamID + wc.lastMessageStreamID = &v1 + v2 := msg.Type + wc.lastType = &v2 + v3 := bodyLen + wc.lastBodyLen = &v3 + v4 := timestamp + wc.lastTimestamp = &v4 + wc.lastTimestampDelta = timestampDelta + + return wc.mw.bw.Flush() } // Writer is a raw message writer. diff --git a/internal/protocols/rtmp/rawmessage/writer_test.go b/internal/protocols/rtmp/rawmessage/writer_test.go index ff5b7ea1..adb746d0 100644 --- a/internal/protocols/rtmp/rawmessage/writer_test.go +++ b/internal/protocols/rtmp/rawmessage/writer_test.go @@ -60,6 +60,8 @@ func TestWriter(t *testing.T) { require.Equal(t, cach, ch) hasExtendedTimestamp = chunkHasExtendedTimestamp(cach) } + + require.Zero(t, buf.Len()) }) } } diff --git a/internal/protocols/rtmp/reader_test.go b/internal/protocols/rtmp/reader_test.go index 20ae75cb..10f33b16 100644 --- a/internal/protocols/rtmp/reader_test.go +++ b/internal/protocols/rtmp/reader_test.go @@ -2,6 +2,7 @@ package rtmp import ( "bytes" + "io" "testing" "time" @@ -18,59 +19,41 @@ import ( "github.com/bluenviron/mediamtx/internal/test" ) +type dummyConn struct { + rw io.ReadWriter + + bc *bytecounter.ReadWriter + mrw *message.ReadWriter +} + +func (c *dummyConn) initialize() { + c.bc = bytecounter.NewReadWriter(c.rw) + c.mrw = message.NewReadWriter(c.bc, c.bc, false) +} + +// BytesReceived returns the number of bytes received. +func (c *dummyConn) BytesReceived() uint64 { + return c.bc.Reader.Count() +} + +// BytesSent returns the number of bytes sent. +func (c *dummyConn) BytesSent() uint64 { + return c.bc.Writer.Count() +} + +func (c *dummyConn) Read() (message.Message, error) { + return c.mrw.Read() +} + +func (c *dummyConn) Write(msg message.Message) error { + return c.mrw.Write(msg) +} + func TestReadTracks(t *testing.T) { var spsp h265.SPS err := spsp.Unmarshal(test.FormatH265.SPS) require.NoError(t, err) - hvcc := &mp4.HvcC{ - ConfigurationVersion: 1, - GeneralProfileIdc: spsp.ProfileTierLevel.GeneralProfileIdc, - GeneralProfileCompatibility: spsp.ProfileTierLevel.GeneralProfileCompatibilityFlag, - GeneralConstraintIndicator: [6]uint8{ - test.FormatH265.SPS[7], test.FormatH265.SPS[8], test.FormatH265.SPS[9], - test.FormatH265.SPS[10], test.FormatH265.SPS[11], test.FormatH265.SPS[12], - }, - GeneralLevelIdc: spsp.ProfileTierLevel.GeneralLevelIdc, - // MinSpatialSegmentationIdc - // ParallelismType - ChromaFormatIdc: uint8(spsp.ChromaFormatIdc), - BitDepthLumaMinus8: uint8(spsp.BitDepthLumaMinus8), - BitDepthChromaMinus8: uint8(spsp.BitDepthChromaMinus8), - // AvgFrameRate - // ConstantFrameRate - NumTemporalLayers: 1, - // TemporalIdNested - LengthSizeMinusOne: 3, - NumOfNaluArrays: 3, - NaluArrays: []mp4.HEVCNaluArray{ - { - NaluType: byte(h265.NALUType_VPS_NUT), - NumNalus: 1, - Nalus: []mp4.HEVCNalu{{ - Length: uint16(len(test.FormatH265.VPS)), - NALUnit: test.FormatH265.VPS, - }}, - }, - { - NaluType: byte(h265.NALUType_SPS_NUT), - NumNalus: 1, - Nalus: []mp4.HEVCNalu{{ - Length: uint16(len(test.FormatH265.SPS)), - NALUnit: test.FormatH265.SPS, - }}, - }, - { - NaluType: byte(h265.NALUType_PPS_NUT), - NumNalus: 1, - Nalus: []mp4.HEVCNalu{{ - Length: uint16(len(test.FormatH265.PPS)), - NALUnit: test.FormatH265.PPS, - }}, - }, - }, - } - for _, ca := range []struct { name string tracks []format.Format @@ -521,7 +504,7 @@ func TestReadTracks(t *testing.T) { ChunkStreamID: 4, MessageStreamID: 0x1000000, FourCC: message.FourCCHEVC, - HEVCHeader: hvcc, + HEVCHeader: generateHvcC(test.FormatH265.VPS, test.FormatH265.SPS, test.FormatH265.PPS), }, &message.VideoExCodedFrames{ ChunkStreamID: 4, @@ -577,7 +560,7 @@ func TestReadTracks(t *testing.T) { ChunkStreamID: 4, MessageStreamID: 0x1000000, FourCC: message.FourCCHEVC, - HEVCHeader: hvcc, + HEVCHeader: generateHvcC(test.FormatH265.VPS, test.FormatH265.SPS, test.FormatH265.PPS), }, &message.VideoExCodedFrames{ ChunkStreamID: 6, diff --git a/internal/protocols/rtmp/server_conn.go b/internal/protocols/rtmp/server_conn.go index 72cb7c22..b5307e5d 100644 --- a/internal/protocols/rtmp/server_conn.go +++ b/internal/protocols/rtmp/server_conn.go @@ -110,10 +110,11 @@ type ServerConn struct { RW io.ReadWriter // filled by Initialize - connectCmd *message.CommandAMF0 - connectObject amf0.Object - app string - tcURL string + connectChunkStreamID byte + connectCommandID int + app string + tcURL string + FourCcList amf0.StrictArray // filled by Accept URL *url.URL @@ -144,40 +145,47 @@ func (c *ServerConn) Initialize() error { c.mrw = message.NewReadWriter(rw, c.bc, false) - c.connectCmd, err = readCommand(c.mrw) + connectCmd, err := readCommand(c.mrw) if err != nil { return err } - if c.connectCmd.Name != "connect" { - return fmt.Errorf("unexpected command: %+v", c.connectCmd) + if connectCmd.Name != "connect" { + return fmt.Errorf("unexpected command: %+v", connectCmd) } - if len(c.connectCmd.Arguments) < 1 { - return fmt.Errorf("invalid connect command: %+v", c.connectCmd) + if len(connectCmd.Arguments) < 1 { + return fmt.Errorf("invalid connect command: %+v", connectCmd) } - var ok bool - c.connectObject, ok = objectOrArray(c.connectCmd.Arguments[0]) + c.connectChunkStreamID = connectCmd.ChunkStreamID + c.connectCommandID = connectCmd.CommandID + + connectObject, ok := objectOrArray(connectCmd.Arguments[0]) if !ok { - return fmt.Errorf("invalid connect command: %+v", c.connectCmd) + return fmt.Errorf("invalid connect command: %+v", connectCmd) } - c.app, ok = c.connectObject.GetString("app") + c.app, ok = connectObject.GetString("app") if !ok { - return fmt.Errorf("invalid connect command: %+v", c.connectCmd) + return fmt.Errorf("invalid connect command: %+v", connectCmd) } - c.tcURL, ok = c.connectObject.GetString("tcUrl") + c.tcURL, ok = connectObject.GetString("tcUrl") if !ok { - c.tcURL, ok = c.connectObject.GetString("tcurl") + c.tcURL, ok = connectObject.GetString("tcurl") if !ok { - return fmt.Errorf("invalid connect command: %+v", c.connectCmd) + return fmt.Errorf("invalid connect command: %+v", connectCmd) } } - c.tcURL = strings.Trim(c.tcURL, "'") + if raw, ok2 := connectObject.Get("fourCcList"); ok2 { + if arr, ok3 := raw.(amf0.StrictArray); ok3 { + c.FourCcList = arr + } + } + return nil } @@ -186,9 +194,9 @@ func (c *ServerConn) CheckCredentials(expectedUser string, expectedPass string) i := strings.Index(c.app, "?authmod=adobe") if i < 0 { err := c.mrw.Write(&message.CommandAMF0{ - ChunkStreamID: c.connectCmd.ChunkStreamID, + ChunkStreamID: c.connectChunkStreamID, Name: "_error", - CommandID: c.connectCmd.CommandID, + CommandID: c.connectCommandID, Arguments: []interface{}{ nil, amf0.Object{ @@ -218,9 +226,9 @@ func (c *ServerConn) CheckCredentials(expectedUser string, expectedPass string) if clientChallenge == "" || response == "" { err := c.mrw.Write(&message.CommandAMF0{ - ChunkStreamID: c.connectCmd.ChunkStreamID, + ChunkStreamID: c.connectChunkStreamID, Name: "_error", - CommandID: c.connectCmd.CommandID, + CommandID: c.connectCommandID, Arguments: []interface{}{ nil, amf0.Object{ @@ -244,9 +252,9 @@ func (c *ServerConn) CheckCredentials(expectedUser string, expectedPass string) expectedResponse := authResponse(expectedUser, expectedPass, serverSalt, "", serverChallenge, clientChallenge) if expectedResponse != response { err := c.mrw.Write(&message.CommandAMF0{ - ChunkStreamID: c.connectCmd.ChunkStreamID, + ChunkStreamID: c.connectChunkStreamID, Name: "_error", - CommandID: c.connectCmd.CommandID, + CommandID: c.connectCommandID, Arguments: []interface{}{ nil, amf0.Object{ @@ -301,12 +309,10 @@ func (c *ServerConn) Accept() error { return err } - oe, _ := c.connectObject.GetFloat64("objectEncoding") - err = c.mrw.Write(&message.CommandAMF0{ - ChunkStreamID: c.connectCmd.ChunkStreamID, + ChunkStreamID: c.connectChunkStreamID, Name: "_result", - CommandID: c.connectCmd.CommandID, + CommandID: c.connectCommandID, Arguments: []interface{}{ amf0.Object{ {Key: "fmsVer", Value: "LNX 9,0,124,2"}, @@ -316,7 +322,7 @@ func (c *ServerConn) Accept() error { {Key: "level", Value: "status"}, {Key: "code", Value: "NetConnection.Connect.Success"}, {Key: "description", Value: "Connection succeeded."}, - {Key: "objectEncoding", Value: oe}, + {Key: "objectEncoding", Value: float64(encodingAMF0)}, }, }, }) diff --git a/internal/protocols/rtmp/server_conn_test.go b/internal/protocols/rtmp/server_conn_test.go index 4628cb73..1cf2b5cf 100644 --- a/internal/protocols/rtmp/server_conn_test.go +++ b/internal/protocols/rtmp/server_conn_test.go @@ -727,3 +727,65 @@ func TestServerConnPath(t *testing.T) { }) } } + +func TestServerConnFourCcList(t *testing.T) { + ln, err := net.Listen("tcp", "127.0.0.1:9121") + require.NoError(t, err) + defer ln.Close() + + done := make(chan struct{}) + + go func() { + defer close(done) + + nconn, err2 := ln.Accept() + require.NoError(t, err2) + defer nconn.Close() + + conn := &ServerConn{ + RW: nconn, + } + err2 = conn.Initialize() + require.NoError(t, err2) + + require.Equal(t, amf0.StrictArray{ + "av01", + "Avc1", + }, conn.FourCcList) + }() + + conn, err := net.Dial("tcp", "127.0.0.1:9121") + require.NoError(t, err) + defer conn.Close() + bc := bytecounter.NewReadWriter(conn) + + _, _, err = handshake.DoClient(bc, false, false) + require.NoError(t, err) + + mrw := message.NewReadWriter(bc, bc, true) + + err = mrw.Write(&message.CommandAMF0{ + ChunkStreamID: 3, + Name: "connect", + CommandID: 1, + Arguments: []interface{}{ + amf0.Object{ + {Key: "app", Value: "stream?key=val"}, + {Key: "flashVer", Value: "LNX 9,0,124,2"}, + {Key: "tcUrl", Value: "rtmp://127.0.0.1:9121/stream?key=val"}, + {Key: "fpad", Value: false}, + {Key: "capabilities", Value: float64(15)}, + {Key: "audioCodecs", Value: float64(4071)}, + {Key: "videoCodecs", Value: float64(252)}, + {Key: "videoFunction", Value: float64(1)}, + {Key: "fourCcList", Value: amf0.StrictArray{ + "av01", + "Avc1", + }}, + }, + }, + }) + require.NoError(t, err) + + <-done +} diff --git a/internal/protocols/rtmp/writer.go b/internal/protocols/rtmp/writer.go index efecb7b6..657fd594 100644 --- a/internal/protocols/rtmp/writer.go +++ b/internal/protocols/rtmp/writer.go @@ -1,18 +1,155 @@ package rtmp import ( + "fmt" + "strconv" "time" + "github.com/abema/go-mp4" "github.com/bluenviron/gortsplib/v4/pkg/format" + "github.com/bluenviron/mediacommon/v2/pkg/codecs/av1" "github.com/bluenviron/mediacommon/v2/pkg/codecs/h264" + "github.com/bluenviron/mediacommon/v2/pkg/codecs/h265" "github.com/bluenviron/mediacommon/v2/pkg/codecs/mpeg1audio" - "github.com/bluenviron/mediacommon/v2/pkg/codecs/mpeg4audio" + "github.com/bluenviron/mediamtx/internal/formatprocessor" "github.com/bluenviron/mediamtx/internal/protocols/rtmp/amf0" "github.com/bluenviron/mediamtx/internal/protocols/rtmp/h264conf" "github.com/bluenviron/mediamtx/internal/protocols/rtmp/message" ) +func codecID(track format.Format) float64 { + switch track := track.(type) { + // video + + case *format.AV1: + return float64(message.FourCCAV1) + + case *format.VP9: + return float64(message.FourCCVP9) + + case *format.H265: + return float64(message.FourCCHEVC) + + case *format.H264: + return message.CodecH264 + + // audio + + case *format.Opus: + return float64(message.FourCCOpus) + + case *format.MPEG4Audio, *format.MPEG4AudioLATM: + return message.CodecMPEG4Audio + + case *format.MPEG1Audio: + return message.CodecMPEG1Audio + + case *format.AC3: + return float64(message.FourCCAC3) + + case *format.G711: + if track.MULaw { + return message.CodecPCMU + } + return message.CodecPCMA + + case *format.LPCM: + return message.CodecLPCM + } + + return 0 +} + +func generateHvcC(vps, sps, pps []byte) *mp4.HvcC { + var psps h265.SPS + err := psps.Unmarshal(sps) + if err != nil { + panic(err) + } + + return &mp4.HvcC{ + ConfigurationVersion: 1, + GeneralProfileIdc: psps.ProfileTierLevel.GeneralProfileIdc, + GeneralProfileCompatibility: psps.ProfileTierLevel.GeneralProfileCompatibilityFlag, + GeneralConstraintIndicator: [6]uint8{ + sps[7], sps[8], sps[9], + sps[10], sps[11], sps[12], + }, + GeneralLevelIdc: psps.ProfileTierLevel.GeneralLevelIdc, + // MinSpatialSegmentationIdc + // ParallelismType + ChromaFormatIdc: uint8(psps.ChromaFormatIdc), + BitDepthLumaMinus8: uint8(psps.BitDepthLumaMinus8), + BitDepthChromaMinus8: uint8(psps.BitDepthChromaMinus8), + // AvgFrameRate + // ConstantFrameRate + NumTemporalLayers: 1, + // TemporalIdNested + LengthSizeMinusOne: 3, + NumOfNaluArrays: 3, + NaluArrays: []mp4.HEVCNaluArray{ + { + NaluType: byte(h265.NALUType_VPS_NUT), + NumNalus: 1, + Nalus: []mp4.HEVCNalu{{ + Length: uint16(len(vps)), + NALUnit: vps, + }}, + }, + { + NaluType: byte(h265.NALUType_SPS_NUT), + NumNalus: 1, + Nalus: []mp4.HEVCNalu{{ + Length: uint16(len(sps)), + NALUnit: sps, + }}, + }, + { + NaluType: byte(h265.NALUType_PPS_NUT), + NumNalus: 1, + Nalus: []mp4.HEVCNalu{{ + Length: uint16(len(pps)), + NALUnit: pps, + }}, + }, + }, + } +} + +func generateAvcC(sps, pps []byte) *mp4.AVCDecoderConfiguration { + var psps h264.SPS + err := psps.Unmarshal(sps) + if err != nil { + panic(err) + } + + return &mp4.AVCDecoderConfiguration{ // + AnyTypeBox: mp4.AnyTypeBox{ + Type: mp4.BoxTypeAvcC(), + }, + ConfigurationVersion: 1, + Profile: psps.ProfileIdc, + ProfileCompatibility: sps[2], + Level: psps.LevelIdc, + LengthSizeMinusOne: 3, + NumOfSequenceParameterSets: 1, + SequenceParameterSets: []mp4.AVCParameterSet{ + { + Length: uint16(len(sps)), + NALUnit: sps, + }, + }, + NumOfPictureParameterSets: 1, + PictureParameterSets: []mp4.AVCParameterSet{ + { + Length: uint16(len(pps)), + NALUnit: pps, + }, + }, + } +} + func audioRateRTMPToInt(v uint8) int { switch v { case message.Rate5512: @@ -26,17 +163,19 @@ func audioRateRTMPToInt(v uint8) int { } } -func audioRateIntToRTMP(v int) uint8 { +func audioRateIntToRTMP(v int) (uint8, bool) { switch v { case 5512: - return message.Rate5512 + return message.Rate5512, true case 11025: - return message.Rate11025 + return message.Rate11025, true case 22050: - return message.Rate22050 - default: - return message.Rate44100 + return message.Rate22050, true + case 44100: + return message.Rate44100, true } + + return 0, false } func mpeg1AudioChannels(m mpeg1audio.ChannelMode) bool { @@ -45,9 +184,11 @@ func mpeg1AudioChannels(m mpeg1audio.ChannelMode) bool { // Writer provides functions to write outgoing data. type Writer struct { - Conn Conn - VideoTrack format.Format - AudioTrack format.Format + Conn Conn + Tracks []format.Format + + videoTrackToID map[format.Format]uint8 + audioTrackToID map[format.Format]uint8 } // Initialize initializes Writer. @@ -61,155 +202,728 @@ func (w *Writer) Initialize() error { } func (w *Writer) writeTracks() error { + w.videoTrackToID = make(map[format.Format]uint8) + w.audioTrackToID = make(map[format.Format]uint8) + + var videoTracks []format.Format + var audioTracks []format.Format + + for _, track := range w.Tracks { + switch track.(type) { + case *format.AV1, *format.VP9, *format.H265, *format.H264: + w.videoTrackToID[track] = uint8(len(videoTracks)) + videoTracks = append(videoTracks, track) + + case *format.Opus, *format.MPEG4Audio, *format.MPEG4AudioLATM, + *format.MPEG1Audio, *format.AC3, *format.G711, *format.LPCM: + w.audioTrackToID[track] = uint8(len(audioTracks)) + audioTracks = append(audioTracks, track) + + default: + return fmt.Errorf("unsupported track: %T", track) + } + } + + metadata := amf0.Object{ + { + Key: "videocodecid", + Value: func() float64 { + if len(videoTracks) != 0 { + return codecID(videoTracks[0]) + } + return 0 + }(), + }, + { + Key: "videodatarate", + Value: float64(0), + }, + { + Key: "audiocodecid", + Value: func() float64 { + if len(audioTracks) != 0 { + return codecID(audioTracks[0]) + } + return 0 + }(), + }, + { + Key: "audiodatarate", + Value: float64(0), + }, + } + + if len(videoTracks) > 1 { + var val amf0.Object + + for id, track := range videoTracks[1:] { + val = append(val, amf0.ObjectEntry{ + Key: strconv.FormatInt(int64(id+2), 10), + Value: amf0.Object{ + { + Key: "videocodecid", + Value: codecID(track), + }, + { + Key: "videodatarate", + Value: float64(0), + }, + }, + }) + } + + metadata = append(metadata, amf0.ObjectEntry{ + Key: "videoTrackIdInfoMap", + Value: val, + }) + } + + if len(audioTracks) > 1 { + var val amf0.Object + + for id, track := range audioTracks[1:] { + val = append(val, amf0.ObjectEntry{ + Key: strconv.FormatInt(int64(id+2), 10), + Value: amf0.Object{ + { + Key: "audiocodecid", + Value: codecID(track), + }, + { + Key: "audiodatarate", + Value: float64(0), + }, + }, + }) + } + + metadata = append(metadata, amf0.ObjectEntry{ + Key: "audioTrackIdInfoMap", + Value: val, + }) + } + err := w.Conn.Write(&message.DataAMF0{ ChunkStreamID: 4, MessageStreamID: 0x1000000, Payload: []interface{}{ "@setDataFrame", "onMetaData", - amf0.Object{ - { - Key: "videodatarate", - Value: float64(0), - }, - { - Key: "videocodecid", - Value: func() float64 { - switch w.VideoTrack.(type) { - case *format.H264: - return message.CodecH264 - - default: - return 0 - } - }(), - }, - { - Key: "audiodatarate", - Value: float64(0), - }, - { - Key: "audiocodecid", - Value: func() float64 { - switch w.AudioTrack.(type) { - case *format.MPEG1Audio: - return message.CodecMPEG1Audio - - case *format.MPEG4Audio, *format.MPEG4AudioLATM: - return message.CodecMPEG4Audio - - default: - return 0 - } - }(), - }, - }, + metadata, }, }) if err != nil { return err } - if videoTrack, ok := w.VideoTrack.(*format.H264); ok { - // write decoder config only if SPS and PPS are available. - // if they're not available yet, they're sent later. - if sps, pps := videoTrack.SafeParams(); sps != nil && pps != nil { - buf, _ := h264conf.Conf{ - SPS: sps, - PPS: pps, - }.Marshal() - - err = w.Conn.Write(&message.Video{ + for id, track := range videoTracks { + switch track := track.(type) { + case *format.AV1: + // TODO: fill properly. + // unfortunately, AV1 config is not available at this stage. + var msg message.Message = &message.VideoExSequenceStart{ ChunkStreamID: message.VideoChunkStreamID, MessageStreamID: 0x1000000, - Codec: message.CodecH264, - IsKeyFrame: true, - Type: message.VideoTypeConfig, - Payload: buf, - }) + FourCC: message.FourCCAV1, + AV1Header: &mp4.Av1C{ + Marker: 0x1, + Version: 0x1, + SeqLevelIdx0: 0x8, + ChromaSubsamplingX: 0x1, + ChromaSubsamplingY: 0x1, + ConfigOBUs: []uint8{0xa, 0xb, 0x0, 0x0, 0x0, 0x42, 0xab, 0xbf, 0xc3, 0x70, 0xb, 0xe0, 0x1}, + }, + } + + if id != 0 { + msg = &message.VideoExMultitrack{ + MultitrackType: 0x0, + TrackID: uint8(id), + Wrapped: msg, + } + } + + err = w.Conn.Write(msg) if err != nil { return err } + + case *format.VP9: + // TODO: fill properly. + // unfortunately, VP9 config is not available at this stage. + var msg message.Message = &message.VideoExSequenceStart{ + ChunkStreamID: message.VideoChunkStreamID, + MessageStreamID: 0x1000000, + FourCC: message.FourCCVP9, + VP9Header: &mp4.VpcC{ + FullBox: mp4.FullBox{Version: 0x1}, + Level: 0x28, + BitDepth: 0x8, + ChromaSubsampling: 0x1, + ColourPrimaries: 0x2, + TransferCharacteristics: 0x2, + MatrixCoefficients: 0x2, + CodecInitializationData: []uint8{}, + }, + } + + if id != 0 { + msg = &message.VideoExMultitrack{ + MultitrackType: 0x0, + TrackID: uint8(id), + Wrapped: msg, + } + } + + err = w.Conn.Write(msg) + if err != nil { + return err + } + + case *format.H265: + vps, sps, pps := track.SafeParams() + if vps == nil || sps == nil || pps == nil { + vps = formatprocessor.H265DefaultVPS + sps = formatprocessor.H265DefaultSPS + pps = formatprocessor.H265DefaultPPS + } + + var msg message.Message = &message.VideoExSequenceStart{ + ChunkStreamID: message.VideoChunkStreamID, + MessageStreamID: 0x1000000, + FourCC: message.FourCCHEVC, + HEVCHeader: generateHvcC(vps, sps, pps), + } + + if id != 0 { + msg = &message.VideoExMultitrack{ + MultitrackType: 0x0, + TrackID: uint8(id), + Wrapped: msg, + } + } + + err = w.Conn.Write(msg) + if err != nil { + return err + } + + case *format.H264: + sps, pps := track.SafeParams() + if sps == nil || pps == nil { + sps = formatprocessor.H264DefaultSPS + pps = formatprocessor.H264DefaultPPS + } + + if id == 0 { + buf, _ := h264conf.Conf{ + SPS: sps, + PPS: pps, + }.Marshal() + + err = w.Conn.Write(&message.Video{ + ChunkStreamID: message.VideoChunkStreamID, + MessageStreamID: 0x1000000, + Codec: message.CodecH264, + IsKeyFrame: true, + Type: message.VideoTypeConfig, + Payload: buf, + }) + if err != nil { + return err + } + } else { + err = w.Conn.Write(&message.VideoExMultitrack{ + MultitrackType: 0x0, + TrackID: uint8(id), + Wrapped: &message.VideoExSequenceStart{ + ChunkStreamID: message.VideoChunkStreamID, + MessageStreamID: 0x1000000, + FourCC: message.FourCCAVC, + AVCHeader: generateAvcC(sps, pps), + }, + }) + if err != nil { + return err + } + } } } - var audioConf *mpeg4audio.AudioSpecificConfig + for id, track := range audioTracks { + switch track := track.(type) { + case *format.Opus: + var msg message.Message = &message.AudioExSequenceStart{ + ChunkStreamID: message.AudioChunkStreamID, + MessageStreamID: 0x1000000, + FourCC: message.FourCCOpus, + OpusHeader: &message.OpusIDHeader{ + Version: 0x1, + ChannelCount: uint8(track.ChannelCount), + PreSkip: 3840, + }, + } - if track, ok := w.AudioTrack.(*format.MPEG4Audio); ok { - audioConf = track.Config - } else if track, ok2 := w.AudioTrack.(*format.MPEG4AudioLATM); ok2 { - audioConf = track.StreamMuxConfig.Programs[0].Layers[0].AudioSpecificConfig - } + if id != 0 { + msg = &message.AudioExMultitrack{ + MultitrackType: 0x0, + TrackID: uint8(id), + Wrapped: msg, + } + } - if audioConf != nil { - var enc []byte - enc, err = audioConf.Marshal() - if err != nil { - return err - } + err = w.Conn.Write(msg) + if err != nil { + return err + } - err = w.Conn.Write(&message.Audio{ - ChunkStreamID: message.AudioChunkStreamID, - MessageStreamID: 0x1000000, - Codec: message.CodecMPEG4Audio, - Rate: message.Rate44100, - Depth: message.Depth16, - IsStereo: true, - AACType: message.AudioAACTypeConfig, - Payload: enc, - }) - if err != nil { - return err + case *format.MPEG4Audio: + audioConf := track.Config + + if id == 0 { + var enc []byte + enc, err = audioConf.Marshal() + if err != nil { + return err + } + + err = w.Conn.Write(&message.Audio{ + ChunkStreamID: message.AudioChunkStreamID, + MessageStreamID: 0x1000000, + Codec: message.CodecMPEG4Audio, + Rate: message.Rate44100, + Depth: message.Depth16, + IsStereo: true, + AACType: message.AudioAACTypeConfig, + Payload: enc, + }) + if err != nil { + return err + } + } else { + err = w.Conn.Write(&message.AudioExMultitrack{ + MultitrackType: 0x0, + TrackID: uint8(id), + Wrapped: &message.AudioExSequenceStart{ + ChunkStreamID: message.VideoChunkStreamID, + MessageStreamID: 0x1000000, + FourCC: message.FourCCMP4A, + AACHeader: audioConf, + }, + }) + if err != nil { + return err + } + } + + case *format.MPEG4AudioLATM: + audioConf := track.StreamMuxConfig.Programs[0].Layers[0].AudioSpecificConfig + + if id == 0 { + var enc []byte + enc, err = audioConf.Marshal() + if err != nil { + return err + } + + err = w.Conn.Write(&message.Audio{ + ChunkStreamID: message.AudioChunkStreamID, + MessageStreamID: 0x1000000, + Codec: message.CodecMPEG4Audio, + Rate: message.Rate44100, + Depth: message.Depth16, + IsStereo: true, + AACType: message.AudioAACTypeConfig, + Payload: enc, + }) + if err != nil { + return err + } + } else { + err = w.Conn.Write(&message.AudioExMultitrack{ + MultitrackType: 0x0, + TrackID: uint8(id), + Wrapped: &message.AudioExSequenceStart{ + ChunkStreamID: message.VideoChunkStreamID, + MessageStreamID: 0x1000000, + FourCC: message.FourCCMP4A, + AACHeader: audioConf, + }, + }) + if err != nil { + return err + } + } + + case *format.AC3: + var msg message.Message = &message.AudioExSequenceStart{ + ChunkStreamID: message.AudioChunkStreamID, + MessageStreamID: 0x1000000, + FourCC: message.FourCCAC3, + } + + if id != 0 { + msg = &message.AudioExMultitrack{ + MultitrackType: 0x0, + TrackID: uint8(id), + Wrapped: msg, + } + } + + err = w.Conn.Write(msg) + if err != nil { + return err + } + + case *format.G711: + if id != 0 { + return fmt.Errorf("it is not possible to use G711 tracks as secondary tracks") + } + + case *format.LPCM: + if id != 0 { + return fmt.Errorf("it is not possible to use LPCM tracks as secondary tracks") + } } } return nil } -// WriteH264 writes H264 data. -func (w *Writer) WriteH264(pts time.Duration, dts time.Duration, au [][]byte) error { +// WriteAV1 writes a AV1 temporal unit. +func (w *Writer) WriteAV1(track *format.AV1, pts time.Duration, dts time.Duration, tu [][]byte) error { + // FFmpeg requires this. + tu = append([][]byte{{byte(av1.OBUTypeTemporalDelimiter << 3)}}, tu...) + + enc, err := av1.Bitstream(tu).Marshal() + if err != nil { + return err + } + + var msg message.Message + + if pts == dts { + msg = &message.VideoExFramesX{ + ChunkStreamID: message.VideoChunkStreamID, + MessageStreamID: 0x1000000, + FourCC: message.FourCCAV1, + DTS: dts, + Payload: enc, + } + } else { + msg = &message.VideoExCodedFrames{ + ChunkStreamID: message.VideoChunkStreamID, + MessageStreamID: 0x1000000, + FourCC: message.FourCCAV1, + DTS: dts, + PTSDelta: pts - dts, + Payload: enc, + } + } + + id := w.videoTrackToID[track] + + if id != 0 { + msg = &message.VideoExMultitrack{ + MultitrackType: 0x0, + TrackID: id, + Wrapped: msg, + } + } + + return w.Conn.Write(msg) +} + +// WriteVP9 writes a VP9 frame. +func (w *Writer) WriteVP9(track *format.VP9, pts time.Duration, dts time.Duration, frame []byte) error { + var msg message.Message + + if pts == dts { + msg = &message.VideoExFramesX{ + ChunkStreamID: message.VideoChunkStreamID, + MessageStreamID: 0x1000000, + FourCC: message.FourCCVP9, + DTS: dts, + Payload: frame, + } + } else { + msg = &message.VideoExCodedFrames{ + ChunkStreamID: message.VideoChunkStreamID, + MessageStreamID: 0x1000000, + FourCC: message.FourCCVP9, + DTS: dts, + PTSDelta: pts - dts, + Payload: frame, + } + } + + id := w.videoTrackToID[track] + + if id != 0 { + msg = &message.VideoExMultitrack{ + MultitrackType: 0x0, + TrackID: id, + Wrapped: msg, + } + } + + return w.Conn.Write(msg) +} + +// WriteH265 writes a H265 access unit. +func (w *Writer) WriteH265(track *format.H265, pts time.Duration, dts time.Duration, au [][]byte) error { avcc, err := h264.AVCC(au).Marshal() if err != nil { return err } - return w.Conn.Write(&message.Video{ - ChunkStreamID: message.VideoChunkStreamID, - MessageStreamID: 0x1000000, - Codec: message.CodecH264, - IsKeyFrame: h264.IsRandomAccess(au), - Type: message.VideoTypeAU, - Payload: avcc, - DTS: dts, - PTSDelta: pts - dts, + var msg message.Message + + if pts == dts { + msg = &message.VideoExFramesX{ + ChunkStreamID: message.VideoChunkStreamID, + MessageStreamID: 0x1000000, + FourCC: message.FourCCHEVC, + DTS: dts, + Payload: avcc, + } + } else { + msg = &message.VideoExCodedFrames{ + ChunkStreamID: message.VideoChunkStreamID, + MessageStreamID: 0x1000000, + FourCC: message.FourCCHEVC, + DTS: dts, + PTSDelta: pts - dts, + Payload: avcc, + } + } + + id := w.videoTrackToID[track] + + if id != 0 { + msg = &message.VideoExMultitrack{ + MultitrackType: 0x0, + TrackID: id, + Wrapped: msg, + } + } + + return w.Conn.Write(msg) +} + +// WriteH264 writes a H264 access unit. +func (w *Writer) WriteH264(track *format.H264, pts time.Duration, dts time.Duration, au [][]byte) error { + avcc, err := h264.AVCC(au).Marshal() + if err != nil { + return err + } + + id := w.videoTrackToID[track] + + if id == 0 { + return w.Conn.Write(&message.Video{ + ChunkStreamID: message.VideoChunkStreamID, + MessageStreamID: 0x1000000, + Codec: message.CodecH264, + IsKeyFrame: h264.IsRandomAccess(au), + Type: message.VideoTypeAU, + Payload: avcc, + DTS: dts, + PTSDelta: pts - dts, + }) + } + + var msg message.Message + + if pts == dts { + msg = &message.VideoExFramesX{ + ChunkStreamID: message.VideoChunkStreamID, + MessageStreamID: 0x1000000, + FourCC: message.FourCCAVC, + DTS: dts, + Payload: avcc, + } + } else { + msg = &message.VideoExCodedFrames{ + ChunkStreamID: message.VideoChunkStreamID, + MessageStreamID: 0x1000000, + FourCC: message.FourCCAVC, + DTS: dts, + PTSDelta: pts - dts, + Payload: avcc, + } + } + + return w.Conn.Write(&message.VideoExMultitrack{ + MultitrackType: 0x0, + TrackID: id, + Wrapped: msg, }) } -// WriteMPEG4Audio writes MPEG-4 Audio data. -func (w *Writer) WriteMPEG4Audio(pts time.Duration, au []byte) error { - return w.Conn.Write(&message.Audio{ +// WriteOpus writes a Opus packet. +func (w *Writer) WriteOpus(track *format.Opus, pts time.Duration, pkt []byte) error { + var msg message.Message = &message.AudioExCodedFrames{ ChunkStreamID: message.AudioChunkStreamID, MessageStreamID: 0x1000000, - Codec: message.CodecMPEG4Audio, - Rate: message.Rate44100, - Depth: message.Depth16, - IsStereo: true, - AACType: message.AudioAACTypeAU, - Payload: au, DTS: pts, + FourCC: message.FourCCOpus, + Payload: pkt, + } + + id := w.audioTrackToID[track] + + if id != 0 { + msg = &message.AudioExMultitrack{ + MultitrackType: 0x0, + TrackID: id, + Wrapped: msg, + } + } + + return w.Conn.Write(msg) +} + +// WriteMPEG4Audio writes a MPEG-4 Audio access unit. +func (w *Writer) WriteMPEG4Audio(track format.Format, pts time.Duration, au []byte) error { + id := w.audioTrackToID[track] + + if id == 0 { + return w.Conn.Write(&message.Audio{ + ChunkStreamID: message.AudioChunkStreamID, + MessageStreamID: 0x1000000, + Codec: message.CodecMPEG4Audio, + Rate: message.Rate44100, + Depth: message.Depth16, + IsStereo: true, + AACType: message.AudioAACTypeAU, + Payload: au, + DTS: pts, + }) + } + + return w.Conn.Write(&message.AudioExMultitrack{ + MultitrackType: 0x0, + TrackID: id, + Wrapped: &message.AudioExCodedFrames{ + ChunkStreamID: message.AudioChunkStreamID, + MessageStreamID: 0x1000000, + DTS: pts, + FourCC: message.FourCCMP4A, + Payload: au, + }, }) } -// WriteMPEG1Audio writes MPEG-1 Audio data. -func (w *Writer) WriteMPEG1Audio(pts time.Duration, h *mpeg1audio.FrameHeader, frame []byte) error { - return w.Conn.Write(&message.Audio{ +// WriteMPEG1Audio writes a MPEG-1 Audio frame. +func (w *Writer) WriteMPEG1Audio(track *format.MPEG1Audio, pts time.Duration, + h *mpeg1audio.FrameHeader, frame []byte, +) error { + id := w.audioTrackToID[track] + + if id == 0 { + rate, ok := audioRateIntToRTMP(h.SampleRate) + if !ok { + return fmt.Errorf("unsupported sample rate: %v", h.SampleRate) + } + + return w.Conn.Write(&message.Audio{ + ChunkStreamID: message.AudioChunkStreamID, + MessageStreamID: 0x1000000, + Codec: message.CodecMPEG1Audio, + Rate: rate, + Depth: message.Depth16, + IsStereo: mpeg1AudioChannels(h.ChannelMode), + Payload: frame, + DTS: pts, + }) + } + + return w.Conn.Write(&message.AudioExMultitrack{ + MultitrackType: 0x0, + TrackID: id, + Wrapped: &message.AudioExCodedFrames{ + ChunkStreamID: message.AudioChunkStreamID, + MessageStreamID: 0x1000000, + DTS: pts, + FourCC: message.FourCCMP3, + Payload: frame, + }, + }) +} + +// WriteAC3 writes an AC-3 frame. +func (w *Writer) WriteAC3(track *format.AC3, pts time.Duration, frame []byte) error { + var msg message.Message = &message.AudioExCodedFrames{ ChunkStreamID: message.AudioChunkStreamID, MessageStreamID: 0x1000000, - Codec: message.CodecMPEG1Audio, - Rate: audioRateIntToRTMP(h.SampleRate), - Depth: message.Depth16, - IsStereo: mpeg1AudioChannels(h.ChannelMode), + FourCC: message.FourCCAC3, Payload: frame, DTS: pts, + } + + id := w.audioTrackToID[track] + + if id != 0 { + msg = &message.AudioExMultitrack{ + MultitrackType: 0x0, + TrackID: id, + Wrapped: msg, + } + } + + return w.Conn.Write(msg) +} + +// WriteG711 writes G711 samples. +func (w *Writer) WriteG711(track *format.G711, pts time.Duration, samples []byte) error { + var codec uint8 + + if track.MULaw { + codec = message.CodecPCMU + } else { + codec = message.CodecPCMA + } + + return w.Conn.Write(&message.Audio{ + ChunkStreamID: message.AudioChunkStreamID, + MessageStreamID: 0x1000000, + Codec: codec, + Rate: message.Rate5512, + Depth: message.Depth16, + IsStereo: track.ChannelCount == 2, + Payload: samples, + DTS: pts, + }) +} + +// WriteLPCM writes LPCM samples. +func (w *Writer) WriteLPCM(track *format.LPCM, pts time.Duration, samples []byte) error { + rate, ok := audioRateIntToRTMP(track.SampleRate) + if !ok { + return fmt.Errorf("unsupported sample rate: %v", track.SampleRate) + } + + le := len(samples) + if le%2 != 0 { + return fmt.Errorf("invalid payload length: %d", le) + } + + samplesCopy := append([]byte(nil), samples...) + + // convert from big endian to little endian + for i := 0; i < le; i += 2 { + samplesCopy[i], samplesCopy[i+1] = samplesCopy[i+1], samplesCopy[i] + } + + return w.Conn.Write(&message.Audio{ + ChunkStreamID: message.AudioChunkStreamID, + MessageStreamID: 0x1000000, + Codec: message.CodecLPCM, + Rate: rate, + Depth: message.Depth16, + IsStereo: (track.ChannelCount == 2), + Payload: samplesCopy, + DTS: pts, }) } diff --git a/internal/protocols/rtmp/writer_test.go b/internal/protocols/rtmp/writer_test.go index ea3e355e..2bf2f773 100644 --- a/internal/protocols/rtmp/writer_test.go +++ b/internal/protocols/rtmp/writer_test.go @@ -3,104 +3,683 @@ package rtmp import ( "bytes" "testing" + "time" + "github.com/abema/go-mp4" "github.com/bluenviron/gortsplib/v4/pkg/format" + "github.com/bluenviron/mediacommon/v2/pkg/codecs/mpeg1audio" "github.com/bluenviron/mediacommon/v2/pkg/codecs/mpeg4audio" "github.com/stretchr/testify/require" + "github.com/bluenviron/mediamtx/internal/formatprocessor" "github.com/bluenviron/mediamtx/internal/protocols/rtmp/amf0" "github.com/bluenviron/mediamtx/internal/protocols/rtmp/bytecounter" "github.com/bluenviron/mediamtx/internal/protocols/rtmp/message" + "github.com/bluenviron/mediamtx/internal/test" ) -func TestWriteTracks(t *testing.T) { - videoTrack := &format.H264{ - PayloadTyp: 96, - SPS: []byte{ - 0x67, 0x64, 0x00, 0x0c, 0xac, 0x3b, 0x50, 0xb0, - 0x4b, 0x42, 0x00, 0x00, 0x03, 0x00, 0x02, 0x00, - 0x00, 0x03, 0x00, 0x3d, 0x08, - }, - PPS: []byte{ - 0x68, 0xee, 0x3c, 0x80, - }, - PacketizationMode: 1, +func TestWriter(t *testing.T) { + for _, ca := range []string{ + "h264 + aac", + "av1", + "vp9", + "h265", + "h265 no params", + "h264 no params", + "opus", + "mp3", + "ac-3", + "pcma", + "pcmu", + "lpcm", + } { + t.Run(ca, func(t *testing.T) { + var tracks []format.Format + + switch ca { + case "h264 + aac": + tracks = append(tracks, &format.H264{ + PayloadTyp: 96, + SPS: []byte{ + 0x67, 0x64, 0x00, 0x0c, 0xac, 0x3b, 0x50, 0xb0, + 0x4b, 0x42, 0x00, 0x00, 0x03, 0x00, 0x02, 0x00, + 0x00, 0x03, 0x00, 0x3d, 0x08, + }, + PPS: []byte{ + 0x68, 0xee, 0x3c, 0x80, + }, + PacketizationMode: 1, + }) + + tracks = append(tracks, &format.MPEG4Audio{ + PayloadTyp: 96, + Config: &mpeg4audio.AudioSpecificConfig{ + Type: 2, + SampleRate: 44100, + ChannelCount: 2, + }, + SizeLength: 13, + IndexLength: 3, + IndexDeltaLength: 3, + }) + + case "av1": + tracks = append(tracks, &format.AV1{ + PayloadTyp: 96, + }) + + case "vp9": + tracks = append(tracks, &format.VP9{ + PayloadTyp: 96, + }) + + case "h265": + tracks = append(tracks, test.FormatH265) + + case "h265 no params": + tracks = append(tracks, &format.H265{}) + + case "h264 no params": + tracks = append(tracks, &format.H264{}) + + case "opus": + tracks = append(tracks, &format.Opus{ + PayloadTyp: 96, + ChannelCount: 2, + }) + + case "mp3": + tracks = append(tracks, &format.MPEG1Audio{}) + + case "ac-3": + tracks = append(tracks, &format.AC3{ + SampleRate: 44100, + ChannelCount: 2, + }) + + case "pcma": + tracks = append(tracks, &format.G711{ + MULaw: false, + SampleRate: 8000, + ChannelCount: 1, + }) + + case "pcmu": + tracks = append(tracks, &format.G711{ + MULaw: true, + SampleRate: 8000, + ChannelCount: 1, + }) + + case "lpcm": + tracks = append(tracks, &format.LPCM{ + BitDepth: 16, + SampleRate: 44100, + ChannelCount: 1, + }) + } + + var buf bytes.Buffer + c := &dummyConn{ + rw: &buf, + } + c.initialize() + + w := &Writer{ + Conn: c, + Tracks: tracks, + } + err := w.Initialize() + require.NoError(t, err) + + bc := bytecounter.NewReadWriter(&buf) + mrw := message.NewReadWriter(bc, bc, true) + + msg, err := mrw.Read() + require.NoError(t, err) + + switch ca { + case "h264 + aac": + require.Equal(t, &message.DataAMF0{ + ChunkStreamID: 4, + MessageStreamID: 0x1000000, + Payload: []interface{}{ + "@setDataFrame", + "onMetaData", + amf0.Object{ + {Key: "videocodecid", Value: float64(7)}, + {Key: "videodatarate", Value: float64(0)}, + {Key: "audiocodecid", Value: float64(10)}, + {Key: "audiodatarate", Value: float64(0)}, + }, + }, + }, msg) + + case "av1": + require.Equal(t, &message.DataAMF0{ + ChunkStreamID: 4, + MessageStreamID: 0x1000000, + Payload: []interface{}{ + "@setDataFrame", + "onMetaData", + amf0.Object{ + {Key: "videocodecid", Value: float64(1.635135537e+09)}, + {Key: "videodatarate", Value: float64(0)}, + {Key: "audiocodecid", Value: float64(0)}, + {Key: "audiodatarate", Value: float64(0)}, + }, + }, + }, msg) + + case "vp9": + require.Equal(t, &message.DataAMF0{ + ChunkStreamID: 4, + MessageStreamID: 0x1000000, + Payload: []interface{}{ + "@setDataFrame", + "onMetaData", + amf0.Object{ + {Key: "videocodecid", Value: float64(1.987063865e+09)}, + {Key: "videodatarate", Value: float64(0)}, + {Key: "audiocodecid", Value: float64(0)}, + {Key: "audiodatarate", Value: float64(0)}, + }, + }, + }, msg) + + case "h265": + require.Equal(t, &message.DataAMF0{ + ChunkStreamID: 4, + MessageStreamID: 0x1000000, + Payload: []interface{}{ + "@setDataFrame", + "onMetaData", + amf0.Object{ + {Key: "videocodecid", Value: float64(1.752589105e+09)}, + {Key: "videodatarate", Value: float64(0)}, + {Key: "audiocodecid", Value: float64(0)}, + {Key: "audiodatarate", Value: float64(0)}, + }, + }, + }, msg) + + case "h265 no params": + require.Equal(t, &message.DataAMF0{ + ChunkStreamID: 4, + MessageStreamID: 0x1000000, + Payload: []interface{}{ + "@setDataFrame", + "onMetaData", + amf0.Object{ + {Key: "videocodecid", Value: float64(1.752589105e+09)}, + {Key: "videodatarate", Value: float64(0)}, + {Key: "audiocodecid", Value: float64(0)}, + {Key: "audiodatarate", Value: float64(0)}, + }, + }, + }, msg) + + case "h264 no params": + require.Equal(t, &message.DataAMF0{ + ChunkStreamID: 4, + MessageStreamID: 0x1000000, + Payload: []interface{}{ + "@setDataFrame", + "onMetaData", + amf0.Object{ + {Key: "videocodecid", Value: float64(7)}, + {Key: "videodatarate", Value: float64(0)}, + {Key: "audiocodecid", Value: float64(0)}, + {Key: "audiodatarate", Value: float64(0)}, + }, + }, + }, msg) + + case "opus": + require.Equal(t, &message.DataAMF0{ + ChunkStreamID: 4, + MessageStreamID: 0x1000000, + Payload: []interface{}{ + "@setDataFrame", + "onMetaData", + amf0.Object{ + {Key: "videocodecid", Value: float64(0)}, + {Key: "videodatarate", Value: float64(0)}, + {Key: "audiocodecid", Value: float64(1.332770163e+09)}, + {Key: "audiodatarate", Value: float64(0)}, + }, + }, + }, msg) + + case "mp3": + require.Equal(t, &message.DataAMF0{ + ChunkStreamID: 4, + MessageStreamID: 0x1000000, + Payload: []interface{}{ + "@setDataFrame", + "onMetaData", + amf0.Object{ + {Key: "videocodecid", Value: float64(0)}, + {Key: "videodatarate", Value: float64(0)}, + {Key: "audiocodecid", Value: float64(2)}, + {Key: "audiodatarate", Value: float64(0)}, + }, + }, + }, msg) + + case "ac-3": + require.Equal(t, &message.DataAMF0{ + ChunkStreamID: 4, + MessageStreamID: 0x1000000, + Payload: []interface{}{ + "@setDataFrame", + "onMetaData", + amf0.Object{ + {Key: "videocodecid", Value: float64(0)}, + {Key: "videodatarate", Value: float64(0)}, + {Key: "audiocodecid", Value: float64(1.633889587e+09)}, + {Key: "audiodatarate", Value: float64(0)}, + }, + }, + }, msg) + + case "pcma": + require.Equal(t, &message.DataAMF0{ + ChunkStreamID: 4, + MessageStreamID: 0x1000000, + Payload: []interface{}{ + "@setDataFrame", + "onMetaData", + amf0.Object{ + {Key: "videocodecid", Value: float64(0)}, + {Key: "videodatarate", Value: float64(0)}, + {Key: "audiocodecid", Value: float64(7)}, + {Key: "audiodatarate", Value: float64(0)}, + }, + }, + }, msg) + + case "pcmu": + require.Equal(t, &message.DataAMF0{ + ChunkStreamID: 4, + MessageStreamID: 0x1000000, + Payload: []interface{}{ + "@setDataFrame", + "onMetaData", + amf0.Object{ + {Key: "videocodecid", Value: float64(0)}, + {Key: "videodatarate", Value: float64(0)}, + {Key: "audiocodecid", Value: float64(8)}, + {Key: "audiodatarate", Value: float64(0)}, + }, + }, + }, msg) + + case "lpcm": + require.Equal(t, &message.DataAMF0{ + ChunkStreamID: 4, + MessageStreamID: 0x1000000, + Payload: []interface{}{ + "@setDataFrame", + "onMetaData", + amf0.Object{ + {Key: "videocodecid", Value: float64(0)}, + {Key: "videodatarate", Value: float64(0)}, + {Key: "audiocodecid", Value: float64(3)}, + {Key: "audiodatarate", Value: float64(0)}, + }, + }, + }, msg) + } + + switch ca { + case "h264 + aac": + msg, err = mrw.Read() + require.NoError(t, err) + require.Equal(t, &message.Video{ + ChunkStreamID: message.VideoChunkStreamID, + MessageStreamID: 0x1000000, + Codec: message.CodecH264, + IsKeyFrame: true, + Type: message.VideoTypeConfig, + Payload: []byte{ + 0x01, 0x64, 0x00, 0x0c, 0xff, 0xe1, 0x00, 0x15, + 0x67, 0x64, 0x00, 0x0c, 0xac, 0x3b, 0x50, 0xb0, + 0x4b, 0x42, 0x00, 0x00, 0x03, 0x00, 0x02, 0x00, + 0x00, 0x03, 0x00, 0x3d, 0x08, 0x01, 0x00, 0x04, + 0x68, 0xee, 0x3c, 0x80, + }, + }, msg) + + msg, err = mrw.Read() + require.NoError(t, err) + require.Equal(t, &message.Audio{ + ChunkStreamID: message.AudioChunkStreamID, + MessageStreamID: 0x1000000, + Codec: message.CodecMPEG4Audio, + Rate: message.Rate44100, + Depth: message.Depth16, + IsStereo: true, + AACType: message.AudioAACTypeConfig, + Payload: []byte{0x12, 0x10}, + }, msg) + + err = w.WriteH264(tracks[0].(*format.H264), 100*time.Millisecond, 0, [][]byte{{5, 1}}) + require.NoError(t, err) + + err = w.WriteMPEG4Audio(tracks[1], 0, []byte{1, 2}) + require.NoError(t, err) + + msg, err = mrw.Read() + require.NoError(t, err) + require.Equal(t, &message.Video{ + ChunkStreamID: message.VideoChunkStreamID, + MessageStreamID: 0x1000000, + Codec: message.CodecH264, + IsKeyFrame: true, + Type: message.VideoTypeAU, + PTSDelta: 100 * time.Millisecond, + Payload: []byte{0, 0, 0, 2, 5, 1}, + }, msg) + + case "av1": + msg, err = mrw.Read() + require.NoError(t, err) + require.Equal(t, &message.VideoExSequenceStart{ + ChunkStreamID: message.VideoChunkStreamID, + MessageStreamID: 0x1000000, + FourCC: message.FourCCAV1, + AV1Header: &mp4.Av1C{ + Marker: 0x1, + Version: 0x1, + SeqLevelIdx0: 0x8, + ChromaSubsamplingX: 0x1, + ChromaSubsamplingY: 0x1, + ConfigOBUs: []uint8{0xa, 0xb, 0x0, 0x0, 0x0, 0x42, 0xab, 0xbf, 0xc3, 0x70, 0xb, 0xe0, 0x1}, + }, + }, msg) + + err = w.WriteAV1(tracks[0].(*format.AV1), 0, 0, [][]byte{{1, 2}}) + require.NoError(t, err) + + msg, err = mrw.Read() + require.NoError(t, err) + require.Equal(t, &message.VideoExFramesX{ + ChunkStreamID: message.VideoChunkStreamID, + MessageStreamID: 0x1000000, + FourCC: message.FourCCAV1, + Payload: []byte{0x12, 0x0, 0x3, 0x1, 0x2}, + }, msg) + + case "vp9": + msg, err = mrw.Read() + require.NoError(t, err) + require.Equal(t, &message.VideoExSequenceStart{ + ChunkStreamID: message.VideoChunkStreamID, + MessageStreamID: 0x1000000, + FourCC: message.FourCCVP9, + VP9Header: &mp4.VpcC{ + FullBox: mp4.FullBox{Version: 0x1}, + Level: 0x28, + BitDepth: 0x8, + ChromaSubsampling: 0x1, + ColourPrimaries: 0x2, + TransferCharacteristics: 0x2, + MatrixCoefficients: 0x2, + CodecInitializationData: []uint8{}, + }, + }, msg) + + err = w.WriteVP9(tracks[0].(*format.VP9), 0, 0, []byte{1, 2}) + require.NoError(t, err) + + msg, err = mrw.Read() + require.NoError(t, err) + require.Equal(t, &message.VideoExFramesX{ + ChunkStreamID: message.VideoChunkStreamID, + MessageStreamID: 0x1000000, + FourCC: message.FourCCVP9, + Payload: []byte{0x1, 0x2}, + }, msg) + + case "h265": + msg, err = mrw.Read() + require.NoError(t, err) + + require.Equal(t, &message.VideoExSequenceStart{ + ChunkStreamID: message.VideoChunkStreamID, + MessageStreamID: 0x1000000, + FourCC: message.FourCCHEVC, + HEVCHeader: &mp4.HvcC{ + ConfigurationVersion: 0x1, + GeneralProfileIdc: 2, + GeneralProfileCompatibility: [32]bool{ + false, false, true, false, false, false, false, false, + false, false, false, false, false, false, false, false, + false, false, false, false, false, false, false, false, + false, false, false, false, false, false, false, false, + }, + GeneralConstraintIndicator: [6]uint8{0x03, 0x0, 0xb0, 0x0, 0x0, 0x03}, + GeneralLevelIdc: 0x7b, + ChromaFormatIdc: 0x1, + LengthSizeMinusOne: 0x3, + NumOfNaluArrays: 0x3, + BitDepthLumaMinus8: 2, + BitDepthChromaMinus8: 2, + NumTemporalLayers: 1, + NaluArrays: []mp4.HEVCNaluArray{ + { + NaluType: 0x20, + NumNalus: 0x1, + Nalus: []mp4.HEVCNalu{{ + Length: 24, + NALUnit: test.FormatH265.VPS, + }}, + }, + { + NaluType: 0x21, + NumNalus: 0x1, + Nalus: []mp4.HEVCNalu{{ + Length: 60, + NALUnit: test.FormatH265.SPS, + }}, + }, + { + NaluType: 0x22, + NumNalus: 0x1, + Nalus: []mp4.HEVCNalu{{ + Length: 8, + NALUnit: test.FormatH265.PPS, + }}, + }, + }, + }, + }, msg) + + err = w.WriteH265(tracks[0].(*format.H265), 0, 0, [][]byte{{1, 2}}) + require.NoError(t, err) + + msg, err = mrw.Read() + require.NoError(t, err) + require.Equal(t, &message.VideoExFramesX{ + ChunkStreamID: message.VideoChunkStreamID, + MessageStreamID: 0x1000000, + FourCC: message.FourCCHEVC, + Payload: []byte{0, 0, 0, 2, 1, 2}, + }, msg) + + case "h265 no params": + msg, err = mrw.Read() + require.NoError(t, err) + + require.Equal(t, &message.VideoExSequenceStart{ + ChunkStreamID: message.VideoChunkStreamID, + MessageStreamID: 0x1000000, + FourCC: message.FourCCHEVC, + HEVCHeader: generateHvcC(formatprocessor.H265DefaultVPS, + formatprocessor.H265DefaultSPS, formatprocessor.H265DefaultPPS), + }, msg) + + err = w.WriteH265(tracks[0].(*format.H265), 0, 0, [][]byte{{1, 2}}) + require.NoError(t, err) + + msg, err = mrw.Read() + require.NoError(t, err) + require.Equal(t, &message.VideoExFramesX{ + ChunkStreamID: message.VideoChunkStreamID, + MessageStreamID: 0x1000000, + FourCC: message.FourCCHEVC, + Payload: []byte{0, 0, 0, 2, 1, 2}, + }, msg) + + case "h264 no params": + msg, err = mrw.Read() + require.NoError(t, err) + require.Equal(t, &message.Video{ + ChunkStreamID: message.VideoChunkStreamID, + MessageStreamID: 0x1000000, + Codec: message.CodecH264, + IsKeyFrame: true, + Type: message.VideoTypeConfig, + Payload: []byte{ + 0x01, 0x42, 0xc0, 0x28, 0xff, 0xe1, 0x00, 0x19, + 0x67, 0x42, 0xc0, 0x28, 0xd9, 0x00, 0x78, 0x02, + 0x27, 0xe5, 0x84, 0x00, 0x00, 0x03, 0x00, 0x04, + 0x00, 0x00, 0x03, 0x00, 0xf0, 0x3c, 0x60, 0xc9, + 0x20, 0x01, 0x00, 0x04, 0x08, 0x06, 0x07, 0x08, + }, + }, msg) + + err = w.WriteH264(tracks[0].(*format.H264), 0, 0, [][]byte{{1, 2}}) + require.NoError(t, err) + + msg, err = mrw.Read() + require.NoError(t, err) + require.Equal(t, &message.Video{ + ChunkStreamID: message.VideoChunkStreamID, + MessageStreamID: 0x1000000, + Codec: message.CodecH264, + Type: message.VideoTypeAU, + Payload: []byte{0, 0, 0, 2, 1, 2}, + }, msg) + + case "opus": + msg, err = mrw.Read() + require.NoError(t, err) + require.Equal(t, &message.AudioExSequenceStart{ + ChunkStreamID: message.AudioChunkStreamID, + MessageStreamID: 0x1000000, + FourCC: message.FourCCOpus, + OpusHeader: &message.OpusIDHeader{ + Version: 1, + PreSkip: 3840, + ChannelCount: 2, + ChannelMappingTable: []uint8{}, + }, + }, msg) + + err = w.WriteOpus(tracks[0].(*format.Opus), 0, []byte{1, 2}) + require.NoError(t, err) + + msg, err = mrw.Read() + require.NoError(t, err) + require.Equal(t, &message.AudioExCodedFrames{ + ChunkStreamID: message.AudioChunkStreamID, + MessageStreamID: 0x1000000, + FourCC: message.FourCCOpus, + Payload: []byte{1, 2}, + }, msg) + + case "mp3": + fr := []byte{ + 0xff, 0xfa, 0x52, 0x04, 0x00, + } + + var h mpeg1audio.FrameHeader + err = h.Unmarshal(fr) + require.NoError(t, err) + + err = w.WriteMPEG1Audio(tracks[0].(*format.MPEG1Audio), 0, &h, fr) + require.NoError(t, err) + + msg, err = mrw.Read() + require.NoError(t, err) + require.Equal(t, &message.Audio{ + ChunkStreamID: message.AudioChunkStreamID, + MessageStreamID: 0x1000000, + Codec: message.CodecMPEG1Audio, + Rate: message.Rate44100, + Depth: message.Depth16, + IsStereo: true, + Payload: []byte{ + 0xff, 0xfa, 0x52, 0x04, 0x00, + }, + }, msg) + + case "ac-3": + msg, err = mrw.Read() + require.NoError(t, err) + require.Equal(t, &message.AudioExSequenceStart{ + ChunkStreamID: message.AudioChunkStreamID, + MessageStreamID: 0x1000000, + FourCC: message.FourCCAC3, + }, msg) + + err = w.WriteAC3(tracks[0].(*format.AC3), 0, []byte{1, 2}) + require.NoError(t, err) + + msg, err = mrw.Read() + require.NoError(t, err) + require.Equal(t, &message.AudioExCodedFrames{ + ChunkStreamID: message.AudioChunkStreamID, + MessageStreamID: 0x1000000, + FourCC: message.FourCCAC3, + Payload: []byte{1, 2}, + }, msg) + + case "pcma": + err = w.WriteG711(tracks[0].(*format.G711), 0, []byte{1, 2}) + require.NoError(t, err) + + msg, err = mrw.Read() + require.NoError(t, err) + require.Equal(t, &message.Audio{ + ChunkStreamID: message.AudioChunkStreamID, + MessageStreamID: 0x1000000, + Codec: message.CodecPCMA, + Depth: message.Depth16, + Payload: []byte{1, 2}, + }, msg) + + case "pcmu": + err = w.WriteG711(tracks[0].(*format.G711), 0, []byte{1, 2}) + require.NoError(t, err) + + msg, err = mrw.Read() + require.NoError(t, err) + require.Equal(t, &message.Audio{ + ChunkStreamID: message.AudioChunkStreamID, + MessageStreamID: 0x1000000, + Codec: message.CodecPCMU, + Depth: message.Depth16, + Payload: []byte{1, 2}, + }, msg) + + case "lpcm": + err = w.WriteLPCM(tracks[0].(*format.LPCM), 0, []byte{1, 2}) + require.NoError(t, err) + + msg, err = mrw.Read() + require.NoError(t, err) + require.Equal(t, &message.Audio{ + ChunkStreamID: message.AudioChunkStreamID, + MessageStreamID: 0x1000000, + Codec: message.CodecLPCM, + Rate: message.Rate44100, + Depth: message.Depth16, + Payload: []byte{2, 1}, + }, msg) + } + }) } - - audioTrack := &format.MPEG4Audio{ - PayloadTyp: 96, - Config: &mpeg4audio.AudioSpecificConfig{ - Type: 2, - SampleRate: 44100, - ChannelCount: 2, - }, - SizeLength: 13, - IndexLength: 3, - IndexDeltaLength: 3, - } - - var buf bytes.Buffer - c := &dummyConn{ - rw: &buf, - } - c.initialize() - - w := &Writer{ - Conn: c, - VideoTrack: videoTrack, - AudioTrack: audioTrack, - } - err := w.Initialize() - require.NoError(t, err) - - bc := bytecounter.NewReadWriter(&buf) - mrw := message.NewReadWriter(bc, bc, true) - - msg, err := mrw.Read() - require.NoError(t, err) - require.Equal(t, &message.DataAMF0{ - ChunkStreamID: 4, - MessageStreamID: 0x1000000, - Payload: []interface{}{ - "@setDataFrame", - "onMetaData", - amf0.Object{ - {Key: "videodatarate", Value: float64(0)}, - {Key: "videocodecid", Value: float64(7)}, - {Key: "audiodatarate", Value: float64(0)}, - {Key: "audiocodecid", Value: float64(10)}, - }, - }, - }, msg) - - msg, err = mrw.Read() - require.NoError(t, err) - require.Equal(t, &message.Video{ - ChunkStreamID: message.VideoChunkStreamID, - MessageStreamID: 0x1000000, - Codec: message.CodecH264, - IsKeyFrame: true, - Type: message.VideoTypeConfig, - Payload: []byte{ - 0x1, 0x64, 0x0, - 0xc, 0xff, 0xe1, 0x0, 0x15, 0x67, 0x64, 0x0, - 0xc, 0xac, 0x3b, 0x50, 0xb0, 0x4b, 0x42, 0x0, - 0x0, 0x3, 0x0, 0x2, 0x0, 0x0, 0x3, 0x0, - 0x3d, 0x8, 0x1, 0x0, 0x4, 0x68, 0xee, 0x3c, - 0x80, - }, - }, msg) - - msg, err = mrw.Read() - require.NoError(t, err) - require.Equal(t, &message.Audio{ - ChunkStreamID: message.AudioChunkStreamID, - MessageStreamID: 0x1000000, - Codec: message.CodecMPEG4Audio, - Rate: message.Rate44100, - Depth: message.Depth16, - IsStereo: true, - AACType: message.AudioAACTypeConfig, - Payload: []byte{0x12, 0x10}, - }, msg) } diff --git a/internal/recorder/format_fmp4.go b/internal/recorder/format_fmp4.go index 0e432794..95a113f7 100644 --- a/internal/recorder/format_fmp4.go +++ b/internal/recorder/format_fmp4.go @@ -158,13 +158,9 @@ func (f *formatFMP4) initialize() bool { paramsChanged := false for _, obu := range tunit.TU { - var h av1.OBUHeader - err := h.Unmarshal(obu) - if err != nil { - return err - } + typ := av1.OBUType((obu[0] >> 3) & 0b1111) - if h.Type == av1.OBUTypeSequenceHeader { + if typ == av1.OBUTypeSequenceHeader { if !bytes.Equal(codec.SequenceHeader, obu) { codec.SequenceHeader = obu paramsChanged = true @@ -285,7 +281,6 @@ func (f *formatFMP4) initialize() bool { case *rtspformat.H265: vps, sps, pps := forma.SafeParams() - if vps == nil || sps == nil || pps == nil { vps = formatprocessor.H265DefaultVPS sps = formatprocessor.H265DefaultSPS @@ -376,7 +371,6 @@ func (f *formatFMP4) initialize() bool { case *rtspformat.H264: sps, pps := forma.SafeParams() - if sps == nil || pps == nil { sps = formatprocessor.H264DefaultSPS pps = formatprocessor.H264DefaultPPS diff --git a/internal/recorder/recorder_test.go b/internal/recorder/recorder_test.go index f80899b5..55054d93 100644 --- a/internal/recorder/recorder_test.go +++ b/internal/recorder/recorder_test.go @@ -9,7 +9,6 @@ import ( "github.com/bluenviron/gortsplib/v4/pkg/description" rtspformat "github.com/bluenviron/gortsplib/v4/pkg/format" - "github.com/bluenviron/mediacommon/v2/pkg/codecs/h265" "github.com/bluenviron/mediacommon/v2/pkg/codecs/mpeg4audio" "github.com/bluenviron/mediacommon/v2/pkg/formats/fmp4" "github.com/bluenviron/mediacommon/v2/pkg/formats/mp4" @@ -96,7 +95,7 @@ func TestRecorder(t *testing.T) { test.FormatH265.VPS, test.FormatH265.SPS, test.FormatH265.PPS, - {byte(h265.NALUType_CRA_NUT) << 1, 0}, // IDR + {0x26, 0x1, 0xaf, 0x8, 0x42, 0x23, 0x48, 0x8a, 0x43, 0xe2}, }, }) diff --git a/internal/servers/rtmp/server_test.go b/internal/servers/rtmp/server_test.go index 8a7390e0..9a808fb7 100644 --- a/internal/servers/rtmp/server_test.go +++ b/internal/servers/rtmp/server_test.go @@ -128,14 +128,14 @@ func TestServerPublish(t *testing.T) { defer conn.Close() w := &rtmp.Writer{ - Conn: conn, - VideoTrack: test.FormatH264, - AudioTrack: test.FormatMPEG4Audio, + Conn: conn, + Tracks: []format.Format{test.FormatH264, test.FormatMPEG4Audio}, } err = w.Initialize() require.NoError(t, err) err = w.WriteH264( + test.FormatH264, 2*time.Second, 2*time.Second, [][]byte{ {5, 2, 3, 4}, }) @@ -165,6 +165,7 @@ func TestServerPublish(t *testing.T) { defer strm.RemoveReader(reader) err = w.WriteH264( + test.FormatH264, 3*time.Second, 3*time.Second, [][]byte{ {5, 2, 3, 4}, }) diff --git a/internal/staticsources/rtmp/source_test.go b/internal/staticsources/rtmp/source_test.go index 16f7f402..24308f5d 100644 --- a/internal/staticsources/rtmp/source_test.go +++ b/internal/staticsources/rtmp/source_test.go @@ -10,6 +10,7 @@ import ( "github.com/stretchr/testify/require" + "github.com/bluenviron/gortsplib/v4/pkg/format" "github.com/bluenviron/mediamtx/internal/conf" "github.com/bluenviron/mediamtx/internal/defs" "github.com/bluenviron/mediamtx/internal/protocols/rtmp" @@ -117,17 +118,16 @@ func TestSource(t *testing.T) { require.NoError(t, err) w := &rtmp.Writer{ - Conn: conn, - VideoTrack: test.FormatH264, - AudioTrack: test.FormatMPEG4Audio, + Conn: conn, + Tracks: []format.Format{test.FormatH264, test.FormatMPEG4Audio}, } err = w.Initialize() require.NoError(t, err) - err = w.WriteH264(2*time.Second, 2*time.Second, [][]byte{{5, 2, 3, 4}}) + err = w.WriteH264(test.FormatH264, 2*time.Second, 2*time.Second, [][]byte{{5, 2, 3, 4}}) require.NoError(t, err) - err = w.WriteH264(3*time.Second, 3*time.Second, [][]byte{{5, 2, 3, 4}}) + err = w.WriteH264(test.FormatH264, 3*time.Second, 3*time.Second, [][]byte{{5, 2, 3, 4}}) require.NoError(t, err) break