diff --git a/internal/playback/session_controller.go b/internal/playback/session_controller.go new file mode 100644 index 0000000..94181b7 --- /dev/null +++ b/internal/playback/session_controller.go @@ -0,0 +1,303 @@ +package playback + +import ( + "context" + "errors" + "fmt" +) + +type VideoSlotRunner interface { + Run( + context.Context, + FeedConfig, + <-chan FeedConfig, + ) error +} + +type AudioSlotRunner interface { + Run( + context.Context, + FeedConfig, + <-chan FeedConfig, + ) error +} + +type SyncSlotRunner interface { + Run( + context.Context, + SyncPairConfig, + <-chan SyncPairConfig, + ) error +} + +var _ VideoSlotRunner = (*VideoSlot)(nil) +var _ AudioSlotRunner = (*AudioSlot)(nil) +var _ SyncSlotRunner = (*SyncSlot)(nil) + +var ( + ErrVideoSlotRequired = errors.New("video slot is required") + ErrAudioSlotRequired = errors.New("audio slot is required") + ErrSyncSlotRequired = errors.New("sync slot is required") +) + +type SessionController struct { + videoSlot VideoSlotRunner + audioSlot AudioSlotRunner + syncSlot SyncSlotRunner + canSync SyncPredicate +} + +func NewSessionController( + videoSlot VideoSlotRunner, + audioSlot AudioSlotRunner, + syncSlot SyncSlotRunner, + canSync SyncPredicate, +) (*SessionController, error) { + if videoSlot == nil { + return nil, ErrVideoSlotRequired + } + if audioSlot == nil { + return nil, ErrAudioSlotRequired + } + if syncSlot == nil { + return nil, ErrSyncSlotRequired + } + return &SessionController{ + videoSlot: videoSlot, + audioSlot: audioSlot, + syncSlot: syncSlot, + canSync: canSync, + }, nil +} + +type sessionRuntime struct { + topology SessionTopology + cancel context.CancelFunc + done chan struct{} + result error + + videoCommands chan FeedConfig + audioCommands chan FeedConfig + syncCommands chan SyncPairConfig +} + +var ( + ErrSessionRuntimeStopped = errors.New( + "session slot runtime stopped unexpectedly", + ) +) + +func (c *SessionController) Run( + ctx context.Context, + initial SessionConfig, + commands <-chan SessionCommand, +) error { + plan, err := BuildSessionPlan(initial, c.canSync) + if err != nil { + return fmt.Errorf("build initial session plan: %w", err) + } + + desired := initial + runtime := c.startSessionRuntime(ctx, plan) + + for { + var runtimeDone <-chan struct{} + if runtime != nil { + runtimeDone = runtime.done + } + + select { + case <-ctx.Done(): + stopSessionRuntime(runtime) + return ctx.Err() + + case <-runtimeDone: + return unexpectedSessionRuntimeError(runtime.result) + + case command, ok := <-commands: + if !ok { + stopSessionRuntime(runtime) + return nil + } + + nextDesired, err := ApplySessionCommand(desired, command) + if err != nil { + // Invalid commands must not disturb the current valid runtime. + continue + } + nextPlan, err := BuildSessionPlan(nextDesired, c.canSync) + if err != nil { + continue + } + + nextRuntime, err := c.reconcileSessionRuntime( + ctx, + runtime, + plan, + nextPlan, + ) + if err != nil { + stopSessionRuntime(runtime) + if ctx.Err() != nil { + return ctx.Err() + } + return err + } + + desired = nextDesired + plan = nextPlan + runtime = nextRuntime + } + } +} + +func (c *SessionController) startSessionRuntime( + ctx context.Context, + plan SessionPlan, +) *sessionRuntime { + if plan.Topology == TopologyIdle { + return &sessionRuntime{topology: TopologyIdle} + } + + runtimeCtx, cancel := context.WithCancel(ctx) + runtime := &sessionRuntime{ + topology: plan.Topology, + cancel: cancel, + done: make(chan struct{}), + } + + switch plan.Topology { + case TopologyIndependent: + runtime.videoCommands = make(chan FeedConfig) + runtime.audioCommands = make(chan FeedConfig) + results := make(chan error, 2) + + go func() { + results <- c.videoSlot.Run( + runtimeCtx, + plan.Video, + runtime.videoCommands, + ) + }() + go func() { + results <- c.audioSlot.Run( + runtimeCtx, + plan.Audio, + runtime.audioCommands, + ) + }() + go func() { + first := <-results + cancel() + second := <-results + runtime.result = errors.Join(first, second) + close(runtime.done) + }() + + case TopologySynchronized: + runtime.syncCommands = make(chan SyncPairConfig) + go func() { + runtime.result = c.syncSlot.Run( + runtimeCtx, + plan.Sync, + runtime.syncCommands, + ) + close(runtime.done) + }() + } + + return runtime +} + +func stopSessionRuntime(runtime *sessionRuntime) { + if runtime == nil || runtime.done == nil { + return + } + runtime.cancel() + <-runtime.done +} + +func unexpectedSessionRuntimeError(err error) error { + if err == nil { + return ErrSessionRuntimeStopped + } + return fmt.Errorf("%w: %v", ErrSessionRuntimeStopped, err) +} + +func (c *SessionController) reconcileSessionRuntime( + ctx context.Context, + runtime *sessionRuntime, + current SessionPlan, + next SessionPlan, +) (*sessionRuntime, error) { + if current.Topology != next.Topology { + stopSessionRuntime(runtime) + if err := ctx.Err(); err != nil { + return runtime, err + } + return c.startSessionRuntime(ctx, next), nil + } + + switch next.Topology { + case TopologyIndependent: + if current.Video != next.Video { + if !sendFeedConfig(ctx, runtime.done, runtime.videoCommands, next.Video) { + return runtime, sessionRuntimeSendError(ctx, runtime) + } + } + if current.Audio != next.Audio { + if !sendFeedConfig(ctx, runtime.done, runtime.audioCommands, next.Audio) { + return runtime, sessionRuntimeSendError(ctx, runtime) + } + } + + case TopologySynchronized: + if current.Sync != next.Sync { + if !sendSyncPairConfig(ctx, runtime.done, runtime.syncCommands, next.Sync) { + return runtime, sessionRuntimeSendError(ctx, runtime) + } + } + } + + return runtime, nil +} + +func sendFeedConfig( + ctx context.Context, + done <-chan struct{}, + commands chan<- FeedConfig, + config FeedConfig, +) bool { + select { + case commands <- config: + return true + case <-ctx.Done(): + return false + case <-done: + return false + } +} + +func sendSyncPairConfig( + ctx context.Context, + done <-chan struct{}, + commands chan<- SyncPairConfig, + config SyncPairConfig, +) bool { + select { + case commands <- config: + return true + case <-ctx.Done(): + return false + case <-done: + return false + } +} + +func sessionRuntimeSendError(ctx context.Context, runtime *sessionRuntime) error { + if err := ctx.Err(); err != nil { + return err + } + <-runtime.done + return unexpectedSessionRuntimeError(runtime.result) +} diff --git a/internal/playback/session_controller_test.go b/internal/playback/session_controller_test.go new file mode 100644 index 0000000..38ad8c9 --- /dev/null +++ b/internal/playback/session_controller_test.go @@ -0,0 +1,279 @@ +package playback + +import ( + "context" + "errors" + "testing" + "time" +) + +type stubVideoSlot struct{} + +func (stubVideoSlot) Run(context.Context, FeedConfig, <-chan FeedConfig) error { + return nil +} + +type stubAudioSlot struct{} + +func (stubAudioSlot) Run(context.Context, FeedConfig, <-chan FeedConfig) error { + return nil +} + +type stubSyncSlot struct{} + +func (stubSyncSlot) Run(context.Context, SyncPairConfig, <-chan SyncPairConfig) error { + return nil +} + +func TestNewSessionControllerValidatesSlots(t *testing.T) { + video := stubVideoSlot{} + audio := stubAudioSlot{} + sync := stubSyncSlot{} + tests := []struct { + name string + video VideoSlotRunner + audio AudioSlotRunner + sync SyncSlotRunner + want error + }{ + {"missing video", nil, audio, sync, ErrVideoSlotRequired}, + {"missing audio", video, nil, sync, ErrAudioSlotRequired}, + {"missing sync", video, audio, nil, ErrSyncSlotRequired}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + controller, err := NewSessionController(tt.video, tt.audio, tt.sync, nil) + if controller != nil { + t.Fatalf("NewSessionController() controller = %#v, want nil", controller) + } + if !errors.Is(err, tt.want) { + t.Fatalf("NewSessionController() error = %v, want %v", err, tt.want) + } + }) + } +} + +func TestNewSessionControllerAllowsUnavailableSynchronization(t *testing.T) { + video := stubVideoSlot{} + audio := stubAudioSlot{} + sync := stubSyncSlot{} + controller, err := NewSessionController(video, audio, sync, nil) + if err != nil { + t.Fatalf("NewSessionController() error = %v", err) + } + if controller == nil { + t.Fatal("NewSessionController() controller is nil") + } + if controller.videoSlot != video || controller.audioSlot != audio || controller.syncSlot != sync { + t.Fatalf("NewSessionController() = %#v", controller) + } + if controller.canSync != nil { + t.Fatal("nil sync predicate was not preserved") + } +} + +func TestNewSessionControllerStoresSyncPredicate(t *testing.T) { + predicate := func(FeedConfig, FeedConfig) bool { return true } + controller, err := NewSessionController( + stubVideoSlot{}, stubAudioSlot{}, stubSyncSlot{}, predicate, + ) + if err != nil { + t.Fatalf("NewSessionController() error = %v", err) + } + if controller.canSync == nil || !controller.canSync(FeedConfig{}, FeedConfig{}) { + t.Fatal("sync predicate was not stored") + } +} + +type controllerEvent struct { + unit Unit + action string + feed FeedConfig + pair SyncPairConfig +} + +type recordingVideoSlot struct{ events chan<- controllerEvent } + +func (s recordingVideoSlot) Run( + ctx context.Context, + initial FeedConfig, + commands <-chan FeedConfig, +) error { + s.events <- controllerEvent{unit: UnitVideo, action: "start", feed: initial} + for { + select { + case config := <-commands: + s.events <- controllerEvent{unit: UnitVideo, action: "command", feed: config} + case <-ctx.Done(): + s.events <- controllerEvent{unit: UnitVideo, action: "stop"} + return ctx.Err() + } + } +} + +type recordingAudioSlot struct{ events chan<- controllerEvent } + +func (s recordingAudioSlot) Run( + ctx context.Context, + initial FeedConfig, + commands <-chan FeedConfig, +) error { + s.events <- controllerEvent{unit: UnitAudio, action: "start", feed: initial} + for { + select { + case config := <-commands: + s.events <- controllerEvent{unit: UnitAudio, action: "command", feed: config} + case <-ctx.Done(): + s.events <- controllerEvent{unit: UnitAudio, action: "stop"} + return ctx.Err() + } + } +} + +type recordingSyncSlot struct{ events chan<- controllerEvent } + +func (s recordingSyncSlot) Run( + ctx context.Context, + initial SyncPairConfig, + commands <-chan SyncPairConfig, +) error { + s.events <- controllerEvent{unit: UnitSync, action: "start", pair: initial} + for { + select { + case config := <-commands: + s.events <- controllerEvent{unit: UnitSync, action: "command", pair: config} + case <-ctx.Done(): + s.events <- controllerEvent{unit: UnitSync, action: "stop"} + return ctx.Err() + } + } +} + +func newRecordingController(t *testing.T, events chan<- controllerEvent) *SessionController { + t.Helper() + controller, err := NewSessionController( + recordingVideoSlot{events}, + recordingAudioSlot{events}, + recordingSyncSlot{events}, + func(FeedConfig, FeedConfig) bool { return true }, + ) + if err != nil { + t.Fatal(err) + } + return controller +} + +func receiveControllerEvent(t *testing.T, events <-chan controllerEvent) controllerEvent { + t.Helper() + select { + case event := <-events: + return event + case <-time.After(time.Second): + t.Fatal("controller event timed out") + return controllerEvent{} + } +} + +func receiveIndependentStarts(t *testing.T, events <-chan controllerEvent) { + t.Helper() + seen := map[Unit]bool{} + for len(seen) < 2 { + event := receiveControllerEvent(t, events) + if event.action != "start" || (event.unit != UnitVideo && event.unit != UnitAudio) { + t.Fatalf("unexpected initial event: %+v", event) + } + seen[event.unit] = true + } +} + +func TestSessionControllerUpdatesOnlyChangedIndependentSlot(t *testing.T) { + events := make(chan controllerEvent, 32) + controller := newRecordingController(t, events) + initial := validCommandSession() + initial.SyncRequested = false + commands := make(chan SessionCommand) + done := make(chan error, 1) + go func() { done <- controller.Run(context.Background(), initial, commands) }() + receiveIndependentStarts(t, events) + + want := FeedConfig{Domain: "/new-video", UUID: "new-video", Active: true} + commands <- SessionCommand{Kind: CommandSetVideo, Config: want} + event := receiveControllerEvent(t, events) + if event.unit != UnitVideo || event.action != "command" || event.feed != want { + t.Fatalf("replacement event = %+v", event) + } + select { + case event := <-events: + t.Fatalf("unchanged audio slot was disturbed: %+v", event) + 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") + } +} + +func TestSessionControllerStopsIndependentSlotsBeforeStartingSync(t *testing.T) { + events := make(chan controllerEvent, 32) + controller := newRecordingController(t, events) + initial := validCommandSession() + initial.SyncRequested = false + commands := make(chan SessionCommand) + done := make(chan error, 1) + go func() { done <- controller.Run(context.Background(), initial, commands) }() + receiveIndependentStarts(t, events) + + commands <- SessionCommand{Kind: CommandEnableSync} + stopped := map[Unit]bool{} + for { + event := receiveControllerEvent(t, events) + if event.unit == UnitSync && event.action == "start" { + if !stopped[UnitVideo] || !stopped[UnitAudio] { + t.Fatalf("sync started before both independent slots stopped: %v", stopped) + } + break + } + if event.action != "stop" || (event.unit != UnitVideo && event.unit != UnitAudio) { + t.Fatalf("unexpected transition event: %+v", event) + } + stopped[event.unit] = true + } + + 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") + } +} + +func TestSessionControllerCancellationStopsAndJoinsRuntime(t *testing.T) { + events := make(chan controllerEvent, 32) + controller := newRecordingController(t, events) + initial := validCommandSession() + initial.SyncRequested = false + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- controller.Run(ctx, initial, make(chan SessionCommand)) }() + receiveIndependentStarts(t, events) + cancel() + + select { + case err := <-done: + if !errors.Is(err, context.Canceled) { + t.Fatalf("Run() error = %v, want context.Canceled", err) + } + case <-time.After(time.Second): + t.Fatal("Run() did not stop after cancellation") + } +}