Implement NormalizeH264AccessUnit for H.264 payload normalization and update live stream publishing to use the new function

machinery/data/tus/1785961483_3-259_encrypted_0-0-0-0_-1_20373.mp4.primary.json

@@ -0,0 +1 @@
+{"upload_url":"https://vault.kerberos.io/api/storage/tus/7a192d02cd47ede3939ddfa5f8819465","vault_uri":"https://vault.kerberos.io/api/storage/tus/","size":4739581}

machinery/src/cloud/livemoq/annexb.go

@@ -1,6 +1,11 @@
 package livemoq
 
-import "strings"
+import (
+	"bytes"
+	"strings"
+
+	"github.com/bluenviron/mediacommon/pkg/codecs/h264"
+)
 
 var annexBStartCode = []byte{0x00, 0x00, 0x00, 0x01}
 
@@ -15,6 +20,37 @@ func EnsureAnnexB(payload []byte) []byte {
 	return append(framed, payload...)
 }
 
+// NormalizeH264AccessUnit removes delimiters and duplicate parameter sets that
+// can make older MoQ splitters emit a parameter-only frame before the IDR.
+func NormalizeH264AccessUnit(payload []byte) ([]byte, error) {
+	nalus, err := h264.AnnexBUnmarshal(EnsureAnnexB(payload))
+	if err != nil {
+		return nil, err
+	}
+
+	normalized := make([][]byte, 0, len(nalus))
+	for _, nalu := range nalus {
+		if len(nalu) == 0 || nalu[0]&0x1f == 9 {
+			continue
+		}
+		if nalu[0]&0x1f == 7 || nalu[0]&0x1f == 8 {
+			duplicate := false
+			for _, existing := range normalized {
+				if bytes.Equal(existing, nalu) {
+					duplicate = true
+					break
+				}
+			}
+			if duplicate {
+				continue
+			}
+		}
+		normalized = append(normalized, nalu)
+	}
+
+	return h264.AnnexBMarshal(normalized)
+}
+
 func BroadcastPath(prefix string, deviceKey string) string {
 	prefix = strings.Trim(prefix, "/")
 	if prefix == "" {

machinery/src/cloud/livemoq/annexb_test.go

@@ -37,6 +37,33 @@ func TestEnsureAnnexB(t *testing.T) {
 	}
 }
 
+func TestNormalizeH264AccessUnit(t *testing.T) {
+	startCode := []byte{0x00, 0x00, 0x00, 0x01}
+	sps := []byte{0x67, 0x42, 0x00, 0x1f}
+	pps := []byte{0x68, 0xce, 0x06, 0xe2}
+	aud := []byte{0x09, 0xf0}
+	idr := []byte{0x65, 0x88, 0x84}
+
+	payload := make([]byte, 0)
+	for _, nalu := range [][]byte{sps, pps, aud, sps, pps, idr} {
+		payload = append(payload, startCode...)
+		payload = append(payload, nalu...)
+	}
+
+	got, err := NormalizeH264AccessUnit(payload)
+	if err != nil {
+		t.Fatal(err)
+	}
+	want := make([]byte, 0)
+	for _, nalu := range [][]byte{sps, pps, idr} {
+		want = append(want, startCode...)
+		want = append(want, nalu...)
+	}
+	if !bytes.Equal(got, want) {
+		t.Fatalf("NormalizeH264AccessUnit() = %x, want %x", got, want)
+	}
+}
+
 func TestBroadcastPath(t *testing.T) {
 	if got := BroadcastPath("/devices/", "/camera-1/"); got != "devices/camera-1/live.hang" {
 		t.Fatalf("BroadcastPath() = %q", got)

machinery/src/cloud/livemoq_enabled.go

@@ -129,7 +129,7 @@ func publishLiveStreamMoQ(ctx context.Context, config liveMoQConfig) error {
 	}
 	defer broadcast.Finish()
 
-	stream, err := broadcast.PublishMediaStream("avc3")
+	stream, err := broadcast.PublishMedia("avc3", nil)
 	if err != nil {
 		return fmt.Errorf("create H.264 media stream: %w", err)
 	}
@@ -152,7 +152,19 @@ func publishLiveStreamMoQ(ctx context.Context, config liveMoQConfig) error {
 			writing = true
 			log.Log.Info("cloud.publishLiveStreamMoQ(): first H.264 keyframe received; broadcast is live")
 		}
-		if err := stream.Write(livemoq.EnsureAnnexB(packet.Data)); err != nil {
+		presentationTimeMs := packet.Time + packet.CompositionTime
+		if presentationTimeMs < 0 {
+			presentationTimeMs = 0
+		}
+		payload, err := livemoq.NormalizeH264AccessUnit(packet.Data)
+		if err != nil {
+			return fmt.Errorf("normalize H.264 access unit: %w", err)
+		}
+		frame := moq.Frame{
+			Payload:     payload,
+			TimestampUs: uint64(presentationTimeMs) * 1000,
+		}
+		if err := stream.WriteFrame(frame); err != nil {
 			return fmt.Errorf("write H.264 access unit: %w", err)
 		}
 	}