rtmp: support clients that publish with empty metadata or no metadata (#386) (#769)

internal/core/rtmp_conn.go

@@ -283,7 +283,7 @@ func (c *rtmpConn) runRead(ctx context.Context) error {
 	}
 
 	c.conn.SetWriteDeadline(time.Now().Add(time.Duration(c.writeTimeout)))
-	err := c.conn.WriteMetadata(videoTrack, audioTrack)
+	err := c.conn.WriteTracks(videoTrack, audioTrack)
 	if err != nil {
 		return err
 	}
@@ -441,7 +441,7 @@ func (c *rtmpConn) runRead(ctx context.Context) error {
 
 func (c *rtmpConn) runPublish(ctx context.Context) error {
 	c.conn.SetReadDeadline(time.Now().Add(time.Duration(c.readTimeout)))
-	videoTrack, audioTrack, err := c.conn.ReadMetadata()
+	videoTrack, audioTrack, err := c.conn.ReadTracks()
 	if err != nil {
 		return err
 	}

internal/core/rtmp_server_test.go

@@ -142,7 +142,7 @@ func TestRTMPServerAuth(t *testing.T) {
 			err = conn.ClientHandshake()
 			require.NoError(t, err)
 
-			_, _, err = conn.ReadMetadata()
+			_, _, err = conn.ReadTracks()
 			require.NoError(t, err)
 		})
 	}

internal/core/rtmp_source.go

@@ -124,7 +124,7 @@ func (s *rtmpSource) runInner() bool {
 
 					conn.SetWriteDeadline(time.Time{})
 					conn.SetReadDeadline(time.Now().Add(time.Duration(s.readTimeout)))
-					videoTrack, audioTrack, err := conn.ReadMetadata()
+					videoTrack, audioTrack, err := conn.ReadTracks()
 					if err != nil {
 						return err
 					}

internal/rtmp/conn.go

@@ -1,6 +1,7 @@
 package rtmp
 
 import (
+	"errors"
 	"fmt"
 	"net"
 	"net/url"
@@ -81,39 +82,30 @@ func (c *Conn) WritePacket(pkt av.Packet) error {
 	return c.rconn.FlushWrite()
 }
 
-// ReadMetadata reads track informations.
+func trackFromH264DecoderConfig(data []byte) (*gortsplib.TrackH264, error) {
-func (c *Conn) ReadMetadata() (*gortsplib.TrackH264, *gortsplib.TrackAAC, error) {
+	codec, err := nh264.FromDecoderConfig(data)
-	var videoTrack *gortsplib.TrackH264
-	var audioTrack *gortsplib.TrackAAC
-
-	md, err := func() (flvio.AMFMap, error) {
-		pkt, err := c.ReadPacket()
 	if err != nil {
 		return nil, err
 	}
 
-		if pkt.Type != av.Metadata {
+	return gortsplib.NewTrackH264(96, codec.SPS[0], codec.PPS[0], nil)
-			return nil, fmt.Errorf("first packet must be metadata")
 }
 
+var errEmptyMetadata = errors.New("metadata is empty")
+
+func (c *Conn) readTracksFromMetadata(pkt av.Packet) (*gortsplib.TrackH264, *gortsplib.TrackAAC, error) {
 	arr, err := flvio.ParseAMFVals(pkt.Data, false)
 	if err != nil {
-			return nil, err
+		return nil, nil, err
 	}
 
 	if len(arr) != 1 {
-			return nil, fmt.Errorf("invalid metadata")
+		return nil, nil, fmt.Errorf("invalid metadata")
 	}
 
-		ma, ok := arr[0].(flvio.AMFMap)
+	md, ok := arr[0].(flvio.AMFMap)
 	if !ok {
-			return nil, fmt.Errorf("invalid metadata")
+		return nil, nil, fmt.Errorf("invalid metadata")
-		}
-
-		return ma, nil
-	}()
-	if err != nil {
-		return nil, nil, err
 	}
 
 	hasVideo, err := func() (bool, error) {
@@ -173,9 +165,12 @@ func (c *Conn) ReadMetadata() (*gortsplib.TrackH264, *gortsplib.TrackAAC, error)
 	}
 
 	if !hasVideo && !hasAudio {
-		return nil, nil, fmt.Errorf("stream doesn't contain tracks with supported codecs (H264 or AAC)")
+		return nil, nil, errEmptyMetadata
 	}
 
+	var videoTrack *gortsplib.TrackH264
+	var audioTrack *gortsplib.TrackAAC
+
 	for {
 		var pkt av.Packet
 		pkt, err = c.ReadPacket()
@@ -193,12 +188,7 @@ func (c *Conn) ReadMetadata() (*gortsplib.TrackH264, *gortsplib.TrackAAC, error)
 				return nil, nil, fmt.Errorf("video track setupped twice")
 			}
 
-			codec, err := nh264.FromDecoderConfig(pkt.Data)
+			videoTrack, err = trackFromH264DecoderConfig(pkt.Data)
-			if err != nil {
-				return nil, nil, err
-			}
-
-			videoTrack, err = gortsplib.NewTrackH264(96, codec.SPS[0], codec.PPS[0], nil)
 			if err != nil {
 				return nil, nil, err
 			}
@@ -232,8 +222,55 @@ func (c *Conn) ReadMetadata() (*gortsplib.TrackH264, *gortsplib.TrackAAC, error)
 	}
 }
 
-// WriteMetadata writes track informations.
+// ReadTracks reads track informations.
-func (c *Conn) WriteMetadata(videoTrack *gortsplib.TrackH264, audioTrack *gortsplib.TrackAAC) error {
+func (c *Conn) ReadTracks() (*gortsplib.TrackH264, *gortsplib.TrackAAC, error) {
+	pkt, err := c.ReadPacket()
+	if err != nil {
+		return nil, nil, err
+	}
+
+	switch pkt.Type {
+	case av.Metadata:
+		videoTrack, audioTrack, err := c.readTracksFromMetadata(pkt)
+		if err != nil {
+			if err == errEmptyMetadata {
+				pkt, err := c.ReadPacket()
+				if err != nil {
+					return nil, nil, err
+				}
+
+				if pkt.Type != av.H264DecoderConfig {
+					return nil, nil, fmt.Errorf("unexpected packet (%v)", pkt.Type)
+				}
+
+				videoTrack, err := trackFromH264DecoderConfig(pkt.Data)
+				if err != nil {
+					return nil, nil, err
+				}
+
+				return videoTrack, nil, nil
+			}
+
+			return nil, nil, err
+		}
+
+		return videoTrack, audioTrack, nil
+
+	case av.H264DecoderConfig:
+		videoTrack, err := trackFromH264DecoderConfig(pkt.Data)
+		if err != nil {
+			return nil, nil, err
+		}
+
+		return videoTrack, nil, nil
+
+	default:
+		return nil, nil, fmt.Errorf("unexpected packet (%v)", pkt.Type)
+	}
+}
+
+// WriteTracks writes track informations.
+func (c *Conn) WriteTracks(videoTrack *gortsplib.TrackH264, audioTrack *gortsplib.TrackAAC) error {
 	err := c.WritePacket(av.Packet{
 		Type: av.Metadata,
 		Data: flvio.FillAMF0ValMalloc(flvio.AMFMap{

internal/rtmp/conn_test.go

@@ -237,7 +237,13 @@ func (m chunk3) write(w io.Writer) error {
 	return err
 }
 
-func TestReadMetadata(t *testing.T) {
+func TestReadTracks(t *testing.T) {
+	for _, ca := range []string{
+		"standard",
+		"empty metadata",
+		"no metadata",
+	} {
+		t.Run(ca, func(t *testing.T) {
 			ln, err := net.Listen("tcp", "127.0.0.1:9121")
 			require.NoError(t, err)
 			defer ln.Close()
@@ -253,9 +259,11 @@ func TestReadMetadata(t *testing.T) {
 				err = rconn.ServerHandshake()
 				require.NoError(t, err)
 
-		videoTrack, audioTrack, err := rconn.ReadMetadata()
+				videoTrack, audioTrack, err := rconn.ReadTracks()
 				require.NoError(t, err)
 
+				switch ca {
+				case "standard":
 					videoTrack2, err := gortsplib.NewTrackH264(96,
 						[]byte{
 							0x67, 0x64, 0x00, 0x0c, 0xac, 0x3b, 0x50, 0xb0,
@@ -273,6 +281,40 @@ func TestReadMetadata(t *testing.T) {
 					require.NoError(t, err)
 					require.Equal(t, audioTrack2, audioTrack)
 
+				case "empty metadata":
+					videoTrack2, err := gortsplib.NewTrackH264(96,
+						[]byte{
+							0x67, 0x64, 0x00, 0x32, 0xac, 0x2c, 0x6a, 0x80,
+							0xa8, 0x02, 0xfe, 0x9b, 0x82, 0x80, 0x82, 0xa0,
+							0x00, 0x00, 0x03, 0x00, 0x20, 0x00, 0x00, 0x06,
+							0x50, 0x80,
+						},
+						[]byte{
+							0x68, 0xee, 0x31, 0xb2, 0x1b,
+						},
+						nil)
+					require.NoError(t, err)
+					require.Equal(t, videoTrack2, videoTrack)
+
+					require.Equal(t, (*gortsplib.TrackAAC)(nil), audioTrack)
+
+				case "no metadata":
+					videoTrack2, err := gortsplib.NewTrackH264(96,
+						[]byte{
+							0x27, 0x42, 0xe0, 0x1f, 0x8d, 0x68, 0x05, 0x00,
+							0x5b, 0xa1, 0x00, 0x00, 0x03, 0x00, 0x01, 0x00,
+							0x00, 0x03, 0x00, 0x1e, 0x0f, 0x10, 0x7a, 0x80,
+						},
+						[]byte{
+							0x28, 0xce, 0x32, 0x48,
+						},
+						nil)
+					require.NoError(t, err)
+					require.Equal(t, videoTrack2, videoTrack)
+
+					require.Equal(t, (*gortsplib.TrackAAC)(nil), audioTrack)
+				}
+
 				close(done)
 			}()
 
@@ -469,6 +511,8 @@ func TestReadMetadata(t *testing.T) {
 				},
 			}, arr)
 
+			switch ca {
+			case "standard":
 				// C->S metadata
 				byts = flvio.FillAMF0ValsMalloc([]interface{}{
 					"@setDataFrame",
@@ -548,10 +592,79 @@ func TestReadMetadata(t *testing.T) {
 				}.write(conn)
 				require.NoError(t, err)
 
+			case "empty metadata":
+				// C->S metadata
+				byts = flvio.FillAMF0ValsMalloc([]interface{}{
+					"@setDataFrame",
+					"onMetaData",
+					flvio.AMFMap{
+						{
+							K: "width",
+							V: float64(2688),
+						},
+						{
+							K: "height",
+							V: float64(1520),
+						},
+						{
+							K: "framerate",
+							V: float64(0o25),
+						},
+					},
+				})
+				err = chunk0{
+					chunkStreamID: 4,
+					typ:           0x12,
+					streamID:      1,
+					bodyLen:       uint32(len(byts)),
+					body:          byts,
+				}.write(conn)
+				require.NoError(t, err)
+
+				// C->S H264 decoder config
+				byts := []byte{
+					0x17, 0x00, 0x00, 0x00, 0x00, 0x01, 0x64, 0x00,
+					0x32, 0xff, 0xe1, 0x00, 0x1a, 0x67, 0x64, 0x00,
+					0x32, 0xac, 0x2c, 0x6a, 0x80, 0xa8, 0x02, 0xfe,
+					0x9b, 0x82, 0x80, 0x82, 0xa0, 0x00, 0x00, 0x03,
+					0x00, 0x20, 0x00, 0x00, 0x06, 0x50, 0x80, 0x01,
+					0x00, 0x05, 0x68, 0xee, 0x31, 0xb2, 0x1b,
+				}
+				err = chunk0{
+					chunkStreamID: 21,
+					typ:           0x09,
+					streamID:      1,
+					bodyLen:       uint32(len(byts)),
+					body:          byts,
+				}.write(conn)
+				require.NoError(t, err)
+
+			case "no metadata":
+				// C->S H264 decoder config
+				byts := []byte{
+					0x17, 0x00, 0x00, 0x00, 0x00, 0x01, 0x42, 0xe0,
+					0x1f, 0xff, 0xe1, 0x00, 0x18, 0x27, 0x42, 0xe0,
+					0x1f, 0x8d, 0x68, 0x05, 0x00, 0x5b, 0xa1, 0x00,
+					0x00, 0x03, 0x00, 0x01, 0x00, 0x00, 0x03, 0x00,
+					0x1e, 0x0f, 0x10, 0x7a, 0x80, 0x01, 0x00, 0x04,
+					0x28, 0xce, 0x32, 0x48,
+				}
+				err = chunk0{
+					chunkStreamID: 4,
+					typ:           0x09,
+					streamID:      1,
+					bodyLen:       uint32(len(byts)),
+					body:          byts,
+				}.write(conn)
+				require.NoError(t, err)
+			}
+
 			<-done
+		})
+	}
 }
 
-func TestWriteMetadata(t *testing.T) {
+func TestWriteTracks(t *testing.T) {
 	ln, err := net.Listen("tcp", "127.0.0.1:9121")
 	require.NoError(t, err)
 	defer ln.Close()
@@ -580,7 +693,7 @@ func TestWriteMetadata(t *testing.T) {
 		audioTrack, err := gortsplib.NewTrackAAC(96, 2, 44100, 2, nil)
 		require.NoError(t, err)
 
-		err = rconn.WriteMetadata(videoTrack, audioTrack)
+		err = rconn.WriteTracks(videoTrack, audioTrack)
 		require.NoError(t, err)
 	}()