add independent audio worker

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
+}

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)
+	}
+}