From 8fe8233e531f78791507e5f22dda372f37713ada Mon Sep 17 00:00:00 2001
From: Eli Mallon <eli@iame.li>
Date: Wed, 30 Sep 2026 17:04:38 -0700
Subject: [PATCH] rtmp: test the relay writes the replayed header as a sequence
 header

Pull the playback write loop out as relayRTMPSession, against gortmplib's
Conn interface, and check what it writes for a stream's start: the
replayed SPS/PPS access unit goes out as a sequence header, and the first
keyframe is the only frame at its timestamp. A session without an audio
track no longer hands the writer a nil track.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 pkg/api/rtmp_server.go      | 20 +++++++++--
 pkg/api/rtmp_server_test.go | 69 +++++++++++++++++++++++++++++++++++++
 2 files changed, 86 insertions(+), 3 deletions(-)

diff --git a/pkg/api/rtmp_server.go b/pkg/api/rtmp_server.go
index c8f216d6f..8c11fcaf5 100644
--- a/pkg/api/rtmp_server.go
+++ b/pkg/api/rtmp_server.go
@@ -134,9 +134,23 @@ func (a *StreamplaceAPI) HandleRTMPPlayback(ctx context.Context, sc *gortmplib.S
 		return fmt.Errorf("RTMP session not found for streamer %s", streamer)
 	}
 
+	return relayRTMPSession(ctx, sc, session)
+}
+
+// relayRTMPSession writes a publisher's session out to conn, the internal
+// playback connection the ingest pipeline reads with rtmp2src, until the
+// session closes or ctx ends.
+func relayRTMPSession(ctx context.Context, conn gortmplib.Conn, session *media.RTMPSession) error {
+	tracks := []format.Format{}
+	if session.VideoTrack != nil {
+		tracks = append(tracks, session.VideoTrack)
+	}
+	if session.AudioTrack != nil {
+		tracks = append(tracks, session.AudioTrack)
+	}
 	w := &gortmplib.Writer{
-		Conn:   sc,
-		Tracks: []format.Format{session.VideoTrack, session.AudioTrack},
+		Conn:   conn,
+		Tracks: tracks,
 	}
 	err := w.Initialize()
 	if err != nil {
@@ -159,7 +173,7 @@ func (a *StreamplaceAPI) HandleRTMPPlayback(ctx context.Context, sc *gortmplib.S
 				// h264parse drops that keyframe's PTS, which fails the
 				// segmenter's muxer. Write it as the sequence header it was.
 				if msg, ok := h264ConfigMessage(event.AU, event.DTS); ok {
-					if err := sc.Write(msg); err != nil {
+					if err := conn.Write(msg); err != nil {
 						return fmt.Errorf("error writing H264 config: %w", err)
 					}
 					continue
diff --git a/pkg/api/rtmp_server_test.go b/pkg/api/rtmp_server_test.go
index 3eff83f5b..fba4f7b84 100644
--- a/pkg/api/rtmp_server_test.go
+++ b/pkg/api/rtmp_server_test.go
@@ -1,12 +1,16 @@
 package api
 
 import (
+	"context"
+	"errors"
 	"testing"
 	"time"
 
 	"github.com/bluenviron/gortmplib/pkg/h264conf"
 	"github.com/bluenviron/gortmplib/pkg/message"
+	"github.com/bluenviron/gortsplib/v5/pkg/format"
 	"github.com/stretchr/testify/require"
+	"stream.place/streamplace/pkg/media"
 )
 
 func TestH264ConfigMessage(t *testing.T) {
@@ -38,3 +42,68 @@ func TestH264ConfigMessage(t *testing.T) {
 		require.False(t, ok, name)
 	}
 }
+
+// recordingConn is a gortmplib.Conn that keeps what's written to it.
+type recordingConn struct {
+	msgs []message.Message
+}
+
+func (c *recordingConn) BytesReceived() uint64          { return 0 }
+func (c *recordingConn) BytesSent() uint64              { return 0 }
+func (c *recordingConn) Read() (message.Message, error) { return nil, errors.New("write-only") }
+func (c *recordingConn) Write(msg message.Message) error {
+	c.msgs = append(c.msgs, msg)
+	return nil
+}
+
+// What the relay writes for a stream's start as gortmplib's reader hands it
+// on: the replayed sequence header as an SPS/PPS access unit, then the first
+// keyframe at the same timestamp. The header has to go out as a sequence
+// header, not a frame of its own, so the keyframe is the only frame at that
+// timestamp.
+func TestRelayRTMPSessionSequenceHeader(t *testing.T) {
+	sps := []byte{0x67, 0x42, 0xc0, 0x15, 0xda, 0x06, 0xc7, 0xf9}
+	pps := []byte{0x68, 0xce, 0x3c, 0x80}
+	idr := []byte{0x65, 0x88, 0x84, 0x00}
+	slice := []byte{0x41, 0x9a, 0x02}
+
+	session := &media.RTMPSession{
+		EventChan:  make(chan any, 8),
+		VideoTrack: &format.H264{PayloadTyp: 96, SPS: sps, PPS: pps, PacketizationMode: 1},
+	}
+	session.EventChan <- &media.RTMPH264Data{AU: [][]byte{sps, pps}}
+	session.EventChan <- &media.RTMPH264Data{AU: [][]byte{idr}}
+	session.EventChan <- &media.RTMPH264Data{AU: [][]byte{slice}, PTS: 33 * time.Millisecond, DTS: 33 * time.Millisecond}
+	close(session.EventChan)
+
+	conn := &recordingConn{}
+	err := relayRTMPSession(context.Background(), conn, session)
+	require.ErrorContains(t, err, "session closed")
+
+	var video []*message.Video
+	for _, msg := range conn.msgs {
+		if v, ok := msg.(*message.Video); ok {
+			video = append(video, v)
+		}
+	}
+	// the writer's own header for the track, the relayed one, then frames
+	require.Len(t, video, 4)
+	for i, want := range []struct {
+		typ message.VideoType
+		dts time.Duration
+		key bool
+	}{
+		{message.VideoTypeConfig, 0, true},
+		{message.VideoTypeConfig, 0, true},
+		{message.VideoTypeAU, 0, true},
+		{message.VideoTypeAU, 33 * time.Millisecond, false},
+	} {
+		require.Equal(t, want.typ, video[i].Type, "message %d", i)
+		require.Equal(t, want.dts, video[i].DTS, "message %d", i)
+		require.Equal(t, want.key, video[i].IsKeyFrame, "message %d", i)
+	}
+	var conf h264conf.Conf
+	require.NoError(t, conf.Unmarshal(video[1].Payload))
+	require.Equal(t, sps, conf.SPS)
+	require.Equal(t, pps, conf.PPS)
+}
-- 
knotmirror 0.0.1