detached f248e04

feat: wire real PipeWire player runtime

commit
f248e0464f2acd7accc91b3bae1f45261ab1f261
author
brice · · 2026-04-10 14:24 UTC
parent
715769b
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
4 files · +849 -25
Split
UnifiedSplit
M internal/core/player.go +103 -6
@@ -2,10 +2,12 @@ package core
22
33 import (
44 "errors"
5+ "fmt"
56 "sync"
67 "sync/atomic"
78 "unsafe"
89
10+ "github.com/bnema/purego-pipewire/internal/capi"
911 portout "github.com/bnema/purego-pipewire/internal/ports/out"
1012 )
1113
@@ -100,9 +102,8 @@ func isValidTransition(current, next PlayerState) bool {
100102 }
101103 }
102104
103-// Start begins playback, transitioning from Idle or Stopped to Playing
105+// Start begins playback, transitioning from Idle or Stopped to Playing.
104106 func (p *player) Start() error {
105- // Check if player is in terminal state
106107 current := p.State()
107108 if current == PlayerStateClosed || current == PlayerStateClosing {
108109 return ErrPlayerClosed
@@ -113,16 +114,38 @@ func (p *player) Start() error {
113114 return ErrInvalidPlayerState
114115 }
115116
116- // Ensure runtime is available (placeholder)
117+ // Ensure runtime is available
117118 if err := p.ensureRuntime(); err != nil {
118119 return err
119120 }
120121
121- // Transition to Starting then Playing
122+ // Transition to Starting
122123 if err := p.transition(PlayerStateStarting); err != nil {
123124 return err
124125 }
125126
127+ // Check if we already have PipeWire resources (restart after Stop)
128+ p.mu.Lock()
129+ hasResources := p.loopPtr != nil && p.streamPtr != nil
130+ p.mu.Unlock()
131+
132+ if !hasResources {
133+ // First start — create PipeWire resources
134+ if err := p.startCreateResourcesAndActivate(); err != nil {
135+ return err
136+ }
137+ } else {
138+ // Restart — just reactivate existing stream
139+ p.mu.Lock()
140+ streamPtr := p.streamPtr
141+ p.mu.Unlock()
142+
143+ if err := p.streamOps.SetStreamActive(streamPtr, true); err != nil {
144+ p.transition(PlayerStateError)
145+ return err
146+ }
147+ }
148+
126149 // Clear paused flag
127150 p.setPaused(false)
128151
@@ -130,6 +153,67 @@ func (p *player) Start() error {
130153 return p.transition(PlayerStatePlaying)
131154 }
132155
156+// startCreateResourcesAndActivate creates the main loop, playback stream, connects
157+// and activates it, stores owned resources, and starts the main loop goroutine.
158+// On failure, all created resources are destroyed and the state transitions to Error.
159+// No owned pointers are left set on failure.
160+func (p *player) startCreateResourcesAndActivate() error {
161+ // Create main loop
162+ loopPtr, err := p.streamOps.CreateMainLoop()
163+ if err != nil {
164+ p.transition(PlayerStateError)
165+ return err
166+ }
167+
168+ // Create playback stream
169+ streamPtr, err := p.streamOps.CreatePlaybackStream(loopPtr, "purego-pipewire-player", p.onProcess)
170+ if err != nil {
171+ p.streamOps.DestroyMainLoop(loopPtr)
172+ p.transition(PlayerStateError)
173+ return err
174+ }
175+
176+ // Connect stream
177+ format := portout.PlaybackFormat{
178+ SampleRate: p.config.SampleRate,
179+ Channels: p.config.Channels,
180+ FramesPerBuffer: p.config.FramesPerBuffer,
181+ }
182+ if err := p.streamOps.ConnectPlaybackStream(streamPtr, format); err != nil {
183+ p.streamOps.DestroyStream(streamPtr)
184+ p.streamOps.DestroyMainLoop(loopPtr)
185+ p.transition(PlayerStateError)
186+ return err
187+ }
188+
189+ // Activate stream
190+ if err := p.streamOps.SetStreamActive(streamPtr, true); err != nil {
191+ // Best-effort disconnect before destroying
192+ _ = p.streamOps.DisconnectStream(streamPtr)
193+ p.streamOps.DestroyStream(streamPtr)
194+ p.streamOps.DestroyMainLoop(loopPtr)
195+ p.transition(PlayerStateError)
196+ return err
197+ }
198+
199+ // All steps succeeded — store owned resources
200+ p.mu.Lock()
201+ p.loopPtr = loopPtr
202+ p.streamPtr = streamPtr
203+ p.mu.Unlock()
204+
205+ // Start main loop in an internal goroutine
206+ go func() {
207+ if err := p.streamOps.RunMainLoop(loopPtr); err != nil {
208+ if state := p.State(); state != PlayerStateClosing && state != PlayerStateClosed {
209+ p.fail(err)
210+ }
211+ }
212+ }()
213+
214+ return nil
215+}
216+
133217 // Pause temporarily pauses playback
134218 func (p *player) Pause() error {
135219 current := p.State()
@@ -193,9 +277,22 @@ func (p *player) Close() error {
193277 return p.transition(PlayerStateClosed)
194278 }
195279
196-// ensureRuntime ensures the runtime is available (placeholder)
280+// ensureRuntime ensures the player has a usable StreamOps implementation.
281+// If streamOps is already set (e.g., by tests), it is used as-is.
282+// Otherwise, PipeWire is registered via capi.Register() and the default
283+// StreamOps implementation is obtained.
197284 func (p *player) ensureRuntime() error {
198- // Minimal placeholder - will be implemented in future tasks
285+ if p.streamOps != nil {
286+ return nil
287+ }
288+ if err := capi.Register(); err != nil {
289+ return fmt.Errorf("register pipewire: %w", err)
290+ }
291+ ops := capi.DefaultStreamOps()
292+ if ops == nil {
293+ return errors.New("failed to obtain default stream operations")
294+ }
295+ p.streamOps = ops
199296 return nil
200297 }
201298
@@ -2,10 +2,12 @@ package core
2 2
3 import (3 import (
4 "errors"4 "errors"
5+ "fmt"
5 "sync"6 "sync"
6 "sync/atomic"7 "sync/atomic"
7 "unsafe"8 "unsafe"
8 9
10+ "github.com/bnema/purego-pipewire/internal/capi"
9 portout "github.com/bnema/purego-pipewire/internal/ports/out"11 portout "github.com/bnema/purego-pipewire/internal/ports/out"
10 )12 )
11 13
@@ -100,9 +102,8 @@ func isValidTransition(current, next PlayerState) bool {
100 }102 }
101 }103 }
102 104
103-// Start begins playback, transitioning from Idle or Stopped to Playing105+// Start begins playback, transitioning from Idle or Stopped to Playing.
104 func (p *player) Start() error {106 func (p *player) Start() error {
105- // Check if player is in terminal state
106 current := p.State()107 current := p.State()
107 if current == PlayerStateClosed || current == PlayerStateClosing {108 if current == PlayerStateClosed || current == PlayerStateClosing {
108 return ErrPlayerClosed109 return ErrPlayerClosed
@@ -113,16 +114,38 @@ func (p *player) Start() error {
113 return ErrInvalidPlayerState114 return ErrInvalidPlayerState
114 }115 }
115 116
116- // Ensure runtime is available (placeholder)117+ // Ensure runtime is available
117 if err := p.ensureRuntime(); err != nil {118 if err := p.ensureRuntime(); err != nil {
118 return err119 return err
119 }120 }
120 121
121- // Transition to Starting then Playing122+ // Transition to Starting
122 if err := p.transition(PlayerStateStarting); err != nil {123 if err := p.transition(PlayerStateStarting); err != nil {
123 return err124 return err
124 }125 }
125 126
127+ // Check if we already have PipeWire resources (restart after Stop)
128+ p.mu.Lock()
129+ hasResources := p.loopPtr != nil && p.streamPtr != nil
130+ p.mu.Unlock()
131+
132+ if !hasResources {
133+ // First start — create PipeWire resources
134+ if err := p.startCreateResourcesAndActivate(); err != nil {
135+ return err
136+ }
137+ } else {
138+ // Restart — just reactivate existing stream
139+ p.mu.Lock()
140+ streamPtr := p.streamPtr
141+ p.mu.Unlock()
142+
143+ if err := p.streamOps.SetStreamActive(streamPtr, true); err != nil {
144+ p.transition(PlayerStateError)
145+ return err
146+ }
147+ }
148+
126 // Clear paused flag149 // Clear paused flag
127 p.setPaused(false)150 p.setPaused(false)
128 151
@@ -130,6 +153,67 @@ func (p *player) Start() error {
130 return p.transition(PlayerStatePlaying)153 return p.transition(PlayerStatePlaying)
131 }154 }
132 155
156+// startCreateResourcesAndActivate creates the main loop, playback stream, connects
157+// and activates it, stores owned resources, and starts the main loop goroutine.
158+// On failure, all created resources are destroyed and the state transitions to Error.
159+// No owned pointers are left set on failure.
160+func (p *player) startCreateResourcesAndActivate() error {
161+ // Create main loop
162+ loopPtr, err := p.streamOps.CreateMainLoop()
163+ if err != nil {
164+ p.transition(PlayerStateError)
165+ return err
166+ }
167+
168+ // Create playback stream
169+ streamPtr, err := p.streamOps.CreatePlaybackStream(loopPtr, "purego-pipewire-player", p.onProcess)
170+ if err != nil {
171+ p.streamOps.DestroyMainLoop(loopPtr)
172+ p.transition(PlayerStateError)
173+ return err
174+ }
175+
176+ // Connect stream
177+ format := portout.PlaybackFormat{
178+ SampleRate: p.config.SampleRate,
179+ Channels: p.config.Channels,
180+ FramesPerBuffer: p.config.FramesPerBuffer,
181+ }
182+ if err := p.streamOps.ConnectPlaybackStream(streamPtr, format); err != nil {
183+ p.streamOps.DestroyStream(streamPtr)
184+ p.streamOps.DestroyMainLoop(loopPtr)
185+ p.transition(PlayerStateError)
186+ return err
187+ }
188+
189+ // Activate stream
190+ if err := p.streamOps.SetStreamActive(streamPtr, true); err != nil {
191+ // Best-effort disconnect before destroying
192+ _ = p.streamOps.DisconnectStream(streamPtr)
193+ p.streamOps.DestroyStream(streamPtr)
194+ p.streamOps.DestroyMainLoop(loopPtr)
195+ p.transition(PlayerStateError)
196+ return err
197+ }
198+
199+ // All steps succeeded — store owned resources
200+ p.mu.Lock()
201+ p.loopPtr = loopPtr
202+ p.streamPtr = streamPtr
203+ p.mu.Unlock()
204+
205+ // Start main loop in an internal goroutine
206+ go func() {
207+ if err := p.streamOps.RunMainLoop(loopPtr); err != nil {
208+ if state := p.State(); state != PlayerStateClosing && state != PlayerStateClosed {
209+ p.fail(err)
210+ }
211+ }
212+ }()
213+
214+ return nil
215+}
216+
133 // Pause temporarily pauses playback217 // Pause temporarily pauses playback
134 func (p *player) Pause() error {218 func (p *player) Pause() error {
135 current := p.State()219 current := p.State()
@@ -193,9 +277,22 @@ func (p *player) Close() error {
193 return p.transition(PlayerStateClosed)277 return p.transition(PlayerStateClosed)
194 }278 }
195 279
196-// ensureRuntime ensures the runtime is available (placeholder)280+// ensureRuntime ensures the player has a usable StreamOps implementation.
281+// If streamOps is already set (e.g., by tests), it is used as-is.
282+// Otherwise, PipeWire is registered via capi.Register() and the default
283+// StreamOps implementation is obtained.
197 func (p *player) ensureRuntime() error {284 func (p *player) ensureRuntime() error {
198- // Minimal placeholder - will be implemented in future tasks285+ if p.streamOps != nil {
286+ return nil
287+ }
288+ if err := capi.Register(); err != nil {
289+ return fmt.Errorf("register pipewire: %w", err)
290+ }
291+ ops := capi.DefaultStreamOps()
292+ if ops == nil {
293+ return errors.New("failed to obtain default stream operations")
294+ }
295+ p.streamOps = ops
199 return nil296 return nil
200 }297 }
201 298
M internal/core/player_rt.go +40 -0
@@ -2,6 +2,7 @@ package core
22
33 import (
44 "errors"
5+ "fmt"
56 "io"
67 )
78
@@ -153,3 +154,42 @@ func (p *player) fail(err error) {
153154 p.callbacks.OnError(err)
154155 }
155156 }
157+
158+// onProcess is the PipeWire process callback. It is invoked by PipeWire
159+// each time the stream needs more audio data. It dequeues a buffer,
160+// fills it with PCM data via processPCM, commits the result, and
161+// queues the buffer back. If any step fails, it routes the error
162+// through p.fail.
163+func (p *player) onProcess() {
164+ p.mu.Lock()
165+ ops := p.streamOps
166+ sp := p.streamPtr
167+ p.mu.Unlock()
168+
169+ if ops == nil || sp == nil {
170+ return
171+ }
172+
173+ bufPtr := ops.DequeueBuffer(sp)
174+ if bufPtr == nil {
175+ return
176+ }
177+
178+ view, err := newPWBufferView(bufPtr, p.config.Channels, p.config.FramesPerBuffer)
179+ if err != nil {
180+ p.fail(fmt.Errorf("buffer view: %w", err))
181+ return
182+ }
183+
184+ frames, err := p.processPCM(view.PCM())
185+ if err != nil {
186+ // processPCM error: do not queue the buffer; let existing
187+ // state/error behavior stand.
188+ return
189+ }
190+
191+ view.Commit(frames)
192+ if err := ops.QueueBuffer(sp, bufPtr); err != nil {
193+ p.fail(fmt.Errorf("queue buffer: %w", err))
194+ }
195+}
@@ -2,6 +2,7 @@ package core
2 2
3 import (3 import (
4 "errors"4 "errors"
5+ "fmt"
5 "io"6 "io"
6 )7 )
7 8
@@ -153,3 +154,42 @@ func (p *player) fail(err error) {
153 p.callbacks.OnError(err)154 p.callbacks.OnError(err)
154 }155 }
155 }156 }
157+
158+// onProcess is the PipeWire process callback. It is invoked by PipeWire
159+// each time the stream needs more audio data. It dequeues a buffer,
160+// fills it with PCM data via processPCM, commits the result, and
161+// queues the buffer back. If any step fails, it routes the error
162+// through p.fail.
163+func (p *player) onProcess() {
164+ p.mu.Lock()
165+ ops := p.streamOps
166+ sp := p.streamPtr
167+ p.mu.Unlock()
168+
169+ if ops == nil || sp == nil {
170+ return
171+ }
172+
173+ bufPtr := ops.DequeueBuffer(sp)
174+ if bufPtr == nil {
175+ return
176+ }
177+
178+ view, err := newPWBufferView(bufPtr, p.config.Channels, p.config.FramesPerBuffer)
179+ if err != nil {
180+ p.fail(fmt.Errorf("buffer view: %w", err))
181+ return
182+ }
183+
184+ frames, err := p.processPCM(view.PCM())
185+ if err != nil {
186+ // processPCM error: do not queue the buffer; let existing
187+ // state/error behavior stand.
188+ return
189+ }
190+
191+ view.Commit(frames)
192+ if err := ops.QueueBuffer(sp, bufPtr); err != nil {
193+ p.fail(fmt.Errorf("queue buffer: %w", err))
194+ }
195+}
M internal/core/player_rt_test.go +564 -0
@@ -4,6 +4,11 @@ import (
44 "errors"
55 "io"
66 "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"
712 )
813
914 // TestProcessCallbackSilenceFillUnderrun tests that when Fill returns fewer frames
@@ -485,3 +490,562 @@ func TestPCMBufferAllocate(t *testing.T) {
485490 }
486491 }
487492 }
493+
494+// --- Task 6: Start and process callback integration tests ---
495+
496+// TestPlayerStartCreatesLoopConnectsStreamAndRunsMainLoop verifies that Start()
497+// creates the main loop, creates a playback stream, connects it with the right
498+// format, activates it, stores owned pointers, and starts the main loop in a
499+// goroutine — then transitions to Playing.
500+func TestPlayerStartCreatesLoopConnectsStreamAndRunsMainLoop(t *testing.T) {
501+ mockOps := mocks.NewMockStreamOps(t)
502+ fakeLoop := unsafe.Pointer(uintptr(0xCAFE))
503+ fakeStream := unsafe.Pointer(uintptr(0xBEEF))
504+ cfg := PlayerConfig{
505+ SampleRate: 48000,
506+ Channels: 2,
507+ FramesPerBuffer: 256,
508+ }
509+
510+ wg := expectStartWithSync(mockOps, fakeLoop, fakeStream, cfg)
511+
512+ p := newPlayer(cfg, PlayerCallbacks{})
513+ p.streamOps = mockOps
514+
515+ if err := p.Start(); err != nil {
516+ t.Fatalf("Start failed: %v", err)
517+ }
518+
519+ if p.State() != PlayerStatePlaying {
520+ t.Fatalf("expected Playing state after Start, got %v", p.State())
521+ }
522+
523+ // Verify owned pointers are set
524+ p.mu.Lock()
525+ loopPtr := p.loopPtr
526+ streamPtr := p.streamPtr
527+ p.mu.Unlock()
528+
529+ if loopPtr != fakeLoop {
530+ t.Fatalf("expected loopPtr = %v, got %v", fakeLoop, loopPtr)
531+ }
532+ if streamPtr != fakeStream {
533+ t.Fatalf("expected streamPtr = %v, got %v", fakeStream, streamPtr)
534+ }
535+
536+ // Wait for the RunMainLoop goroutine to finish
537+ waitOnMainLoop(t, wg)
538+}
539+
540+// TestProcessCallbackDequeuesProcessesAndQueues verifies the full onProcess
541+// path: dequeue buffer, build pwBufferView, call processPCM, commit, and
542+// queue the buffer back.
543+func TestProcessCallbackDequeuesProcessesAndQueues(t *testing.T) {
544+ mockOps := mocks.NewMockStreamOps(t)
545+ fakeLoop := unsafe.Pointer(uintptr(0xCAFE))
546+ fakeStream := unsafe.Pointer(uintptr(0xBEEF))
547+ cfg := PlayerConfig{
548+ SampleRate: 48000,
549+ Channels: 2,
550+ FramesPerBuffer: 64,
551+ }
552+
553+ // Build a real pwBuffer that onProcess will dequeue
554+ const channels = 2
555+ const frames = 64
556+ channelData := make([][]float32, channels)
557+ chunkSlice := make([]spaChunk, channels)
558+ dataSlice := make([]spaData, channels)
559+ floatPtrs := make([]unsafe.Pointer, channels)
560+ for ch := 0; ch < channels; ch++ {
561+ channelData[ch] = make([]float32, frames)
562+ floatPtrs[ch] = unsafe.Pointer(&channelData[ch][0])
563+ dataSlice[ch] = spaData{
564+ Type: spaDataTypeMemPtr,
565+ Flags: spaDataFlagRW,
566+ Maxsize: uint32(frames * 4),
567+ Data: floatPtrs[ch],
568+ Chunk: &chunkSlice[ch],
569+ }
570+ }
571+ spaBuf := spaBuffer{
572+ NDatas: uint32(channels),
573+ Datas: unsafe.Pointer(&dataSlice[0]),
574+ }
575+ pwBuf := pwBuffer{
576+ Buffer: &spaBuf,
577+ }
578+ bufPtr := unsafe.Pointer(&pwBuf)
579+
580+ wg := expectStartWithSync(mockOps, fakeLoop, fakeStream, cfg)
581+
582+ // Expect DequeueBuffer, then QueueBuffer after processing
583+ mockOps.EXPECT().DequeueBuffer(fakeStream).Return(bufPtr)
584+ mockOps.EXPECT().QueueBuffer(fakeStream, bufPtr).Return(nil)
585+
586+ fillCalled := false
587+ p := newPlayer(cfg, PlayerCallbacks{
588+ Fill: func(buf *PCMBuffer) (int, error) {
589+ fillCalled = true
590+ // Write data to verify it propagates
591+ for ch := 0; ch < buf.Channels; ch++ {
592+ for f := 0; f < buf.Frames; f++ {
593+ buf.Samples[ch][f] = float32(f)
594+ }
595+ }
596+ return buf.Frames, nil
597+ },
598+ })
599+ p.streamOps = mockOps
600+
601+ if err := p.Start(); err != nil {
602+ t.Fatalf("Start failed: %v", err)
603+ }
604+
605+ if p.State() != PlayerStatePlaying {
606+ t.Fatalf("expected Playing state, got %v", p.State())
607+ }
608+
609+ // Call onProcess directly (simulating PipeWire calling it)
610+ p.onProcess()
611+
612+ if !fillCalled {
613+ t.Error("Fill callback was not called during onProcess")
614+ }
615+
616+ // Verify chunk sizes were updated by Commit
617+ for ch := 0; ch < channels; ch++ {
618+ if chunkSlice[ch].Size != uint32(frames*4) {
619+ t.Errorf("chunk[%d].Size = %d, want %d", ch, chunkSlice[ch].Size, frames*4)
620+ }
621+ if chunkSlice[ch].Stride != 4 {
622+ t.Errorf("chunk[%d].Stride = %d, want 4", ch, chunkSlice[ch].Stride)
623+ }
624+ }
625+
626+ // Verify pwBuffer.Size reflects frame count
627+ if pwBuf.Size != uint64(frames) {
628+ t.Errorf("pwBuf.Size = %d, want %d", pwBuf.Size, frames)
629+ }
630+
631+ waitOnMainLoop(t, wg)
632+}
633+
634+// TestProcessCallbackNilBufferReturnsEarly verifies that onProcess returns
635+// early when DequeueBuffer returns nil.
636+func TestProcessCallbackNilBufferReturnsEarly(t *testing.T) {
637+ mockOps := mocks.NewMockStreamOps(t)
638+ fakeLoop := unsafe.Pointer(uintptr(0xCAFE))
639+ fakeStream := unsafe.Pointer(uintptr(0xBEEF))
640+ cfg := defaultTestConfig()
641+
642+ wg := expectStartWithSync(mockOps, fakeLoop, fakeStream, cfg)
643+
644+ fillCalled := false
645+ p := newPlayer(cfg, PlayerCallbacks{
646+ Fill: func(buf *PCMBuffer) (int, error) {
647+ fillCalled = true
648+ return buf.Frames, nil
649+ },
650+ })
651+ p.streamOps = mockOps
652+
653+ if err := p.Start(); err != nil {
654+ t.Fatalf("Start failed: %v", err)
655+ }
656+ waitOnMainLoop(t, wg)
657+
658+ // DequeueBuffer returns nil — onProcess should return early
659+ mockOps.EXPECT().DequeueBuffer(fakeStream).Return(nil)
660+
661+ p.onProcess()
662+
663+ if fillCalled {
664+ t.Error("Fill should not be called when DequeueBuffer returns nil")
665+ }
666+}
667+
668+// TestProcessCallbackQueueBufferFailsRoutesThroughFail verifies that when
669+// QueueBuffer returns an error, onProcess routes it through p.fail().
670+func TestProcessCallbackQueueBufferFailsRoutesThroughFail(t *testing.T) {
671+ mockOps := mocks.NewMockStreamOps(t)
672+ fakeLoop := unsafe.Pointer(uintptr(0xCAFE))
673+ fakeStream := unsafe.Pointer(uintptr(0xBEEF))
674+ cfg := PlayerConfig{
675+ SampleRate: 48000,
676+ Channels: 2,
677+ FramesPerBuffer: 64,
678+ }
679+
680+ // Build a real pwBuffer
681+ const channels = 2
682+ const frames = 64
683+ channelData := make([][]float32, channels)
684+ floatPtrs := make([]unsafe.Pointer, channels)
685+ for ch := 0; ch < channels; ch++ {
686+ channelData[ch] = make([]float32, frames)
687+ floatPtrs[ch] = unsafe.Pointer(&channelData[ch][0])
688+ }
689+ chunkSlice := make([]spaChunk, channels)
690+ dataSlice := make([]spaData, channels)
691+ for ch := 0; ch < channels; ch++ {
692+ dataSlice[ch] = spaData{
693+ Type: spaDataTypeMemPtr,
694+ Flags: spaDataFlagRW,
695+ Maxsize: uint32(frames * 4),
696+ Data: floatPtrs[ch],
697+ Chunk: &chunkSlice[ch],
698+ }
699+ }
700+ spaBuf := spaBuffer{
701+ NDatas: uint32(channels),
702+ Datas: unsafe.Pointer(&dataSlice[0]),
703+ }
704+ pwBuf := pwBuffer{
705+ Buffer: &spaBuf,
706+ }
707+ bufPtr := unsafe.Pointer(&pwBuf)
708+
709+ wg := expectStartWithSync(mockOps, fakeLoop, fakeStream, cfg)
710+
711+ queueErr := errors.New("queue buffer failed")
712+ errCalled := false
713+ var errReceived error
714+
715+ p := newPlayer(cfg, PlayerCallbacks{
716+ Fill: func(buf *PCMBuffer) (int, error) {
717+ return buf.Frames, nil
718+ },
719+ OnError: func(err error) {
720+ errCalled = true
721+ errReceived = err
722+ },
723+ })
724+ p.streamOps = mockOps
725+
726+ if err := p.Start(); err != nil {
727+ t.Fatalf("Start failed: %v", err)
728+ }
729+ waitOnMainLoop(t, wg)
730+
731+ // Expect DequeueBuffer to return a buffer, then QueueBuffer to fail
732+ mockOps.EXPECT().DequeueBuffer(fakeStream).Return(bufPtr)
733+ mockOps.EXPECT().QueueBuffer(fakeStream, bufPtr).Return(queueErr)
734+
735+ p.onProcess()
736+
737+ if !errCalled {
738+ t.Error("OnError should have been called when QueueBuffer fails")
739+ }
740+ if errReceived == nil {
741+ t.Error("expected non-nil error from OnError")
742+ }
743+ if p.State() != PlayerStateError {
744+ t.Errorf("expected Error state after QueueBuffer failure, got %v", p.State())
745+ }
746+}
747+
748+// TestProcessCallbackBufferViewFailsRoutesThroughFail verifies that when
749+// newPWBufferView fails, onProcess routes the error through p.fail()
750+// and does not attempt to queue the buffer.
751+func TestProcessCallbackBufferViewFailsRoutesThroughFail(t *testing.T) {
752+ mockOps := mocks.NewMockStreamOps(t)
753+ fakeLoop := unsafe.Pointer(uintptr(0xCAFE))
754+ fakeStream := unsafe.Pointer(uintptr(0xBEEF))
755+ cfg := PlayerConfig{
756+ SampleRate: 48000,
757+ Channels: 2,
758+ FramesPerBuffer: 64,
759+ }
760+
761+ // Build a buffer with NDatas mismatch to trigger ErrChannelMismatch
762+ // NDatas = 1 but config.Channels = 2
763+ singleData := spaData{
764+ Type: spaDataTypeMemPtr,
765+ Maxsize: 256,
766+ Data: unsafe.Pointer(new(float32)),
767+ Chunk: &spaChunk{},
768+ }
769+ spaBuf := spaBuffer{
770+ NDatas: 1, // mismatch with Channels=2
771+ Datas: unsafe.Pointer(&singleData),
772+ }
773+ pwBuf := pwBuffer{
774+ Buffer: &spaBuf,
775+ }
776+ bufPtr := unsafe.Pointer(&pwBuf)
777+
778+ wg := expectStartWithSync(mockOps, fakeLoop, fakeStream, cfg)
779+
780+ errCalled := false
781+ var errReceived error
782+
783+ p := newPlayer(cfg, PlayerCallbacks{
784+ Fill: func(buf *PCMBuffer) (int, error) {
785+ return buf.Frames, nil
786+ },
787+ OnError: func(err error) {
788+ errCalled = true
789+ errReceived = err
790+ },
791+ })
792+ p.streamOps = mockOps
793+
794+ if err := p.Start(); err != nil {
795+ t.Fatalf("Start failed: %v", err)
796+ }
797+ waitOnMainLoop(t, wg)
798+
799+ // DequeueBuffer returns a buffer that will fail newPWBufferView
800+ // QueueBuffer should NOT be called
801+ mockOps.EXPECT().DequeueBuffer(fakeStream).Return(bufPtr)
802+
803+ p.onProcess()
804+
805+ if !errCalled {
806+ t.Error("OnError should have been called when buffer view fails")
807+ }
808+ if errReceived == nil {
809+ t.Error("expected non-nil error from OnError")
810+ }
811+ if !errors.Is(errReceived, ErrChannelMismatch) {
812+ t.Errorf("expected ErrChannelMismatch, got %v", errReceived)
813+ }
814+ if p.State() != PlayerStateError {
815+ t.Errorf("expected Error state after buffer view failure, got %v", p.State())
816+ }
817+}
818+
819+// TestProcessCallbackProcessCMErrorSkipsQueue verifies that when processPCM
820+// returns an error, onProcess does not queue the buffer and lets the existing
821+// error behavior stand.
822+func TestProcessCallbackProcessCMErrorSkipsQueue(t *testing.T) {
823+ mockOps := mocks.NewMockStreamOps(t)
824+ fakeLoop := unsafe.Pointer(uintptr(0xCAFE))
825+ fakeStream := unsafe.Pointer(uintptr(0xBEEF))
826+ cfg := PlayerConfig{
827+ SampleRate: 48000,
828+ Channels: 2,
829+ FramesPerBuffer: 64,
830+ }
831+
832+ // Build a real pwBuffer
833+ const channels = 2
834+ const frames = 64
835+ channelData := make([][]float32, channels)
836+ floatPtrs := make([]unsafe.Pointer, channels)
837+ for ch := 0; ch < channels; ch++ {
838+ channelData[ch] = make([]float32, frames)
839+ floatPtrs[ch] = unsafe.Pointer(&channelData[ch][0])
840+ }
841+ chunkSlice := make([]spaChunk, channels)
842+ dataSlice := make([]spaData, channels)
843+ for ch := 0; ch < channels; ch++ {
844+ dataSlice[ch] = spaData{
845+ Type: spaDataTypeMemPtr,
846+ Flags: spaDataFlagRW,
847+ Maxsize: uint32(frames * 4),
848+ Data: floatPtrs[ch],
849+ Chunk: &chunkSlice[ch],
850+ }
851+ }
852+ spaBuf := spaBuffer{
853+ NDatas: uint32(channels),
854+ Datas: unsafe.Pointer(&dataSlice[0]),
855+ }
856+ pwBuf := pwBuffer{
857+ Buffer: &spaBuf,
858+ }
859+ bufPtr := unsafe.Pointer(&pwBuf)
860+
861+ wg := expectStartWithSync(mockOps, fakeLoop, fakeStream, cfg)
862+
863+ fillErr := errors.New("fill error")
864+ errCalled := false
865+ var errReceived error
866+
867+ p := newPlayer(cfg, PlayerCallbacks{
868+ Fill: func(buf *PCMBuffer) (int, error) {
869+ return 0, fillErr
870+ },
871+ OnError: func(err error) {
872+ errCalled = true
873+ errReceived = err
874+ },
875+ })
876+ p.streamOps = mockOps
877+
878+ if err := p.Start(); err != nil {
879+ t.Fatalf("Start failed: %v", err)
880+ }
881+ waitOnMainLoop(t, wg)
882+
883+ // DequeueBuffer returns a buffer; QueueBuffer should NOT be called
884+ // because processPCM (via Fill) returns an error
885+ mockOps.EXPECT().DequeueBuffer(fakeStream).Return(bufPtr)
886+ // Note: no QueueBuffer expectation
887+
888+ p.onProcess()
889+
890+ if !errCalled {
891+ t.Error("OnError should have been called when Fill returns an error")
892+ }
893+ if !errors.Is(errReceived, fillErr) {
894+ t.Errorf("expected fillErr, got %v", errReceived)
895+ }
896+ if p.State() != PlayerStateError {
897+ t.Errorf("expected Error state after Fill error, got %v", p.State())
898+ }
899+}
900+
901+// TestPlayerStartFailedConnectDestroysResources verifies that when
902+// ConnectPlaybackStream fails after stream creation, the stream
903+// and loop are destroyed and no stale pointers are left.
904+func TestPlayerStartFailedConnectDestroysResources(t *testing.T) {
905+ mockOps := mocks.NewMockStreamOps(t)
906+ fakeLoop := unsafe.Pointer(uintptr(0xCAFE))
907+ fakeStream := unsafe.Pointer(uintptr(0xBEEF))
908+ cfg := defaultTestConfig()
909+
910+ connectErr := errors.New("connect failed")
911+
912+ mockOps.EXPECT().CreateMainLoop().Return(fakeLoop, nil)
913+ mockOps.EXPECT().CreatePlaybackStream(fakeLoop, "purego-pipewire-player", mock.AnythingOfType("func()")).Return(fakeStream, nil)
914+ mockOps.EXPECT().ConnectPlaybackStream(fakeStream, portout.PlaybackFormat{
915+ SampleRate: cfg.SampleRate,
916+ Channels: cfg.Channels,
917+ FramesPerBuffer: cfg.FramesPerBuffer,
918+ }).Return(connectErr)
919+ // On connect failure, stream and loop must be destroyed
920+ mockOps.EXPECT().DestroyStream(fakeStream).Return()
921+ mockOps.EXPECT().DestroyMainLoop(fakeLoop).Return()
922+
923+ p := newPlayer(cfg, PlayerCallbacks{})
924+ p.streamOps = mockOps
925+
926+ err := p.Start()
927+ if err == nil {
928+ t.Fatal("expected Start to fail when connect fails")
929+ }
930+ if !errors.Is(err, connectErr) {
931+ t.Fatalf("expected connectErr, got %v", err)
932+ }
933+
934+ // Player should be in Error state
935+ if p.State() != PlayerStateError {
936+ t.Errorf("expected Error state after failed connect, got %v", p.State())
937+ }
938+
939+ // No stale pointers should remain
940+ p.mu.Lock()
941+ loopPtr := p.loopPtr
942+ streamPtr := p.streamPtr
943+ p.mu.Unlock()
944+
945+ if loopPtr != nil {
946+ t.Error("expected loopPtr to be nil after failed start")
947+ }
948+ if streamPtr != nil {
949+ t.Error("expected streamPtr to be nil after failed start")
950+ }
951+}
952+
953+// TestPlayerStartFailedActivateDestroysResources verifies that when
954+// SetStreamActive fails after stream creation and connection, the
955+// stream is disconnected (best-effort), destroyed, the loop is destroyed,
956+// and no stale pointers are left.
957+func TestPlayerStartFailedActivateDestroysResources(t *testing.T) {
958+ mockOps := mocks.NewMockStreamOps(t)
959+ fakeLoop := unsafe.Pointer(uintptr(0xCAFE))
960+ fakeStream := unsafe.Pointer(uintptr(0xBEEF))
961+ cfg := defaultTestConfig()
962+
963+ activateErr := errors.New("activate failed")
964+
965+ mockOps.EXPECT().CreateMainLoop().Return(fakeLoop, nil)
966+ mockOps.EXPECT().CreatePlaybackStream(fakeLoop, "purego-pipewire-player", mock.AnythingOfType("func()")).Return(fakeStream, nil)
967+ mockOps.EXPECT().ConnectPlaybackStream(fakeStream, portout.PlaybackFormat{
968+ SampleRate: cfg.SampleRate,
969+ Channels: cfg.Channels,
970+ FramesPerBuffer: cfg.FramesPerBuffer,
971+ }).Return(nil)
972+ mockOps.EXPECT().SetStreamActive(fakeStream, true).Return(activateErr)
973+ // On activate failure: disconnect (best-effort), destroy stream, destroy loop
974+ mockOps.EXPECT().DisconnectStream(fakeStream).Return(nil)
975+ mockOps.EXPECT().DestroyStream(fakeStream).Return()
976+ mockOps.EXPECT().DestroyMainLoop(fakeLoop).Return()
977+
978+ p := newPlayer(cfg, PlayerCallbacks{})
979+ p.streamOps = mockOps
980+
981+ err := p.Start()
982+ if err == nil {
983+ t.Fatal("expected Start to fail when activate fails")
984+ }
985+ if !errors.Is(err, activateErr) {
986+ t.Fatalf("expected activateErr, got %v", err)
987+ }
988+
989+ // Player should be in Error state
990+ if p.State() != PlayerStateError {
991+ t.Errorf("expected Error state after failed activate, got %v", p.State())
992+ }
993+
994+ // No stale pointers should remain
995+ p.mu.Lock()
996+ loopPtr := p.loopPtr
997+ streamPtr := p.streamPtr
998+ p.mu.Unlock()
999+
1000+ if loopPtr != nil {
1001+ t.Error("expected loopPtr to be nil after failed start")
1002+ }
1003+ if streamPtr != nil {
1004+ t.Error("expected streamPtr to be nil after failed start")
1005+ }
1006+}
1007+
1008+// TestPlayerStartFailedCreateStreamDestroysLoop verifies that when
1009+// CreatePlaybackStream fails after main loop creation, the loop is
1010+// destroyed and no stale pointers are left.
1011+func TestPlayerStartFailedCreateStreamDestroysLoop(t *testing.T) {
1012+ mockOps := mocks.NewMockStreamOps(t)
1013+ fakeLoop := unsafe.Pointer(uintptr(0xCAFE))
1014+ cfg := defaultTestConfig()
1015+
1016+ createErr := errors.New("stream creation failed")
1017+
1018+ mockOps.EXPECT().CreateMainLoop().Return(fakeLoop, nil)
1019+ mockOps.EXPECT().CreatePlaybackStream(fakeLoop, "purego-pipewire-player", mock.AnythingOfType("func()")).Return(nil, createErr)
1020+ // On stream creation failure, destroy the loop
1021+ mockOps.EXPECT().DestroyMainLoop(fakeLoop).Return()
1022+
1023+ p := newPlayer(cfg, PlayerCallbacks{})
1024+ p.streamOps = mockOps
1025+
1026+ err := p.Start()
1027+ if err == nil {
1028+ t.Fatal("expected Start to fail when stream creation fails")
1029+ }
1030+ if !errors.Is(err, createErr) {
1031+ t.Fatalf("expected createErr, got %v", err)
1032+ }
1033+
1034+ // Player should be in Error state
1035+ if p.State() != PlayerStateError {
1036+ t.Errorf("expected Error state after failed stream creation, got %v", p.State())
1037+ }
1038+
1039+ // No stale pointers should remain
1040+ p.mu.Lock()
1041+ loopPtr := p.loopPtr
1042+ streamPtr := p.streamPtr
1043+ p.mu.Unlock()
1044+
1045+ if loopPtr != nil {
1046+ t.Error("expected loopPtr to be nil after failed start")
1047+ }
1048+ if streamPtr != nil {
1049+ t.Error("expected streamPtr to be nil after failed start")
1050+ }
1051+}
@@ -4,6 +4,11 @@ import (
4 "errors"4 "errors"
5 "io"5 "io"
6 "testing"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"
7 )12 )
8 13
9 // TestProcessCallbackSilenceFillUnderrun tests that when Fill returns fewer frames14 // TestProcessCallbackSilenceFillUnderrun tests that when Fill returns fewer frames
@@ -485,3 +490,562 @@ func TestPCMBufferAllocate(t *testing.T) {
485 }490 }
486 }491 }
487 }492 }
493+
494+// --- Task 6: Start and process callback integration tests ---
495+
496+// TestPlayerStartCreatesLoopConnectsStreamAndRunsMainLoop verifies that Start()
497+// creates the main loop, creates a playback stream, connects it with the right
498+// format, activates it, stores owned pointers, and starts the main loop in a
499+// goroutine — then transitions to Playing.
500+func TestPlayerStartCreatesLoopConnectsStreamAndRunsMainLoop(t *testing.T) {
501+ mockOps := mocks.NewMockStreamOps(t)
502+ fakeLoop := unsafe.Pointer(uintptr(0xCAFE))
503+ fakeStream := unsafe.Pointer(uintptr(0xBEEF))
504+ cfg := PlayerConfig{
505+ SampleRate: 48000,
506+ Channels: 2,
507+ FramesPerBuffer: 256,
508+ }
509+
510+ wg := expectStartWithSync(mockOps, fakeLoop, fakeStream, cfg)
511+
512+ p := newPlayer(cfg, PlayerCallbacks{})
513+ p.streamOps = mockOps
514+
515+ if err := p.Start(); err != nil {
516+ t.Fatalf("Start failed: %v", err)
517+ }
518+
519+ if p.State() != PlayerStatePlaying {
520+ t.Fatalf("expected Playing state after Start, got %v", p.State())
521+ }
522+
523+ // Verify owned pointers are set
524+ p.mu.Lock()
525+ loopPtr := p.loopPtr
526+ streamPtr := p.streamPtr
527+ p.mu.Unlock()
528+
529+ if loopPtr != fakeLoop {
530+ t.Fatalf("expected loopPtr = %v, got %v", fakeLoop, loopPtr)
531+ }
532+ if streamPtr != fakeStream {
533+ t.Fatalf("expected streamPtr = %v, got %v", fakeStream, streamPtr)
534+ }
535+
536+ // Wait for the RunMainLoop goroutine to finish
537+ waitOnMainLoop(t, wg)
538+}
539+
540+// TestProcessCallbackDequeuesProcessesAndQueues verifies the full onProcess
541+// path: dequeue buffer, build pwBufferView, call processPCM, commit, and
542+// queue the buffer back.
543+func TestProcessCallbackDequeuesProcessesAndQueues(t *testing.T) {
544+ mockOps := mocks.NewMockStreamOps(t)
545+ fakeLoop := unsafe.Pointer(uintptr(0xCAFE))
546+ fakeStream := unsafe.Pointer(uintptr(0xBEEF))
547+ cfg := PlayerConfig{
548+ SampleRate: 48000,
549+ Channels: 2,
550+ FramesPerBuffer: 64,
551+ }
552+
553+ // Build a real pwBuffer that onProcess will dequeue
554+ const channels = 2
555+ const frames = 64
556+ channelData := make([][]float32, channels)
557+ chunkSlice := make([]spaChunk, channels)
558+ dataSlice := make([]spaData, channels)
559+ floatPtrs := make([]unsafe.Pointer, channels)
560+ for ch := 0; ch < channels; ch++ {
561+ channelData[ch] = make([]float32, frames)
562+ floatPtrs[ch] = unsafe.Pointer(&channelData[ch][0])
563+ dataSlice[ch] = spaData{
564+ Type: spaDataTypeMemPtr,
565+ Flags: spaDataFlagRW,
566+ Maxsize: uint32(frames * 4),
567+ Data: floatPtrs[ch],
568+ Chunk: &chunkSlice[ch],
569+ }
570+ }
571+ spaBuf := spaBuffer{
572+ NDatas: uint32(channels),
573+ Datas: unsafe.Pointer(&dataSlice[0]),
574+ }
575+ pwBuf := pwBuffer{
576+ Buffer: &spaBuf,
577+ }
578+ bufPtr := unsafe.Pointer(&pwBuf)
579+
580+ wg := expectStartWithSync(mockOps, fakeLoop, fakeStream, cfg)
581+
582+ // Expect DequeueBuffer, then QueueBuffer after processing
583+ mockOps.EXPECT().DequeueBuffer(fakeStream).Return(bufPtr)
584+ mockOps.EXPECT().QueueBuffer(fakeStream, bufPtr).Return(nil)
585+
586+ fillCalled := false
587+ p := newPlayer(cfg, PlayerCallbacks{
588+ Fill: func(buf *PCMBuffer) (int, error) {
589+ fillCalled = true
590+ // Write data to verify it propagates
591+ for ch := 0; ch < buf.Channels; ch++ {
592+ for f := 0; f < buf.Frames; f++ {
593+ buf.Samples[ch][f] = float32(f)
594+ }
595+ }
596+ return buf.Frames, nil
597+ },
598+ })
599+ p.streamOps = mockOps
600+
601+ if err := p.Start(); err != nil {
602+ t.Fatalf("Start failed: %v", err)
603+ }
604+
605+ if p.State() != PlayerStatePlaying {
606+ t.Fatalf("expected Playing state, got %v", p.State())
607+ }
608+
609+ // Call onProcess directly (simulating PipeWire calling it)
610+ p.onProcess()
611+
612+ if !fillCalled {
613+ t.Error("Fill callback was not called during onProcess")
614+ }
615+
616+ // Verify chunk sizes were updated by Commit
617+ for ch := 0; ch < channels; ch++ {
618+ if chunkSlice[ch].Size != uint32(frames*4) {
619+ t.Errorf("chunk[%d].Size = %d, want %d", ch, chunkSlice[ch].Size, frames*4)
620+ }
621+ if chunkSlice[ch].Stride != 4 {
622+ t.Errorf("chunk[%d].Stride = %d, want 4", ch, chunkSlice[ch].Stride)
623+ }
624+ }
625+
626+ // Verify pwBuffer.Size reflects frame count
627+ if pwBuf.Size != uint64(frames) {
628+ t.Errorf("pwBuf.Size = %d, want %d", pwBuf.Size, frames)
629+ }
630+
631+ waitOnMainLoop(t, wg)
632+}
633+
634+// TestProcessCallbackNilBufferReturnsEarly verifies that onProcess returns
635+// early when DequeueBuffer returns nil.
636+func TestProcessCallbackNilBufferReturnsEarly(t *testing.T) {
637+ mockOps := mocks.NewMockStreamOps(t)
638+ fakeLoop := unsafe.Pointer(uintptr(0xCAFE))
639+ fakeStream := unsafe.Pointer(uintptr(0xBEEF))
640+ cfg := defaultTestConfig()
641+
642+ wg := expectStartWithSync(mockOps, fakeLoop, fakeStream, cfg)
643+
644+ fillCalled := false
645+ p := newPlayer(cfg, PlayerCallbacks{
646+ Fill: func(buf *PCMBuffer) (int, error) {
647+ fillCalled = true
648+ return buf.Frames, nil
649+ },
650+ })
651+ p.streamOps = mockOps
652+
653+ if err := p.Start(); err != nil {
654+ t.Fatalf("Start failed: %v", err)
655+ }
656+ waitOnMainLoop(t, wg)
657+
658+ // DequeueBuffer returns nil — onProcess should return early
659+ mockOps.EXPECT().DequeueBuffer(fakeStream).Return(nil)
660+
661+ p.onProcess()
662+
663+ if fillCalled {
664+ t.Error("Fill should not be called when DequeueBuffer returns nil")
665+ }
666+}
667+
668+// TestProcessCallbackQueueBufferFailsRoutesThroughFail verifies that when
669+// QueueBuffer returns an error, onProcess routes it through p.fail().
670+func TestProcessCallbackQueueBufferFailsRoutesThroughFail(t *testing.T) {
671+ mockOps := mocks.NewMockStreamOps(t)
672+ fakeLoop := unsafe.Pointer(uintptr(0xCAFE))
673+ fakeStream := unsafe.Pointer(uintptr(0xBEEF))
674+ cfg := PlayerConfig{
675+ SampleRate: 48000,
676+ Channels: 2,
677+ FramesPerBuffer: 64,
678+ }
679+
680+ // Build a real pwBuffer
681+ const channels = 2
682+ const frames = 64
683+ channelData := make([][]float32, channels)
684+ floatPtrs := make([]unsafe.Pointer, channels)
685+ for ch := 0; ch < channels; ch++ {
686+ channelData[ch] = make([]float32, frames)
687+ floatPtrs[ch] = unsafe.Pointer(&channelData[ch][0])
688+ }
689+ chunkSlice := make([]spaChunk, channels)
690+ dataSlice := make([]spaData, channels)
691+ for ch := 0; ch < channels; ch++ {
692+ dataSlice[ch] = spaData{
693+ Type: spaDataTypeMemPtr,
694+ Flags: spaDataFlagRW,
695+ Maxsize: uint32(frames * 4),
696+ Data: floatPtrs[ch],
697+ Chunk: &chunkSlice[ch],
698+ }
699+ }
700+ spaBuf := spaBuffer{
701+ NDatas: uint32(channels),
702+ Datas: unsafe.Pointer(&dataSlice[0]),
703+ }
704+ pwBuf := pwBuffer{
705+ Buffer: &spaBuf,
706+ }
707+ bufPtr := unsafe.Pointer(&pwBuf)
708+
709+ wg := expectStartWithSync(mockOps, fakeLoop, fakeStream, cfg)
710+
711+ queueErr := errors.New("queue buffer failed")
712+ errCalled := false
713+ var errReceived error
714+
715+ p := newPlayer(cfg, PlayerCallbacks{
716+ Fill: func(buf *PCMBuffer) (int, error) {
717+ return buf.Frames, nil
718+ },
719+ OnError: func(err error) {
720+ errCalled = true
721+ errReceived = err
722+ },
723+ })
724+ p.streamOps = mockOps
725+
726+ if err := p.Start(); err != nil {
727+ t.Fatalf("Start failed: %v", err)
728+ }
729+ waitOnMainLoop(t, wg)
730+
731+ // Expect DequeueBuffer to return a buffer, then QueueBuffer to fail
732+ mockOps.EXPECT().DequeueBuffer(fakeStream).Return(bufPtr)
733+ mockOps.EXPECT().QueueBuffer(fakeStream, bufPtr).Return(queueErr)
734+
735+ p.onProcess()
736+
737+ if !errCalled {
738+ t.Error("OnError should have been called when QueueBuffer fails")
739+ }
740+ if errReceived == nil {
741+ t.Error("expected non-nil error from OnError")
742+ }
743+ if p.State() != PlayerStateError {
744+ t.Errorf("expected Error state after QueueBuffer failure, got %v", p.State())
745+ }
746+}
747+
748+// TestProcessCallbackBufferViewFailsRoutesThroughFail verifies that when
749+// newPWBufferView fails, onProcess routes the error through p.fail()
750+// and does not attempt to queue the buffer.
751+func TestProcessCallbackBufferViewFailsRoutesThroughFail(t *testing.T) {
752+ mockOps := mocks.NewMockStreamOps(t)
753+ fakeLoop := unsafe.Pointer(uintptr(0xCAFE))
754+ fakeStream := unsafe.Pointer(uintptr(0xBEEF))
755+ cfg := PlayerConfig{
756+ SampleRate: 48000,
757+ Channels: 2,
758+ FramesPerBuffer: 64,
759+ }
760+
761+ // Build a buffer with NDatas mismatch to trigger ErrChannelMismatch
762+ // NDatas = 1 but config.Channels = 2
763+ singleData := spaData{
764+ Type: spaDataTypeMemPtr,
765+ Maxsize: 256,
766+ Data: unsafe.Pointer(new(float32)),
767+ Chunk: &spaChunk{},
768+ }
769+ spaBuf := spaBuffer{
770+ NDatas: 1, // mismatch with Channels=2
771+ Datas: unsafe.Pointer(&singleData),
772+ }
773+ pwBuf := pwBuffer{
774+ Buffer: &spaBuf,
775+ }
776+ bufPtr := unsafe.Pointer(&pwBuf)
777+
778+ wg := expectStartWithSync(mockOps, fakeLoop, fakeStream, cfg)
779+
780+ errCalled := false
781+ var errReceived error
782+
783+ p := newPlayer(cfg, PlayerCallbacks{
784+ Fill: func(buf *PCMBuffer) (int, error) {
785+ return buf.Frames, nil
786+ },
787+ OnError: func(err error) {
788+ errCalled = true
789+ errReceived = err
790+ },
791+ })
792+ p.streamOps = mockOps
793+
794+ if err := p.Start(); err != nil {
795+ t.Fatalf("Start failed: %v", err)
796+ }
797+ waitOnMainLoop(t, wg)
798+
799+ // DequeueBuffer returns a buffer that will fail newPWBufferView
800+ // QueueBuffer should NOT be called
801+ mockOps.EXPECT().DequeueBuffer(fakeStream).Return(bufPtr)
802+
803+ p.onProcess()
804+
805+ if !errCalled {
806+ t.Error("OnError should have been called when buffer view fails")
807+ }
808+ if errReceived == nil {
809+ t.Error("expected non-nil error from OnError")
810+ }
811+ if !errors.Is(errReceived, ErrChannelMismatch) {
812+ t.Errorf("expected ErrChannelMismatch, got %v", errReceived)
813+ }
814+ if p.State() != PlayerStateError {
815+ t.Errorf("expected Error state after buffer view failure, got %v", p.State())
816+ }
817+}
818+
819+// TestProcessCallbackProcessCMErrorSkipsQueue verifies that when processPCM
820+// returns an error, onProcess does not queue the buffer and lets the existing
821+// error behavior stand.
822+func TestProcessCallbackProcessCMErrorSkipsQueue(t *testing.T) {
823+ mockOps := mocks.NewMockStreamOps(t)
824+ fakeLoop := unsafe.Pointer(uintptr(0xCAFE))
825+ fakeStream := unsafe.Pointer(uintptr(0xBEEF))
826+ cfg := PlayerConfig{
827+ SampleRate: 48000,
828+ Channels: 2,
829+ FramesPerBuffer: 64,
830+ }
831+
832+ // Build a real pwBuffer
833+ const channels = 2
834+ const frames = 64
835+ channelData := make([][]float32, channels)
836+ floatPtrs := make([]unsafe.Pointer, channels)
837+ for ch := 0; ch < channels; ch++ {
838+ channelData[ch] = make([]float32, frames)
839+ floatPtrs[ch] = unsafe.Pointer(&channelData[ch][0])
840+ }
841+ chunkSlice := make([]spaChunk, channels)
842+ dataSlice := make([]spaData, channels)
843+ for ch := 0; ch < channels; ch++ {
844+ dataSlice[ch] = spaData{
845+ Type: spaDataTypeMemPtr,
846+ Flags: spaDataFlagRW,
847+ Maxsize: uint32(frames * 4),
848+ Data: floatPtrs[ch],
849+ Chunk: &chunkSlice[ch],
850+ }
851+ }
852+ spaBuf := spaBuffer{
853+ NDatas: uint32(channels),
854+ Datas: unsafe.Pointer(&dataSlice[0]),
855+ }
856+ pwBuf := pwBuffer{
857+ Buffer: &spaBuf,
858+ }
859+ bufPtr := unsafe.Pointer(&pwBuf)
860+
861+ wg := expectStartWithSync(mockOps, fakeLoop, fakeStream, cfg)
862+
863+ fillErr := errors.New("fill error")
864+ errCalled := false
865+ var errReceived error
866+
867+ p := newPlayer(cfg, PlayerCallbacks{
868+ Fill: func(buf *PCMBuffer) (int, error) {
869+ return 0, fillErr
870+ },
871+ OnError: func(err error) {
872+ errCalled = true
873+ errReceived = err
874+ },
875+ })
876+ p.streamOps = mockOps
877+
878+ if err := p.Start(); err != nil {
879+ t.Fatalf("Start failed: %v", err)
880+ }
881+ waitOnMainLoop(t, wg)
882+
883+ // DequeueBuffer returns a buffer; QueueBuffer should NOT be called
884+ // because processPCM (via Fill) returns an error
885+ mockOps.EXPECT().DequeueBuffer(fakeStream).Return(bufPtr)
886+ // Note: no QueueBuffer expectation
887+
888+ p.onProcess()
889+
890+ if !errCalled {
891+ t.Error("OnError should have been called when Fill returns an error")
892+ }
893+ if !errors.Is(errReceived, fillErr) {
894+ t.Errorf("expected fillErr, got %v", errReceived)
895+ }
896+ if p.State() != PlayerStateError {
897+ t.Errorf("expected Error state after Fill error, got %v", p.State())
898+ }
899+}
900+
901+// TestPlayerStartFailedConnectDestroysResources verifies that when
902+// ConnectPlaybackStream fails after stream creation, the stream
903+// and loop are destroyed and no stale pointers are left.
904+func TestPlayerStartFailedConnectDestroysResources(t *testing.T) {
905+ mockOps := mocks.NewMockStreamOps(t)
906+ fakeLoop := unsafe.Pointer(uintptr(0xCAFE))
907+ fakeStream := unsafe.Pointer(uintptr(0xBEEF))
908+ cfg := defaultTestConfig()
909+
910+ connectErr := errors.New("connect failed")
911+
912+ mockOps.EXPECT().CreateMainLoop().Return(fakeLoop, nil)
913+ mockOps.EXPECT().CreatePlaybackStream(fakeLoop, "purego-pipewire-player", mock.AnythingOfType("func()")).Return(fakeStream, nil)
914+ mockOps.EXPECT().ConnectPlaybackStream(fakeStream, portout.PlaybackFormat{
915+ SampleRate: cfg.SampleRate,
916+ Channels: cfg.Channels,
917+ FramesPerBuffer: cfg.FramesPerBuffer,
918+ }).Return(connectErr)
919+ // On connect failure, stream and loop must be destroyed
920+ mockOps.EXPECT().DestroyStream(fakeStream).Return()
921+ mockOps.EXPECT().DestroyMainLoop(fakeLoop).Return()
922+
923+ p := newPlayer(cfg, PlayerCallbacks{})
924+ p.streamOps = mockOps
925+
926+ err := p.Start()
927+ if err == nil {
928+ t.Fatal("expected Start to fail when connect fails")
929+ }
930+ if !errors.Is(err, connectErr) {
931+ t.Fatalf("expected connectErr, got %v", err)
932+ }
933+
934+ // Player should be in Error state
935+ if p.State() != PlayerStateError {
936+ t.Errorf("expected Error state after failed connect, got %v", p.State())
937+ }
938+
939+ // No stale pointers should remain
940+ p.mu.Lock()
941+ loopPtr := p.loopPtr
942+ streamPtr := p.streamPtr
943+ p.mu.Unlock()
944+
945+ if loopPtr != nil {
946+ t.Error("expected loopPtr to be nil after failed start")
947+ }
948+ if streamPtr != nil {
949+ t.Error("expected streamPtr to be nil after failed start")
950+ }
951+}
952+
953+// TestPlayerStartFailedActivateDestroysResources verifies that when
954+// SetStreamActive fails after stream creation and connection, the
955+// stream is disconnected (best-effort), destroyed, the loop is destroyed,
956+// and no stale pointers are left.
957+func TestPlayerStartFailedActivateDestroysResources(t *testing.T) {
958+ mockOps := mocks.NewMockStreamOps(t)
959+ fakeLoop := unsafe.Pointer(uintptr(0xCAFE))
960+ fakeStream := unsafe.Pointer(uintptr(0xBEEF))
961+ cfg := defaultTestConfig()
962+
963+ activateErr := errors.New("activate failed")
964+
965+ mockOps.EXPECT().CreateMainLoop().Return(fakeLoop, nil)
966+ mockOps.EXPECT().CreatePlaybackStream(fakeLoop, "purego-pipewire-player", mock.AnythingOfType("func()")).Return(fakeStream, nil)
967+ mockOps.EXPECT().ConnectPlaybackStream(fakeStream, portout.PlaybackFormat{
968+ SampleRate: cfg.SampleRate,
969+ Channels: cfg.Channels,
970+ FramesPerBuffer: cfg.FramesPerBuffer,
971+ }).Return(nil)
972+ mockOps.EXPECT().SetStreamActive(fakeStream, true).Return(activateErr)
973+ // On activate failure: disconnect (best-effort), destroy stream, destroy loop
974+ mockOps.EXPECT().DisconnectStream(fakeStream).Return(nil)
975+ mockOps.EXPECT().DestroyStream(fakeStream).Return()
976+ mockOps.EXPECT().DestroyMainLoop(fakeLoop).Return()
977+
978+ p := newPlayer(cfg, PlayerCallbacks{})
979+ p.streamOps = mockOps
980+
981+ err := p.Start()
982+ if err == nil {
983+ t.Fatal("expected Start to fail when activate fails")
984+ }
985+ if !errors.Is(err, activateErr) {
986+ t.Fatalf("expected activateErr, got %v", err)
987+ }
988+
989+ // Player should be in Error state
990+ if p.State() != PlayerStateError {
991+ t.Errorf("expected Error state after failed activate, got %v", p.State())
992+ }
993+
994+ // No stale pointers should remain
995+ p.mu.Lock()
996+ loopPtr := p.loopPtr
997+ streamPtr := p.streamPtr
998+ p.mu.Unlock()
999+
1000+ if loopPtr != nil {
1001+ t.Error("expected loopPtr to be nil after failed start")
1002+ }
1003+ if streamPtr != nil {
1004+ t.Error("expected streamPtr to be nil after failed start")
1005+ }
1006+}
1007+
1008+// TestPlayerStartFailedCreateStreamDestroysLoop verifies that when
1009+// CreatePlaybackStream fails after main loop creation, the loop is
1010+// destroyed and no stale pointers are left.
1011+func TestPlayerStartFailedCreateStreamDestroysLoop(t *testing.T) {
1012+ mockOps := mocks.NewMockStreamOps(t)
1013+ fakeLoop := unsafe.Pointer(uintptr(0xCAFE))
1014+ cfg := defaultTestConfig()
1015+
1016+ createErr := errors.New("stream creation failed")
1017+
1018+ mockOps.EXPECT().CreateMainLoop().Return(fakeLoop, nil)
1019+ mockOps.EXPECT().CreatePlaybackStream(fakeLoop, "purego-pipewire-player", mock.AnythingOfType("func()")).Return(nil, createErr)
1020+ // On stream creation failure, destroy the loop
1021+ mockOps.EXPECT().DestroyMainLoop(fakeLoop).Return()
1022+
1023+ p := newPlayer(cfg, PlayerCallbacks{})
1024+ p.streamOps = mockOps
1025+
1026+ err := p.Start()
1027+ if err == nil {
1028+ t.Fatal("expected Start to fail when stream creation fails")
1029+ }
1030+ if !errors.Is(err, createErr) {
1031+ t.Fatalf("expected createErr, got %v", err)
1032+ }
1033+
1034+ // Player should be in Error state
1035+ if p.State() != PlayerStateError {
1036+ t.Errorf("expected Error state after failed stream creation, got %v", p.State())
1037+ }
1038+
1039+ // No stale pointers should remain
1040+ p.mu.Lock()
1041+ loopPtr := p.loopPtr
1042+ streamPtr := p.streamPtr
1043+ p.mu.Unlock()
1044+
1045+ if loopPtr != nil {
1046+ t.Error("expected loopPtr to be nil after failed start")
1047+ }
1048+ if streamPtr != nil {
1049+ t.Error("expected streamPtr to be nil after failed start")
1050+ }
1051+}
M internal/core/player_test.go +142 -19
@@ -2,20 +2,91 @@ package core
22
33 import (
44 "errors"
5+ "sync"
56 "testing"
7+ "time"
68 "unsafe"
79
10+ portout "github.com/bnema/purego-pipewire/internal/ports/out"
811 "github.com/bnema/purego-pipewire/internal/ports/out/mocks"
12+ "github.com/stretchr/testify/mock"
913 )
1014
15+// defaultTestConfig returns a PlayerConfig suitable for tests that need
16+// valid PipeWire format parameters.
17+func defaultTestConfig() PlayerConfig {
18+ return PlayerConfig{
19+ SampleRate: 48000,
20+ Channels: 2,
21+ FramesPerBuffer: 256,
22+ }
23+}
24+
25+// expectStartWithSync sets up mock expectations for a first-time Start()
26+// and returns a pointer to a WaitGroup that is signaled when RunMainLoop completes.
27+// The caller must call waitOnMainLoop(t, wg) after Start() returns.
28+func expectStartWithSync(mockOps *mocks.MockStreamOps, fakeLoop, fakeStream unsafe.Pointer, cfg PlayerConfig) *sync.WaitGroup {
29+ wg := &sync.WaitGroup{}
30+ wg.Add(1)
31+
32+ mockOps.EXPECT().CreateMainLoop().Return(fakeLoop, nil)
33+ mockOps.EXPECT().CreatePlaybackStream(fakeLoop, "purego-pipewire-player", mock.AnythingOfType("func()")).Return(fakeStream, nil)
34+ mockOps.EXPECT().ConnectPlaybackStream(fakeStream, portout.PlaybackFormat{
35+ SampleRate: cfg.SampleRate,
36+ Channels: cfg.Channels,
37+ FramesPerBuffer: cfg.FramesPerBuffer,
38+ }).Return(nil)
39+ mockOps.EXPECT().SetStreamActive(fakeStream, true).Return(nil)
40+ mockOps.EXPECT().RunMainLoop(fakeLoop).RunAndReturn(func(unsafe.Pointer) error {
41+ wg.Done()
42+ return nil
43+ })
44+
45+ return wg
46+}
47+
48+// waitOnMainLoop waits for the RunMainLoop goroutine to finish, with a timeout.
49+func waitOnMainLoop(t *testing.T, wg *sync.WaitGroup) {
50+ t.Helper()
51+ done := make(chan struct{})
52+ go func() {
53+ wg.Wait()
54+ close(done)
55+ }()
56+ select {
57+ case <-done:
58+ case <-time.After(2 * time.Second):
59+ t.Fatal("timed out waiting for RunMainLoop goroutine")
60+ }
61+}
62+
1163 func TestPlayerStopIsRestartableButCloseIsTerminal(t *testing.T) {
12- p := newPlayer(PlayerConfig{}, PlayerCallbacks{})
64+ mockOps := mocks.NewMockStreamOps(t)
65+ fakeLoop := unsafe.Pointer(uintptr(0xBEEF))
66+ fakeStream := unsafe.Pointer(uintptr(0xDEAD))
67+ cfg := defaultTestConfig()
68+
69+ // First start creates resources
70+ wg := expectStartWithSync(mockOps, fakeLoop, fakeStream, cfg)
71+ // Stop deactivates the stream
72+ mockOps.EXPECT().SetStreamActive(fakeStream, false).Return(nil)
73+ // Restart reactivates existing stream
74+ mockOps.EXPECT().SetStreamActive(fakeStream, true).Return(nil)
75+ // Close tears down resources
76+ mockOps.EXPECT().DisconnectStream(fakeStream).Return(nil)
77+ mockOps.EXPECT().DestroyStream(fakeStream).Return()
78+ mockOps.EXPECT().QuitMainLoop(fakeLoop).Return()
79+ mockOps.EXPECT().DestroyMainLoop(fakeLoop).Return()
80+
81+ p := newPlayer(cfg, PlayerCallbacks{})
82+ p.streamOps = mockOps
1383
1484 // Start from Stopped state should work
1585 p.setState(PlayerStateStopped)
1686 if err := p.Start(); err != nil {
1787 t.Fatalf("Start from Stopped failed: %v", err)
1888 }
89+ waitOnMainLoop(t, wg)
1990
2091 // Should be in Playing state after Start
2192 if p.State() != PlayerStatePlaying {
@@ -79,7 +150,15 @@ func TestPlayerStartFromClosingReturnsError(t *testing.T) {
79150 }
80151
81152 func TestPlayerStartTransitionsThroughStartingToPlaying(t *testing.T) {
82- p := newPlayer(PlayerConfig{}, PlayerCallbacks{})
153+ mockOps := mocks.NewMockStreamOps(t)
154+ fakeLoop := unsafe.Pointer(uintptr(0xBEEF))
155+ fakeStream := unsafe.Pointer(uintptr(0xDEAD))
156+ cfg := defaultTestConfig()
157+
158+ wg := expectStartWithSync(mockOps, fakeLoop, fakeStream, cfg)
159+
160+ p := newPlayer(cfg, PlayerCallbacks{})
161+ p.streamOps = mockOps
83162
84163 // Initial state should be Idle
85164 if p.State() != PlayerStateIdle {
@@ -90,6 +169,7 @@ func TestPlayerStartTransitionsThroughStartingToPlaying(t *testing.T) {
90169 if err := p.Start(); err != nil {
91170 t.Fatalf("Start failed: %v", err)
92171 }
172+ waitOnMainLoop(t, wg)
93173
94174 if p.State() != PlayerStatePlaying {
95175 t.Fatalf("expected Playing state after Start, got %v", p.State())
@@ -97,12 +177,21 @@ func TestPlayerStartTransitionsThroughStartingToPlaying(t *testing.T) {
97177 }
98178
99179 func TestPlayerPauseTransitionsToPaused(t *testing.T) {
100- p := newPlayer(PlayerConfig{}, PlayerCallbacks{})
180+ mockOps := mocks.NewMockStreamOps(t)
181+ fakeLoop := unsafe.Pointer(uintptr(0xBEEF))
182+ fakeStream := unsafe.Pointer(uintptr(0xDEAD))
183+ cfg := defaultTestConfig()
184+
185+ wg := expectStartWithSync(mockOps, fakeLoop, fakeStream, cfg)
186+
187+ p := newPlayer(cfg, PlayerCallbacks{})
188+ p.streamOps = mockOps
101189
102190 // Start first
103191 if err := p.Start(); err != nil {
104192 t.Fatalf("Start failed: %v", err)
105193 }
194+ waitOnMainLoop(t, wg)
106195
107196 // Pause should transition to Paused
108197 if err := p.Pause(); err != nil {
@@ -115,12 +204,23 @@ func TestPlayerPauseTransitionsToPaused(t *testing.T) {
115204 }
116205
117206 func TestPlayerStopClearsPausedState(t *testing.T) {
118- p := newPlayer(PlayerConfig{}, PlayerCallbacks{})
207+ mockOps := mocks.NewMockStreamOps(t)
208+ fakeLoop := unsafe.Pointer(uintptr(0xBEEF))
209+ fakeStream := unsafe.Pointer(uintptr(0xDEAD))
210+ cfg := defaultTestConfig()
211+
212+ wg := expectStartWithSync(mockOps, fakeLoop, fakeStream, cfg)
213+ mockOps.EXPECT().SetStreamActive(fakeStream, false).Return(nil)
214+
215+ p := newPlayer(cfg, PlayerCallbacks{})
216+ p.streamOps = mockOps
119217
120218 // Start then pause
121219 if err := p.Start(); err != nil {
122220 t.Fatalf("Start failed: %v", err)
123221 }
222+ waitOnMainLoop(t, wg)
223+
124224 if err := p.Pause(); err != nil {
125225 t.Fatalf("Pause failed: %v", err)
126226 }
@@ -139,12 +239,25 @@ func TestPlayerStopClearsPausedState(t *testing.T) {
139239 }
140240
141241 func TestPlayerCloseIsIdempotent(t *testing.T) {
142- p := newPlayer(PlayerConfig{}, PlayerCallbacks{})
242+ mockOps := mocks.NewMockStreamOps(t)
243+ fakeLoop := unsafe.Pointer(uintptr(0xBEEF))
244+ fakeStream := unsafe.Pointer(uintptr(0xDEAD))
245+ cfg := defaultTestConfig()
246+
247+ wg := expectStartWithSync(mockOps, fakeLoop, fakeStream, cfg)
248+ mockOps.EXPECT().DisconnectStream(fakeStream).Return(nil)
249+ mockOps.EXPECT().DestroyStream(fakeStream).Return()
250+ mockOps.EXPECT().QuitMainLoop(fakeLoop).Return()
251+ mockOps.EXPECT().DestroyMainLoop(fakeLoop).Return()
252+
253+ p := newPlayer(cfg, PlayerCallbacks{})
254+ p.streamOps = mockOps
143255
144256 // Start first
145257 if err := p.Start(); err != nil {
146258 t.Fatalf("Start failed: %v", err)
147259 }
260+ waitOnMainLoop(t, wg)
148261
149262 // Close should work
150263 if err := p.Close(); err != nil {
@@ -167,18 +280,21 @@ func TestPlayerCloseIsIdempotent(t *testing.T) {
167280 // SetStreamActive(false) via StreamOps and transitions to Stopped.
168281 func TestPlayerStopDeactivatesStream(t *testing.T) {
169282 mockOps := mocks.NewMockStreamOps(t)
170- fakeStream := unsafe.Pointer(uintptr(0x1234))
283+ fakeLoop := unsafe.Pointer(uintptr(0x1234))
284+ fakeStream := unsafe.Pointer(uintptr(0x5678))
285+ cfg := defaultTestConfig()
171286
287+ wg := expectStartWithSync(mockOps, fakeLoop, fakeStream, cfg)
172288 mockOps.EXPECT().SetStreamActive(fakeStream, false).Return(nil)
173289
174- p := newPlayer(PlayerConfig{}, PlayerCallbacks{})
290+ p := newPlayer(cfg, PlayerCallbacks{})
175291 p.streamOps = mockOps
176- p.streamPtr = fakeStream
177292
178293 // Start the player
179294 if err := p.Start(); err != nil {
180295 t.Fatalf("Start failed: %v", err)
181296 }
297+ waitOnMainLoop(t, wg)
182298 if p.State() != PlayerStatePlaying {
183299 t.Fatalf("expected Playing state, got %v", p.State())
184300 }
@@ -196,19 +312,22 @@ func TestPlayerStopDeactivatesStream(t *testing.T) {
196312 // SetStreamActive error and leaves the player in the pre-stop state.
197313 func TestPlayerStopReturnsDeactivateError(t *testing.T) {
198314 mockOps := mocks.NewMockStreamOps(t)
199- fakeStream := unsafe.Pointer(uintptr(0x5678))
315+ fakeLoop := unsafe.Pointer(uintptr(0x5678))
316+ fakeStream := unsafe.Pointer(uintptr(0x9ABC))
200317 deactivateErr := errors.New("deactivate failed")
318+ cfg := defaultTestConfig()
201319
320+ wg := expectStartWithSync(mockOps, fakeLoop, fakeStream, cfg)
202321 mockOps.EXPECT().SetStreamActive(fakeStream, false).Return(deactivateErr)
203322
204- p := newPlayer(PlayerConfig{}, PlayerCallbacks{})
323+ p := newPlayer(cfg, PlayerCallbacks{})
205324 p.streamOps = mockOps
206- p.streamPtr = fakeStream
207325
208326 // Start the player
209327 if err := p.Start(); err != nil {
210328 t.Fatalf("Start failed: %v", err)
211329 }
330+ waitOnMainLoop(t, wg)
212331 if p.State() != PlayerStatePlaying {
213332 t.Fatalf("expected Playing state, got %v", p.State())
214333 }
@@ -228,16 +347,20 @@ func TestPlayerStopReturnsDeactivateError(t *testing.T) {
228347 }
229348 }
230349
231-// TestPlayerStopWithoutStreamOpsIsNoop verifies that Stop() works
232-// when no StreamOps is configured (backward compat).
233-func TestPlayerStopWithoutStreamOpsIsNoop(t *testing.T) {
234- p := newPlayer(PlayerConfig{}, PlayerCallbacks{})
350+// TestPlayerStopWithNilStreamPtrIsNoop verifies that Stop() works
351+// when streamOps is set but streamPtr is nil (backward compat).
352+func TestPlayerStopWithNilStreamPtrIsNoop(t *testing.T) {
353+ mockOps := mocks.NewMockStreamOps(t)
354+ cfg := defaultTestConfig()
235355
236- if err := p.Start(); err != nil {
237- t.Fatalf("Start failed: %v", err)
238- }
356+ p := newPlayer(cfg, PlayerCallbacks{})
357+ p.streamOps = mockOps
358+
359+ // Set state to Playing manually (without going through Start)
360+ // to test Stop without an active stream pointer
361+ p.setState(PlayerStatePlaying)
239362
240- // Stop without StreamOps should succeed (deactivation is a no-op)
363+ // Stop without a streamPtr should succeed (deactivation is a no-op)
241364 if err := p.Stop(); err != nil {
242365 t.Fatalf("Stop failed: %v", err)
243366 }
@@ -2,20 +2,91 @@ package core
2 2
3 import (3 import (
4 "errors"4 "errors"
5+ "sync"
5 "testing"6 "testing"
7+ "time"
6 "unsafe"8 "unsafe"
7 9
10+ portout "github.com/bnema/purego-pipewire/internal/ports/out"
8 "github.com/bnema/purego-pipewire/internal/ports/out/mocks"11 "github.com/bnema/purego-pipewire/internal/ports/out/mocks"
12+ "github.com/stretchr/testify/mock"
9 )13 )
10 14
15+// defaultTestConfig returns a PlayerConfig suitable for tests that need
16+// valid PipeWire format parameters.
17+func defaultTestConfig() PlayerConfig {
18+ return PlayerConfig{
19+ SampleRate: 48000,
20+ Channels: 2,
21+ FramesPerBuffer: 256,
22+ }
23+}
24+
25+// expectStartWithSync sets up mock expectations for a first-time Start()
26+// and returns a pointer to a WaitGroup that is signaled when RunMainLoop completes.
27+// The caller must call waitOnMainLoop(t, wg) after Start() returns.
28+func expectStartWithSync(mockOps *mocks.MockStreamOps, fakeLoop, fakeStream unsafe.Pointer, cfg PlayerConfig) *sync.WaitGroup {
29+ wg := &sync.WaitGroup{}
30+ wg.Add(1)
31+
32+ mockOps.EXPECT().CreateMainLoop().Return(fakeLoop, nil)
33+ mockOps.EXPECT().CreatePlaybackStream(fakeLoop, "purego-pipewire-player", mock.AnythingOfType("func()")).Return(fakeStream, nil)
34+ mockOps.EXPECT().ConnectPlaybackStream(fakeStream, portout.PlaybackFormat{
35+ SampleRate: cfg.SampleRate,
36+ Channels: cfg.Channels,
37+ FramesPerBuffer: cfg.FramesPerBuffer,
38+ }).Return(nil)
39+ mockOps.EXPECT().SetStreamActive(fakeStream, true).Return(nil)
40+ mockOps.EXPECT().RunMainLoop(fakeLoop).RunAndReturn(func(unsafe.Pointer) error {
41+ wg.Done()
42+ return nil
43+ })
44+
45+ return wg
46+}
47+
48+// waitOnMainLoop waits for the RunMainLoop goroutine to finish, with a timeout.
49+func waitOnMainLoop(t *testing.T, wg *sync.WaitGroup) {
50+ t.Helper()
51+ done := make(chan struct{})
52+ go func() {
53+ wg.Wait()
54+ close(done)
55+ }()
56+ select {
57+ case <-done:
58+ case <-time.After(2 * time.Second):
59+ t.Fatal("timed out waiting for RunMainLoop goroutine")
60+ }
61+}
62+
11 func TestPlayerStopIsRestartableButCloseIsTerminal(t *testing.T) {63 func TestPlayerStopIsRestartableButCloseIsTerminal(t *testing.T) {
12- p := newPlayer(PlayerConfig{}, PlayerCallbacks{})64+ mockOps := mocks.NewMockStreamOps(t)
65+ fakeLoop := unsafe.Pointer(uintptr(0xBEEF))
66+ fakeStream := unsafe.Pointer(uintptr(0xDEAD))
67+ cfg := defaultTestConfig()
68+
69+ // First start creates resources
70+ wg := expectStartWithSync(mockOps, fakeLoop, fakeStream, cfg)
71+ // Stop deactivates the stream
72+ mockOps.EXPECT().SetStreamActive(fakeStream, false).Return(nil)
73+ // Restart reactivates existing stream
74+ mockOps.EXPECT().SetStreamActive(fakeStream, true).Return(nil)
75+ // Close tears down resources
76+ mockOps.EXPECT().DisconnectStream(fakeStream).Return(nil)
77+ mockOps.EXPECT().DestroyStream(fakeStream).Return()
78+ mockOps.EXPECT().QuitMainLoop(fakeLoop).Return()
79+ mockOps.EXPECT().DestroyMainLoop(fakeLoop).Return()
80+
81+ p := newPlayer(cfg, PlayerCallbacks{})
82+ p.streamOps = mockOps
13 83
14 // Start from Stopped state should work84 // Start from Stopped state should work
15 p.setState(PlayerStateStopped)85 p.setState(PlayerStateStopped)
16 if err := p.Start(); err != nil {86 if err := p.Start(); err != nil {
17 t.Fatalf("Start from Stopped failed: %v", err)87 t.Fatalf("Start from Stopped failed: %v", err)
18 }88 }
89+ waitOnMainLoop(t, wg)
19 90
20 // Should be in Playing state after Start91 // Should be in Playing state after Start
21 if p.State() != PlayerStatePlaying {92 if p.State() != PlayerStatePlaying {
@@ -79,7 +150,15 @@ func TestPlayerStartFromClosingReturnsError(t *testing.T) {
79 }150 }
80 151
81 func TestPlayerStartTransitionsThroughStartingToPlaying(t *testing.T) {152 func TestPlayerStartTransitionsThroughStartingToPlaying(t *testing.T) {
82- p := newPlayer(PlayerConfig{}, PlayerCallbacks{})153+ mockOps := mocks.NewMockStreamOps(t)
154+ fakeLoop := unsafe.Pointer(uintptr(0xBEEF))
155+ fakeStream := unsafe.Pointer(uintptr(0xDEAD))
156+ cfg := defaultTestConfig()
157+
158+ wg := expectStartWithSync(mockOps, fakeLoop, fakeStream, cfg)
159+
160+ p := newPlayer(cfg, PlayerCallbacks{})
161+ p.streamOps = mockOps
83 162
84 // Initial state should be Idle163 // Initial state should be Idle
85 if p.State() != PlayerStateIdle {164 if p.State() != PlayerStateIdle {
@@ -90,6 +169,7 @@ func TestPlayerStartTransitionsThroughStartingToPlaying(t *testing.T) {
90 if err := p.Start(); err != nil {169 if err := p.Start(); err != nil {
91 t.Fatalf("Start failed: %v", err)170 t.Fatalf("Start failed: %v", err)
92 }171 }
172+ waitOnMainLoop(t, wg)
93 173
94 if p.State() != PlayerStatePlaying {174 if p.State() != PlayerStatePlaying {
95 t.Fatalf("expected Playing state after Start, got %v", p.State())175 t.Fatalf("expected Playing state after Start, got %v", p.State())
@@ -97,12 +177,21 @@ func TestPlayerStartTransitionsThroughStartingToPlaying(t *testing.T) {
97 }177 }
98 178
99 func TestPlayerPauseTransitionsToPaused(t *testing.T) {179 func TestPlayerPauseTransitionsToPaused(t *testing.T) {
100- p := newPlayer(PlayerConfig{}, PlayerCallbacks{})180+ mockOps := mocks.NewMockStreamOps(t)
181+ fakeLoop := unsafe.Pointer(uintptr(0xBEEF))
182+ fakeStream := unsafe.Pointer(uintptr(0xDEAD))
183+ cfg := defaultTestConfig()
184+
185+ wg := expectStartWithSync(mockOps, fakeLoop, fakeStream, cfg)
186+
187+ p := newPlayer(cfg, PlayerCallbacks{})
188+ p.streamOps = mockOps
101 189
102 // Start first190 // Start first
103 if err := p.Start(); err != nil {191 if err := p.Start(); err != nil {
104 t.Fatalf("Start failed: %v", err)192 t.Fatalf("Start failed: %v", err)
105 }193 }
194+ waitOnMainLoop(t, wg)
106 195
107 // Pause should transition to Paused196 // Pause should transition to Paused
108 if err := p.Pause(); err != nil {197 if err := p.Pause(); err != nil {
@@ -115,12 +204,23 @@ func TestPlayerPauseTransitionsToPaused(t *testing.T) {
115 }204 }
116 205
117 func TestPlayerStopClearsPausedState(t *testing.T) {206 func TestPlayerStopClearsPausedState(t *testing.T) {
118- p := newPlayer(PlayerConfig{}, PlayerCallbacks{})207+ mockOps := mocks.NewMockStreamOps(t)
208+ fakeLoop := unsafe.Pointer(uintptr(0xBEEF))
209+ fakeStream := unsafe.Pointer(uintptr(0xDEAD))
210+ cfg := defaultTestConfig()
211+
212+ wg := expectStartWithSync(mockOps, fakeLoop, fakeStream, cfg)
213+ mockOps.EXPECT().SetStreamActive(fakeStream, false).Return(nil)
214+
215+ p := newPlayer(cfg, PlayerCallbacks{})
216+ p.streamOps = mockOps
119 217
120 // Start then pause218 // Start then pause
121 if err := p.Start(); err != nil {219 if err := p.Start(); err != nil {
122 t.Fatalf("Start failed: %v", err)220 t.Fatalf("Start failed: %v", err)
123 }221 }
222+ waitOnMainLoop(t, wg)
223+
124 if err := p.Pause(); err != nil {224 if err := p.Pause(); err != nil {
125 t.Fatalf("Pause failed: %v", err)225 t.Fatalf("Pause failed: %v", err)
126 }226 }
@@ -139,12 +239,25 @@ func TestPlayerStopClearsPausedState(t *testing.T) {
139 }239 }
140 240
141 func TestPlayerCloseIsIdempotent(t *testing.T) {241 func TestPlayerCloseIsIdempotent(t *testing.T) {
142- p := newPlayer(PlayerConfig{}, PlayerCallbacks{})242+ mockOps := mocks.NewMockStreamOps(t)
243+ fakeLoop := unsafe.Pointer(uintptr(0xBEEF))
244+ fakeStream := unsafe.Pointer(uintptr(0xDEAD))
245+ cfg := defaultTestConfig()
246+
247+ wg := expectStartWithSync(mockOps, fakeLoop, fakeStream, cfg)
248+ mockOps.EXPECT().DisconnectStream(fakeStream).Return(nil)
249+ mockOps.EXPECT().DestroyStream(fakeStream).Return()
250+ mockOps.EXPECT().QuitMainLoop(fakeLoop).Return()
251+ mockOps.EXPECT().DestroyMainLoop(fakeLoop).Return()
252+
253+ p := newPlayer(cfg, PlayerCallbacks{})
254+ p.streamOps = mockOps
143 255
144 // Start first256 // Start first
145 if err := p.Start(); err != nil {257 if err := p.Start(); err != nil {
146 t.Fatalf("Start failed: %v", err)258 t.Fatalf("Start failed: %v", err)
147 }259 }
260+ waitOnMainLoop(t, wg)
148 261
149 // Close should work262 // Close should work
150 if err := p.Close(); err != nil {263 if err := p.Close(); err != nil {
@@ -167,18 +280,21 @@ func TestPlayerCloseIsIdempotent(t *testing.T) {
167 // SetStreamActive(false) via StreamOps and transitions to Stopped.280 // SetStreamActive(false) via StreamOps and transitions to Stopped.
168 func TestPlayerStopDeactivatesStream(t *testing.T) {281 func TestPlayerStopDeactivatesStream(t *testing.T) {
169 mockOps := mocks.NewMockStreamOps(t)282 mockOps := mocks.NewMockStreamOps(t)
170- fakeStream := unsafe.Pointer(uintptr(0x1234))283+ fakeLoop := unsafe.Pointer(uintptr(0x1234))
284+ fakeStream := unsafe.Pointer(uintptr(0x5678))
285+ cfg := defaultTestConfig()
171 286
287+ wg := expectStartWithSync(mockOps, fakeLoop, fakeStream, cfg)
172 mockOps.EXPECT().SetStreamActive(fakeStream, false).Return(nil)288 mockOps.EXPECT().SetStreamActive(fakeStream, false).Return(nil)
173 289
174- p := newPlayer(PlayerConfig{}, PlayerCallbacks{})290+ p := newPlayer(cfg, PlayerCallbacks{})
175 p.streamOps = mockOps291 p.streamOps = mockOps
176- p.streamPtr = fakeStream
177 292
178 // Start the player293 // Start the player
179 if err := p.Start(); err != nil {294 if err := p.Start(); err != nil {
180 t.Fatalf("Start failed: %v", err)295 t.Fatalf("Start failed: %v", err)
181 }296 }
297+ waitOnMainLoop(t, wg)
182 if p.State() != PlayerStatePlaying {298 if p.State() != PlayerStatePlaying {
183 t.Fatalf("expected Playing state, got %v", p.State())299 t.Fatalf("expected Playing state, got %v", p.State())
184 }300 }
@@ -196,19 +312,22 @@ func TestPlayerStopDeactivatesStream(t *testing.T) {
196 // SetStreamActive error and leaves the player in the pre-stop state.312 // SetStreamActive error and leaves the player in the pre-stop state.
197 func TestPlayerStopReturnsDeactivateError(t *testing.T) {313 func TestPlayerStopReturnsDeactivateError(t *testing.T) {
198 mockOps := mocks.NewMockStreamOps(t)314 mockOps := mocks.NewMockStreamOps(t)
199- fakeStream := unsafe.Pointer(uintptr(0x5678))315+ fakeLoop := unsafe.Pointer(uintptr(0x5678))
316+ fakeStream := unsafe.Pointer(uintptr(0x9ABC))
200 deactivateErr := errors.New("deactivate failed")317 deactivateErr := errors.New("deactivate failed")
318+ cfg := defaultTestConfig()
201 319
320+ wg := expectStartWithSync(mockOps, fakeLoop, fakeStream, cfg)
202 mockOps.EXPECT().SetStreamActive(fakeStream, false).Return(deactivateErr)321 mockOps.EXPECT().SetStreamActive(fakeStream, false).Return(deactivateErr)
203 322
204- p := newPlayer(PlayerConfig{}, PlayerCallbacks{})323+ p := newPlayer(cfg, PlayerCallbacks{})
205 p.streamOps = mockOps324 p.streamOps = mockOps
206- p.streamPtr = fakeStream
207 325
208 // Start the player326 // Start the player
209 if err := p.Start(); err != nil {327 if err := p.Start(); err != nil {
210 t.Fatalf("Start failed: %v", err)328 t.Fatalf("Start failed: %v", err)
211 }329 }
330+ waitOnMainLoop(t, wg)
212 if p.State() != PlayerStatePlaying {331 if p.State() != PlayerStatePlaying {
213 t.Fatalf("expected Playing state, got %v", p.State())332 t.Fatalf("expected Playing state, got %v", p.State())
214 }333 }
@@ -228,16 +347,20 @@ func TestPlayerStopReturnsDeactivateError(t *testing.T) {
228 }347 }
229 }348 }
230 349
231-// TestPlayerStopWithoutStreamOpsIsNoop verifies that Stop() works350+// TestPlayerStopWithNilStreamPtrIsNoop verifies that Stop() works
232-// when no StreamOps is configured (backward compat).351+// when streamOps is set but streamPtr is nil (backward compat).
233-func TestPlayerStopWithoutStreamOpsIsNoop(t *testing.T) {352+func TestPlayerStopWithNilStreamPtrIsNoop(t *testing.T) {
234- p := newPlayer(PlayerConfig{}, PlayerCallbacks{})353+ mockOps := mocks.NewMockStreamOps(t)
354+ cfg := defaultTestConfig()
235 355
236- if err := p.Start(); err != nil {356+ p := newPlayer(cfg, PlayerCallbacks{})
237- t.Fatalf("Start failed: %v", err)357+ p.streamOps = mockOps
238- }358+
359+ // Set state to Playing manually (without going through Start)
360+ // to test Stop without an active stream pointer
361+ p.setState(PlayerStatePlaying)
239 362
240- // Stop without StreamOps should succeed (deactivation is a no-op)363+ // Stop without a streamPtr should succeed (deactivation is a no-op)
241 if err := p.Stop(); err != nil {364 if err := p.Stop(); err != nil {
242 t.Fatalf("Stop failed: %v", err)365 t.Fatalf("Stop failed: %v", err)
243 }366 }

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 1

Tags 8

↑↓ moveenter openesc close