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