1package core
2
3import (
4 "errors"
5 "fmt"
6 "io"
7 "sync/atomic"
8 "unsafe"
9)
10
11// UnderrunPolicy determines how to handle underruns
12type UnderrunPolicy int
13
14const (
15 UnderrunFillSilence UnderrunPolicy = iota
16 UnderrunFailFast
17)
18
19// PCMBuffer represents a PCM audio buffer with deinterleaved samples
20// Samples[channel][frame] format for cache efficiency
21type PCMBuffer struct {
22 Frames int
23 Channels int
24 Stride int
25 Samples [][]float32
26}
27
28// allocate allocates the Samples slice structure
29func (b *PCMBuffer) allocate() {
30 if len(b.Samples) != b.Channels {
31 b.Samples = make([][]float32, b.Channels)
32 }
33 for ch := 0; ch < b.Channels; ch++ {
34 if len(b.Samples[ch]) != b.Frames {
35 b.Samples[ch] = make([]float32, b.Frames)
36 }
37 }
38}
39
40// PlayerConfig holds configuration for the player
41type PlayerConfig struct {
42 // SampleRate is the playback sample rate used when opening the stream.
43 SampleRate int
44 FramesPerBuffer int
45 Channels int
46 UnderrunPolicy UnderrunPolicy
47}
48
49// PlayerCallbacks holds callbacks for player events
50type PlayerCallbacks struct {
51 OnStateChange func(PlayerState)
52 Fill func(*PCMBuffer) (int, error)
53 OnUnderrun func(int)
54 OnDrain func()
55 OnError func(error)
56}
57
58// playerConfig is the internal alias for PlayerConfig
59type playerConfig = PlayerConfig
60
61// playerCallbacks is the internal alias for PlayerCallbacks
62type playerCallbacks = PlayerCallbacks
63
64// processPCM processes PCM audio data for the given buffer
65// Returns the number of frames processed, or an error
66func (p *player) processPCM(buf *PCMBuffer) (int, error) {
67 // Ensure buffer is allocated
68 buf.allocate()
69
70 state := p.State()
71
72 // If paused or stopped, fill with silence and return
73 if state == PlayerStatePaused || state == PlayerStateStopped || state == PlayerStateIdle {
74 p.fillSilence(buf, 0, buf.Frames)
75 return buf.Frames, nil
76 }
77
78 // If not playing, fill with silence
79 if state != PlayerStatePlaying {
80 p.fillSilence(buf, 0, buf.Frames)
81 return buf.Frames, nil
82 }
83
84 // Call Fill callback
85 if p.callbacks.Fill == nil {
86 p.fillSilence(buf, 0, buf.Frames)
87 return buf.Frames, nil
88 }
89
90 frames, err := p.callbacks.Fill(buf)
91 if err != nil {
92 if errors.Is(err, io.EOF) {
93 // Drain condition - emit drain callback and return frames
94 p.emitDrain()
95 return frames, nil
96 }
97 // Other error - fail the player
98 p.fail(err)
99 return 0, err
100 }
101
102 // Handle underrun if Fill returned fewer frames than requested
103 if frames < buf.Frames {
104 return p.handleUnderrun(buf, frames)
105 }
106
107 return frames, nil
108}
109
110// fillSilence fills the buffer with silence (0.0) from startFrame to endFrame
111func (p *player) fillSilence(buf *PCMBuffer, startFrame, endFrame int) {
112 if startFrame < 0 {
113 startFrame = 0
114 }
115 if endFrame > buf.Frames {
116 endFrame = buf.Frames
117 }
118 for ch := 0; ch < buf.Channels; ch++ {
119 clear(buf.Samples[ch][startFrame:endFrame])
120 }
121}
122
123// handleUnderrun applies the underrun policy when Fill returns fewer frames than requested
124func (p *player) handleUnderrun(buf *PCMBuffer, filledFrames int) (int, error) {
125 // Emit underrun callback
126 if p.callbacks.OnUnderrun != nil {
127 p.callbacks.OnUnderrun(buf.Frames - filledFrames)
128 }
129
130 switch p.config.UnderrunPolicy {
131 case UnderrunFailFast:
132 err := errors.New("underrun occurred")
133 p.fail(err)
134 return 0, err
135 case UnderrunFillSilence:
136 fallthrough
137 default:
138 p.fillSilence(buf, filledFrames, buf.Frames)
139 return buf.Frames, nil
140 }
141}
142
143// emitDrain invokes the OnDrain callback
144func (p *player) emitDrain() {
145 if p.callbacks.OnDrain != nil {
146 p.callbacks.OnDrain()
147 }
148}
149
150// fail transitions the player to error state and invokes OnError
151func (p *player) fail(err error) {
152 if transitionErr := p.transition(PlayerStateError); transitionErr != nil {
153 err = errors.Join(err, fmt.Errorf("transition player to error: %w", transitionErr))
154 }
155 if p.callbacks.OnError != nil {
156 p.callbacks.OnError(err)
157 }
158}
159
160// onProcess is the PipeWire process callback. It is invoked by PipeWire
161// each time the stream needs more audio data. It dequeues a buffer,
162// fills it with PCM data via processPCM, commits the result, and
163// queues the buffer back. If any step fails, it routes the error
164// through p.fail.
165func (p *player) onProcess() {
166 // Drop native references before returning from each callback.
167 defer p.pwView.clear()
168
169 ops := p.streamOps
170 if ops == nil {
171 return
172 }
173
174 sp := atomic.LoadPointer((*unsafe.Pointer)(unsafe.Pointer(&p.streamPtr)))
175 if sp == nil {
176 return
177 }
178
179 bufPtr := ops.DequeueBuffer(sp)
180 if bufPtr == nil {
181 return
182 }
183
184 queueOnError := true
185 defer func() {
186 err := ops.QueueBuffer(sp, bufPtr)
187 if err != nil && queueOnError {
188 // Clear before OnError can reenter player methods.
189 p.pwView.clear()
190 p.fail(fmt.Errorf("queue buffer: %w", err))
191 }
192 }()
193
194 if err := p.pwView.refresh(bufPtr); err != nil {
195 queueOnError = false
196 p.fail(fmt.Errorf("buffer view: %w", err))
197 return
198 }
199
200 frames, err := p.processPCM(p.pwView.PCM())
201 if err != nil {
202 queueOnError = false
203 return
204 }
205
206 p.pwView.Commit(frames)
207}