internal/playback/video_bridge.go

@@ -0,0 +1,60 @@
+package playback
+
+import (
+	"context"
+	"sync"
+)
+
+type PendingVideoFrame struct {
+	Frame VideoFrame
+
+	completeOnce sync.Once
+	result       chan error
+}
+
+type VideoBridge struct {
+	requests chan *PendingVideoFrame
+}
+
+func NewVideoBridge() *VideoBridge {
+	return &VideoBridge{
+		requests: make(chan *PendingVideoFrame),
+	}
+}
+
+func (b *VideoBridge) ConsumeVideo(
+	ctx context.Context,
+	frame VideoFrame,
+) error {
+	pending := &PendingVideoFrame{
+		Frame:  frame,
+		result: make(chan error, 1),
+	}
+
+	select {
+	case b.requests <- pending:
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+
+	// The render thread now owns temporary access to the borrowed payload.
+	// We must wait for Complete even if ctx is canceled.
+	return <-pending.result
+}
+
+func (b *VideoBridge) Next(
+	ctx context.Context,
+) (*PendingVideoFrame, error) {
+	select {
+	case pending := <-b.requests:
+		return pending, nil
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+}
+
+func (f *PendingVideoFrame) Complete(err error) {
+	f.completeOnce.Do(func() {
+		f.result <- err
+	})
+}

internal/playback/video_bridge_test.go

@@ -0,0 +1,138 @@
+package playback
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+)
+
+const videoBridgeTestTimeout = time.Second
+
+func TestVideoBridgeDeliversFrameAndCompletionResult(t *testing.T) {
+	bridge := NewVideoBridge()
+	wantErr := errors.New("stage frame")
+	wantFrame := VideoFrame{
+		Index:   42,
+		Width:   1920,
+		Height:  1080,
+		Stride:  7680,
+		Payload: []byte{1, 2, 3},
+	}
+	consumeResult := make(chan error, 1)
+
+	go func() {
+		consumeResult <- bridge.ConsumeVideo(context.Background(), wantFrame)
+	}()
+
+	ctx, cancel := context.WithTimeout(context.Background(), videoBridgeTestTimeout)
+	defer cancel()
+	pending, err := bridge.Next(ctx)
+	if err != nil {
+		t.Fatalf("Next() error = %v", err)
+	}
+	if pending.Frame.Index != wantFrame.Index {
+		t.Fatalf("Next() frame index = %d, want %d", pending.Frame.Index, wantFrame.Index)
+	}
+	if &pending.Frame.Payload[0] != &wantFrame.Payload[0] {
+		t.Fatal("Next() copied the borrowed payload")
+	}
+
+	pending.Complete(wantErr)
+	select {
+	case err := <-consumeResult:
+		if !errors.Is(err, wantErr) {
+			t.Fatalf("ConsumeVideo() error = %v, want %v", err, wantErr)
+		}
+	case <-time.After(videoBridgeTestTimeout):
+		t.Fatal("ConsumeVideo() did not return after completion")
+	}
+}
+
+func TestVideoBridgeConsumeHonorsCancellationBeforeDelivery(t *testing.T) {
+	bridge := NewVideoBridge()
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	err := bridge.ConsumeVideo(ctx, VideoFrame{})
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("ConsumeVideo() error = %v, want %v", err, context.Canceled)
+	}
+}
+
+func TestVideoBridgeConsumeWaitsForCompletionAfterDelivery(t *testing.T) {
+	bridge := NewVideoBridge()
+	ctx, cancel := context.WithCancel(context.Background())
+	consumeResult := make(chan error, 1)
+
+	go func() {
+		consumeResult <- bridge.ConsumeVideo(ctx, VideoFrame{Index: 7})
+	}()
+
+	nextCtx, nextCancel := context.WithTimeout(context.Background(), videoBridgeTestTimeout)
+	defer nextCancel()
+	pending, err := bridge.Next(nextCtx)
+	if err != nil {
+		t.Fatalf("Next() error = %v", err)
+	}
+	cancel()
+
+	select {
+	case err := <-consumeResult:
+		t.Fatalf("ConsumeVideo() returned before completion: %v", err)
+	case <-time.After(20 * time.Millisecond):
+	}
+
+	pending.Complete(nil)
+	select {
+	case err := <-consumeResult:
+		if err != nil {
+			t.Fatalf("ConsumeVideo() error = %v, want nil", err)
+		}
+	case <-time.After(videoBridgeTestTimeout):
+		t.Fatal("ConsumeVideo() did not return after completion")
+	}
+}
+
+func TestPendingVideoFrameCompleteIsIdempotent(t *testing.T) {
+	bridge := NewVideoBridge()
+	consumeResult := make(chan error, 1)
+
+	go func() {
+		consumeResult <- bridge.ConsumeVideo(context.Background(), VideoFrame{})
+	}()
+
+	ctx, cancel := context.WithTimeout(context.Background(), videoBridgeTestTimeout)
+	defer cancel()
+	pending, err := bridge.Next(ctx)
+	if err != nil {
+		t.Fatalf("Next() error = %v", err)
+	}
+
+	firstErr := errors.New("first")
+	pending.Complete(firstErr)
+	pending.Complete(errors.New("second"))
+
+	select {
+	case err := <-consumeResult:
+		if !errors.Is(err, firstErr) {
+			t.Fatalf("ConsumeVideo() error = %v, want %v", err, firstErr)
+		}
+	case <-time.After(videoBridgeTestTimeout):
+		t.Fatal("ConsumeVideo() did not return")
+	}
+}
+
+func TestVideoBridgeNextHonorsCancellation(t *testing.T) {
+	bridge := NewVideoBridge()
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	pending, err := bridge.Next(ctx)
+	if pending != nil {
+		t.Fatalf("Next() pending = %#v, want nil", pending)
+	}
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("Next() error = %v, want %v", err, context.Canceled)
+	}
+}