diff --git a/internal/playback/audio_slot.go b/internal/playback/audio_slot.go new file mode 100644 index 0000000..4708c22 --- /dev/null +++ b/internal/playback/audio_slot.go @@ -0,0 +1,93 @@ +package playback + +import ( + "context" + "errors" + "fmt" +) + +var ErrAudioWorkerRequired = errors.New("audio worker is required") + +type AudioSlot struct { + worker *AudioWorker +} + +func NewAudioSlot(worker *AudioWorker) (*AudioSlot, error) { + if worker == nil { + return nil, ErrAudioWorkerRequired + } + return &AudioSlot{worker: worker}, nil +} + +func (s *AudioSlot) Run( + ctx context.Context, + initial FeedConfig, + commands <-chan FeedConfig, +) error { + if err := initial.Validate(); err != nil { + return fmt.Errorf("validate initial audio 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. + workerCancel() + workerCancel = nil + workerDone = nil + } + } +} diff --git a/internal/playback/audio_slot_test.go b/internal/playback/audio_slot_test.go new file mode 100644 index 0000000..ea306bd --- /dev/null +++ b/internal/playback/audio_slot_test.go @@ -0,0 +1,202 @@ +package playback + +import ( + "context" + "errors" + "sync" + "testing" + "time" +) + +type slotAudioFactory struct { + opened chan FeedConfig + + mu sync.Mutex + active int + maxActive int + closeCount int +} + +func newSlotAudioFactory() *slotAudioFactory { + return &slotAudioFactory{opened: make(chan FeedConfig, 8)} +} + +func (f *slotAudioFactory) OpenAudio( + _ context.Context, + config FeedConfig, +) (AudioReader, error) { + f.mu.Lock() + f.active++ + if f.active > f.maxActive { + f.maxActive = f.active + } + f.mu.Unlock() + f.opened <- config + return &slotAudioReader{factory: f}, nil +} + +func (f *slotAudioFactory) counts() (active, maxActive, closeCount int) { + f.mu.Lock() + defer f.mu.Unlock() + return f.active, f.maxActive, f.closeCount +} + +type slotAudioReader struct { + factory *slotAudioFactory +} + +func (r *slotAudioReader) ReadAudio(ctx context.Context) (AudioFrame, error) { + <-ctx.Done() + return AudioFrame{}, ctx.Err() +} + +func (r *slotAudioReader) Close() error { + r.factory.mu.Lock() + defer r.factory.mu.Unlock() + r.factory.active-- + r.factory.closeCount++ + return nil +} + +func newAudioSlotWorker(t *testing.T, factory AudioReaderFactory) *AudioWorker { + t.Helper() + return newAudioWorkerForTest( + t, + factory, + &fakeAudioSink{}, + 1, + func(error) bool { return false }, + nil, + ) +} + +func receiveAudioSlotOpen(t *testing.T, opened <-chan FeedConfig) FeedConfig { + t.Helper() + select { + case config := <-opened: + return config + case <-time.After(time.Second): + t.Fatal("audio worker did not open") + return FeedConfig{} + } +} + +func TestNewAudioSlotRequiresWorker(t *testing.T) { + slot, err := NewAudioSlot(nil) + if slot != nil { + t.Fatalf("NewAudioSlot(nil) slot = %#v, want nil", slot) + } + if !errors.Is(err, ErrAudioWorkerRequired) { + t.Fatalf("NewAudioSlot(nil) error = %v, want %v", err, ErrAudioWorkerRequired) + } +} + +func TestAudioSlotStartsAndJoinsInitialWorker(t *testing.T) { + factory := newSlotAudioFactory() + slot, err := NewAudioSlot(newAudioSlotWorker(t, factory)) + if err != nil { + t.Fatalf("NewAudioSlot() error = %v", err) + } + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + want := FeedConfig{Domain: "/audio", UUID: "first", Active: true} + + go func() { done <- slot.Run(ctx, want, make(chan FeedConfig)) }() + if got := receiveAudioSlotOpen(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, maxActive, closeCount := factory.counts() + if active != 0 || maxActive != 1 || closeCount != 1 { + t.Fatalf("reader counts = %d, %d, %d; want 0, 1, 1", active, maxActive, closeCount) + } +} + +func TestAudioSlotReplacesWithoutOverlapAndStops(t *testing.T) { + factory := newSlotAudioFactory() + slot, err := NewAudioSlot(newAudioSlotWorker(t, factory)) + if err != nil { + t.Fatalf("NewAudioSlot() error = %v", err) + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + commands := make(chan FeedConfig) + done := make(chan error, 1) + first := FeedConfig{Domain: "/audio", UUID: "first", Active: true} + second := FeedConfig{Domain: "/audio", UUID: "second", Active: true} + + go func() { done <- slot.Run(ctx, first, commands) }() + receiveAudioSlotOpen(t, factory.opened) + commands <- second + if got := receiveAudioSlotOpen(t, factory.opened); got != second { + t.Fatalf("replacement config = %#v, want %#v", got, second) + } + commands <- FeedConfig{Domain: "/audio", UUID: "second", Active: false} + + deadline := time.Now().Add(time.Second) + for { + active, maxActive, closeCount := factory.counts() + if active == 0 && closeCount == 2 { + if maxActive != 1 { + t.Fatalf("maximum active readers = %d, want 1", maxActive) + } + break + } + if time.Now().After(deadline) { + t.Fatalf("reader counts = %d, %d, %d; want 0, 1, 2", active, maxActive, closeCount) + } + time.Sleep(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") + } +} + +func TestAudioSlotIgnoresInvalidCommand(t *testing.T) { + factory := newSlotAudioFactory() + slot, err := NewAudioSlot(newAudioSlotWorker(t, factory)) + if err != nil { + t.Fatalf("NewAudioSlot() error = %v", err) + } + ctx, cancel := context.WithCancel(context.Background()) + commands := make(chan FeedConfig) + done := make(chan error, 1) + + go func() { + done <- slot.Run( + ctx, + FeedConfig{Domain: "/audio", UUID: "first", Active: true}, + commands, + ) + }() + receiveAudioSlotOpen(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 +}