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 p.transition(PlayerStateError)
153 if p.callbacks.OnError != nil {
154 p.callbacks.OnError(err)
155 }
156}
157
158// onProcess is the PipeWire process callback. It is invoked by PipeWire
159// each time the stream needs more audio data. It dequeues a buffer,
160// fills it with PCM data via processPCM, commits the result, and
161// queues the buffer back. If any step fails, it routes the error
162// through p.fail.
163func (p *player) onProcess() {
164 ops := p.streamOps
165 if ops == nil {
166 return
167 }
168
169 sp := atomic.LoadPointer((*unsafe.Pointer)(unsafe.Pointer(&p.streamPtr)))
170 if sp == nil {
171 return
172 }
173
174 bufPtr := ops.DequeueBuffer(sp)
175 if bufPtr == nil {
176 return
177 }
178
179 queueOnError := true
180 defer func() {
181 if err := ops.QueueBuffer(sp, bufPtr); err != nil && queueOnError {
182 p.fail(fmt.Errorf("queue buffer: %w", err))
183 }
184 }()
185
186 frames := p.config.FramesPerBuffer
187 view, err := newPWBufferView(bufPtr, p.config.Channels, frames)
188 if err != nil {
189 queueOnError = false
190 p.fail(fmt.Errorf("buffer view: %w", err))
191 return
192 }
193
194 frames, err = p.processPCM(view.PCM())
195 if err != nil {
196 queueOnError = false
197 return
198 }
199
200 view.Commit(frames)
201}