ai/live: Gateway-native WHEP server.

Author: j0sh

core/livepeernode.go

@@ -19,6 +19,7 @@ import (
 	"time"
 
 	"github.com/golang/glog"
+	"github.com/livepeer/go-livepeer/media"
 	"github.com/livepeer/go-livepeer/pm"
 	"github.com/livepeer/go-livepeer/trickle"
 
@@ -174,6 +175,9 @@ type LivePipeline struct {
 	Pipeline    string
 	ControlPub  *trickle.TricklePublisher
 	StopControl func()
+	OutCond     *sync.Cond
+	OutWriter   *media.RingBuffer
+	Closed      bool
 }
 
 // NewLivepeerNode creates a new Livepeer Node. Eth can be nil.

media/rtp_timestamper.go

@@ -0,0 +1,37 @@
+package media
+
+const (
+	// MPEG‑TS PTS is 33 bits wide; it wraps at 2^33.
+	mpegtsMaxPts = 1 << 33
+	halfPts      = mpegtsMaxPts / 2
+)
+
+// RTPTimestamper keeps track of wrap‑arounds in a 33‑bit PTS stream.
+type RTPTimestamper struct {
+	lastRaw     int64 // last raw PTS seen
+	wrapCount   int64 // how many times we've wrapped
+	initialized bool  // false until first Unwrap call
+}
+
+// NewRTPTimestamper creates a fresh unwrapper.
+func NewRTPTimestamper() *RTPTimestamper {
+	return &RTPTimestamper{}
+}
+
+// Unwrap returns a monotonic 64‑bit timeline by accounting for wrap‑arounds.
+func (u *RTPTimestamper) ToTS(rawPts int64) uint32 {
+	if u.initialized {
+		delta := rawPts - u.lastRaw
+		if delta < -halfPts {
+			// wrapped forward
+			u.wrapCount++
+		} else if delta > halfPts {
+			// wrapped backward (unlikely in normal streams)
+			u.wrapCount--
+		}
+	} else {
+		u.initialized = true
+	}
+	u.lastRaw = rawPts
+	return uint32(rawPts + u.wrapCount*mpegtsMaxPts)
+}

media/rtp_timestamper_test.go

@@ -0,0 +1,96 @@
+package media
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+const (
+	mod32 = 1 << 32 // 4 294 967 296
+)
+
+func TestRTPTimestamper(t *testing.T) {
+	tests := []struct {
+		name    string
+		inputs  []int64
+		wantRtp []uint32
+	}{
+		{
+			name:    "Initialization",
+			inputs:  []int64{0},
+			wantRtp: []uint32{0},
+		},
+		{
+			name:    "Monotonic increase",
+			inputs:  []int64{0, 1, 2, 1000},
+			wantRtp: []uint32{0, 1, 2, 1000},
+		},
+		{
+			name:    "Small backwards jump (no wrap)",
+			inputs:  []int64{1000, 990},
+			wantRtp: []uint32{1000, 990},
+		},
+		{
+			name:   "Forward 33-bit PTS wrap",
+			inputs: []int64{mpegtsMaxPts - 2, 2},
+			wantRtp: []uint32{
+				uint32((mpegtsMaxPts - 2) % mod32),
+				uint32((2 + mpegtsMaxPts) % mod32),
+			},
+		},
+		{
+			name:   "Backward 33-bit PTS wrap",
+			inputs: []int64{5, mpegtsMaxPts - 3},
+			wantRtp: []uint32{
+				5,
+				uint32(mod32 - 3),
+			},
+		},
+		{
+			name:    "Crossing 32-bit boundary (wrap in RTP)",
+			inputs:  []int64{halfPts - 1, halfPts, halfPts + 1},
+			wantRtp: []uint32{uint32(halfPts - 1), 0, 1},
+		},
+		{
+			name:   "Random jitter near wrap threshold",
+			inputs: []int64{mpegtsMaxPts - halfPts/2, 2},
+			wantRtp: []uint32{
+				uint32((mpegtsMaxPts - halfPts/2) % mod32),
+				2,
+			},
+		},
+		{
+			name:    "Large monotonic advance (< halfPts)",
+			inputs:  []int64{100, halfPts - 1},
+			wantRtp: []uint32{100, uint32(halfPts - 1)},
+		},
+		{
+			name:   "Multiple successive wraps",
+			inputs: []int64{mpegtsMaxPts - 1, 2, mpegtsMaxPts - 1, 2},
+			wantRtp: []uint32{
+				uint32((mpegtsMaxPts - 1) % mod32),
+				uint32((2 + mpegtsMaxPts) % mod32),
+				uint32((mpegtsMaxPts - 1) % mod32),
+				uint32((2 + mpegtsMaxPts) % mod32),
+			},
+		},
+		{
+			name:    "Idempotent Unwrap (same raw twice)",
+			inputs:  []int64{500, 500},
+			wantRtp: []uint32{500, 500},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			u := NewRTPTimestamper()
+			gotRtp := make([]uint32, len(tc.inputs))
+			for i, raw := range tc.inputs {
+				ext := u.ToTS(raw)
+				gotRtp[i] = ext
+			}
+			require.Equal(t, tc.wantRtp, gotRtp)
+		})
+	}
+}