diff --git a/internal/playback/video_slot.go b/internal/playback/video_slot.go new file mode 100644 index 0000000..b8b8a44 --- /dev/null +++ b/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 + } + } +} diff --git a/internal/playback/video_slot_test.go b/internal/playback/video_slot_test.go new file mode 100644 index 0000000..fe92aa3 --- /dev/null +++ b/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") + } +}