package playback import ( "context" "errors" "fmt" "sync" ) 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 SessionSnapshot struct { Desired SessionConfig Plan SessionPlan Generation uint64 } type SessionController struct { videoSlot VideoSlotRunner audioSlot AudioSlotRunner syncSlot SyncSlotRunner canSync SyncPredicate mu sync.RWMutex snapshot SessionSnapshot hasSnapshot bool } 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) generation := uint64(1) c.publish(SessionSnapshot{ Desired: initial, Plan: plan, Generation: generation, }) 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 } if plan.Topology != nextPlan.Topology { generation++ } desired = nextDesired plan = nextPlan runtime = nextRuntime c.publish(SessionSnapshot{ Desired: desired, Plan: plan, Generation: generation, }) } } } 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) } func (c *SessionController) Snapshot() (SessionSnapshot, bool) { c.mu.RLock() snapshot := c.snapshot ok := c.hasSnapshot c.mu.RUnlock() return snapshot, ok } func (c *SessionController) publish(snapshot SessionSnapshot) { c.mu.Lock() c.snapshot = snapshot c.hasSnapshot = true c.mu.Unlock() }