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

package playback

import (
	"context"
	"errors"
	"sync"
	"testing"
	"time"
)

type slotSyncFactory struct {
	opened chan SyncPairConfig

	mu         sync.Mutex
	active     int
	maxActive  int
	closeCount int
}

func newSlotSyncFactory() *slotSyncFactory {
	return &slotSyncFactory{opened: make(chan SyncPairConfig, 8)}
}

func (f *slotSyncFactory) OpenSync(
	_ context.Context,
	video FeedConfig,
	audio FeedConfig,
) (SyncReader, error) {
	f.mu.Lock()
	f.active++
	if f.active > f.maxActive {
		f.maxActive = f.active
	}
	f.mu.Unlock()
	f.opened <- SyncPairConfig{Video: video, Audio: audio}
	return &slotSyncReader{factory: f}, nil
}

func (f *slotSyncFactory) counts() (active, maxActive, closeCount int) {
	f.mu.Lock()
	defer f.mu.Unlock()
	return f.active, f.maxActive, f.closeCount
}

type slotSyncReader struct {
	factory *slotSyncFactory
}

func (r *slotSyncReader) ReadSync(ctx context.Context) (SyncFrame, error) {
	<-ctx.Done()
	return SyncFrame{}, ctx.Err()
}

func (r *slotSyncReader) Close() error {
	r.factory.mu.Lock()
	defer r.factory.mu.Unlock()
	r.factory.active--
	r.factory.closeCount++
	return nil
}

func newSlotTestSyncWorker(t *testing.T, factory SyncReaderFactory) *SyncWorker {
	t.Helper()
	worker, err := NewSyncWorker(
		factory,
		&fakeVideoSink{},
		&fakeAudioSink{},
		testRetryPolicy(1),
		func(error) bool { return false },
		nil,
	)
	if err != nil {
		t.Fatalf("NewSyncWorker() error = %v", err)
	}
	return worker
}

func receiveSyncSlotOpen(t *testing.T, opened <-chan SyncPairConfig) SyncPairConfig {
	t.Helper()
	select {
	case config := <-opened:
		return config
	case <-time.After(time.Second):
		t.Fatal("sync worker did not open")
		return SyncPairConfig{}
	}
}

func testSyncPair(name string, active bool) SyncPairConfig {
	return SyncPairConfig{
		Video: FeedConfig{Domain: "/video", UUID: name + "-video", Active: active},
		Audio: FeedConfig{Domain: "/audio", UUID: name + "-audio", Active: active},
	}
}

func TestSyncPairConfigRejectsActivityMismatch(t *testing.T) {
	config := testSyncPair("pair", true)
	config.Audio.Active = false
	if err := config.Validate(); !errors.Is(err, ErrSyncActivityMismatch) {
		t.Fatalf("Validate() error = %v, want %v", err, ErrSyncActivityMismatch)
	}
}

func TestNewSyncSlotRequiresWorker(t *testing.T) {
	slot, err := NewSyncSlot(nil)
	if slot != nil {
		t.Fatalf("NewSyncSlot(nil) slot = %#v, want nil", slot)
	}
	if !errors.Is(err, ErrSyncWorkerRequired) {
		t.Fatalf("NewSyncSlot(nil) error = %v, want %v", err, ErrSyncWorkerRequired)
	}
}

func TestSyncSlotStartsInitialActivePair(t *testing.T) {
	factory := newSlotSyncFactory()
	slot, err := NewSyncSlot(newSlotTestSyncWorker(t, factory))
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan error, 1)
	want := testSyncPair("first", true)
	go func() { done <- slot.Run(ctx, want, make(chan SyncPairConfig)) }()

	if got := receiveSyncSlotOpen(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 context.Canceled", err)
		}
	case <-time.After(time.Second):
		t.Fatal("Run() did not stop after cancellation")
	}
	active, _, closed := factory.counts()
	if active != 0 || closed != 1 {
		t.Fatalf("reader counts = active %d, closed %d; want 0, 1", active, closed)
	}
}

func TestSyncSlotReplacesWithoutOverlappingWorkers(t *testing.T) {
	factory := newSlotSyncFactory()
	slot, err := NewSyncSlot(newSlotTestSyncWorker(t, factory))
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	commands := make(chan SyncPairConfig)
	done := make(chan error, 1)
	first := testSyncPair("first", true)
	second := testSyncPair("second", true)
	go func() { done <- slot.Run(ctx, first, commands) }()
	receiveSyncSlotOpen(t, factory.opened)
	commands <- second
	if got := receiveSyncSlotOpen(t, factory.opened); got != second {
		t.Fatalf("replacement = %#v, want %#v", got, second)
	}
	cancel()
	<-done
	active, maxActive, closed := factory.counts()
	if active != 0 || maxActive != 1 || closed != 2 {
		t.Fatalf("counts = active %d, maximum %d, closed %d; want 0, 1, 2", active, maxActive, closed)
	}
}

func TestSyncSlotIgnoresInvalidCommand(t *testing.T) {
	factory := newSlotSyncFactory()
	slot, err := NewSyncSlot(newSlotTestSyncWorker(t, factory))
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	commands := make(chan SyncPairConfig)
	done := make(chan error, 1)
	initial := testSyncPair("first", true)
	go func() { done <- slot.Run(ctx, initial, commands) }()
	receiveSyncSlotOpen(t, factory.opened)

	invalid := testSyncPair("invalid", true)
	invalid.Audio.Active = false
	commands <- invalid
	select {
	case config := <-factory.opened:
		t.Fatalf("invalid command opened config %#v", config)
	case <-time.After(20 * time.Millisecond):
	}
	active, _, closed := factory.counts()
	if active != 1 || closed != 0 {
		t.Fatalf("invalid command disturbed reader: active %d, closed %d", active, closed)
	}
	cancel()
	<-done
}

func TestSyncSlotInactivePairStopsWithoutRestart(t *testing.T) {
	factory := newSlotSyncFactory()
	slot, err := NewSyncSlot(newSlotTestSyncWorker(t, factory))
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	commands := make(chan SyncPairConfig)
	done := make(chan error, 1)
	go func() { done <- slot.Run(ctx, testSyncPair("first", true), commands) }()
	receiveSyncSlotOpen(t, factory.opened)
	commands <- testSyncPair("first", false)

	deadline := time.Now().Add(time.Second)
	for {
		active, _, closed := factory.counts()
		if active == 0 && closed == 1 {
			break
		}
		if time.Now().After(deadline) {
			t.Fatal("inactive pair did not stop reader")
		}
		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")
	}
}