1package core
2
3import (
4 "errors"
5 "fmt"
6 "sync"
7 "sync/atomic"
8 "unsafe"
9
10 "github.com/bnema/purego-pipewire/internal/capi"
11 portout "github.com/bnema/purego-pipewire/internal/ports/out"
12)
13
14var (
15 ErrInvalidPlayerState = errors.New("invalid player state transition")
16 ErrPlayerClosed = errors.New("player is closed")
17
18 registerPipeWire = capi.Register
19 defaultPipeWireCAPI = capi.Default
20 defaultPipeWireOps = capi.DefaultStreamOps
21 initializePipeWire = initPipeWire
22)
23
24// player is the internal player shell with state machine.
25type player struct {
26 // config is immutable after construction. It is safe to read from
27 // any goroutine — including the PipeWire process callback — without
28 // holding mu.
29 config playerConfig
30 callbacks playerCallbacks
31 state atomic.Int32
32 paused atomic.Bool
33 mu sync.Mutex
34
35 // pwView reuses Go slice headers for PipeWire callback buffers. Its native
36 // references are cleared when each callback completes.
37 pwView pwBufferView
38
39 // Stream cleanup fields — nil when no stream is active.
40 streamOps portout.StreamOps
41 streamPtr unsafe.Pointer
42 loopPtr unsafe.Pointer
43 loopDone chan struct{}
44}
45
46// newPlayer creates a new player instance
47func newPlayer(config PlayerConfig, callbacks PlayerCallbacks) *player {
48 p := &player{
49 config: config,
50 callbacks: callbacks,
51 pwView: newReusablePWBufferView(config.Channels, config.FramesPerBuffer),
52 }
53 p.state.Store(int32(PlayerStateIdle))
54 return p
55}
56
57// State returns the current player state
58func (p *player) State() PlayerState {
59 return PlayerState(p.state.Load())
60}
61
62// setState atomically sets the player state
63func (p *player) setState(next PlayerState) {
64 p.state.Store(int32(next))
65}
66
67// transition attempts to transition to the next state
68func (p *player) transition(next PlayerState) error {
69 p.mu.Lock()
70
71 current := p.State()
72
73 // Validate transition
74 if !isValidTransition(current, next) {
75 p.mu.Unlock()
76 return ErrInvalidPlayerState
77 }
78
79 // Perform transition
80 p.setState(next)
81
82 // Capture callback under lock
83 cb := p.callbacks.OnStateChange
84
85 p.mu.Unlock()
86
87 // Invoke callback after releasing lock to avoid deadlock on reentry
88 if cb != nil {
89 cb(next)
90 }
91
92 return nil
93}
94
95// isValidTransition checks if a state transition is valid
96func isValidTransition(current, next PlayerState) bool {
97 switch current {
98 case PlayerStateIdle:
99 return next == PlayerStateStarting || next == PlayerStateClosing
100 case PlayerStateStarting:
101 return next == PlayerStatePlaying || next == PlayerStateError
102 case PlayerStatePlaying:
103 return next == PlayerStatePaused || next == PlayerStateStopped || next == PlayerStateClosing || next == PlayerStateError
104 case PlayerStatePaused:
105 return next == PlayerStatePlaying || next == PlayerStateStopped || next == PlayerStateClosing || next == PlayerStateError
106 case PlayerStateStopped:
107 return next == PlayerStateStarting || next == PlayerStateClosing
108 case PlayerStateClosing:
109 return next == PlayerStateClosed
110 case PlayerStateClosed:
111 return false // Terminal state
112 case PlayerStateError:
113 return next == PlayerStateClosing
114 default:
115 return false
116 }
117}
118
119// Start begins playback, transitioning from Idle or Stopped to Playing.
120func (p *player) Start() error {
121 current := p.State()
122 if current == PlayerStateClosed || current == PlayerStateClosing {
123 return ErrPlayerClosed
124 }
125
126 // Only valid from Idle or Stopped
127 if current != PlayerStateIdle && current != PlayerStateStopped {
128 return ErrInvalidPlayerState
129 }
130
131 // Ensure runtime is available
132 if err := p.ensureRuntime(); err != nil {
133 return err
134 }
135
136 // Transition to Starting
137 if err := p.transition(PlayerStateStarting); err != nil {
138 return err
139 }
140
141 // Check if we already have PipeWire resources (restart after Stop)
142 p.mu.Lock()
143 hasResources := p.loopPtr != nil && atomic.LoadPointer((*unsafe.Pointer)(unsafe.Pointer(&p.streamPtr))) != nil
144 p.mu.Unlock()
145
146 if !hasResources {
147 // First start — create PipeWire resources
148 if err := p.startCreateResourcesAndActivate(); err != nil {
149 return err
150 }
151 } else {
152 // Restart — just reactivate existing stream
153 streamPtr := atomic.LoadPointer((*unsafe.Pointer)(unsafe.Pointer(&p.streamPtr)))
154
155 if err := p.streamOps.SetStreamActive(streamPtr, true); err != nil {
156 // Activation failed — clean up owned resources so stale pointers
157 // are not left behind, then transition to Error.
158 p.teardown()
159 return p.failWithErrorState("reactivate stream", err)
160 }
161 }
162
163 // Clear paused flag
164 p.setPaused(false)
165
166 // Transition to Playing
167 return p.transition(PlayerStatePlaying)
168}
169
170// startCreateResourcesAndActivate creates the main loop, playback stream, connects
171// and activates it, stores owned resources, and starts the main loop goroutine.
172// On failure, all created resources are destroyed and the state transitions to Error.
173// No owned pointers are left set on failure.
174func (p *player) startCreateResourcesAndActivate() error {
175 // Create main loop
176 loopPtr, err := p.streamOps.CreateMainLoop()
177 if err != nil {
178 return p.failWithErrorState("create main loop", err)
179 }
180
181 // Create playback stream
182 streamPtr, err := p.streamOps.CreatePlaybackStream(loopPtr, "purego-pipewire-player", p.onProcess)
183 if err != nil {
184 p.streamOps.DestroyMainLoop(loopPtr)
185 return p.failWithErrorState("create playback stream", err)
186 }
187
188 // Connect stream
189 format := portout.PlaybackFormat{
190 SampleRate: p.config.SampleRate,
191 Channels: p.config.Channels,
192 FramesPerBuffer: p.config.FramesPerBuffer,
193 }
194 if err := p.streamOps.ConnectPlaybackStream(streamPtr, format); err != nil {
195 p.streamOps.DestroyStream(streamPtr)
196 p.streamOps.DestroyMainLoop(loopPtr)
197 return p.failWithErrorState("connect playback stream", err)
198 }
199
200 // Activate stream
201 if err := p.streamOps.SetStreamActive(streamPtr, true); err != nil {
202 // Best-effort disconnect before destroying
203 _ = p.streamOps.DisconnectStream(streamPtr)
204 p.streamOps.DestroyStream(streamPtr)
205 p.streamOps.DestroyMainLoop(loopPtr)
206 return p.failWithErrorState("activate playback stream", err)
207 }
208
209 // All steps succeeded — store owned resources
210 loopDone := make(chan struct{})
211 p.mu.Lock()
212 p.loopPtr = loopPtr
213 p.loopDone = loopDone
214 atomic.StorePointer((*unsafe.Pointer)(unsafe.Pointer(&p.streamPtr)), streamPtr)
215 p.mu.Unlock()
216
217 // Start main loop in an internal goroutine
218 go func() {
219 defer close(loopDone)
220 if err := p.streamOps.RunMainLoop(loopPtr); err != nil {
221 if state := p.State(); state != PlayerStateClosing && state != PlayerStateClosed {
222 p.fail(err)
223 }
224 }
225 }()
226
227 return nil
228}
229
230// Pause temporarily pauses playback
231func (p *player) Pause() error {
232 current := p.State()
233 if current == PlayerStateClosed || current == PlayerStateClosing {
234 return ErrPlayerClosed
235 }
236
237 // Only valid from Playing
238 if current != PlayerStatePlaying {
239 return ErrInvalidPlayerState
240 }
241
242 p.setPaused(true)
243 return p.transition(PlayerStatePaused)
244}
245
246// Stop stops playback but allows restart.
247// Returns an error if stream deactivation fails, leaving the state unchanged.
248func (p *player) Stop() error {
249 current := p.State()
250 if current == PlayerStateClosed || current == PlayerStateClosing {
251 return ErrPlayerClosed
252 }
253
254 // Valid from Playing or Paused.
255 if current != PlayerStatePlaying && current != PlayerStatePaused {
256 return ErrInvalidPlayerState
257 }
258
259 // Deactivate stream - if this fails, return the error and don't transition
260 if err := p.deactivateStream(); err != nil {
261 return err
262 }
263
264 // Clear paused flag only after successful deactivation
265 p.setPaused(false)
266
267 return p.transition(PlayerStateStopped)
268}
269
270// Close permanently shuts down the player
271func (p *player) Close() error {
272 current := p.State()
273
274 // Idempotent on already closed
275 if current == PlayerStateClosed {
276 return nil
277 }
278
279 // Transition to Closing if not already there
280 if current != PlayerStateClosing {
281 if err := p.transition(PlayerStateClosing); err != nil {
282 return err
283 }
284 }
285
286 // Teardown the player-owned stream and main-loop resources.
287 p.teardown()
288
289 // Transition to Closed
290 return p.transition(PlayerStateClosed)
291}
292
293// ensureRuntime ensures the player has a usable StreamOps implementation.
294// If streamOps is already set (e.g., by tests), it is used as-is.
295// Otherwise, PipeWire is registered via capi.Register(), the default CAPI
296// instance is retrieved, and the shared process-level init guard is used
297// before obtaining the default StreamOps implementation.
298func (p *player) ensureRuntime() error {
299 if p.streamOps != nil {
300 return nil
301 }
302 if err := registerPipeWire(); err != nil {
303 return fmt.Errorf("register pipewire: %w", err)
304 }
305 capiInstance := defaultPipeWireCAPI()
306 if capiInstance == nil {
307 return errors.New("failed to get default CAPI instance")
308 }
309 if err := initializePipeWire(capiInstance); err != nil {
310 return fmt.Errorf("initialize pipewire runtime: %w", err)
311 }
312 ops := defaultPipeWireOps()
313 if ops == nil {
314 return errors.New("failed to obtain default stream operations")
315 }
316 p.streamOps = ops
317 return nil
318}
319
320func (p *player) failWithErrorState(context string, err error) error {
321 if transitionErr := p.transition(PlayerStateError); transitionErr != nil {
322 return errors.Join(fmt.Errorf("%s: %w", context, err), fmt.Errorf("transition player to error: %w", transitionErr))
323 }
324 return fmt.Errorf("%s: %w", context, err)
325}
326
327// setPaused sets the paused flag
328func (p *player) setPaused(paused bool) {
329 p.paused.Store(paused)
330}
331
332// deactivateStream deactivates the stream via StreamOps.
333// Returns an error if deactivation fails. No-op when streamOps is nil.
334func (p *player) deactivateStream() error {
335 p.mu.Lock()
336 streamOps := p.streamOps
337 streamPtr := atomic.LoadPointer((*unsafe.Pointer)(unsafe.Pointer(&p.streamPtr)))
338 p.mu.Unlock()
339
340 if streamOps == nil || streamPtr == nil {
341 return nil
342 }
343 return streamOps.SetStreamActive(streamPtr, false)
344}
345
346// teardown releases player-owned stream and main-loop resources through
347// StreamOps. Safe to call when fields are nil or on repeated invocation.
348// When a main loop exists, DisconnectStream is issued first while the loop is
349// still running, then the loop is quit and awaited, and only then is the
350// stream destroyed. This ordering avoids destroying a pw_stream while its
351// main loop is still active.
352func (p *player) teardown() {
353 p.mu.Lock()
354 streamOps := p.streamOps
355 streamPtr := atomic.LoadPointer((*unsafe.Pointer)(unsafe.Pointer(&p.streamPtr)))
356 loopPtr := p.loopPtr
357 loopDone := p.loopDone
358 atomic.StorePointer((*unsafe.Pointer)(unsafe.Pointer(&p.streamPtr)), nil)
359 p.loopPtr = nil
360 p.loopDone = nil
361 p.mu.Unlock()
362
363 if streamOps == nil {
364 return
365 }
366
367 // If a stream exists, disconnect it first while the loop is still alive.
368 if streamPtr != nil {
369 _ = streamOps.DisconnectStream(streamPtr) // Best-effort cleanup
370 }
371
372 // Quit then destroy the main loop.
373 // Never destroy the main loop until RunMainLoop has actually returned;
374 // destroying while pw_main_loop_run is still active can crash in PipeWire.
375 if loopPtr != nil {
376 streamOps.QuitMainLoop(loopPtr)
377 if loopDone != nil {
378 <-loopDone
379 }
380 }
381
382 // Only destroy the stream after the loop has stopped so pw_stream_destroy
383 // can safely flush any queued loop callbacks before freeing stream state.
384 if streamPtr != nil {
385 streamOps.DestroyStream(streamPtr)
386 }
387 if loopPtr != nil {
388 streamOps.DestroyMainLoop(loopPtr)
389 }
390}
391
392// Player is the exported wrapper for the internal player
393type Player struct {
394 p *player
395}
396
397// NewPlayer creates a new exported Player instance
398func NewPlayer(config playerConfig, callbacks playerCallbacks) *Player {
399 return &Player{p: newPlayer(config, callbacks)}
400}
401
402// State returns the current player state
403func (p *Player) State() PlayerState {
404 if p.p == nil {
405 return PlayerStateIdle
406 }
407 return p.p.State()
408}
409
410// Start begins playback
411func (p *Player) Start() error {
412 if p.p == nil {
413 return ErrPlayerClosed
414 }
415 return p.p.Start()
416}
417
418// Pause temporarily pauses playback
419func (p *Player) Pause() error {
420 if p.p == nil {
421 return ErrPlayerClosed
422 }
423 return p.p.Pause()
424}
425
426// Stop stops playback but allows restart
427func (p *Player) Stop() error {
428 if p.p == nil {
429 return ErrPlayerClosed
430 }
431 return p.p.Stop()
432}
433
434// Close permanently shuts down the player
435func (p *Player) Close() error {
436 if p.p == nil {
437 return nil
438 }
439 return p.p.Close()
440}