add audio slot lifecycle

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

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