1package pipewire
2
3import (
4 "context"
5 "errors"
6 "sync"
7 "testing"
8 "time"
9
10 "github.com/bnema/dumber/internal/application/port"
11 pipewiremocks "github.com/bnema/dumber/internal/infrastructure/audio/pipewire/mocks"
12 pwpipewire "github.com/bnema/purego-pipewire/pipewire"
13 "github.com/stretchr/testify/assert"
14 "github.com/stretchr/testify/require"
15)
16
17type playerRecorder struct {
18 *pipewiremocks.MockPlayer
19 started, stopped, closedFlag bool
20 startErr, stopErr, closeErr error
21}
22
23func newPlayerRecorder(t *testing.T) *playerRecorder {
24 t.Helper()
25 p := &playerRecorder{MockPlayer: pipewiremocks.NewMockPlayer(t)}
26 p.EXPECT().Start().RunAndReturn(func() error {
27 if p.startErr != nil {
28 return p.startErr
29 }
30 p.started = true
31 return nil
32 }).Maybe()
33 p.EXPECT().Stop().RunAndReturn(func() error {
34 if p.stopErr != nil {
35 return p.stopErr
36 }
37 p.stopped = true
38 return nil
39 }).Maybe()
40 p.EXPECT().Close().RunAndReturn(func() error {
41 if p.closeErr != nil {
42 return p.closeErr
43 }
44 p.closedFlag = true
45 return nil
46 }).Maybe()
47 return p
48}
49
50func testAdapter(frames int) *playerStreamAdapter {
51 return newPlayerStreamAdapter(context.Background(), 2, frames)
52}
53
54func stereoPCM(frames int) *pwpipewire.PCMBuffer {
55 return &pwpipewire.PCMBuffer{Frames: frames, Channels: 2, Samples: [][]float32{make([]float32, frames), make([]float32, frames)}}
56}
57
58func TestFactory_NewStreamValidatesAndStartsPlayer(t *testing.T) {
59 player := newPlayerRecorder(t)
60 factory := &Factory{createPlayer: func(config pwpipewire.PlayerConfig, callbacks pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
61 assert.Equal(t, 48000, config.SampleRate)
62 assert.Equal(t, 2, config.Channels)
63 assert.NotNil(t, callbacks.Fill)
64 return player, nil
65 }}
66 stream, err := factory.NewStream(context.Background(), port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 512})
67 require.NoError(t, err)
68 assert.True(t, player.started)
69 require.NoError(t, stream.Close())
70}
71
72func TestFactory_NewStreamRejectsInvalidFormat(t *testing.T) {
73 factory := &Factory{createPlayer: func(pwpipewire.PlayerConfig, pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
74 t.Fatal("unexpected player")
75 return nil, nil
76 }}
77 for _, format := range []port.AudioStreamFormat{
78 {SampleRate: 0, ChannelCount: 2, FramesPerBuffer: 512},
79 {SampleRate: 48000, ChannelCount: 0, FramesPerBuffer: 512},
80 {SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 0},
81 } {
82 stream, err := factory.NewStream(context.Background(), format)
83 require.ErrorIs(t, err, ErrInvalidFormat)
84 assert.Nil(t, stream)
85 }
86}
87
88func TestFactory_NewStreamClosesPlayerWhenStartFails(t *testing.T) {
89 player := newPlayerRecorder(t)
90 player.startErr = errors.New("start failed")
91 factory := &Factory{createPlayer: func(pwpipewire.PlayerConfig, pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
92 return player, nil
93 }}
94 stream, err := factory.NewStream(context.Background(), port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 512})
95 require.Error(t, err)
96 assert.Nil(t, stream)
97 assert.True(t, player.closedFlag)
98}
99
100func TestPlayerStreamAdapter_FillGetsOwnedPacketAfterCallbackMutation(t *testing.T) {
101 adapter := testAdapter(3)
102 samples := [][]float32{{1, 2, 3}, {4, 5, 6}}
103 require.NoError(t, adapter.Write(samples))
104 samples[0][0], samples[1][0] = 99, 99
105 pcm := stereoPCM(3)
106 frames, err := adapter.fill(pcm)
107 require.NoError(t, err)
108 assert.Equal(t, 3, frames)
109 assert.InEpsilon(t, 1, pcm.Samples[0][0], 0.0001)
110 assert.InEpsilon(t, 4, pcm.Samples[1][0], 0.0001)
111}
112
113func TestPlayerStreamAdapter_SaturationDropsWithoutBlocking(t *testing.T) {
114 adapter := testAdapter(2)
115 for i := 0; i < packetQueueCapacity; i++ {
116 require.NoError(t, adapter.Write([][]float32{{float32(i)}, {float32(i)}}))
117 }
118 done := make(chan error, 1)
119 go func() { done <- adapter.Write([][]float32{{99}, {99}}) }()
120 select {
121 case err := <-done:
122 require.NoError(t, err)
123 case <-time.After(time.Second):
124 t.Fatal("Write blocked while the bounded queue was saturated")
125 }
126 assert.Equal(t, uint64(1), adapter.dropCount.Load())
127 assert.Len(t, adapter.ready, packetQueueCapacity)
128}
129
130func TestPlayerStreamAdapter_EmptyFillCountsUnderrun(t *testing.T) {
131 adapter := testAdapter(2)
132 frames, err := adapter.fill(stereoPCM(2))
133 require.NoError(t, err)
134 assert.Zero(t, frames)
135 assert.Equal(t, uint64(1), adapter.underrunCount.Load())
136}
137
138func TestPlayerStreamAdapter_CopyToPCMHandlesMismatches(t *testing.T) {
139 pcm := &pwpipewire.PCMBuffer{Frames: 4, Channels: 1, Samples: [][]float32{make([]float32, 4)}}
140 assert.Equal(t, 2, copyToPCM(pcm, [][]float32{{1, 2}, {3, 4}}, 2))
141 assert.InEpsilon(t, 1, pcm.Samples[0][0], 0.0001)
142 assert.Zero(t, copyToPCM(nil, [][]float32{{1}}, 1))
143}
144
145func TestPlayerStreamAdapter_CloseVsWrite(t *testing.T) {
146 adapter := testAdapter(2)
147 player := newPlayerRecorder(t)
148 adapter.player = player
149 require.NoError(t, adapter.Close())
150 require.ErrorIs(t, adapter.Write([][]float32{{1}, {2}}), ErrStreamClosed)
151 assert.True(t, player.stopped)
152 assert.True(t, player.closedFlag)
153}
154
155func TestPlayerStreamAdapter_ConcurrentWriteClose_NoRace(t *testing.T) {
156 adapter := testAdapter(4)
157 var writers sync.WaitGroup
158 for range 4 {
159 writers.Add(1)
160 go func() {
161 defer writers.Done()
162 for range 1000 {
163 err := adapter.Write([][]float32{{1, 2}, {3, 4}})
164 if err != nil {
165 assert.ErrorIs(t, err, ErrStreamClosed)
166 return
167 }
168 }
169 }()
170 }
171 adapter.Close()
172 writers.Wait()
173}
174
175func TestStream_WriteFlowsThroughToFill(t *testing.T) {
176 var callbacks pwpipewire.PlayerCallbacks
177 player := newPlayerRecorder(t)
178 factory := &Factory{createPlayer: func(_ pwpipewire.PlayerConfig, got pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
179 callbacks = got
180 return player, nil
181 }}
182 stream, err := factory.NewStream(context.Background(), port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 4})
183 require.NoError(t, err)
184 require.NoError(t, stream.Write([][]float32{{.1, .2, .3, .4}, {.5, .6, .7, .8}}))
185 pcm := stereoPCM(4)
186 frames, err := callbacks.Fill(pcm)
187 require.NoError(t, err)
188 assert.Equal(t, 4, frames)
189 assert.InEpsilon(t, .1, pcm.Samples[0][0], .0001)
190 assert.InEpsilon(t, .5, pcm.Samples[1][0], .0001)
191 require.NoError(t, stream.Close())
192}
193
194func BenchmarkPlayerStreamAdapterStereo(b *testing.B) {
195 adapter := testAdapter(512)
196 samples := [][]float32{make([]float32, 512), make([]float32, 512)}
197 pcm := stereoPCM(512)
198 b.ReportAllocs()
199 b.SetBytes(2 * 512 * 4)
200 for i := 0; i < b.N; i++ {
201 if err := adapter.Write(samples); err != nil {
202 b.Fatal(err)
203 }
204 if _, err := adapter.fill(pcm); err != nil {
205 b.Fatal(err)
206 }
207 }
208}