1package core
2
3import (
4 "errors"
5 "io"
6 "testing"
7 "unsafe"
8
9 portout "github.com/bnema/purego-pipewire/internal/ports/out"
10 "github.com/bnema/purego-pipewire/internal/ports/out/mocks"
11 "github.com/stretchr/testify/mock"
12)
13
14func opaqueTestPtr() unsafe.Pointer {
15 return unsafe.Pointer(new(byte))
16}
17
18// TestProcessCallbackSilenceFillUnderrun tests that when Fill returns fewer frames
19// than requested, the remaining frames are filled with silence
20func TestProcessCallbackSilenceFillUnderrun(t *testing.T) {
21 fillCallCount := 0
22 var receivedBuf *PCMBuffer
23
24 p := newPlayer(
25 PlayerConfig{
26 FramesPerBuffer: 4,
27 Channels: 2,
28 UnderrunPolicy: UnderrunFillSilence,
29 },
30 PlayerCallbacks{
31 Fill: func(buf *PCMBuffer) (int, error) {
32 fillCallCount++
33 receivedBuf = buf
34 // Return only 2 frames, leaving 2 frames as underrun
35 return 2, nil
36 },
37 },
38 )
39
40 // Set player to playing state
41 p.setState(PlayerStatePlaying)
42
43 buf := &PCMBuffer{
44 Frames: 4,
45 Channels: 2,
46 }
47 buf.allocate()
48
49 frames, err := p.processPCM(buf)
50 if err != nil {
51 t.Fatalf("processPCM returned error: %v", err)
52 }
53
54 if frames != 4 {
55 t.Errorf("processPCM returned %d frames, want 4", frames)
56 }
57
58 if fillCallCount != 1 {
59 t.Errorf("Fill callback called %d times, want 1", fillCallCount)
60 }
61
62 // Verify the buffer was silence-filled
63 // Frames 0-1 should have data (we don't verify the actual values since Fill could write anything)
64 // Frames 2-3 should be silence (0.0)
65 for ch := 0; ch < 2; ch++ {
66 for frame := 2; frame < 4; frame++ {
67 if buf.Samples[ch][frame] != 0.0 {
68 t.Errorf("Expected silence at channel %d frame %d, got %f", ch, frame, buf.Samples[ch][frame])
69 }
70 }
71 }
72
73 if receivedBuf != buf {
74 t.Error("Fill callback received different buffer than passed to processPCM")
75 }
76}
77
78// TestProcessCallbackFailFastUnderrun tests that when configured to fail on underrun,
79// the player transitions to error state and invokes OnError callback
80func TestProcessCallbackFailFastUnderrun(t *testing.T) {
81 errCalled := false
82 var errReceived error
83
84 p := newPlayer(
85 PlayerConfig{
86 FramesPerBuffer: 4,
87 Channels: 2,
88 UnderrunPolicy: UnderrunFailFast,
89 },
90 PlayerCallbacks{
91 Fill: func(buf *PCMBuffer) (int, error) {
92 // Return only 2 frames, triggering underrun
93 return 2, nil
94 },
95 OnError: func(err error) {
96 errCalled = true
97 errReceived = err
98 },
99 },
100 )
101
102 // Set player to playing state
103 p.setState(PlayerStatePlaying)
104
105 buf := &PCMBuffer{
106 Frames: 4,
107 Channels: 2,
108 }
109 buf.allocate()
110
111 _, err := p.processPCM(buf)
112 if err == nil {
113 t.Fatal("processPCM should have returned error for underrun with FailFast policy")
114 }
115
116 if p.State() != PlayerStateError {
117 t.Errorf("expected state Error after underrun, got %v", p.State())
118 }
119
120 if !errCalled {
121 t.Error("OnError callback was not called")
122 }
123
124 if errReceived == nil {
125 t.Error("OnError callback received nil error")
126 }
127}
128
129// TestProcessCallbackPausedSuppressesFillAndEmitsSilence tests that when paused,
130// Fill is not called and the buffer is filled with silence
131func TestProcessCallbackPausedSuppressesFillAndEmitsSilence(t *testing.T) {
132 fillCallCount := 0
133
134 p := newPlayer(
135 PlayerConfig{
136 FramesPerBuffer: 4,
137 Channels: 2,
138 UnderrunPolicy: UnderrunFillSilence,
139 },
140 PlayerCallbacks{
141 Fill: func(buf *PCMBuffer) (int, error) {
142 fillCallCount++
143 return 4, nil
144 },
145 },
146 )
147
148 // Set player to paused state
149 p.setState(PlayerStatePaused)
150
151 buf := &PCMBuffer{
152 Frames: 4,
153 Channels: 2,
154 }
155 buf.allocate()
156
157 frames, err := p.processPCM(buf)
158 if err != nil {
159 t.Fatalf("processPCM returned error: %v", err)
160 }
161
162 if frames != 4 {
163 t.Errorf("processPCM returned %d frames, want 4", frames)
164 }
165
166 if fillCallCount != 0 {
167 t.Errorf("Fill callback called %d times while paused, want 0", fillCallCount)
168 }
169
170 // Verify the entire buffer is silence
171 for ch := 0; ch < 2; ch++ {
172 for frame := 0; frame < 4; frame++ {
173 if buf.Samples[ch][frame] != 0.0 {
174 t.Errorf("Expected silence at channel %d frame %d while paused, got %f", ch, frame, buf.Samples[ch][frame])
175 }
176 }
177 }
178}
179
180// TestProcessCallbackStoppedFillsSilence tests that when stopped,
181// Fill is not called and the buffer is filled with silence
182func TestProcessCallbackStoppedFillsSilence(t *testing.T) {
183 fillCallCount := 0
184
185 p := newPlayer(
186 PlayerConfig{
187 FramesPerBuffer: 4,
188 Channels: 2,
189 UnderrunPolicy: UnderrunFillSilence,
190 },
191 PlayerCallbacks{
192 Fill: func(buf *PCMBuffer) (int, error) {
193 fillCallCount++
194 return 4, nil
195 },
196 },
197 )
198
199 // Set player to stopped state
200 p.setState(PlayerStateStopped)
201
202 buf := &PCMBuffer{
203 Frames: 4,
204 Channels: 2,
205 }
206 buf.allocate()
207
208 frames, err := p.processPCM(buf)
209 if err != nil {
210 t.Fatalf("processPCM returned error: %v", err)
211 }
212
213 if frames != 4 {
214 t.Errorf("processPCM returned %d frames, want 4", frames)
215 }
216
217 if fillCallCount != 0 {
218 t.Errorf("Fill callback called %d times while stopped, want 0", fillCallCount)
219 }
220
221 // Verify the entire buffer is silence
222 for ch := 0; ch < 2; ch++ {
223 for frame := 0; frame < 4; frame++ {
224 if buf.Samples[ch][frame] != 0.0 {
225 t.Errorf("Expected silence at channel %d frame %d while stopped, got %f", ch, frame, buf.Samples[ch][frame])
226 }
227 }
228 }
229}
230
231// TestProcessCallbackDrainOnEOF tests that io.EOF from Fill triggers drain callback
232func TestProcessCallbackDrainOnEOF(t *testing.T) {
233 drainCalled := false
234 fillCallCount := 0
235
236 p := newPlayer(
237 PlayerConfig{
238 FramesPerBuffer: 4,
239 Channels: 2,
240 UnderrunPolicy: UnderrunFillSilence,
241 },
242 PlayerCallbacks{
243 Fill: func(buf *PCMBuffer) (int, error) {
244 fillCallCount++
245 // Return EOF to signal drain
246 return 2, io.EOF
247 },
248 OnDrain: func() {
249 drainCalled = true
250 },
251 },
252 )
253
254 // Set player to playing state
255 p.setState(PlayerStatePlaying)
256
257 buf := &PCMBuffer{
258 Frames: 4,
259 Channels: 2,
260 }
261 buf.allocate()
262
263 frames, err := p.processPCM(buf)
264 if err != nil {
265 t.Fatalf("processPCM returned error: %v", err)
266 }
267
268 if frames != 2 {
269 t.Errorf("processPCM returned %d frames, want 2", frames)
270 }
271
272 if !drainCalled {
273 t.Error("OnDrain callback was not called on io.EOF")
274 }
275
276 if fillCallCount != 1 {
277 t.Errorf("Fill callback called %d times, want 1", fillCallCount)
278 }
279}
280
281// TestProcessCallbackErrorHandling tests that errors from Fill (other than EOF)
282// fail the player and invoke OnError callback
283func TestProcessCallbackErrorHandling(t *testing.T) {
284 testErr := errors.New("fill error")
285 errCalled := false
286 var errReceived error
287
288 p := newPlayer(
289 PlayerConfig{
290 FramesPerBuffer: 4,
291 Channels: 2,
292 UnderrunPolicy: UnderrunFillSilence,
293 },
294 PlayerCallbacks{
295 Fill: func(buf *PCMBuffer) (int, error) {
296 return 0, testErr
297 },
298 OnError: func(err error) {
299 errCalled = true
300 errReceived = err
301 },
302 },
303 )
304
305 // Set player to playing state
306 p.setState(PlayerStatePlaying)
307
308 buf := &PCMBuffer{
309 Frames: 4,
310 Channels: 2,
311 }
312 buf.allocate()
313
314 _, err := p.processPCM(buf)
315 if err == nil {
316 t.Fatal("processPCM should have returned error")
317 }
318
319 if p.State() != PlayerStateError {
320 t.Errorf("expected state Error after fill error, got %v", p.State())
321 }
322
323 if !errCalled {
324 t.Error("OnError callback was not called")
325 }
326
327 if !errors.Is(errReceived, testErr) {
328 t.Errorf("OnError received error %v, want %v", errReceived, testErr)
329 }
330}
331
332// TestProcessCallbackIdleFillsSilence tests that when idle,
333// Fill is not called and the buffer is filled with silence
334func TestProcessCallbackIdleFillsSilence(t *testing.T) {
335 fillCallCount := 0
336
337 p := newPlayer(
338 PlayerConfig{
339 FramesPerBuffer: 4,
340 Channels: 2,
341 UnderrunPolicy: UnderrunFillSilence,
342 },
343 PlayerCallbacks{
344 Fill: func(buf *PCMBuffer) (int, error) {
345 fillCallCount++
346 return 4, nil
347 },
348 },
349 )
350
351 // Player starts in Idle state
352 buf := &PCMBuffer{
353 Frames: 4,
354 Channels: 2,
355 }
356 buf.allocate()
357
358 frames, err := p.processPCM(buf)
359 if err != nil {
360 t.Fatalf("processPCM returned error: %v", err)
361 }
362
363 if frames != 4 {
364 t.Errorf("processPCM returned %d frames, want 4", frames)
365 }
366
367 if fillCallCount != 0 {
368 t.Errorf("Fill callback called %d times while idle, want 0", fillCallCount)
369 }
370}
371
372// TestProcessCallbackUnderrunCallback tests that OnUnderrun is called when underrun occurs
373func TestProcessCallbackUnderrunCallback(t *testing.T) {
374 underrunCalled := false
375 underrunFrames := 0
376
377 p := newPlayer(
378 PlayerConfig{
379 FramesPerBuffer: 4,
380 Channels: 2,
381 UnderrunPolicy: UnderrunFillSilence,
382 },
383 PlayerCallbacks{
384 Fill: func(buf *PCMBuffer) (int, error) {
385 return 2, nil
386 },
387 OnUnderrun: func(frames int) {
388 underrunCalled = true
389 underrunFrames = frames
390 },
391 },
392 )
393
394 // Set player to playing state
395 p.setState(PlayerStatePlaying)
396
397 buf := &PCMBuffer{
398 Frames: 4,
399 Channels: 2,
400 }
401 buf.allocate()
402
403 _, err := p.processPCM(buf)
404 if err != nil {
405 t.Fatalf("processPCM returned error: %v", err)
406 }
407
408 if !underrunCalled {
409 t.Error("OnUnderrun callback was not called")
410 }
411
412 if underrunFrames != 2 {
413 t.Errorf("OnUnderrun received %d frames, want 2", underrunFrames)
414 }
415}
416
417// TestProcessCallbackFillReturnsFullFrames tests normal operation when Fill returns all requested frames
418func TestProcessCallbackFillReturnsFullFrames(t *testing.T) {
419 fillCallCount := 0
420
421 p := newPlayer(
422 PlayerConfig{
423 FramesPerBuffer: 4,
424 Channels: 2,
425 UnderrunPolicy: UnderrunFillSilence,
426 },
427 PlayerCallbacks{
428 Fill: func(buf *PCMBuffer) (int, error) {
429 fillCallCount++
430 // Write some data to verify Fill was called
431 for ch := 0; ch < buf.Channels; ch++ {
432 for frame := 0; frame < buf.Frames; frame++ {
433 buf.Samples[ch][frame] = float32(frame + ch*10)
434 }
435 }
436 return 4, nil
437 },
438 },
439 )
440
441 // Set player to playing state
442 p.setState(PlayerStatePlaying)
443
444 buf := &PCMBuffer{
445 Frames: 4,
446 Channels: 2,
447 }
448 buf.allocate()
449
450 frames, err := p.processPCM(buf)
451 if err != nil {
452 t.Fatalf("processPCM returned error: %v", err)
453 }
454
455 if frames != 4 {
456 t.Errorf("processPCM returned %d frames, want 4", frames)
457 }
458
459 if fillCallCount != 1 {
460 t.Errorf("Fill callback called %d times, want 1", fillCallCount)
461 }
462
463 // Verify the data was written by Fill
464 for ch := 0; ch < 2; ch++ {
465 for frame := 0; frame < 4; frame++ {
466 expected := float32(frame + ch*10)
467 if buf.Samples[ch][frame] != expected {
468 t.Errorf("Expected %f at channel %d frame %d, got %f", expected, ch, frame, buf.Samples[ch][frame])
469 }
470 }
471 }
472}
473
474// TestPCMBufferAllocate tests the allocate method
475func TestPCMBufferAllocate(t *testing.T) {
476 buf := &PCMBuffer{
477 Frames: 4,
478 Channels: 2,
479 }
480
481 buf.allocate()
482
483 if buf.Samples == nil {
484 t.Fatal("Samples is nil after allocate")
485 }
486
487 if len(buf.Samples) != 2 {
488 t.Errorf("len(Samples) = %d, want 2", len(buf.Samples))
489 }
490
491 for ch := 0; ch < 2; ch++ {
492 if len(buf.Samples[ch]) != 4 {
493 t.Errorf("len(Samples[%d]) = %d, want 4", ch, len(buf.Samples[ch]))
494 }
495 }
496}
497
498// --- Task 6: Start and process callback integration tests ---
499
500// TestPlayerStartCreatesLoopConnectsStreamAndRunsMainLoop verifies that Start()
501// creates the main loop, creates a playback stream, connects it with the right
502// format, activates it, stores owned pointers, and starts the main loop in a
503// goroutine — then transitions to Playing.
504func TestPlayerStartCreatesLoopConnectsStreamAndRunsMainLoop(t *testing.T) {
505 mockOps := mocks.NewMockStreamOps(t)
506 fakeLoop := opaqueTestPtr()
507 fakeStream := opaqueTestPtr()
508 cfg := PlayerConfig{
509 SampleRate: 48000,
510 Channels: 2,
511 FramesPerBuffer: 256,
512 }
513
514 wg := expectStartWithSync(mockOps, fakeLoop, fakeStream, cfg)
515
516 p := newPlayer(cfg, PlayerCallbacks{})
517 p.streamOps = mockOps
518
519 if err := p.Start(); err != nil {
520 t.Fatalf("Start failed: %v", err)
521 }
522
523 if p.State() != PlayerStatePlaying {
524 t.Fatalf("expected Playing state after Start, got %v", p.State())
525 }
526
527 // Verify owned pointers are set
528 p.mu.Lock()
529 loopPtr := p.loopPtr
530 streamPtr := p.streamPtr
531 p.mu.Unlock()
532
533 if loopPtr != fakeLoop {
534 t.Fatalf("expected loopPtr = %v, got %v", fakeLoop, loopPtr)
535 }
536 if streamPtr != fakeStream {
537 t.Fatalf("expected streamPtr = %v, got %v", fakeStream, streamPtr)
538 }
539
540 // Wait for the RunMainLoop goroutine to finish
541 waitOnMainLoop(t, wg)
542}
543
544// TestProcessCallbackDequeuesProcessesAndQueues verifies the full onProcess
545// path: dequeue buffer, build pwBufferView, call processPCM, commit, and
546// queue the buffer back.
547func TestProcessCallbackDequeuesProcessesAndQueues(t *testing.T) {
548 mockOps := mocks.NewMockStreamOps(t)
549 fakeLoop := opaqueTestPtr()
550 fakeStream := opaqueTestPtr()
551 cfg := PlayerConfig{
552 SampleRate: 48000,
553 Channels: 2,
554 FramesPerBuffer: 64,
555 }
556
557 // Build a real pwBuffer that onProcess will dequeue
558 const channels = 2
559 const frames = 64
560 channelData := make([][]float32, channels)
561 chunkSlice := make([]spaChunk, channels)
562 dataSlice := make([]spaData, channels)
563 floatPtrs := make([]unsafe.Pointer, channels)
564 for ch := 0; ch < channels; ch++ {
565 channelData[ch] = make([]float32, frames)
566 floatPtrs[ch] = unsafe.Pointer(&channelData[ch][0])
567 dataSlice[ch] = spaData{
568 Type: spaDataTypeMemPtr,
569 Flags: spaDataFlagRW,
570 Maxsize: uint32(frames * 4),
571 Data: floatPtrs[ch],
572 Chunk: &chunkSlice[ch],
573 }
574 }
575 spaBuf := spaBuffer{
576 NDatas: uint32(channels),
577 Datas: unsafe.Pointer(&dataSlice[0]),
578 }
579 pwBuf := pwBuffer{
580 Buffer: &spaBuf,
581 }
582 bufPtr := unsafe.Pointer(&pwBuf)
583
584 wg := expectStartWithSync(mockOps, fakeLoop, fakeStream, cfg)
585
586 // Expect DequeueBuffer, then QueueBuffer after processing
587 mockOps.EXPECT().DequeueBuffer(fakeStream).Return(bufPtr)
588 mockOps.EXPECT().QueueBuffer(fakeStream, bufPtr).Return(nil)
589
590 fillCalled := false
591 p := newPlayer(cfg, PlayerCallbacks{
592 Fill: func(buf *PCMBuffer) (int, error) {
593 fillCalled = true
594 // Write data to verify it propagates
595 for ch := 0; ch < buf.Channels; ch++ {
596 for f := 0; f < buf.Frames; f++ {
597 buf.Samples[ch][f] = float32(f)
598 }
599 }
600 return buf.Frames, nil
601 },
602 })
603 p.streamOps = mockOps
604
605 if err := p.Start(); err != nil {
606 t.Fatalf("Start failed: %v", err)
607 }
608
609 if p.State() != PlayerStatePlaying {
610 t.Fatalf("expected Playing state, got %v", p.State())
611 }
612
613 // Call onProcess directly (simulating PipeWire calling it)
614 p.onProcess()
615
616 if !fillCalled {
617 t.Error("Fill callback was not called during onProcess")
618 }
619
620 // Verify chunk sizes were updated by Commit
621 for ch := 0; ch < channels; ch++ {
622 if chunkSlice[ch].Size != uint32(frames*4) {
623 t.Errorf("chunk[%d].Size = %d, want %d", ch, chunkSlice[ch].Size, frames*4)
624 }
625 if chunkSlice[ch].Stride != 4 {
626 t.Errorf("chunk[%d].Stride = %d, want 4", ch, chunkSlice[ch].Stride)
627 }
628 }
629
630 // Verify pwBuffer.Size reflects frame count
631 if pwBuf.Size != uint64(frames) {
632 t.Errorf("pwBuf.Size = %d, want %d", pwBuf.Size, frames)
633 }
634
635 waitOnMainLoop(t, wg)
636}
637
638// TestProcessCallbackNilBufferReturnsEarly verifies that onProcess returns
639// early when DequeueBuffer returns nil.
640func TestProcessCallbackNilBufferReturnsEarly(t *testing.T) {
641 mockOps := mocks.NewMockStreamOps(t)
642 fakeLoop := opaqueTestPtr()
643 fakeStream := opaqueTestPtr()
644 cfg := defaultTestConfig()
645
646 wg := expectStartWithSync(mockOps, fakeLoop, fakeStream, cfg)
647
648 fillCalled := false
649 p := newPlayer(cfg, PlayerCallbacks{
650 Fill: func(buf *PCMBuffer) (int, error) {
651 fillCalled = true
652 return buf.Frames, nil
653 },
654 })
655 p.streamOps = mockOps
656
657 if err := p.Start(); err != nil {
658 t.Fatalf("Start failed: %v", err)
659 }
660 waitOnMainLoop(t, wg)
661
662 // DequeueBuffer returns nil — onProcess should return early
663 mockOps.EXPECT().DequeueBuffer(fakeStream).Return(nil)
664
665 p.onProcess()
666
667 if fillCalled {
668 t.Error("Fill should not be called when DequeueBuffer returns nil")
669 }
670}
671
672// TestProcessCallbackQueueBufferFailsRoutesThroughFail verifies that when
673// QueueBuffer returns an error, onProcess routes it through p.fail().
674func TestProcessCallbackQueueBufferFailsRoutesThroughFail(t *testing.T) {
675 mockOps := mocks.NewMockStreamOps(t)
676 fakeLoop := opaqueTestPtr()
677 fakeStream := opaqueTestPtr()
678 cfg := PlayerConfig{
679 SampleRate: 48000,
680 Channels: 2,
681 FramesPerBuffer: 64,
682 }
683
684 // Build a real pwBuffer
685 const channels = 2
686 const frames = 64
687 channelData := make([][]float32, channels)
688 floatPtrs := make([]unsafe.Pointer, channels)
689 for ch := 0; ch < channels; ch++ {
690 channelData[ch] = make([]float32, frames)
691 floatPtrs[ch] = unsafe.Pointer(&channelData[ch][0])
692 }
693 chunkSlice := make([]spaChunk, channels)
694 dataSlice := make([]spaData, channels)
695 for ch := 0; ch < channels; ch++ {
696 dataSlice[ch] = spaData{
697 Type: spaDataTypeMemPtr,
698 Flags: spaDataFlagRW,
699 Maxsize: uint32(frames * 4),
700 Data: floatPtrs[ch],
701 Chunk: &chunkSlice[ch],
702 }
703 }
704 spaBuf := spaBuffer{
705 NDatas: uint32(channels),
706 Datas: unsafe.Pointer(&dataSlice[0]),
707 }
708 pwBuf := pwBuffer{
709 Buffer: &spaBuf,
710 }
711 bufPtr := unsafe.Pointer(&pwBuf)
712
713 wg := expectStartWithSync(mockOps, fakeLoop, fakeStream, cfg)
714
715 queueErr := errors.New("queue buffer failed")
716 errCalled := false
717 var errReceived error
718
719 p := newPlayer(cfg, PlayerCallbacks{
720 Fill: func(buf *PCMBuffer) (int, error) {
721 return buf.Frames, nil
722 },
723 OnError: func(err error) {
724 errCalled = true
725 errReceived = err
726 },
727 })
728 p.streamOps = mockOps
729
730 if err := p.Start(); err != nil {
731 t.Fatalf("Start failed: %v", err)
732 }
733 waitOnMainLoop(t, wg)
734
735 // Expect DequeueBuffer to return a buffer, then QueueBuffer to fail
736 mockOps.EXPECT().DequeueBuffer(fakeStream).Return(bufPtr)
737 mockOps.EXPECT().QueueBuffer(fakeStream, bufPtr).Return(queueErr)
738
739 p.onProcess()
740
741 if !errCalled {
742 t.Error("OnError should have been called when QueueBuffer fails")
743 }
744 if errReceived == nil {
745 t.Error("expected non-nil error from OnError")
746 }
747 if p.State() != PlayerStateError {
748 t.Errorf("expected Error state after QueueBuffer failure, got %v", p.State())
749 }
750}
751
752// TestProcessCallbackBufferViewFailsRoutesThroughFail verifies that when
753// newPWBufferView fails, onProcess routes the error through p.fail()
754// and still re-queues the buffer.
755func TestProcessCallbackBufferViewFailsRoutesThroughFail(t *testing.T) {
756 mockOps := mocks.NewMockStreamOps(t)
757 fakeLoop := opaqueTestPtr()
758 fakeStream := opaqueTestPtr()
759 cfg := PlayerConfig{
760 SampleRate: 48000,
761 Channels: 2,
762 FramesPerBuffer: 64,
763 }
764
765 // Build a buffer with NDatas mismatch to trigger ErrChannelMismatch
766 // NDatas = 1 but config.Channels = 2
767 singleData := spaData{
768 Type: spaDataTypeMemPtr,
769 Maxsize: 256,
770 Data: unsafe.Pointer(new(float32)),
771 Chunk: &spaChunk{},
772 }
773 spaBuf := spaBuffer{
774 NDatas: 1, // mismatch with Channels=2
775 Datas: unsafe.Pointer(&singleData),
776 }
777 pwBuf := pwBuffer{
778 Buffer: &spaBuf,
779 }
780 bufPtr := unsafe.Pointer(&pwBuf)
781
782 wg := expectStartWithSync(mockOps, fakeLoop, fakeStream, cfg)
783
784 errCalled := false
785 var errReceived error
786
787 p := newPlayer(cfg, PlayerCallbacks{
788 Fill: func(buf *PCMBuffer) (int, error) {
789 return buf.Frames, nil
790 },
791 OnError: func(err error) {
792 errCalled = true
793 errReceived = err
794 },
795 })
796 p.streamOps = mockOps
797
798 if err := p.Start(); err != nil {
799 t.Fatalf("Start failed: %v", err)
800 }
801 waitOnMainLoop(t, wg)
802
803 // DequeueBuffer returns a buffer that will fail newPWBufferView.
804 // The buffer should still be re-queued.
805 mockOps.EXPECT().DequeueBuffer(fakeStream).Return(bufPtr)
806 mockOps.EXPECT().QueueBuffer(fakeStream, bufPtr).Return(nil)
807
808 p.onProcess()
809
810 if !errCalled {
811 t.Error("OnError should have been called when buffer view fails")
812 }
813 if errReceived == nil {
814 t.Error("expected non-nil error from OnError")
815 }
816 if !errors.Is(errReceived, ErrChannelMismatch) {
817 t.Errorf("expected ErrChannelMismatch, got %v", errReceived)
818 }
819 if p.State() != PlayerStateError {
820 t.Errorf("expected Error state after buffer view failure, got %v", p.State())
821 }
822}
823
824// TestProcessCallbackProcessCMErrorRequeuesBuffer verifies that when processPCM
825// returns an error, onProcess still re-queues the buffer and lets the existing
826// error behavior stand.
827func TestProcessCallbackProcessCMErrorRequeuesBuffer(t *testing.T) {
828 mockOps := mocks.NewMockStreamOps(t)
829 fakeLoop := opaqueTestPtr()
830 fakeStream := opaqueTestPtr()
831 cfg := PlayerConfig{
832 SampleRate: 48000,
833 Channels: 2,
834 FramesPerBuffer: 64,
835 }
836
837 // Build a real pwBuffer
838 const channels = 2
839 const frames = 64
840 channelData := make([][]float32, channels)
841 floatPtrs := make([]unsafe.Pointer, channels)
842 for ch := 0; ch < channels; ch++ {
843 channelData[ch] = make([]float32, frames)
844 floatPtrs[ch] = unsafe.Pointer(&channelData[ch][0])
845 }
846 chunkSlice := make([]spaChunk, channels)
847 dataSlice := make([]spaData, channels)
848 for ch := 0; ch < channels; ch++ {
849 dataSlice[ch] = spaData{
850 Type: spaDataTypeMemPtr,
851 Flags: spaDataFlagRW,
852 Maxsize: uint32(frames * 4),
853 Data: floatPtrs[ch],
854 Chunk: &chunkSlice[ch],
855 }
856 }
857 spaBuf := spaBuffer{
858 NDatas: uint32(channels),
859 Datas: unsafe.Pointer(&dataSlice[0]),
860 }
861 pwBuf := pwBuffer{
862 Buffer: &spaBuf,
863 }
864 bufPtr := unsafe.Pointer(&pwBuf)
865
866 wg := expectStartWithSync(mockOps, fakeLoop, fakeStream, cfg)
867
868 fillErr := errors.New("fill error")
869 errCalled := false
870 var errReceived error
871
872 p := newPlayer(cfg, PlayerCallbacks{
873 Fill: func(buf *PCMBuffer) (int, error) {
874 return 0, fillErr
875 },
876 OnError: func(err error) {
877 errCalled = true
878 errReceived = err
879 },
880 })
881 p.streamOps = mockOps
882
883 if err := p.Start(); err != nil {
884 t.Fatalf("Start failed: %v", err)
885 }
886 waitOnMainLoop(t, wg)
887
888 // DequeueBuffer returns a buffer; the buffer should still be re-queued
889 // even though processPCM (via Fill) returns an error.
890 mockOps.EXPECT().DequeueBuffer(fakeStream).Return(bufPtr)
891 mockOps.EXPECT().QueueBuffer(fakeStream, bufPtr).Return(nil)
892
893 p.onProcess()
894
895 if !errCalled {
896 t.Error("OnError should have been called when Fill returns an error")
897 }
898 if !errors.Is(errReceived, fillErr) {
899 t.Errorf("expected fillErr, got %v", errReceived)
900 }
901 if p.State() != PlayerStateError {
902 t.Errorf("expected Error state after Fill error, got %v", p.State())
903 }
904}
905
906// TestPlayerStartFailedConnectDestroysResources verifies that when
907// ConnectPlaybackStream fails after stream creation, the stream
908// and loop are destroyed and no stale pointers are left.
909func TestPlayerStartFailedConnectDestroysResources(t *testing.T) {
910 mockOps := mocks.NewMockStreamOps(t)
911 fakeLoop := opaqueTestPtr()
912 fakeStream := opaqueTestPtr()
913 cfg := defaultTestConfig()
914
915 connectErr := errors.New("connect failed")
916
917 mockOps.EXPECT().CreateMainLoop().Return(fakeLoop, nil)
918 mockOps.EXPECT().CreatePlaybackStream(fakeLoop, "purego-pipewire-player", mock.AnythingOfType("func()")).Return(fakeStream, nil)
919 mockOps.EXPECT().ConnectPlaybackStream(fakeStream, portout.PlaybackFormat{
920 SampleRate: cfg.SampleRate,
921 Channels: cfg.Channels,
922 FramesPerBuffer: cfg.FramesPerBuffer,
923 }).Return(connectErr)
924 // On connect failure, stream and loop must be destroyed
925 mockOps.EXPECT().DestroyStream(fakeStream).Return()
926 mockOps.EXPECT().DestroyMainLoop(fakeLoop).Return()
927
928 p := newPlayer(cfg, PlayerCallbacks{})
929 p.streamOps = mockOps
930
931 err := p.Start()
932 if err == nil {
933 t.Fatal("expected Start to fail when connect fails")
934 }
935 if !errors.Is(err, connectErr) {
936 t.Fatalf("expected connectErr, got %v", err)
937 }
938
939 // Player should be in Error state
940 if p.State() != PlayerStateError {
941 t.Errorf("expected Error state after failed connect, got %v", p.State())
942 }
943
944 // No stale pointers should remain
945 p.mu.Lock()
946 loopPtr := p.loopPtr
947 streamPtr := p.streamPtr
948 p.mu.Unlock()
949
950 if loopPtr != nil {
951 t.Error("expected loopPtr to be nil after failed start")
952 }
953 if streamPtr != nil {
954 t.Error("expected streamPtr to be nil after failed start")
955 }
956}
957
958// TestPlayerStartFailedActivateDestroysResources verifies that when
959// SetStreamActive fails after stream creation and connection, the
960// stream is disconnected (best-effort), destroyed, the loop is destroyed,
961// and no stale pointers are left.
962func TestPlayerStartFailedActivateDestroysResources(t *testing.T) {
963 mockOps := mocks.NewMockStreamOps(t)
964 fakeLoop := opaqueTestPtr()
965 fakeStream := opaqueTestPtr()
966 cfg := defaultTestConfig()
967
968 activateErr := errors.New("activate failed")
969
970 mockOps.EXPECT().CreateMainLoop().Return(fakeLoop, nil)
971 mockOps.EXPECT().CreatePlaybackStream(fakeLoop, "purego-pipewire-player", mock.AnythingOfType("func()")).Return(fakeStream, nil)
972 mockOps.EXPECT().ConnectPlaybackStream(fakeStream, portout.PlaybackFormat{
973 SampleRate: cfg.SampleRate,
974 Channels: cfg.Channels,
975 FramesPerBuffer: cfg.FramesPerBuffer,
976 }).Return(nil)
977 mockOps.EXPECT().SetStreamActive(fakeStream, true).Return(activateErr)
978 // On activate failure: disconnect (best-effort), destroy stream, destroy loop
979 mockOps.EXPECT().DisconnectStream(fakeStream).Return(nil)
980 mockOps.EXPECT().DestroyStream(fakeStream).Return()
981 mockOps.EXPECT().DestroyMainLoop(fakeLoop).Return()
982
983 p := newPlayer(cfg, PlayerCallbacks{})
984 p.streamOps = mockOps
985
986 err := p.Start()
987 if err == nil {
988 t.Fatal("expected Start to fail when activate fails")
989 }
990 if !errors.Is(err, activateErr) {
991 t.Fatalf("expected activateErr, got %v", err)
992 }
993
994 // Player should be in Error state
995 if p.State() != PlayerStateError {
996 t.Errorf("expected Error state after failed activate, got %v", p.State())
997 }
998
999 // No stale pointers should remain
1000 p.mu.Lock()
1001 loopPtr := p.loopPtr
1002 streamPtr := p.streamPtr
1003 p.mu.Unlock()
1004
1005 if loopPtr != nil {
1006 t.Error("expected loopPtr to be nil after failed start")
1007 }
1008 if streamPtr != nil {
1009 t.Error("expected streamPtr to be nil after failed start")
1010 }
1011}
1012
1013// TestPlayerStartFailedCreateStreamDestroysLoop verifies that when
1014// CreatePlaybackStream fails after main loop creation, the loop is
1015// destroyed and no stale pointers are left.
1016func TestPlayerStartFailedCreateStreamDestroysLoop(t *testing.T) {
1017 mockOps := mocks.NewMockStreamOps(t)
1018 fakeLoop := opaqueTestPtr()
1019 cfg := defaultTestConfig()
1020
1021 createErr := errors.New("stream creation failed")
1022
1023 mockOps.EXPECT().CreateMainLoop().Return(fakeLoop, nil)
1024 mockOps.EXPECT().CreatePlaybackStream(fakeLoop, "purego-pipewire-player", mock.AnythingOfType("func()")).Return(nil, createErr)
1025 // On stream creation failure, destroy the loop
1026 mockOps.EXPECT().DestroyMainLoop(fakeLoop).Return()
1027
1028 p := newPlayer(cfg, PlayerCallbacks{})
1029 p.streamOps = mockOps
1030
1031 err := p.Start()
1032 if err == nil {
1033 t.Fatal("expected Start to fail when stream creation fails")
1034 }
1035 if !errors.Is(err, createErr) {
1036 t.Fatalf("expected createErr, got %v", err)
1037 }
1038
1039 // Player should be in Error state
1040 if p.State() != PlayerStateError {
1041 t.Errorf("expected Error state after failed stream creation, got %v", p.State())
1042 }
1043
1044 // No stale pointers should remain
1045 p.mu.Lock()
1046 loopPtr := p.loopPtr
1047 streamPtr := p.streamPtr
1048 p.mu.Unlock()
1049
1050 if loopPtr != nil {
1051 t.Error("expected loopPtr to be nil after failed start")
1052 }
1053 if streamPtr != nil {
1054 t.Error("expected streamPtr to be nil after failed start")
1055 }
1056}
1057
1058// TestProcessCallbackUsesConfigFramesPerBuffer verifies that onProcess uses
1059// config.FramesPerBuffer even when pwBuffer.Requested is smaller.
1060func TestProcessCallbackUsesConfigFramesPerBuffer(t *testing.T) {
1061 mockOps := mocks.NewMockStreamOps(t)
1062 fakeLoop := opaqueTestPtr()
1063 fakeStream := opaqueTestPtr()
1064
1065 // Config requests 1024 frames per buffer, but PipeWire will request
1066 // fewer (941) due to rate conversion.
1067 const configFrames = 1024
1068 const requestedFrames = 941
1069 const channels = 2
1070
1071 cfg := PlayerConfig{
1072 SampleRate: 48000,
1073 Channels: channels,
1074 FramesPerBuffer: configFrames,
1075 }
1076
1077 // Build a real pwBuffer with backing storage for the full configFrames
1078 // so that a view up to configFrames would succeed — but set
1079 // pwBuf.Requested smaller to ensure onProcess still uses configFrames.
1080 channelData := make([][]float32, channels)
1081 floatPtrs := make([]unsafe.Pointer, channels)
1082 for ch := 0; ch < channels; ch++ {
1083 channelData[ch] = make([]float32, configFrames)
1084 floatPtrs[ch] = unsafe.Pointer(&channelData[ch][0])
1085 }
1086 chunkSlice := make([]spaChunk, channels)
1087 dataSlice := make([]spaData, channels)
1088 for ch := 0; ch < channels; ch++ {
1089 dataSlice[ch] = spaData{
1090 Type: spaDataTypeMemPtr,
1091 Flags: spaDataFlagRW,
1092 Maxsize: uint32(configFrames * 4),
1093 Data: floatPtrs[ch],
1094 Chunk: &chunkSlice[ch],
1095 }
1096 }
1097 spaBuf := spaBuffer{
1098 NDatas: uint32(channels),
1099 Datas: unsafe.Pointer(&dataSlice[0]),
1100 }
1101 pwBuf := pwBuffer{
1102 Buffer: &spaBuf,
1103 Requested: uint64(requestedFrames), // PipeWire asks for 941, not 1024
1104 }
1105 bufPtr := unsafe.Pointer(&pwBuf)
1106
1107 wg := expectStartWithSync(mockOps, fakeLoop, fakeStream, cfg)
1108 mockOps.EXPECT().DequeueBuffer(fakeStream).Return(bufPtr)
1109 mockOps.EXPECT().QueueBuffer(fakeStream, bufPtr).Return(nil)
1110
1111 var receivedFrames int
1112 p := newPlayer(cfg, PlayerCallbacks{
1113 Fill: func(buf *PCMBuffer) (int, error) {
1114 receivedFrames = buf.Frames
1115 // Write data into the buffer to verify propagation
1116 for ch := 0; ch < buf.Channels; ch++ {
1117 for f := 0; f < buf.Frames; f++ {
1118 buf.Samples[ch][f] = float32(f)
1119 }
1120 }
1121 return buf.Frames, nil
1122 },
1123 })
1124 p.streamOps = mockOps
1125
1126 if err := p.Start(); err != nil {
1127 t.Fatalf("Start failed: %v", err)
1128 }
1129
1130 p.onProcess()
1131
1132 // Fill must have been called with the configured frame count.
1133 if receivedFrames != configFrames {
1134 t.Errorf("Fill received %d frames, want %d (config.FramesPerBuffer)", receivedFrames, configFrames)
1135 }
1136
1137 // Commit must set chunk sizes and pwBuf.Size to reflect configFrames.
1138 for ch := 0; ch < channels; ch++ {
1139 wantChunkSize := uint32(configFrames * 4)
1140 if chunkSlice[ch].Size != wantChunkSize {
1141 t.Errorf("chunk[%d].Size = %d, want %d (configFrames*4)", ch, chunkSlice[ch].Size, wantChunkSize)
1142 }
1143 if chunkSlice[ch].Stride != 4 {
1144 t.Errorf("chunk[%d].Stride = %d, want 4", ch, chunkSlice[ch].Stride)
1145 }
1146 }
1147
1148 if pwBuf.Size != uint64(configFrames) {
1149 t.Errorf("pwBuf.Size = %d, want %d (configFrames)", pwBuf.Size, configFrames)
1150 }
1151
1152 _ = channelData // prevent escape
1153 waitOnMainLoop(t, wg)
1154}