diff --git a/internal/playback/video_bridge.go b/internal/playback/video_bridge.go new file mode 100644 index 0000000..fb394d5 --- /dev/null +++ b/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 + }) +} diff --git a/internal/playback/video_bridge_test.go b/internal/playback/video_bridge_test.go new file mode 100644 index 0000000..8205b9e --- /dev/null +++ b/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) + } +}