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
18var (
19 _ port.AudioOutputFactory = (*Factory)(nil)
20 _ port.AudioOutputStream = (*playerStreamAdapter)(nil)
21)
22
23// ErrInvalidFormat is returned when the audio stream format is invalid.
24var ErrInvalidFormat = errors.New("invalid audio stream format")
25
26// ErrStreamClosed is returned when writing to a closed stream.
27var ErrStreamClosed = errors.New("audio stream closed")
28
29const packetQueueCapacity = 4
30
31// playerCreator is a function type that creates a Player from config and callbacks.
32// This exists to allow dependency injection in tests.
33type playerCreator func(config pwpipewire.PlayerConfig, callbacks pwpipewire.PlayerCallbacks) (pwpipewire.Player, error)
34
35// Factory creates PipeWire audio output streams.
36// Each call to NewStream creates an independent purego-pipewire Player.
37type Factory struct {
38 createPlayer playerCreator
39}
40
41// NewFactory creates a new PipeWire audio factory.
42func NewFactory() (*Factory, error) {
43 return &Factory{createPlayer: pwpipewire.NewPlayer}, nil
44}
45
46// NewStream creates a push-to-pull adapter for a PipeWire player.
47func (f *Factory) NewStream(ctx context.Context, format port.AudioStreamFormat) (port.AudioOutputStream, error) {
48 if ctx.Err() != nil {
49 return nil, ctx.Err()
50 }
51 if err := validateFormat(format); err != nil {
52 return nil, err
53 }
54
55 log := logging.FromContext(ctx)
56 log.Info().
57 Int("sample_rate", format.SampleRate).
58 Int("channels", format.ChannelCount).
59 Int("frames_per_buffer", format.FramesPerBuffer).
60 Msg("pipewire: creating player")
61
62 // Every slot owns storage sized from the negotiated CEF format. Write copies
63 // directly into an available slot; Fill copies that same slot into PipeWire.
64 // Callback slices are therefore never queued or retained.
65 adapter := newPlayerStreamAdapter(ctx, format.ChannelCount, format.FramesPerBuffer)
66 config := pwpipewire.PlayerConfig{
67 SampleRate: format.SampleRate, Channels: format.ChannelCount, FramesPerBuffer: format.FramesPerBuffer,
68 SampleFormat: pwpipewire.SampleFormatF32, UnderrunPolicy: pwpipewire.UnderrunFillSilence,
69 }
70 player, err := f.createPlayer(config, pwpipewire.PlayerCallbacks{Fill: adapter.fill})
71 if err != nil {
72 log.Warn().Err(err).Msg("pipewire: failed to create player")
73 return nil, fmt.Errorf("failed to create pipewire player: %w", err)
74 }
75 adapter.player = player
76 if err := player.Start(); err != nil {
77 _ = player.Close()
78 log.Warn().Err(err).Msg("pipewire: failed to start player")
79 return nil, fmt.Errorf("failed to start pipewire player: %w", err)
80 }
81 log.Info().Msg("pipewire: player started successfully")
82 return adapter, nil
83}
84
85func validateFormat(format port.AudioStreamFormat) error {
86 if format.SampleRate < 8000 || format.SampleRate > 384000 {
87 return fmt.Errorf("%w: sample rate %d Hz out of range [8000, 384kHz]", ErrInvalidFormat, format.SampleRate)
88 }
89 if format.ChannelCount < 1 || format.ChannelCount > 1024 {
90 return fmt.Errorf("%w: channel count %d out of range [1, 1024]", ErrInvalidFormat, format.ChannelCount)
91 }
92 if format.FramesPerBuffer < 1 || format.FramesPerBuffer > 8192 {
93 return fmt.Errorf("%w: frames per buffer %d out of range [1, 8192]", ErrInvalidFormat, format.FramesPerBuffer)
94 }
95 return nil
96}
97
98// audioPacket is a reusable, adapter-owned packet slot. Its backing arrays are
99// allocated only when the stream is created and never alias CEF callback data.
100type audioPacket struct {
101 samples [][]float32
102 channels int
103 frames int
104}
105
106// playerStreamAdapter bridges callback-owned CEF audio to PipeWire's Fill
107// callback. available and ready form a bounded reusable packet ring.
108type playerStreamAdapter struct {
109 player pwpipewire.Player
110 ctx context.Context
111
112 available chan *audioPacket
113 ready chan *audioPacket
114 // TryLock serializes only simultaneous producers. The CEF callback never
115 // waits: contention and saturation both drop the packet. Fill never takes it.
116 writeMu sync.Mutex
117 closed atomic.Bool
118
119 closeOnce sync.Once
120
121 fillCount atomic.Uint64
122 underrunCount atomic.Uint64
123 dropCount atomic.Uint64
124}
125
126func newPlayerStreamAdapter(ctx context.Context, channels, frames int) *playerStreamAdapter {
127 a := &playerStreamAdapter{
128 ctx: ctx, available: make(chan *audioPacket, packetQueueCapacity), ready: make(chan *audioPacket, packetQueueCapacity),
129 }
130 for range packetQueueCapacity {
131 packet := &audioPacket{samples: make([][]float32, channels)}
132 for ch := range packet.samples {
133 packet.samples[ch] = make([]float32, frames)
134 }
135 a.available <- packet
136 }
137 return a
138}
139
140// Write makes exactly one owned copy for each accepted packet. It never blocks:
141// full queues or another producer cause a silent drop, preserving CEF timing.
142// The supplied slices are callback-owned and are not retained after return.
143func (a *playerStreamAdapter) Write(samples [][]float32) error {
144 if a.closed.Load() {
145 return ErrStreamClosed
146 }
147 if !a.writeMu.TryLock() {
148 a.recordDrop()
149 return nil
150 }
151 defer a.writeMu.Unlock()
152 if a.closed.Load() {
153 return ErrStreamClosed
154 }
155
156 select {
157 case packet := <-a.available:
158 copyPacket(packet, samples)
159 // A close racing after the first check must not publish a packet to a
160 // stopped player. Returning the slot keeps the ring bounded.
161 if a.closed.Load() {
162 a.available <- packet
163 return ErrStreamClosed
164 }
165 select {
166 case a.ready <- packet:
167 return nil
168 default:
169 // This cannot occur while ring invariants hold, but keep Write
170 // non-blocking if an implementation change violates them.
171 a.available <- packet
172 a.recordDrop()
173 return nil
174 }
175 default:
176 a.recordDrop()
177 return nil
178 }
179}
180
181func copyPacket(packet *audioPacket, samples [][]float32) {
182 packet.channels = min(len(packet.samples), len(samples))
183 packet.frames = 0
184 if packet.channels == 0 {
185 return
186 }
187 frames := len(packet.samples[0])
188 for ch := 0; ch < packet.channels; ch++ {
189 n := min(len(packet.samples[ch]), len(samples[ch]))
190 if n < frames {
191 frames = n
192 }
193 }
194 packet.frames = frames
195 for ch := 0; ch < packet.channels; ch++ {
196 copy(packet.samples[ch][:frames], samples[ch][:frames])
197 }
198}
199
200func (a *playerStreamAdapter) recordDrop() {
201 // Write runs on CEF's real-time callback thread. Diagnostics must stay
202 // allocation-free and must not synchronously emit logs from that path.
203 a.dropCount.Add(1)
204}
205
206// fill is the PlayerCallbacks.Fill function. It performs the only post-handoff
207// full-packet copy, directly into PipeWire's destination buffer.
208func (a *playerStreamAdapter) fill(pcm *pwpipewire.PCMBuffer) (int, error) {
209 select {
210 case packet := <-a.ready:
211 fillNum := a.fillCount.Add(1)
212 if fillNum == 1 && a.ctx != nil {
213 logging.FromContext(a.ctx).Info().Msg("pipewire: first Fill callback with data")
214 }
215 copied := copyToPCM(pcm, packet.samples[:packet.channels], packet.frames)
216 a.available <- packet
217 return copied, nil
218 default:
219 underruns := a.underrunCount.Add(1)
220 if underruns == 1 && a.ctx != nil {
221 logging.FromContext(a.ctx).Debug().Msg("pipewire: first Fill underrun (no data available)")
222 }
223 return 0, nil
224 }
225}
226
227func copyToPCM(pcm *pwpipewire.PCMBuffer, samples [][]float32, packetFrames int) int {
228 if pcm == nil || len(samples) == 0 || packetFrames == 0 {
229 return 0
230 }
231 channels := min(min(len(samples), pcm.Channels), len(pcm.Samples))
232 if channels == 0 {
233 return 0
234 }
235 frames := min(pcm.Frames, packetFrames)
236 for ch := 0; ch < channels; ch++ {
237 frames = min(frames, min(len(pcm.Samples[ch]), len(samples[ch])))
238 }
239 for ch := 0; ch < channels; ch++ {
240 copy(pcm.Samples[ch][:frames], samples[ch][:frames])
241 }
242 return frames
243}
244
245// Close stops the player. It is safe with Write: Write observes closed before
246// publishing, and no channel is closed while a callback may be selecting on it.
247func (a *playerStreamAdapter) Close() error {
248 var err error
249 a.closeOnce.Do(func() {
250 a.closed.Store(true)
251 if a.ctx != nil {
252 logging.FromContext(a.ctx).Info().
253 Uint64("fills", a.fillCount.Load()).
254 Uint64("underruns", a.underrunCount.Load()).
255 Uint64("drops", a.dropCount.Load()).
256 Msg("pipewire: closing player")
257 }
258 if a.player != nil {
259 if stopErr := a.player.Stop(); stopErr != nil {
260 err = stopErr
261 }
262 if closeErr := a.player.Close(); closeErr != nil && err == nil {
263 err = closeErr
264 }
265 }
266 })
267 return err
268}