Rework RTSP and RTMP processing

pkg/h264/README.md

@@ -1,3 +1,11 @@
+# H264
+
+Access Unit (AU) can contain one or multiple NAL Unit:
+
+1. [SEI,] SPS, PPS, IFrame, [IFrame...]
+2. BFrame, [BFrame...]
+3. IFrame, [IFrame...]
+
 ## RTP H264
 
 Camera | NALu

pkg/h264/avc.go

@@ -12,49 +12,33 @@ func IsAVC(codec *streamer.Codec) bool {
 	return codec.PayloadType == PayloadTypeAVC
 }
 
-func EncodeAVC(raw []byte) (avc []byte) {
+func EncodeAVC(nals ...[]byte) (avc []byte) {
-	avc = make([]byte, len(raw)+4)
+	n := 4 * len(nals)
-	binary.BigEndian.PutUint32(avc, uint32(len(raw)))
+	for _, nal := range nals {
-	copy(avc[4:], raw)
+		n += len(nal)
+	}
+
+	avc = make([]byte, n)
+
+	var i int
+	for _, nal := range nals {
+		binary.BigEndian.PutUint32(avc[i:], uint32(len(nal)))
+		i += 4 + copy(avc[i+4:], nal)
+	}
+
 	return
 }
 
 func RepairAVC(track *streamer.Track) streamer.WrapperFunc {
 	sps, pps := GetParameterSet(track.Codec.FmtpLine)
-	sps = EncodeAVC(sps)
+	ps := EncodeAVC(sps, pps)
-	pps = EncodeAVC(pps)
 
 	return func(push streamer.WriterFunc) streamer.WriterFunc {
 		return func(packet *rtp.Packet) (err error) {
-			naluType := NALUType(packet.Payload)
+			if NALUType(packet.Payload) == NALUTypeIFrame {
-			switch naluType {
+				packet.Payload = Join(ps, packet.Payload)
-			case NALUTypeSPS:
-				sps = packet.Payload
-				return
-			case NALUTypePPS:
-				pps = packet.Payload
-				return
 			}
-
+			return push(packet)
-			var clone rtp.Packet
-
-			if naluType == NALUTypeIFrame {
-				clone = *packet
-				clone.Payload = sps
-				if err = push(&clone); err != nil {
-					return
-				}
-
-				clone = *packet
-				clone.Payload = pps
-				if err = push(&clone); err != nil {
-					return
-				}
-			}
-
-			clone = *packet
-			clone.Payload = packet.Payload
-			return push(&clone)
 		}
 	}
 }

pkg/h264/helper.go

@@ -2,6 +2,7 @@ package h264
 
 import (
 	"encoding/base64"
+	"encoding/binary"
 	"github.com/AlexxIT/go2rtc/pkg/streamer"
 	"strings"
 )
@@ -18,8 +19,31 @@ func NALUType(b []byte) byte {
 	return b[4] & 0x1F
 }
 
+// IsKeyframe - check if any NALU in one AU is Keyframe
 func IsKeyframe(b []byte) bool {
-	return NALUType(b) == NALUTypeIFrame
+	for {
+		switch NALUType(b) {
+		case NALUTypePFrame:
+			return false
+		case NALUTypeIFrame:
+			return true
+		}
+
+		size := int(binary.BigEndian.Uint32(b)) + 4
+		if size < len(b) {
+			b = b[size:]
+			continue
+		} else {
+			return false
+		}
+	}
+}
+
+func Join(ps, iframe []byte) []byte {
+	b := make([]byte, len(ps)+len(iframe))
+	i := copy(b, ps)
+	copy(b[i:], iframe)
+	return b
 }
 
 func GetProfileLevelID(fmtp string) string {

pkg/h264/rtp.go

@@ -1,7 +1,6 @@
 package h264
 
 import (
-	"encoding/binary"
 	"github.com/AlexxIT/go2rtc/pkg/streamer"
 	"github.com/pion/rtp"
 	"github.com/pion/rtp/codecs"
@@ -13,8 +12,7 @@ func RTPDepay(track *streamer.Track) streamer.WrapperFunc {
 	depack := &codecs.H264Packet{IsAVC: true}
 
 	sps, pps := GetParameterSet(track.Codec.FmtpLine)
-	sps = EncodeAVC(sps)
+	ps := EncodeAVC(sps, pps)
-	pps = EncodeAVC(pps)
 
 	var buffer []byte
 
@@ -27,83 +25,32 @@ func RTPDepay(track *streamer.Track) streamer.WrapperFunc {
 			//	packet.PayloadType, packet.SSRC, packet.SequenceNumber,
 			//)
 
-			data, err := depack.Unmarshal(packet.Payload)
+			payload, err := depack.Unmarshal(packet.Payload)
-			if len(data) == 0 || err != nil {
+			if len(payload) == 0 || err != nil {
 				return nil
 			}
 
-			for {
+			// ffmpeg with `-tune zerolatency` enable option `-x264opts sliced-threads=1`
-				unitType := NALUType(data)
+			// and every NALU will be sliced to multiple NALUs
-				//fmt.Printf("[H264] nalu: %2d, size: %6d\n", unitType, len(data))
+			if !packet.Marker {
-
+				buffer = append(buffer, payload...)
-				// multiple 5 and 1 in one payload is OK
+				return nil
-				if unitType != NALUTypeIFrame && unitType != NALUTypePFrame {
-					i := int(binary.BigEndian.Uint32(data)) + 4
-					if i < len(data) {
-						data0 := data[:i] // NAL Unit with AVC header
-						data = data[i:]
-						switch unitType {
-						case NALUTypeSPS:
-							sps = data0
-							continue
-						case NALUTypePPS:
-							pps = data0
-							continue
-						case NALUTypeSEI:
-							// some unnecessary text information
-							continue
-						}
-					}
-				}
-
-				switch unitType {
-				case NALUTypeSPS:
-					sps = data
-					return nil
-				case NALUTypePPS:
-					pps = data
-					return nil
-				case NALUTypeSEI:
-					// some unnecessary text information
-					return nil
-				}
-
-				// ffmpeg with `-tune zerolatency` enable option `-x264opts sliced-threads=1`
-				// and every NALU will be sliced to multiple NALUs
-				if !packet.Marker {
-					buffer = append(buffer, data...)
-					return nil
-				}
-
-				if buffer != nil {
-					buffer = append(buffer, data...)
-					data = buffer
-					buffer = nil
-				}
-
-				var clone rtp.Packet
-
-				if unitType == NALUTypeIFrame {
-					clone = *packet
-					clone.Version = RTPPacketVersionAVC
-					clone.Payload = sps
-					if err = push(&clone); err != nil {
-						return err
-					}
-
-					clone = *packet
-					clone.Version = RTPPacketVersionAVC
-					clone.Payload = pps
-					if err = push(&clone); err != nil {
-						return err
-					}
-				}
-
-				clone = *packet
-				clone.Version = RTPPacketVersionAVC
-				clone.Payload = data
-				return push(&clone)
 			}
+
+			if buffer != nil {
+				payload = append(buffer, payload...)
+				buffer = nil
+			}
+
+			switch NALUType(payload) {
+			case NALUTypeIFrame:
+				payload = Join(ps, payload)
+			}
+
+			clone := *packet
+			clone.Version = RTPPacketVersionAVC
+			clone.Payload = payload
+			return push(&clone)
 		}
 	}
 }
@@ -117,11 +64,12 @@ func RTPPay(mtu uint16) streamer.WrapperFunc {
 		return func(packet *rtp.Packet) error {
 			if packet.Version == RTPPacketVersionAVC {
 				payloads := payloader.Payload(mtu, packet.Payload)
+				last := len(payloads) - 1
 				for i, payload := range payloads {
 					clone := rtp.Packet{
 						Header: rtp.Header{
 							Version: 2,
-							Marker:  i == len(payloads)-1,
+							Marker:  i == last,
 							//PayloadType:    packet.PayloadType,
 							SequenceNumber: sequencer.NextSequenceNumber(),
 							Timestamp:      packet.Timestamp,

pkg/ivideon/client.go

@@ -245,14 +245,12 @@ func (c *Client) worker() {
 			time.Sleep(d)
 
 			// can be SPS, PPS and IFrame in one packet
-			for _, payload := range h264.SplitAVC(data[:entry.Size]) {
+			packet := &rtp.Packet{
-				packet := &rtp.Packet{
+				// ivideon clockrate=1000, RTP clockrate=90000
-					// ivideon clockrate=1000, RTP clockrate=90000
+				Header:  rtp.Header{Timestamp: ts * 90},
-					Header:  rtp.Header{Timestamp: ts * 90},
+				Payload: data[:entry.Size],
-					Payload: payload,
-				}
-				_ = track.WriteRTP(packet)
 			}
+			_ = track.WriteRTP(packet)
 
 			data = data[entry.Size:]
 			ts += entry.Duration

pkg/mp4/consumer.go

@@ -53,15 +53,16 @@ func (c *Consumer) AddTrack(media *streamer.Media, track *streamer.Track) *strea
 				return nil
 			}
 
-			switch h264.NALUType(packet.Payload) {
+			if c.muxer == nil {
-			case h264.NALUTypeIFrame:
+				return nil
-				c.start = true
+			}
-			case h264.NALUTypePFrame:
+
-				if !c.start {
+			if !c.start {
+				if h264.IsKeyframe(packet.Payload) {
+					c.start = true
+				} else {
 					return nil
 				}
-			default:
-				return nil
 			}
 
 			buf := c.muxer.Marshal(packet)
@@ -71,10 +72,13 @@ func (c *Consumer) AddTrack(media *streamer.Media, track *streamer.Track) *strea
 			return nil
 		}
 
-		if !h264.IsAVC(codec) {
+		var wrapper streamer.WrapperFunc
-			wrapper := h264.RTPDepay(track)
+		if h264.IsAVC(codec) {
-			push = wrapper(push)
+			wrapper = h264.RepairAVC(track)
+		} else {
+			wrapper = h264.RTPDepay(track)
 		}
+		push = wrapper(push)
 
 		return track.Bind(push)
 

pkg/rtmp/client.go

@@ -136,20 +136,11 @@ func (c *Client) Handle() (err error) {
 		// convert seconds to RTP timestamp
 		timestamp := uint32(pkt.Time * time.Duration(track.Codec.ClockRate) / time.Second)
 
-		var payloads [][]byte
+		packet := &rtp.Packet{
-		if track.Codec.Name == streamer.CodecH264 {
+			Header:  rtp.Header{Timestamp: timestamp},
-			payloads = h264.SplitAVC(pkt.Data)
+			Payload: pkt.Data,
-		} else {
-			payloads = [][]byte{pkt.Data}
-		}
-
-		for _, payload := range payloads {
-			packet := &rtp.Packet{
-				Header:  rtp.Header{Timestamp: timestamp},
-				Payload: payload,
-			}
-			_ = track.WriteRTP(packet)
 		}
+		_ = track.WriteRTP(packet)
 	}
 }