rtmp: support reading AV1, VP9, H265, Opus, AC-3, G711, LPCM and multiple tracks at once (#4168) (#4321)

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):

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

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=

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":

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

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

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)
+}

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 {

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},

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