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