go-mxl-player/internal/playback/audio_slot_test.go

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
}