diff --git a/internal/playback/audio_worker.go b/internal/playback/audio_worker.go new file mode 100644 index 0000000..066db14 --- /dev/null +++ b/internal/playback/audio_worker.go @@ -0,0 +1,184 @@ +package playback + +import ( + "context" + "errors" + "fmt" +) + +var ( + ErrAudioFactoryRequired = errors.New("audio reader factory is required") + ErrAudioSinkRequired = errors.New("audio sink is required") + ErrAudioRetryDeciderRequired = errors.New("audio decider is required") + ErrAudioFeedInactive = errors.New("audio feed is not active") +) + +type AudioWorker struct { + factory AudioReaderFactory + sink AudioSink + retry RetryPolicy + shouldRetry retryDecider + observer StatusObserver + wait waitFunc +} + +func NewAudioWorker( + factory AudioReaderFactory, + sink AudioSink, + retry RetryPolicy, + shouldRetry func(error) bool, + observer StatusObserver, +) (*AudioWorker, error) { + if factory == nil { + return nil, ErrAudioFactoryRequired + } + if sink == nil { + return nil, ErrAudioSinkRequired + } + if shouldRetry == nil { + return nil, ErrAudioRetryDeciderRequired + } + if err := retry.Validate(); err != nil { + return nil, fmt.Errorf("validate audio retry policy: %w", err) + } + + return &AudioWorker{ + factory: factory, + sink: sink, + retry: retry, + shouldRetry: shouldRetry, + observer: observer, + wait: waitForRetry, + }, nil +} + +type stabilityAudioSink struct { + sink AudioSink + onStable func() + stable bool +} + +func (s *stabilityAudioSink) ConsumeAudio( + ctx context.Context, + frame AudioFrame, +) error { + err := s.sink.ConsumeAudio(ctx, frame) + if err == nil && !s.stable { + s.stable = true + if s.onStable != nil { + s.onStable() + } + } + return err +} + +func (w *AudioWorker) emit(status Status) { + if w.observer != nil { + w.observer(status) + } +} + +func (w *AudioWorker) Run( + ctx context.Context, + config FeedConfig, +) error { + if err := config.Validate(); err != nil { + return fmt.Errorf("validate audio config: %w", err) + } + if !config.Active { + return ErrAudioFeedInactive + } + + attemptNumber := 0 + var latestRetry retryEvent + + attempt := func(ctx context.Context) (bool, error) { + attemptNumber++ + + state := StateConnecting + if attemptNumber > 1 { + state = StateReconnecting + } + w.emit(Status{ + Unit: UnitAudio, + State: state, + Attempt: attemptNumber, + }) + + attemptSink := &stabilityAudioSink{ + sink: w.sink, + onStable: func() { + w.emit(Status{ + Unit: UnitAudio, + State: StatePlaying, + Attempt: attemptNumber, + }) + }, + } + + err := runAudioAttempt(ctx, w.factory, attemptSink, config) + return attemptSink.stable, err + } + + decide := func(err error) bool { + var sinkErr *audioSinkError + if errors.As(err, &sinkErr) { + return false + } + return w.shouldRetry(err) + } + + observeRetry := func(event retryEvent) { + latestRetry = event + if !event.WillRetry { + return + } + + w.emit(Status{ + Unit: UnitAudio, + State: StateReconnecting, + Attempt: attemptNumber + 1, + FailedAttempts: event.FailedAttempts, + RetryIn: event.RetryIn, + Err: event.Err, + }) + } + + err := runWithRetry( + ctx, + w.retry, + attempt, + decide, + w.wait, + observeRetry, + ) + + if ctx.Err() != nil { + w.emit(Status{ + Unit: UnitAudio, + State: StateStopping, + }) + w.emit(Status{ + Unit: UnitAudio, + State: StateIdle, + }) + return ctx.Err() + } + + if err != nil { + w.emit(Status{ + Unit: UnitAudio, + State: StateFailed, + Attempt: attemptNumber, + FailedAttempts: latestRetry.FailedAttempts, + Err: err, + }) + return err + } + + w.emit(Status{ + Unit: UnitAudio, + State: StateIdle, + }) + return nil +} diff --git a/internal/playback/audio_worker_test.go b/internal/playback/audio_worker_test.go new file mode 100644 index 0000000..82becef --- /dev/null +++ b/internal/playback/audio_worker_test.go @@ -0,0 +1,241 @@ +package playback + +import ( + "context" + "errors" + "testing" + "time" +) + +type queuedAudioFactory struct { + readers []AudioReader + errs []error + calls int +} + +func (f *queuedAudioFactory) OpenAudio( + context.Context, + FeedConfig, +) (AudioReader, error) { + index := f.calls + f.calls++ + if index < len(f.errs) && f.errs[index] != nil { + return nil, f.errs[index] + } + if index < len(f.readers) { + return f.readers[index], nil + } + return nil, errors.New("unexpected audio open") +} + +func newAudioWorkerForTest( + t *testing.T, + factory AudioReaderFactory, + sink AudioSink, + maxAttempts int, + shouldRetry func(error) bool, + observer StatusObserver, +) *AudioWorker { + t.Helper() + worker, err := NewAudioWorker( + factory, + sink, + testRetryPolicy(maxAttempts), + shouldRetry, + observer, + ) + if err != nil { + t.Fatalf("NewAudioWorker() error = %v", err) + } + worker.wait = func(context.Context, time.Duration) error { return nil } + return worker +} + +func TestNewAudioWorkerValidatesDependencies(t *testing.T) { + factory := &fakeAudioFactory{} + sink := &fakeAudioSink{} + retry := testRetryPolicy(3) + decider := func(error) bool { return true } + + tests := []struct { + name string + factory AudioReaderFactory + sink AudioSink + retry RetryPolicy + decider func(error) bool + wantErr error + }{ + {name: "factory", sink: sink, retry: retry, decider: decider, wantErr: ErrAudioFactoryRequired}, + {name: "sink", factory: factory, retry: retry, decider: decider, wantErr: ErrAudioSinkRequired}, + {name: "decider", factory: factory, sink: sink, retry: retry, wantErr: ErrAudioRetryDeciderRequired}, + { + name: "retry policy", + factory: factory, + sink: sink, + retry: RetryPolicy{}, + decider: decider, + wantErr: ErrInvalidRetryDelay, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + worker, err := NewAudioWorker( + tt.factory, + tt.sink, + tt.retry, + tt.decider, + nil, + ) + if worker != nil { + t.Fatalf("NewAudioWorker() worker = %#v, want nil", worker) + } + if !errors.Is(err, tt.wantErr) { + t.Fatalf("NewAudioWorker() error = %v, want %v", err, tt.wantErr) + } + }) + } +} + +func TestAudioWorkerRejectsInactiveFeed(t *testing.T) { + worker := newAudioWorkerForTest( + t, + &fakeAudioFactory{}, + &fakeAudioSink{}, + 1, + func(error) bool { return false }, + nil, + ) + + err := worker.Run( + context.Background(), + FeedConfig{Domain: "/audio", UUID: "audio", Active: false}, + ) + if !errors.Is(err, ErrAudioFeedInactive) { + t.Fatalf("Run() error = %v, want %v", err, ErrAudioFeedInactive) + } +} + +func TestAudioWorkerPublishesPlayingThenFailed(t *testing.T) { + readErr := errors.New("audio disappeared") + reader := &fakeAudioReader{ + frames: []AudioFrame{{Index: 1, Samples: [][]byte{{1, 2, 3, 4}}}}, + readErr: readErr, + } + var statuses []Status + worker := newAudioWorkerForTest( + t, + &fakeAudioFactory{reader: reader}, + &fakeAudioSink{}, + 1, + func(error) bool { return false }, + func(status Status) { statuses = append(statuses, status) }, + ) + + err := worker.Run( + context.Background(), + FeedConfig{Domain: "/audio", UUID: "audio", Active: true}, + ) + if !errors.Is(err, readErr) { + t.Fatalf("Run() error = %v, want %v", err, readErr) + } + wantStates := []State{StateConnecting, StatePlaying, StateFailed} + if len(statuses) != len(wantStates) { + t.Fatalf("status count = %d, want %d: %#v", len(statuses), len(wantStates), statuses) + } + for i, want := range wantStates { + if statuses[i].Unit != UnitAudio || statuses[i].State != want { + t.Errorf("status[%d] = %#v, want audio/%v", i, statuses[i], want) + } + } +} + +func TestAudioWorkerRetriesUpToAttemptLimit(t *testing.T) { + openErr := errors.New("audio unavailable") + factory := &queuedAudioFactory{errs: []error{openErr, openErr, openErr}} + var statuses []Status + worker := newAudioWorkerForTest( + t, + factory, + &fakeAudioSink{}, + 3, + func(error) bool { return true }, + func(status Status) { statuses = append(statuses, status) }, + ) + + err := worker.Run( + context.Background(), + FeedConfig{Domain: "/audio", UUID: "audio", Active: true}, + ) + if !errors.Is(err, openErr) { + t.Fatalf("Run() error = %v, want %v", err, openErr) + } + if factory.calls != 3 { + t.Fatalf("open calls = %d, want 3", factory.calls) + } + last := statuses[len(statuses)-1] + if last.State != StateFailed || last.Attempt != 3 || last.FailedAttempts != 3 { + t.Fatalf("last status = %#v, want failed attempt 3", last) + } +} + +func TestAudioWorkerDoesNotRetrySinkFailure(t *testing.T) { + sinkErr := errors.New("output failed") + factory := &fakeAudioFactory{ + reader: &fakeAudioReader{frames: []AudioFrame{{Index: 1}}}, + } + worker := newAudioWorkerForTest( + t, + factory, + &fakeAudioSink{err: sinkErr}, + 3, + func(error) bool { return true }, + nil, + ) + + err := worker.Run( + context.Background(), + FeedConfig{Domain: "/audio", UUID: "audio", Active: true}, + ) + if !errors.Is(err, sinkErr) { + t.Fatalf("Run() error = %v, want %v", err, sinkErr) + } + if factory.calls != 1 { + t.Fatalf("open calls = %d, want 1", factory.calls) + } +} + +func TestAudioWorkerCancellationPublishesStoppingAndIdle(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + reader := &fakeAudioReader{ + read: func(ctx context.Context) (AudioFrame, error) { + cancel() + <-ctx.Done() + return AudioFrame{}, ctx.Err() + }, + } + var statuses []Status + worker := newAudioWorkerForTest( + t, + &fakeAudioFactory{reader: reader}, + &fakeAudioSink{}, + 1, + func(error) bool { return true }, + func(status Status) { statuses = append(statuses, status) }, + ) + + err := worker.Run( + ctx, + FeedConfig{Domain: "/audio", UUID: "audio", Active: true}, + ) + if !errors.Is(err, context.Canceled) { + t.Fatalf("Run() error = %v, want %v", err, context.Canceled) + } + if len(statuses) < 3 { + t.Fatalf("statuses = %#v, want connecting, stopping, idle", statuses) + } + last := statuses[len(statuses)-2:] + if last[0].State != StateStopping || last[1].State != StateIdle { + t.Fatalf("final statuses = %#v, want stopping then idle", last) + } +}