rtmp: fix compatibility with some dji drones (#928)

internal/core/rtmp_conn.go

@@ -591,9 +591,29 @@ func (c *rtmpConn) runPublish(ctx context.Context) error {
 				return err
 			}
 
+			// skip invalid NALUs sent by DJI
+			n := 0
+			for _, nalu := range nalus {
+				if len(nalu) != 0 {
+					n++
+				}
+			}
+			if n == 0 {
+				continue
+			}
+
+			validNALUs := make([][]byte, n)
+			pos := 0
+			for _, nalu := range nalus {
+				if len(nalu) != 0 {
+					validNALUs[pos] = nalu
+					pos++
+				}
+			}
+
 			pts := pkt.Time + pkt.CTime
 
-			pkts, err := h264Encoder.Encode(nalus, pts)
+			pkts, err := h264Encoder.Encode(validNALUs, pts)
 			if err != nil {
 				return fmt.Errorf("error while encoding H264: %v", err)
 			}
@@ -610,8 +630,8 @@ func (c *rtmpConn) runPublish(ctx context.Context) error {
 					rres.stream.writeData(&data{
 						trackID:      videoTrackID,
 						rtp:          pkt,
-						ptsEqualsDTS: h264.IDRPresent(nalus),
-						h264NALUs:    nalus,
+						ptsEqualsDTS: h264.IDRPresent(validNALUs),
+						h264NALUs:    validNALUs,
 						h264PTS:      pts,
 					})
 				}

internal/rtmp/conn.go

@@ -92,6 +92,17 @@ func trackFromH264DecoderConfig(data []byte) (*gortsplib.TrackH264, error) {
 	return gortsplib.NewTrackH264(96, codec.SPS[0], codec.PPS[0], nil)
 }
 
+func trackFromAACDecoderConfig(data []byte) (*gortsplib.TrackAAC, error) {
+	var mpegConf aac.MPEG4AudioConfig
+	err := mpegConf.Decode(data)
+	if err != nil {
+		return nil, err
+	}
+
+	return gortsplib.NewTrackAAC(96, int(mpegConf.Type), mpegConf.SampleRate,
+		mpegConf.ChannelCount, mpegConf.AOTSpecificConfig, 13, 3, 3)
+}
+
 var errEmptyMetadata = errors.New("metadata is empty")
 
 func (c *Conn) readTracksFromMetadata(pkt av.Packet) (*gortsplib.TrackH264, *gortsplib.TrackAAC, error) {
@@ -203,14 +214,7 @@ func (c *Conn) readTracksFromMetadata(pkt av.Packet) (*gortsplib.TrackH264, *gor
 				return nil, nil, fmt.Errorf("audio track setupped twice")
 			}
 
-			var mpegConf aac.MPEG4AudioConfig
-			err := mpegConf.Decode(pkt.Data)
-			if err != nil {
-				return nil, nil, err
-			}
-
-			audioTrack, err = gortsplib.NewTrackAAC(96, int(mpegConf.Type), mpegConf.SampleRate,
-				mpegConf.ChannelCount, mpegConf.AOTSpecificConfig, 13, 3, 3)
+			audioTrack, err = trackFromAACDecoderConfig(pkt.Data)
 			if err != nil {
 				return nil, nil, err
 			}
@@ -223,6 +227,61 @@ func (c *Conn) readTracksFromMetadata(pkt av.Packet) (*gortsplib.TrackH264, *gor
 	}
 }
 
+func (c *Conn) readTracksFromPackets(pkt av.Packet) (*gortsplib.TrackH264, *gortsplib.TrackAAC, error) {
+	startTime := pkt.Time
+	var videoTrack *gortsplib.TrackH264
+	var audioTrack *gortsplib.TrackAAC
+
+	// analyze 1 second of packets
+	for {
+		switch pkt.Type {
+		case av.H264DecoderConfig:
+			if videoTrack == nil {
+				var err error
+				videoTrack, err = trackFromH264DecoderConfig(pkt.Data)
+				if err != nil {
+					return nil, nil, err
+				}
+
+				// stop the analysis if both tracks are found
+				if videoTrack != nil && audioTrack != nil {
+					return videoTrack, audioTrack, nil
+				}
+			}
+
+		case av.AACDecoderConfig:
+			if audioTrack == nil {
+				var err error
+				audioTrack, err = trackFromAACDecoderConfig(pkt.Data)
+				if err != nil {
+					return nil, nil, err
+				}
+
+				// stop the analysis if both tracks are found
+				if videoTrack != nil && audioTrack != nil {
+					return videoTrack, audioTrack, nil
+				}
+			}
+		}
+
+		if (pkt.Time - startTime) >= 1*time.Second {
+			break
+		}
+
+		var err error
+		pkt, err = c.ReadPacket()
+		if err != nil {
+			return nil, nil, err
+		}
+	}
+
+	if videoTrack == nil && audioTrack == nil {
+		return nil, nil, fmt.Errorf("no tracks found")
+	}
+
+	return videoTrack, audioTrack, nil
+}
+
 // ReadTracks reads track informations.
 func (c *Conn) ReadTracks() (*gortsplib.TrackH264, *gortsplib.TrackAAC, error) {
 	pkt, err := c.ReadPacket()
@@ -230,8 +289,7 @@ func (c *Conn) ReadTracks() (*gortsplib.TrackH264, *gortsplib.TrackAAC, error) {
 		return nil, nil, err
 	}
 
-	switch pkt.Type {
-	case av.Metadata:
+	if pkt.Type == av.Metadata {
 		videoTrack, audioTrack, err := c.readTracksFromMetadata(pkt)
 		if err != nil {
 			if err == errEmptyMetadata {
@@ -240,34 +298,26 @@ func (c *Conn) ReadTracks() (*gortsplib.TrackH264, *gortsplib.TrackAAC, error) {
 					return nil, nil, err
 				}
 
-				if pkt.Type != av.H264DecoderConfig {
-					return nil, nil, fmt.Errorf("unexpected packet (%v)", pkt.Type)
-				}
-
-				videoTrack, err := trackFromH264DecoderConfig(pkt.Data)
+				videoTrack, audioTrack, err := c.readTracksFromPackets(pkt)
 				if err != nil {
 					return nil, nil, err
 				}
 
-				return videoTrack, nil, nil
+				return videoTrack, audioTrack, 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)
+	videoTrack, audioTrack, err := c.readTracksFromPackets(pkt)
+	if err != nil {
+		return nil, nil, err
 	}
+
+	return videoTrack, audioTrack, nil
 }
 
 // WriteTracks writes track informations.

internal/rtmp/conn_test.go

@@ -62,7 +62,7 @@ func TestReadTracks(t *testing.T) {
 	for _, ca := range []string{
 		"standard",
 		"metadata without codec id",
-		"no metadata, video + audio",
+		"missing metadata",
 	} {
 		t.Run(ca, func(t *testing.T) {
 			ln, err := net.Listen("tcp", "127.0.0.1:9121")
@@ -98,14 +98,18 @@ func TestReadTracks(t *testing.T) {
 					require.NoError(t, err)
 					require.Equal(t, videoTrack2, videoTrack)
 
-					require.Equal(t, (*gortsplib.TrackAAC)(nil), audioTrack)
+					audioTrack2, err := gortsplib.NewTrackAAC(96, 2, 44100, 2, nil, 13, 3, 3)
+					require.NoError(t, err)
+					require.Equal(t, audioTrack2, audioTrack)
 
-				case "no metadata, video + audio":
+				case "missing metadata":
 					videoTrack2, err := gortsplib.NewTrackH264(96, sps, pps, nil)
 					require.NoError(t, err)
 					require.Equal(t, videoTrack2, videoTrack)
 
-					require.Equal(t, (*gortsplib.TrackAAC)(nil), audioTrack)
+					audioTrack2, err := gortsplib.NewTrackAAC(96, 2, 44100, 2, nil, 13, 3, 3)
+					require.NoError(t, err)
+					require.Equal(t, audioTrack2, audioTrack)
 				}
 
 				close(done)
@@ -410,7 +414,25 @@ func TestReadTracks(t *testing.T) {
 				})
 				require.NoError(t, err)
 
-			case "no metadata, video + audio":
+				// C->S AAC decoder config
+				enc, err := aac.MPEG4AudioConfig{
+					Type:         2,
+					SampleRate:   44100,
+					ChannelCount: 2,
+				}.Encode()
+				require.NoError(t, err)
+				err = mrw.Write(&message.MsgAudio{
+					ChunkStreamID:   4,
+					MessageStreamID: 1,
+					Rate:            flvio.SOUND_44Khz,
+					Depth:           flvio.SOUND_16BIT,
+					Channels:        flvio.SOUND_STEREO,
+					AACType:         flvio.AAC_SEQHDR,
+					Payload:         enc,
+				})
+				require.NoError(t, err)
+
+			case "missing metadata":
 				// C->S H264 decoder config
 				codec := nh264.Codec{
 					SPS: map[int][]byte{
@@ -431,6 +453,24 @@ func TestReadTracks(t *testing.T) {
 					Payload:         b[:n],
 				})
 				require.NoError(t, err)
+
+				// C->S AAC decoder config
+				enc, err := aac.MPEG4AudioConfig{
+					Type:         2,
+					SampleRate:   44100,
+					ChannelCount: 2,
+				}.Encode()
+				require.NoError(t, err)
+				err = mrw.Write(&message.MsgAudio{
+					ChunkStreamID:   4,
+					MessageStreamID: 1,
+					Rate:            flvio.SOUND_44Khz,
+					Depth:           flvio.SOUND_16BIT,
+					Channels:        flvio.SOUND_STEREO,
+					AACType:         flvio.AAC_SEQHDR,
+					Payload:         enc,
+				})
+				require.NoError(t, err)
 			}
 
 			<-done