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}
40
41// newPlayer creates a new player instance
42func newPlayer(config PlayerConfig, callbacks PlayerCallbacks) *player {
43 p := &player{
44 config: config,
45 callbacks: callbacks,
46 }
47 p.state.Store(int32(PlayerStateIdle))
48 return p
49}
50
51// State returns the current player state
52func (p *player) State() PlayerState {
53 return PlayerState(p.state.Load())
54}
55
56// setState atomically sets the player state
57func (p *player) setState(next PlayerState) {
58 p.state.Store(int32(next))
59}
60
61// transition attempts to transition to the next state
62func (p *player) transition(next PlayerState) error {
63 p.mu.Lock()
64
65 current := p.State()
66
67 // Validate transition
68 if !isValidTransition(current, next) {
69 p.mu.Unlock()
70 return ErrInvalidPlayerState
71 }
72
73 // Perform transition
74 p.setState(next)
75
76 // Capture callback under lock
77 cb := p.callbacks.OnStateChange
78
79 p.mu.Unlock()
80
81 // Invoke callback after releasing lock to avoid deadlock on reentry
82 if cb != nil {
83 cb(next)
84 }
85
86 return nil
87}
88
89// isValidTransition checks if a state transition is valid
90func isValidTransition(current, next PlayerState) bool {
91 switch current {
92 case PlayerStateIdle:
93 return next == PlayerStateStarting || next == PlayerStateClosing
94 case PlayerStateStarting:
95 return next == PlayerStatePlaying || next == PlayerStateError
96 case PlayerStatePlaying:
97 return next == PlayerStatePaused || next == PlayerStateStopped || next == PlayerStateClosing || next == PlayerStateError
98 case PlayerStatePaused:
99 return next == PlayerStatePlaying || next == PlayerStateStopped || next == PlayerStateClosing || next == PlayerStateError
100 case PlayerStateStopped:
101 return next == PlayerStateStarting || next == PlayerStateClosing
102 case PlayerStateClosing:
103 return next == PlayerStateClosed
104 case PlayerStateClosed:
105 return false // Terminal state
106 case PlayerStateError:
107 return next == PlayerStateClosing
108 default:
109 return false
110 }
111}
112
113// Start begins playback, transitioning from Idle or Stopped to Playing.
114func (p *player) Start() error {
115 current := p.State()
116 if current == PlayerStateClosed || current == PlayerStateClosing {
117 return ErrPlayerClosed
118 }
119
120 // Only valid from Idle or Stopped
121 if current != PlayerStateIdle && current != PlayerStateStopped {
122 return ErrInvalidPlayerState
123 }
124
125 // Ensure runtime is available
126 if err := p.ensureRuntime(); err != nil {
127 return err
128 }
129
130 // Transition to Starting
131 if err := p.transition(PlayerStateStarting); err != nil {
132 return err
133 }
134
135 // Check if we already have PipeWire resources (restart after Stop)
136 p.mu.Lock()
137 hasResources := p.loopPtr != nil && atomic.LoadPointer((*unsafe.Pointer)(unsafe.Pointer(&p.streamPtr))) != nil
138 p.mu.Unlock()
139
140 if !hasResources {
141 // First start — create PipeWire resources
142 if err := p.startCreateResourcesAndActivate(); err != nil {
143 return err
144 }
145 } else {
146 // Restart — just reactivate existing stream
147 streamPtr := atomic.LoadPointer((*unsafe.Pointer)(unsafe.Pointer(&p.streamPtr)))
148
149 if err := p.streamOps.SetStreamActive(streamPtr, true); err != nil {
150 // Activation failed — clean up owned resources so stale pointers
151 // are not left behind, then transition to Error.
152 p.teardown()
153 return p.failWithErrorState("reactivate stream", err)
154 }
155 }
156
157 // Clear paused flag
158 p.setPaused(false)
159
160 // Transition to Playing
161 return p.transition(PlayerStatePlaying)
162}
163
164// startCreateResourcesAndActivate creates the main loop, playback stream, connects
165// and activates it, stores owned resources, and starts the main loop goroutine.
166// On failure, all created resources are destroyed and the state transitions to Error.
167// No owned pointers are left set on failure.
168func (p *player) startCreateResourcesAndActivate() error {
169 // Create main loop
170 loopPtr, err := p.streamOps.CreateMainLoop()
171 if err != nil {
172 return p.failWithErrorState("create main loop", err)
173 }
174
175 // Create playback stream
176 streamPtr, err := p.streamOps.CreatePlaybackStream(loopPtr, "purego-pipewire-player", p.onProcess)
177 if err != nil {
178 p.streamOps.DestroyMainLoop(loopPtr)
179 return p.failWithErrorState("create playback stream", err)
180 }
181
182 // Connect stream
183 format := portout.PlaybackFormat{
184 SampleRate: p.config.SampleRate,
185 Channels: p.config.Channels,
186 FramesPerBuffer: p.config.FramesPerBuffer,
187 }
188 if err := p.streamOps.ConnectPlaybackStream(streamPtr, format); err != nil {
189 p.streamOps.DestroyStream(streamPtr)
190 p.streamOps.DestroyMainLoop(loopPtr)
191 return p.failWithErrorState("connect playback stream", err)
192 }
193
194 // Activate stream
195 if err := p.streamOps.SetStreamActive(streamPtr, true); err != nil {
196 // Best-effort disconnect before destroying
197 _ = p.streamOps.DisconnectStream(streamPtr)
198 p.streamOps.DestroyStream(streamPtr)
199 p.streamOps.DestroyMainLoop(loopPtr)
200 return p.failWithErrorState("activate playback stream", 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 := registerPipeWire(); err != nil {
294 return fmt.Errorf("register pipewire: %w", err)
295 }
296 capiInstance := defaultPipeWireCAPI()
297 if capiInstance == nil {
298 return errors.New("failed to get default CAPI instance")
299 }
300 if err := initializePipeWire(capiInstance); err != nil {
301 return fmt.Errorf("initialize pipewire runtime: %w", err)
302 }
303 ops := defaultPipeWireOps()
304 if ops == nil {
305 return errors.New("failed to obtain default stream operations")
306 }
307 p.streamOps = ops
308 return nil
309}
310
311func (p *player) failWithErrorState(context string, err error) error {
312 if transitionErr := p.transition(PlayerStateError); transitionErr != nil {
313 return errors.Join(fmt.Errorf("%s: %w", context, err), fmt.Errorf("transition player to error: %w", transitionErr))
314 }
315 return fmt.Errorf("%s: %w", context, err)
316}
317
318// setPaused sets the paused flag
319func (p *player) setPaused(paused bool) {
320 p.paused.Store(paused)
321}
322
323// deactivateStream deactivates the stream via StreamOps.
324// Returns an error if deactivation fails. No-op when streamOps is nil.
325func (p *player) deactivateStream() error {
326 p.mu.Lock()
327 streamOps := p.streamOps
328 streamPtr := atomic.LoadPointer((*unsafe.Pointer)(unsafe.Pointer(&p.streamPtr)))
329 p.mu.Unlock()
330
331 if streamOps == nil || streamPtr == nil {
332 return nil
333 }
334 return streamOps.SetStreamActive(streamPtr, false)
335}
336
337// teardown releases player-owned stream and main-loop resources through
338// StreamOps. Safe to call when fields are nil or on repeated invocation.
339// DisconnectStream is called before DestroyStream as best-effort cleanup;
340// if disconnect fails, teardown continues (destroy is the definitive release).
341func (p *player) teardown() {
342 p.mu.Lock()
343 streamOps := p.streamOps
344 streamPtr := atomic.LoadPointer((*unsafe.Pointer)(unsafe.Pointer(&p.streamPtr)))
345 loopPtr := p.loopPtr
346 atomic.StorePointer((*unsafe.Pointer)(unsafe.Pointer(&p.streamPtr)), nil)
347 p.loopPtr = nil
348 p.mu.Unlock()
349
350 if streamOps == nil {
351 return
352 }
353
354 // Disconnect then destroy stream (disconnect is best-effort, destroy is definitive).
355 if streamPtr != nil {
356 _ = streamOps.DisconnectStream(streamPtr) // Best-effort cleanup
357 streamOps.DestroyStream(streamPtr)
358 }
359
360 // Quit then destroy the main loop.
361 if loopPtr != nil {
362 streamOps.QuitMainLoop(loopPtr)
363 streamOps.DestroyMainLoop(loopPtr)
364 }
365}
366
367// Player is the exported wrapper for the internal player
368type Player struct {
369 p *player
370}
371
372// NewPlayer creates a new exported Player instance
373func NewPlayer(config playerConfig, callbacks playerCallbacks) *Player {
374 return &Player{p: newPlayer(config, callbacks)}
375}
376
377// State returns the current player state
378func (p *Player) State() PlayerState {
379 if p.p == nil {
380 return PlayerStateIdle
381 }
382 return p.p.State()
383}
384
385// Start begins playback
386func (p *Player) Start() error {
387 if p.p == nil {
388 return ErrPlayerClosed
389 }
390 return p.p.Start()
391}
392
393// Pause temporarily pauses playback
394func (p *Player) Pause() error {
395 if p.p == nil {
396 return ErrPlayerClosed
397 }
398 return p.p.Pause()
399}
400
401// Stop stops playback but allows restart
402func (p *Player) Stop() error {
403 if p.p == nil {
404 return ErrPlayerClosed
405 }
406 return p.p.Stop()
407}
408
409// Close permanently shuts down the player
410func (p *Player) Close() error {
411 if p.p == nil {
412 return nil
413 }
414 return p.p.Close()
415}