203 lines
5.0 KiB
Go
203 lines
5.0 KiB
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
|
|
}
|