add video slot lifecycle

internal/playback/video_slot.go

@@ -0,0 +1,92 @@
+package playback
+
+import (
+	"context"
+	"errors"
+	"fmt"
+)
+
+var ErrVideoWorkerRequired = errors.New("video worker is required")
+
+type VideoSlot struct {
+	worker *VideoWorker
+}
+
+func NewVideoSlot(worker *VideoWorker) (*VideoSlot, error) {
+	if worker == nil {
+		return nil, ErrVideoWorkerRequired
+	}
+	return &VideoSlot{worker: worker}, nil
+}
+
+func (s *VideoSlot) Run(
+	ctx context.Context,
+	initial FeedConfig,
+	commands <-chan FeedConfig,
+) error {
+	if err := initial.Validate(); err != nil {
+		return fmt.Errorf("validate initial video config: %w", err)
+	}
+
+	var (
+		workerCancel context.CancelFunc
+		workerDone   chan error
+	)
+
+	start := func(config FeedConfig) {
+		workerCtx, cancel := context.WithCancel(ctx)
+		done := make(chan error, 1)
+
+		workerCancel = cancel
+		workerDone = done
+
+		go func() {
+			done <- s.worker.Run(workerCtx, config)
+		}()
+	}
+
+	stop := func() {
+		if workerCancel == nil {
+			return
+		}
+
+		workerCancel()
+		<-workerDone
+
+		workerCancel = nil
+		workerDone = nil
+	}
+
+	if initial.Active {
+		start(initial)
+	}
+
+	for {
+		select {
+		case <-ctx.Done():
+			stop()
+			return ctx.Err()
+
+		case config, ok := <-commands:
+			if !ok {
+				stop()
+				return nil
+			}
+			if err := config.Validate(); err != nil {
+				// Ignore invalid commands without disturbing the current worker.
+				continue
+			}
+			stop()
+			if config.Active {
+				start(config)
+			}
+
+		case <-workerDone:
+			// The worker stopped naturally or exhausted its retries.
+			// Clear its lifecycle, but keep the slot alive for future commands.
+			workerCancel()
+			workerCancel = nil
+			workerDone = nil
+		}
+	}
+}

internal/playback/video_slot_test.go

@@ -0,0 +1,233 @@
+package playback
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+)
+
+type slotVideoFactory struct {
+	opened chan FeedConfig
+
+	mu         sync.Mutex
+	active     int
+	maxActive  int
+	closeCount int
+}
+
+func newSlotVideoFactory() *slotVideoFactory {
+	return &slotVideoFactory{opened: make(chan FeedConfig, 8)}
+}
+
+func (f *slotVideoFactory) OpenVideo(
+	_ context.Context,
+	config FeedConfig,
+) (VideoReader, error) {
+	f.mu.Lock()
+	f.active++
+	if f.active > f.maxActive {
+		f.maxActive = f.active
+	}
+	f.mu.Unlock()
+
+	f.opened <- config
+	return &slotVideoReader{factory: f}, nil
+}
+
+func (f *slotVideoFactory) counts() (active, maxActive, closeCount int) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.active, f.maxActive, f.closeCount
+}
+
+type slotVideoReader struct {
+	factory *slotVideoFactory
+}
+
+func (r *slotVideoReader) ReadVideo(ctx context.Context) (VideoFrame, error) {
+	<-ctx.Done()
+	return VideoFrame{}, ctx.Err()
+}
+
+func (r *slotVideoReader) Close() error {
+	r.factory.mu.Lock()
+	defer r.factory.mu.Unlock()
+	r.factory.active--
+	r.factory.closeCount++
+	return nil
+}
+
+func newSlotTestWorker(t *testing.T, factory VideoReaderFactory) *VideoWorker {
+	t.Helper()
+	worker, err := NewVideoWorker(
+		factory,
+		&fakeVideoSink{},
+		testRetryPolicy(1),
+		func(error) bool { return false },
+		nil,
+	)
+	if err != nil {
+		t.Fatalf("NewVideoWorker() error = %v", err)
+	}
+	return worker
+}
+
+func receiveSlotOpen(t *testing.T, opened <-chan FeedConfig) FeedConfig {
+	t.Helper()
+	select {
+	case config := <-opened:
+		return config
+	case <-time.After(time.Second):
+		t.Fatal("video worker did not open")
+		return FeedConfig{}
+	}
+}
+
+func TestNewVideoSlotRequiresWorker(t *testing.T) {
+	slot, err := NewVideoSlot(nil)
+	if slot != nil {
+		t.Fatalf("NewVideoSlot(nil) slot = %#v, want nil", slot)
+	}
+	if !errors.Is(err, ErrVideoWorkerRequired) {
+		t.Fatalf("NewVideoSlot(nil) error = %v, want %v", err, ErrVideoWorkerRequired)
+	}
+}
+
+func TestVideoSlotStartsInitialActiveConfig(t *testing.T) {
+	factory := newSlotVideoFactory()
+	slot, err := NewVideoSlot(newSlotTestWorker(t, factory))
+	if err != nil {
+		t.Fatalf("NewVideoSlot() error = %v", err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	want := FeedConfig{Domain: "/video", UUID: "first", Active: true}
+
+	go func() { done <- slot.Run(ctx, want, make(chan FeedConfig)) }()
+	if got := receiveSlotOpen(t, factory.opened); got != want {
+		t.Fatalf("opened config = %#v, want %#v", got, want)
+	}
+
+	cancel()
+	select {
+	case err := <-done:
+		if !errors.Is(err, context.Canceled) {
+			t.Fatalf("Run() error = %v, want %v", err, context.Canceled)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("Run() did not stop after cancellation")
+	}
+	active, _, closeCount := factory.counts()
+	if active != 0 || closeCount != 1 {
+		t.Fatalf("reader counts = active %d, closed %d; want 0, 1", active, closeCount)
+	}
+}
+
+func TestVideoSlotReplacesWithoutOverlappingWorkers(t *testing.T) {
+	factory := newSlotVideoFactory()
+	slot, err := NewVideoSlot(newSlotTestWorker(t, factory))
+	if err != nil {
+		t.Fatalf("NewVideoSlot() error = %v", err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	commands := make(chan FeedConfig)
+	done := make(chan error, 1)
+	first := FeedConfig{Domain: "/video", UUID: "first", Active: true}
+	second := FeedConfig{Domain: "/video", UUID: "second", Active: true}
+
+	go func() { done <- slot.Run(ctx, first, commands) }()
+	receiveSlotOpen(t, factory.opened)
+	commands <- second
+	if got := receiveSlotOpen(t, factory.opened); got != second {
+		t.Fatalf("replacement config = %#v, want %#v", got, second)
+	}
+
+	cancel()
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("Run() did not stop")
+	}
+	active, maxActive, closeCount := factory.counts()
+	if active != 0 || maxActive != 1 || closeCount != 2 {
+		t.Fatalf(
+			"reader counts = active %d, maximum %d, closed %d; want 0, 1, 2",
+			active, maxActive, closeCount,
+		)
+	}
+}
+
+func TestVideoSlotIgnoresInvalidCommand(t *testing.T) {
+	factory := newSlotVideoFactory()
+	slot, err := NewVideoSlot(newSlotTestWorker(t, factory))
+	if err != nil {
+		t.Fatalf("NewVideoSlot() error = %v", err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	commands := make(chan FeedConfig)
+	done := make(chan error, 1)
+	initial := FeedConfig{Domain: "/video", UUID: "first", Active: true}
+
+	go func() { done <- slot.Run(ctx, initial, commands) }()
+	receiveSlotOpen(t, factory.opened)
+	commands <- FeedConfig{UUID: "invalid", Active: true}
+
+	select {
+	case config := <-factory.opened:
+		t.Fatalf("invalid command opened config %#v", config)
+	case <-time.After(20 * time.Millisecond):
+	}
+
+	active, _, closeCount := factory.counts()
+	if active != 1 || closeCount != 0 {
+		t.Fatalf("invalid command disturbed reader: active %d, closed %d", active, closeCount)
+	}
+	cancel()
+	<-done
+}
+
+func TestVideoSlotInactiveCommandStopsWithoutRestart(t *testing.T) {
+	factory := newSlotVideoFactory()
+	slot, err := NewVideoSlot(newSlotTestWorker(t, factory))
+	if err != nil {
+		t.Fatalf("NewVideoSlot() error = %v", err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	commands := make(chan FeedConfig)
+	done := make(chan error, 1)
+	initial := FeedConfig{Domain: "/video", UUID: "first", Active: true}
+
+	go func() { done <- slot.Run(ctx, initial, commands) }()
+	receiveSlotOpen(t, factory.opened)
+	commands <- FeedConfig{Domain: "/video", UUID: "first", Active: false}
+
+	deadline := time.Now().Add(time.Second)
+	for {
+		active, _, closeCount := factory.counts()
+		if active == 0 && closeCount == 1 {
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatal("inactive command did not stop reader")
+		}
+		time.Sleep(time.Millisecond)
+	}
+	select {
+	case config := <-factory.opened:
+		t.Fatalf("inactive command restarted config %#v", config)
+	case <-time.After(20 * time.Millisecond):
+	}
+
+	close(commands)
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatalf("Run() error = %v, want nil", err)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("Run() did not stop after commands closed")
+	}
+}