1// Package pipewire provides a PipeWire-based audio output implementation.
2// This wraps purego-pipewire's high-level Player API.
3package pipewire
4
5import (
6 "context"
7 "errors"
8 "fmt"
9 "sync"
10 "sync/atomic"
11
12 pwpipewire "github.com/bnema/purego-pipewire/pipewire"
13
14 "github.com/bnema/dumber/internal/application/port"
15 "github.com/bnema/dumber/internal/logging"
16)
17
18// Compile-time interface checks.
19var (
20 _ port.AudioOutputFactory = (*Factory)(nil)
21 _ port.AudioOutputStream = (*playerStreamAdapter)(nil)
22)
23
24// ErrInvalidFormat is returned when the audio stream format is invalid.
25var ErrInvalidFormat = errors.New("invalid audio stream format")
26
27// ErrStreamClosed is returned when writing to a closed stream.
28var ErrStreamClosed = errors.New("audio stream closed")
29
30// playerCreator is a function type that creates a Player from config and callbacks.
31// This exists to allow dependency injection in tests.
32type playerCreator func(config pwpipewire.PlayerConfig, callbacks pwpipewire.PlayerCallbacks) (pwpipewire.Player, error)
33
34// Factory creates PipeWire audio output streams.
35// Each call to NewStream creates an independent purego-pipewire Player.
36type Factory struct {
37 createPlayer playerCreator
38}
39
40// NewFactory creates a new PipeWire audio factory.
41// The factory is stateless; players are created per-stream.
42func NewFactory() (*Factory, error) {
43 return &Factory{
44 createPlayer: pwpipewire.NewPlayer,
45 }, nil
46}
47
48// NewStream creates a new audio output stream with the given format.
49// It validates the Dumber format, creates a purego-pipewire Player configured
50// from the format, and returns a push-to-pull adapter that bridges Write()
51// calls into the Player's Fill callback.
52func (f *Factory) NewStream(ctx context.Context, format port.AudioStreamFormat) (port.AudioOutputStream, error) {
53 // Check context cancellation
54 if ctx.Err() != nil {
55 return nil, ctx.Err()
56 }
57
58 // Validate format
59 if err := validateFormat(format); err != nil {
60 return nil, err
61 }
62
63 log := logging.FromContext(ctx)
64 log.Info().
65 Int("sample_rate", format.SampleRate).
66 Int("channels", format.ChannelCount).
67 Int("frames_per_buffer", format.FramesPerBuffer).
68 Msg("pipewire: creating player")
69
70 // Create the adapter with its ring buffer before creating the player,
71 // because the Fill callback captures the adapter.
72 adapter := &playerStreamAdapter{
73 ctx: ctx,
74 buf: make(chan [][]float32, 4), // small buffer for push→pull bridging
75 closed: make(chan struct{}),
76 }
77
78 config := pwpipewire.PlayerConfig{
79 SampleRate: format.SampleRate,
80 Channels: format.ChannelCount,
81 FramesPerBuffer: format.FramesPerBuffer,
82 SampleFormat: pwpipewire.SampleFormatF32,
83 UnderrunPolicy: pwpipewire.UnderrunFillSilence,
84 }
85
86 callbacks := pwpipewire.PlayerCallbacks{
87 Fill: adapter.fill,
88 }
89
90 player, err := f.createPlayer(config, callbacks)
91 if err != nil {
92 log.Warn().Err(err).Msg("pipewire: failed to create player")
93 return nil, fmt.Errorf("failed to create pipewire player: %w", err)
94 }
95 adapter.player = player
96
97 // Start the player so the Fill callback begins being called.
98 if err := player.Start(); err != nil {
99 _ = player.Close()
100 log.Warn().Err(err).Msg("pipewire: failed to start player")
101 return nil, fmt.Errorf("failed to start pipewire player: %w", err)
102 }
103
104 log.Info().Msg("pipewire: player started successfully")
105
106 return adapter, nil
107}
108
109// validateFormat checks if the audio format parameters are valid.
110func validateFormat(format port.AudioStreamFormat) error {
111 // Sample rate must be reasonable (8kHz to 384kHz)
112 if format.SampleRate < 8000 || format.SampleRate > 384000 {
113 return fmt.Errorf("%w: sample rate %d Hz out of range [8000, 384000]",
114 ErrInvalidFormat, format.SampleRate)
115 }
116
117 // Channel count must be 1 to 1024 (reasonable upper limit)
118 if format.ChannelCount < 1 || format.ChannelCount > 1024 {
119 return fmt.Errorf("%w: channel count %d out of range [1, 1024]",
120 ErrInvalidFormat, format.ChannelCount)
121 }
122
123 // Frames per buffer must be positive and reasonable
124 if format.FramesPerBuffer < 1 || format.FramesPerBuffer > 8192 {
125 return fmt.Errorf("%w: frames per buffer %d out of range [1, 8192]",
126 ErrInvalidFormat, format.FramesPerBuffer)
127 }
128
129 return nil
130}
131
132// playerStreamAdapter bridges Dumber's push-style Write(samples) interface
133// to purego-pipewire's pull-style Fill callback.
134//
135// Write() sends sample buffers into a channel. The Fill callback drains them
136// into the PCMBuffer that purego-pipewire passes in. If no data is available,
137// the player's UnderrunFillSilence policy handles it.
138type playerStreamAdapter struct {
139 player pwpipewire.Player
140 ctx context.Context
141 buf chan [][]float32 // push→pull handoff channel
142
143 closeOnce sync.Once
144 closed chan struct{} // closed when Close is called
145
146 // Diagnostic counters (atomic, no mutex needed).
147 fillCount atomic.Uint64 // total Fill callbacks
148 underrunCount atomic.Uint64 // Fill callbacks with no data available
149 dropCount atomic.Uint64 // Write calls that dropped data (buffer full)
150}
151
152// Write sends audio samples to the PipeWire output.
153// Samples are provided as [channel][frame]float32 matching CEF's planar format.
154// If the internal buffer is full the packet is dropped silently so the caller
155// (CEF audio callback thread) is never blocked.
156// Safe for concurrent use with Close.
157func (a *playerStreamAdapter) Write(samples [][]float32) error {
158 // Fast path: already closed.
159 select {
160 case <-a.closed:
161 return ErrStreamClosed
162 default:
163 }
164 // Non-blocking send: drop packet when buffer is full.
165 select {
166 case a.buf <- samples:
167 return nil
168 case <-a.closed:
169 return ErrStreamClosed
170 default:
171 // Buffer full — drop to avoid blocking the CEF audio thread.
172 n := a.dropCount.Add(1)
173 if n == 1 || n&(n-1) == 0 { // first drop, then powers of two
174 if a.ctx != nil {
175 logging.FromContext(a.ctx).Warn().
176 Uint64("total_drops", n).
177 Msg("pipewire: dropped audio packet (buffer full)")
178 }
179 }
180 return nil
181 }
182}
183
184// fill is the PlayerCallbacks.Fill function called by purego-pipewire
185// when it needs audio data. It pulls from the internal buffer channel.
186func (a *playerStreamAdapter) fill(pcm *pwpipewire.PCMBuffer) (int, error) {
187 select {
188 case samples := <-a.buf:
189 fillNum := a.fillCount.Add(1)
190 if fillNum == 1 && a.ctx != nil {
191 logging.FromContext(a.ctx).Info().
192 Msg("pipewire: first Fill callback with data")
193 }
194 // Copy available samples into the PCM buffer
195 copied := copyToPCM(pcm, samples)
196 return copied, nil
197 default:
198 // No data available — return 0 frames, let underrun policy handle it
199 underruns := a.underrunCount.Add(1)
200 if underruns == 1 && a.ctx != nil {
201 logging.FromContext(a.ctx).Debug().
202 Msg("pipewire: first Fill underrun (no data available)")
203 }
204 return 0, nil
205 }
206}
207
208// copyToPCM copies planar samples into the PCMBuffer.
209// Returns the number of frames actually copied.
210func copyToPCM(pcm *pwpipewire.PCMBuffer, samples [][]float32) int {
211 if pcm == nil || len(samples) == 0 {
212 return 0
213 }
214
215 frames := pcm.Frames
216 channels := len(samples)
217 if channels > pcm.Channels {
218 channels = pcm.Channels
219 }
220 if channels > len(pcm.Samples) {
221 channels = len(pcm.Samples)
222 }
223 if channels == 0 {
224 return 0
225 }
226
227 minFrames := frames
228
229 for ch := 0; ch < channels && ch < len(samples); ch++ {
230 dst := pcm.Samples[ch]
231 src := samples[ch]
232 if len(dst) == 0 || len(src) == 0 {
233 minFrames = 0
234 continue
235 }
236 srcFrames := len(src)
237 n := frames
238 if srcFrames < n {
239 n = srcFrames
240 }
241 if len(dst) < n {
242 n = len(dst)
243 }
244 if n < minFrames {
245 minFrames = n
246 }
247 copy(dst[:n], src[:n])
248 }
249
250 return minFrames
251}
252
253// Close stops the player and releases resources.
254// Safe to call multiple times.
255func (a *playerStreamAdapter) Close() error {
256 var err error
257 a.closeOnce.Do(func() {
258 if a.ctx != nil {
259 logging.FromContext(a.ctx).Info().
260 Uint64("fills", a.fillCount.Load()).
261 Uint64("underruns", a.underrunCount.Load()).
262 Uint64("drops", a.dropCount.Load()).
263 Msg("pipewire: closing player")
264 }
265
266 // Signal writers to stop
267 close(a.closed)
268
269 if a.player != nil {
270 // Stop playback then close the player
271 if stopErr := a.player.Stop(); stopErr != nil {
272 err = stopErr
273 }
274 if closeErr := a.player.Close(); closeErr != nil && err == nil {
275 err = closeErr
276 }
277 }
278 })
279 return err
280}