1package core
2
3import (
4 "errors"
5 "testing"
6 "unsafe"
7
8 portout "github.com/bnema/purego-pipewire/internal/ports/out"
9)
10
11// callbackOps is intentionally allocation-free so callback allocation tests
12// measure player code rather than a mocking framework.
13type callbackOps struct {
14 buffers []unsafe.Pointer
15 dequeueAt int
16 queueErr error
17 queued int
18}
19
20func (o *callbackOps) CreatePlaybackStream(unsafe.Pointer, string, func()) (unsafe.Pointer, error) {
21 return nil, errors.New("not implemented")
22}
23func (o *callbackOps) ConnectPlaybackStream(unsafe.Pointer, portout.PlaybackFormat) error {
24 return errors.New("not implemented")
25}
26func (o *callbackOps) SetStreamActive(unsafe.Pointer, bool) error {
27 return errors.New("not implemented")
28}
29func (o *callbackOps) DequeueBuffer(unsafe.Pointer) unsafe.Pointer {
30 if o.dequeueAt >= len(o.buffers) {
31 return nil
32 }
33 buf := o.buffers[o.dequeueAt]
34 o.dequeueAt++
35 return buf
36}
37func (o *callbackOps) QueueBuffer(unsafe.Pointer, unsafe.Pointer) error {
38 o.queued++
39 return o.queueErr
40}
41func (o *callbackOps) DisconnectStream(unsafe.Pointer) error { return nil }
42func (o *callbackOps) DestroyStream(unsafe.Pointer) {}
43func (o *callbackOps) CreateMainLoop() (unsafe.Pointer, error) {
44 return nil, errors.New("not implemented")
45}
46func (o *callbackOps) RunMainLoop(unsafe.Pointer) error { return nil }
47func (o *callbackOps) QuitMainLoop(unsafe.Pointer) {}
48func (o *callbackOps) DestroyMainLoop(unsafe.Pointer) {}
49
50func callbackBuffer(channels, frames, capacity int) (unsafe.Pointer, [][]float32, []spaData) {
51 samples := make([][]float32, channels)
52 data := make([]spaData, channels)
53 chunks := make([]spaChunk, channels)
54 for ch := range samples {
55 samples[ch] = make([]float32, capacity)
56 data[ch] = spaData{Maxsize: uint32(capacity * 4), Data: unsafe.Pointer(&samples[ch][0]), Chunk: &chunks[ch]}
57 }
58 spa := &spaBuffer{NDatas: uint32(channels), Datas: unsafe.Pointer(&data[0])}
59 return unsafe.Pointer(&pwBuffer{Buffer: spa}), samples, data
60}
61
62func newCallbackPlayer(ops *callbackOps, fill func(*PCMBuffer) (int, error)) *player {
63 p := newPlayer(PlayerConfig{Channels: 2, FramesPerBuffer: 64}, PlayerCallbacks{Fill: fill})
64 p.streamOps = ops
65 p.streamPtr = opaqueTestPtr()
66 p.setState(PlayerStatePlaying)
67 return p
68}
69
70// TestProcessCallbackSteadyStateAllocs is deliberately a RED-first regression
71// test: after its warm-up callback, processing must not allocate.
72func TestProcessCallbackSteadyStateAllocs(t *testing.T) {
73 buffer, _, _ := callbackBuffer(2, 64, 64)
74 ops := &callbackOps{buffers: make([]unsafe.Pointer, 1002)}
75 for i := range ops.buffers {
76 ops.buffers[i] = buffer
77 }
78 p := newCallbackPlayer(ops, func(buf *PCMBuffer) (int, error) { return buf.Frames, nil })
79
80 p.onProcess() // Allocate reusable headers before measuring steady state.
81 allocs := testing.AllocsPerRun(1000, p.onProcess)
82 if allocs != 0 {
83 t.Fatalf("steady-state process callback allocations = %v, want 0", allocs)
84 }
85}
86
87func BenchmarkPlayerOnProcessStereo(b *testing.B) {
88 buffer, _, _ := callbackBuffer(2, 64, 64)
89 ops := &callbackOps{buffers: make([]unsafe.Pointer, b.N+1)}
90 for i := range ops.buffers {
91 ops.buffers[i] = buffer
92 }
93 p := newCallbackPlayer(ops, func(buf *PCMBuffer) (int, error) { return buf.Frames, nil })
94
95 p.onProcess() // Warm reusable callback metadata before reporting allocations.
96 b.ReportAllocs()
97 b.ResetTimer()
98 for i := 0; i < b.N; i++ {
99 p.onProcess()
100 }
101}
102
103func TestProcessCallbackRefreshesChangedPointersAndCapacities(t *testing.T) {
104 first, firstSamples, _ := callbackBuffer(2, 64, 64)
105 second, secondSamples, _ := callbackBuffer(2, 64, 96)
106 ops := &callbackOps{buffers: []unsafe.Pointer{first, second}}
107 calls := 0
108 p := newCallbackPlayer(ops, func(buf *PCMBuffer) (int, error) {
109 calls++
110 for ch := range buf.Samples {
111 buf.Samples[ch][0] = float32(calls)
112 if cap(buf.Samples[ch]) < buf.Frames {
113 t.Fatalf("channel %d capacity = %d, want at least %d", ch, cap(buf.Samples[ch]), buf.Frames)
114 }
115 }
116 return buf.Frames, nil
117 })
118
119 p.onProcess()
120 p.onProcess()
121 if calls != 2 || ops.queued != 2 {
122 t.Fatalf("calls=%d queues=%d, want 2 each", calls, ops.queued)
123 }
124 for ch := range firstSamples {
125 if firstSamples[ch][0] != 1 {
126 t.Errorf("first buffer channel %d = %v, want 1", ch, firstSamples[ch][0])
127 }
128 if secondSamples[ch][0] != 2 {
129 t.Errorf("second buffer channel %d = %v, want 2", ch, secondSamples[ch][0])
130 }
131 }
132}
133
134func TestProcessCallbackInvalidMetadataClearsReusableViewAndRequeuesOnce(t *testing.T) {
135 valid, _, _ := callbackBuffer(2, 64, 64)
136 invalid, _, invalidData := callbackBuffer(2, 64, 64)
137 // Fail after channel zero has already been refreshed to prove partial
138 // validation cannot retain a native pointer from the rejected buffer.
139 invalidData[1].Data = nil
140 ops := &callbackOps{buffers: []unsafe.Pointer{valid, invalid}}
141 fillCalls := 0
142 p := newCallbackPlayer(ops, func(buf *PCMBuffer) (int, error) {
143 fillCalls++
144 return buf.Frames, nil
145 })
146
147 p.onProcess()
148 p.setState(PlayerStatePlaying) // inspect the invalid-buffer behavior independently.
149 p.onProcess()
150
151 if fillCalls != 1 {
152 t.Fatalf("Fill calls = %d, want 1 after invalid metadata", fillCalls)
153 }
154 if ops.queued != 2 {
155 t.Fatalf("QueueBuffer calls = %d, want exactly 2", ops.queued)
156 }
157 if p.pwView.buf != nil || p.pwView.data != nil {
158 t.Fatal("invalid metadata retained native buffer references")
159 }
160 for ch, samples := range p.pwView.samples {
161 if samples != nil {
162 t.Fatalf("invalid metadata retained channel %d samples", ch)
163 }
164 }
165}
166
167func TestProcessCallbackClearsReusableViewAfterQueueAndQueueError(t *testing.T) {
168 buffer, _, _ := callbackBuffer(2, 64, 64)
169 ops := &callbackOps{buffers: []unsafe.Pointer{buffer, buffer}}
170 clearedDuringError := false
171 p := newCallbackPlayer(ops, func(buf *PCMBuffer) (int, error) { return buf.Frames, nil })
172
173 p.onProcess()
174 if p.pwView.buf != nil || p.pwView.data != nil {
175 t.Fatal("successful callback retained reusable native buffer references")
176 }
177 for ch, samples := range p.pwView.samples {
178 if samples != nil {
179 t.Fatalf("successful callback retained channel %d samples", ch)
180 }
181 }
182
183 ops.queueErr = errors.New("queue failed")
184 p.callbacks.OnError = func(error) {
185 clearedDuringError = p.pwView.buf == nil && p.pwView.data == nil
186 for _, samples := range p.pwView.samples {
187 clearedDuringError = clearedDuringError && samples == nil
188 }
189 }
190 p.setState(PlayerStatePlaying)
191 p.onProcess()
192 if !clearedDuringError {
193 t.Fatal("queue error retained reusable native buffer references during OnError")
194 }
195}
196
197var _ portout.StreamOps = (*callbackOps)(nil)