detached b6a9d2e

perf(audio): bound CEF packet handoff

commit
b6a9d2ed88b0f507057571efa922b603dc08ddf6
author
brice · · 2026-07-14 15:41 UTC
committer
brice · · 2026-07-14 17:20 UTC
parent
45b5ee0
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
6 files · +306 -676
Unified
UnifiedSplit
M internal/application/port/audio_output.go +4 -3
@@ -18,9 +18,10 @@ type AudioStreamFormat struct {
1818 // AudioOutputStream represents an active audio output stream.
1919 // Implementations handle the low-level audio playback.
2020 type AudioOutputStream interface {
21- // Write sends audio samples to the output device.
22- // Samples are provided as a 2D slice: [channel][frame]float32.
23- // This matches CEF's planar float32 audio format.
21+ // Write sends callback-owned audio samples to the output device.
22+ // Samples are provided as [channel][frame]float32 matching CEF's planar
23+ // format. Implementations must not retain the slices and must be safe to
24+ // call concurrently with Close; audio callbacks must not block.
2425 Write(samples [][]float32) error
2526 // Close releases the audio stream and associated resources.
2627 Close() error
@@ -18,9 +18,10 @@ type AudioStreamFormat struct {
18 // AudioOutputStream represents an active audio output stream.18 // AudioOutputStream represents an active audio output stream.
19 // Implementations handle the low-level audio playback.19 // Implementations handle the low-level audio playback.
20 type AudioOutputStream interface {20 type AudioOutputStream interface {
21- // Write sends audio samples to the output device.21+ // Write sends callback-owned audio samples to the output device.
22- // Samples are provided as a 2D slice: [channel][frame]float32.22+ // Samples are provided as [channel][frame]float32 matching CEF's planar
23- // This matches CEF's planar float32 audio format.23+ // format. Implementations must not retain the slices and must be safe to
24+ // call concurrently with Close; audio callbacks must not block.
24 Write(samples [][]float32) error25 Write(samples [][]float32) error
25 // Close releases the audio stream and associated resources.26 // Close releases the audio stream and associated resources.
26 Close() error27 Close() error
M internal/infrastructure/audio/pipewire/output.go +124 -123
@@ -15,7 +15,6 @@ import (
1515 "github.com/bnema/dumber/internal/logging"
1616 )
1717
18-// Compile-time interface checks.
1918 var (
2019 _ port.AudioOutputFactory = (*Factory)(nil)
2120 _ port.AudioOutputStream = (*playerStreamAdapter)(nil)
@@ -27,6 +26,8 @@ var ErrInvalidFormat = errors.New("invalid audio stream format")
2726 // ErrStreamClosed is returned when writing to a closed stream.
2827 var ErrStreamClosed = errors.New("audio stream closed")
2928
29+const packetQueueCapacity = 4
30+
3031 // playerCreator is a function type that creates a Player from config and callbacks.
3132 // This exists to allow dependency injection in tests.
3233 type playerCreator func(config pwpipewire.PlayerConfig, callbacks pwpipewire.PlayerCallbacks) (pwpipewire.Player, error)
@@ -38,24 +39,15 @@ type Factory struct {
3839 }
3940
4041 // NewFactory creates a new PipeWire audio factory.
41-// The factory is stateless; players are created per-stream.
4242 func NewFactory() (*Factory, error) {
43- return &Factory{
44- createPlayer: pwpipewire.NewPlayer,
45- }, nil
43+ return &Factory{createPlayer: pwpipewire.NewPlayer}, nil
4644 }
4745
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.
46+// NewStream creates a push-to-pull adapter for a PipeWire player.
5247 func (f *Factory) NewStream(ctx context.Context, format port.AudioStreamFormat) (port.AudioOutputStream, error) {
53- // Check context cancellation
5448 if ctx.Err() != nil {
5549 return nil, ctx.Err()
5650 }
57-
58- // Validate format
5951 if err := validateFormat(format); err != nil {
6052 return nil, err
6153 }
@@ -67,182 +59,196 @@ func (f *Factory) NewStream(ctx context.Context, format port.AudioStreamFormat)
6759 Int("frames_per_buffer", format.FramesPerBuffer).
6860 Msg("pipewire: creating player")
6961
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-
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)
7866 config := pwpipewire.PlayerConfig{
79- SampleRate: format.SampleRate,
80- Channels: format.ChannelCount,
81- FramesPerBuffer: format.FramesPerBuffer,
82- SampleFormat: pwpipewire.SampleFormatF32,
83- UnderrunPolicy: pwpipewire.UnderrunFillSilence,
67+ SampleRate: format.SampleRate, Channels: format.ChannelCount, FramesPerBuffer: format.FramesPerBuffer,
68+ SampleFormat: pwpipewire.SampleFormatF32, UnderrunPolicy: pwpipewire.UnderrunFillSilence,
8469 }
85-
86- callbacks := pwpipewire.PlayerCallbacks{
87- Fill: adapter.fill,
88- }
89-
90- player, err := f.createPlayer(config, callbacks)
70+ player, err := f.createPlayer(config, pwpipewire.PlayerCallbacks{Fill: adapter.fill})
9171 if err != nil {
9272 log.Warn().Err(err).Msg("pipewire: failed to create player")
9373 return nil, fmt.Errorf("failed to create pipewire player: %w", err)
9474 }
9575 adapter.player = player
96-
97- // Start the player so the Fill callback begins being called.
9876 if err := player.Start(); err != nil {
9977 _ = player.Close()
10078 log.Warn().Err(err).Msg("pipewire: failed to start player")
10179 return nil, fmt.Errorf("failed to start pipewire player: %w", err)
10280 }
103-
10481 log.Info().Msg("pipewire: player started successfully")
105-
10682 return adapter, nil
10783 }
10884
109-// validateFormat checks if the audio format parameters are valid.
11085 func validateFormat(format port.AudioStreamFormat) error {
111- // Sample rate must be reasonable (8kHz to 384kHz)
11286 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)
87+ return fmt.Errorf("%w: sample rate %d Hz out of range [8000, 384kHz]", ErrInvalidFormat, format.SampleRate)
11588 }
116-
117- // Channel count must be 1 to 1024 (reasonable upper limit)
11889 if format.ChannelCount < 1 || format.ChannelCount > 1024 {
119- return fmt.Errorf("%w: channel count %d out of range [1, 1024]",
120- ErrInvalidFormat, format.ChannelCount)
90+ return fmt.Errorf("%w: channel count %d out of range [1, 1024]", ErrInvalidFormat, format.ChannelCount)
12191 }
122-
123- // Frames per buffer must be positive and reasonable
12492 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)
93+ return fmt.Errorf("%w: frames per buffer %d out of range [1, 8192]", ErrInvalidFormat, format.FramesPerBuffer)
12794 }
128-
12995 return nil
13096 }
13197
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.
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.
100+type 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.
138108 type playerStreamAdapter struct {
139109 player pwpipewire.Player
140110 ctx context.Context
141- buf chan [][]float32 // push→pull handoff channel
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
142118
143119 closeOnce sync.Once
144- closed chan struct{} // closed when Close is called
145120
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)
121+ fillCount atomic.Uint64
122+ underrunCount atomic.Uint64
123+ dropCount atomic.Uint64
150124 }
151125
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.
126+func 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.
157143 func (a *playerStreamAdapter) Write(samples [][]float32) error {
158- // Fast path: already closed.
159- select {
160- case <-a.closed:
144+ if a.closed.Load() {
161145 return ErrStreamClosed
162- default:
163146 }
164- // Non-blocking send: drop packet when buffer is full.
165- select {
166- case a.buf <- samples:
147+ if !a.writeMu.TryLock() {
148+ a.recordDrop()
167149 return nil
168- case <-a.closed:
150+ }
151+ defer a.writeMu.Unlock()
152+ if a.closed.Load() {
169153 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- }
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
179164 }
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()
180177 return nil
181178 }
182179 }
183180
184-// fill is the PlayerCallbacks.Fill function called by purego-pipewire
185-// when it needs audio data. It pulls from the internal buffer channel.
181+func 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+
200+func (a *playerStreamAdapter) recordDrop() {
201+ n := a.dropCount.Add(1)
202+ if (n == 1 || n&(n-1) == 0) && a.ctx != nil {
203+ logging.FromContext(a.ctx).Warn().Uint64("total_drops", n).Msg("pipewire: dropped audio packet (buffer full)")
204+ }
205+}
206+
207+// fill is the PlayerCallbacks.Fill function. It performs the only post-handoff
208+// full-packet copy, directly into PipeWire's destination buffer.
186209 func (a *playerStreamAdapter) fill(pcm *pwpipewire.PCMBuffer) (int, error) {
187210 select {
188- case samples := <-a.buf:
211+ case packet := <-a.ready:
189212 fillNum := a.fillCount.Add(1)
190213 if fillNum == 1 && a.ctx != nil {
191- logging.FromContext(a.ctx).Info().
192- Msg("pipewire: first Fill callback with data")
214+ logging.FromContext(a.ctx).Info().Msg("pipewire: first Fill callback with data")
193215 }
194- // Copy available samples into the PCM buffer
195- copied := copyToPCM(pcm, samples)
216+ copied := copyToPCM(pcm, packet.samples[:packet.channels], packet.frames)
217+ a.available <- packet
196218 return copied, nil
197219 default:
198- // No data available — return 0 frames, let underrun policy handle it
199220 underruns := a.underrunCount.Add(1)
200221 if underruns == 1 && a.ctx != nil {
201- logging.FromContext(a.ctx).Debug().
202- Msg("pipewire: first Fill underrun (no data available)")
222+ logging.FromContext(a.ctx).Debug().Msg("pipewire: first Fill underrun (no data available)")
203223 }
204224 return 0, nil
205225 }
206226 }
207227
208-// copyToPCM copies planar samples into the PCMBuffer.
209-// Returns the number of frames actually copied.
210-func copyToPCM(pcm *pwpipewire.PCMBuffer, samples [][]float32) int {
211- if pcm == nil || len(samples) == 0 {
228+func copyToPCM(pcm *pwpipewire.PCMBuffer, samples [][]float32, packetFrames int) int {
229+ if pcm == nil || len(samples) == 0 || packetFrames == 0 {
212230 return 0
213231 }
214-
215- frames := pcm.Frames
216232 channels := min(min(len(samples), pcm.Channels), len(pcm.Samples))
217233 if channels == 0 {
218234 return 0
219235 }
220-
221- minFrames := frames
222-
223- for ch := 0; ch < channels && ch < len(samples); ch++ {
224- dst := pcm.Samples[ch]
225- src := samples[ch]
226- if len(dst) == 0 || len(src) == 0 {
227- minFrames = 0
228- continue
229- }
230- srcFrames := len(src)
231- n := min(len(dst), min(srcFrames, frames))
232- if n < minFrames {
233- minFrames = n
234- }
235- copy(dst[:n], src[:n])
236+ frames := min(pcm.Frames, packetFrames)
237+ for ch := 0; ch < channels; ch++ {
238+ frames = min(frames, min(len(pcm.Samples[ch]), len(samples[ch])))
236239 }
237-
238- return minFrames
240+ for ch := 0; ch < channels; ch++ {
241+ copy(pcm.Samples[ch][:frames], samples[ch][:frames])
242+ }
243+ return frames
239244 }
240245
241-// Close stops the player and releases resources.
242-// Safe to call multiple times.
246+// Close stops the player. It is safe with Write: Write observes closed before
247+// publishing, and no channel is closed while a callback may be selecting on it.
243248 func (a *playerStreamAdapter) Close() error {
244249 var err error
245250 a.closeOnce.Do(func() {
251+ a.closed.Store(true)
246252 if a.ctx != nil {
247253 logging.FromContext(a.ctx).Info().
248254 Uint64("fills", a.fillCount.Load()).
@@ -250,12 +256,7 @@ func (a *playerStreamAdapter) Close() error {
250256 Uint64("drops", a.dropCount.Load()).
251257 Msg("pipewire: closing player")
252258 }
253-
254- // Signal writers to stop
255- close(a.closed)
256-
257259 if a.player != nil {
258- // Stop playback then close the player
259260 if stopErr := a.player.Stop(); stopErr != nil {
260261 err = stopErr
261262 }
@@ -15,7 +15,6 @@ import (
15 "github.com/bnema/dumber/internal/logging"15 "github.com/bnema/dumber/internal/logging"
16 )16 )
17 17
18-// Compile-time interface checks.
19 var (18 var (
20 _ port.AudioOutputFactory = (*Factory)(nil)19 _ port.AudioOutputFactory = (*Factory)(nil)
21 _ port.AudioOutputStream = (*playerStreamAdapter)(nil)20 _ port.AudioOutputStream = (*playerStreamAdapter)(nil)
@@ -27,6 +26,8 @@ var ErrInvalidFormat = errors.New("invalid audio stream format")
27 // ErrStreamClosed is returned when writing to a closed stream.26 // ErrStreamClosed is returned when writing to a closed stream.
28 var ErrStreamClosed = errors.New("audio stream closed")27 var ErrStreamClosed = errors.New("audio stream closed")
29 28
29+const packetQueueCapacity = 4
30+
30 // playerCreator is a function type that creates a Player from config and callbacks.31 // playerCreator is a function type that creates a Player from config and callbacks.
31 // This exists to allow dependency injection in tests.32 // This exists to allow dependency injection in tests.
32 type playerCreator func(config pwpipewire.PlayerConfig, callbacks pwpipewire.PlayerCallbacks) (pwpipewire.Player, error)33 type playerCreator func(config pwpipewire.PlayerConfig, callbacks pwpipewire.PlayerCallbacks) (pwpipewire.Player, error)
@@ -38,24 +39,15 @@ type Factory struct {
38 }39 }
39 40
40 // NewFactory creates a new PipeWire audio factory.41 // NewFactory creates a new PipeWire audio factory.
41-// The factory is stateless; players are created per-stream.
42 func NewFactory() (*Factory, error) {42 func NewFactory() (*Factory, error) {
43- return &Factory{43+ return &Factory{createPlayer: pwpipewire.NewPlayer}, nil
44- createPlayer: pwpipewire.NewPlayer,
45- }, nil
46 }44 }
47 45
48-// NewStream creates a new audio output stream with the given format.46+// NewStream creates a push-to-pull adapter for a PipeWire player.
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.
52 func (f *Factory) NewStream(ctx context.Context, format port.AudioStreamFormat) (port.AudioOutputStream, error) {47 func (f *Factory) NewStream(ctx context.Context, format port.AudioStreamFormat) (port.AudioOutputStream, error) {
53- // Check context cancellation
54 if ctx.Err() != nil {48 if ctx.Err() != nil {
55 return nil, ctx.Err()49 return nil, ctx.Err()
56 }50 }
57-
58- // Validate format
59 if err := validateFormat(format); err != nil {51 if err := validateFormat(format); err != nil {
60 return nil, err52 return nil, err
61 }53 }
@@ -67,182 +59,196 @@ func (f *Factory) NewStream(ctx context.Context, format port.AudioStreamFormat)
67 Int("frames_per_buffer", format.FramesPerBuffer).59 Int("frames_per_buffer", format.FramesPerBuffer).
68 Msg("pipewire: creating player")60 Msg("pipewire: creating player")
69 61
70- // Create the adapter with its ring buffer before creating the player,62+ // Every slot owns storage sized from the negotiated CEF format. Write copies
71- // because the Fill callback captures the adapter.63+ // directly into an available slot; Fill copies that same slot into PipeWire.
72- adapter := &playerStreamAdapter{64+ // Callback slices are therefore never queued or retained.
73- ctx: ctx,65+ adapter := newPlayerStreamAdapter(ctx, format.ChannelCount, format.FramesPerBuffer)
74- buf: make(chan [][]float32, 4), // small buffer for push→pull bridging
75- closed: make(chan struct{}),
76- }
77-
78 config := pwpipewire.PlayerConfig{66 config := pwpipewire.PlayerConfig{
79- SampleRate: format.SampleRate,67+ SampleRate: format.SampleRate, Channels: format.ChannelCount, FramesPerBuffer: format.FramesPerBuffer,
80- Channels: format.ChannelCount,68+ SampleFormat: pwpipewire.SampleFormatF32, UnderrunPolicy: pwpipewire.UnderrunFillSilence,
81- FramesPerBuffer: format.FramesPerBuffer,
82- SampleFormat: pwpipewire.SampleFormatF32,
83- UnderrunPolicy: pwpipewire.UnderrunFillSilence,
84 }69 }
85-70+ player, err := f.createPlayer(config, pwpipewire.PlayerCallbacks{Fill: adapter.fill})
86- callbacks := pwpipewire.PlayerCallbacks{
87- Fill: adapter.fill,
88- }
89-
90- player, err := f.createPlayer(config, callbacks)
91 if err != nil {71 if err != nil {
92 log.Warn().Err(err).Msg("pipewire: failed to create player")72 log.Warn().Err(err).Msg("pipewire: failed to create player")
93 return nil, fmt.Errorf("failed to create pipewire player: %w", err)73 return nil, fmt.Errorf("failed to create pipewire player: %w", err)
94 }74 }
95 adapter.player = player75 adapter.player = player
96-
97- // Start the player so the Fill callback begins being called.
98 if err := player.Start(); err != nil {76 if err := player.Start(); err != nil {
99 _ = player.Close()77 _ = player.Close()
100 log.Warn().Err(err).Msg("pipewire: failed to start player")78 log.Warn().Err(err).Msg("pipewire: failed to start player")
101 return nil, fmt.Errorf("failed to start pipewire player: %w", err)79 return nil, fmt.Errorf("failed to start pipewire player: %w", err)
102 }80 }
103-
104 log.Info().Msg("pipewire: player started successfully")81 log.Info().Msg("pipewire: player started successfully")
105-
106 return adapter, nil82 return adapter, nil
107 }83 }
108 84
109-// validateFormat checks if the audio format parameters are valid.
110 func validateFormat(format port.AudioStreamFormat) error {85 func validateFormat(format port.AudioStreamFormat) error {
111- // Sample rate must be reasonable (8kHz to 384kHz)
112 if format.SampleRate < 8000 || format.SampleRate > 384000 {86 if format.SampleRate < 8000 || format.SampleRate > 384000 {
113- return fmt.Errorf("%w: sample rate %d Hz out of range [8000, 384000]",87+ return fmt.Errorf("%w: sample rate %d Hz out of range [8000, 384kHz]", ErrInvalidFormat, format.SampleRate)
114- ErrInvalidFormat, format.SampleRate)
115 }88 }
116-
117- // Channel count must be 1 to 1024 (reasonable upper limit)
118 if format.ChannelCount < 1 || format.ChannelCount > 1024 {89 if format.ChannelCount < 1 || format.ChannelCount > 1024 {
119- return fmt.Errorf("%w: channel count %d out of range [1, 1024]",90+ return fmt.Errorf("%w: channel count %d out of range [1, 1024]", ErrInvalidFormat, format.ChannelCount)
120- ErrInvalidFormat, format.ChannelCount)
121 }91 }
122-
123- // Frames per buffer must be positive and reasonable
124 if format.FramesPerBuffer < 1 || format.FramesPerBuffer > 8192 {92 if format.FramesPerBuffer < 1 || format.FramesPerBuffer > 8192 {
125- return fmt.Errorf("%w: frames per buffer %d out of range [1, 8192]",93+ return fmt.Errorf("%w: frames per buffer %d out of range [1, 8192]", ErrInvalidFormat, format.FramesPerBuffer)
126- ErrInvalidFormat, format.FramesPerBuffer)
127 }94 }
128-
129 return nil95 return nil
130 }96 }
131 97
132-// playerStreamAdapter bridges Dumber's push-style Write(samples) interface98+// audioPacket is a reusable, adapter-owned packet slot. Its backing arrays are
133-// to purego-pipewire's pull-style Fill callback.99+// allocated only when the stream is created and never alias CEF callback data.
134-//100+type audioPacket struct {
135-// Write() sends sample buffers into a channel. The Fill callback drains them101+ samples [][]float32
136-// into the PCMBuffer that purego-pipewire passes in. If no data is available,102+ channels int
137-// the player's UnderrunFillSilence policy handles it.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.
138 type playerStreamAdapter struct {108 type playerStreamAdapter struct {
139 player pwpipewire.Player109 player pwpipewire.Player
140 ctx context.Context110 ctx context.Context
141- buf chan [][]float32 // push→pull handoff channel111+
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
142 118
143 closeOnce sync.Once119 closeOnce sync.Once
144- closed chan struct{} // closed when Close is called
145 120
146- // Diagnostic counters (atomic, no mutex needed).121+ fillCount atomic.Uint64
147- fillCount atomic.Uint64 // total Fill callbacks122+ underrunCount atomic.Uint64
148- underrunCount atomic.Uint64 // Fill callbacks with no data available123+ dropCount atomic.Uint64
149- dropCount atomic.Uint64 // Write calls that dropped data (buffer full)
150 }124 }
151 125
152-// Write sends audio samples to the PipeWire output.126+func newPlayerStreamAdapter(ctx context.Context, channels, frames int) *playerStreamAdapter {
153-// Samples are provided as [channel][frame]float32 matching CEF's planar format.127+ a := &playerStreamAdapter{
154-// If the internal buffer is full the packet is dropped silently so the caller128+ ctx: ctx, available: make(chan *audioPacket, packetQueueCapacity), ready: make(chan *audioPacket, packetQueueCapacity),
155-// (CEF audio callback thread) is never blocked.129+ }
156-// Safe for concurrent use with Close.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.
157 func (a *playerStreamAdapter) Write(samples [][]float32) error {143 func (a *playerStreamAdapter) Write(samples [][]float32) error {
158- // Fast path: already closed.144+ if a.closed.Load() {
159- select {
160- case <-a.closed:
161 return ErrStreamClosed145 return ErrStreamClosed
162- default:
163 }146 }
164- // Non-blocking send: drop packet when buffer is full.147+ if !a.writeMu.TryLock() {
165- select {148+ a.recordDrop()
166- case a.buf <- samples:
167 return nil149 return nil
168- case <-a.closed:150+ }
151+ defer a.writeMu.Unlock()
152+ if a.closed.Load() {
169 return ErrStreamClosed153 return ErrStreamClosed
170- default:154+ }
171- // Buffer full — drop to avoid blocking the CEF audio thread.155+
172- n := a.dropCount.Add(1)156+ select {
173- if n == 1 || n&(n-1) == 0 { // first drop, then powers of two157+ case packet := <-a.available:
174- if a.ctx != nil {158+ copyPacket(packet, samples)
175- logging.FromContext(a.ctx).Warn().159+ // A close racing after the first check must not publish a packet to a
176- Uint64("total_drops", n).160+ // stopped player. Returning the slot keeps the ring bounded.
177- Msg("pipewire: dropped audio packet (buffer full)")161+ if a.closed.Load() {
178- }162+ a.available <- packet
163+ return ErrStreamClosed
179 }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()
180 return nil177 return nil
181 }178 }
182 }179 }
183 180
184-// fill is the PlayerCallbacks.Fill function called by purego-pipewire181+func copyPacket(packet *audioPacket, samples [][]float32) {
185-// when it needs audio data. It pulls from the internal buffer channel.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+
200+func (a *playerStreamAdapter) recordDrop() {
201+ n := a.dropCount.Add(1)
202+ if (n == 1 || n&(n-1) == 0) && a.ctx != nil {
203+ logging.FromContext(a.ctx).Warn().Uint64("total_drops", n).Msg("pipewire: dropped audio packet (buffer full)")
204+ }
205+}
206+
207+// fill is the PlayerCallbacks.Fill function. It performs the only post-handoff
208+// full-packet copy, directly into PipeWire's destination buffer.
186 func (a *playerStreamAdapter) fill(pcm *pwpipewire.PCMBuffer) (int, error) {209 func (a *playerStreamAdapter) fill(pcm *pwpipewire.PCMBuffer) (int, error) {
187 select {210 select {
188- case samples := <-a.buf:211+ case packet := <-a.ready:
189 fillNum := a.fillCount.Add(1)212 fillNum := a.fillCount.Add(1)
190 if fillNum == 1 && a.ctx != nil {213 if fillNum == 1 && a.ctx != nil {
191- logging.FromContext(a.ctx).Info().214+ logging.FromContext(a.ctx).Info().Msg("pipewire: first Fill callback with data")
192- Msg("pipewire: first Fill callback with data")
193 }215 }
194- // Copy available samples into the PCM buffer216+ copied := copyToPCM(pcm, packet.samples[:packet.channels], packet.frames)
195- copied := copyToPCM(pcm, samples)217+ a.available <- packet
196 return copied, nil218 return copied, nil
197 default:219 default:
198- // No data available — return 0 frames, let underrun policy handle it
199 underruns := a.underrunCount.Add(1)220 underruns := a.underrunCount.Add(1)
200 if underruns == 1 && a.ctx != nil {221 if underruns == 1 && a.ctx != nil {
201- logging.FromContext(a.ctx).Debug().222+ logging.FromContext(a.ctx).Debug().Msg("pipewire: first Fill underrun (no data available)")
202- Msg("pipewire: first Fill underrun (no data available)")
203 }223 }
204 return 0, nil224 return 0, nil
205 }225 }
206 }226 }
207 227
208-// copyToPCM copies planar samples into the PCMBuffer.228+func copyToPCM(pcm *pwpipewire.PCMBuffer, samples [][]float32, packetFrames int) int {
209-// Returns the number of frames actually copied.229+ if pcm == nil || len(samples) == 0 || packetFrames == 0 {
210-func copyToPCM(pcm *pwpipewire.PCMBuffer, samples [][]float32) int {
211- if pcm == nil || len(samples) == 0 {
212 return 0230 return 0
213 }231 }
214-
215- frames := pcm.Frames
216 channels := min(min(len(samples), pcm.Channels), len(pcm.Samples))232 channels := min(min(len(samples), pcm.Channels), len(pcm.Samples))
217 if channels == 0 {233 if channels == 0 {
218 return 0234 return 0
219 }235 }
220-236+ frames := min(pcm.Frames, packetFrames)
221- minFrames := frames237+ for ch := 0; ch < channels; ch++ {
222-238+ frames = min(frames, min(len(pcm.Samples[ch]), len(samples[ch])))
223- for ch := 0; ch < channels && ch < len(samples); ch++ {
224- dst := pcm.Samples[ch]
225- src := samples[ch]
226- if len(dst) == 0 || len(src) == 0 {
227- minFrames = 0
228- continue
229- }
230- srcFrames := len(src)
231- n := min(len(dst), min(srcFrames, frames))
232- if n < minFrames {
233- minFrames = n
234- }
235- copy(dst[:n], src[:n])
236 }239 }
237-240+ for ch := 0; ch < channels; ch++ {
238- return minFrames241+ copy(pcm.Samples[ch][:frames], samples[ch][:frames])
242+ }
243+ return frames
239 }244 }
240 245
241-// Close stops the player and releases resources.246+// Close stops the player. It is safe with Write: Write observes closed before
242-// Safe to call multiple times.247+// publishing, and no channel is closed while a callback may be selecting on it.
243 func (a *playerStreamAdapter) Close() error {248 func (a *playerStreamAdapter) Close() error {
244 var err error249 var err error
245 a.closeOnce.Do(func() {250 a.closeOnce.Do(func() {
251+ a.closed.Store(true)
246 if a.ctx != nil {252 if a.ctx != nil {
247 logging.FromContext(a.ctx).Info().253 logging.FromContext(a.ctx).Info().
248 Uint64("fills", a.fillCount.Load()).254 Uint64("fills", a.fillCount.Load()).
@@ -250,12 +256,7 @@ func (a *playerStreamAdapter) Close() error {
250 Uint64("drops", a.dropCount.Load()).256 Uint64("drops", a.dropCount.Load()).
251 Msg("pipewire: closing player")257 Msg("pipewire: closing player")
252 }258 }
253-
254- // Signal writers to stop
255- close(a.closed)
256-
257 if a.player != nil {259 if a.player != nil {
258- // Stop playback then close the player
259 if stopErr := a.player.Stop(); stopErr != nil {260 if stopErr := a.player.Stop(); stopErr != nil {
260 err = stopErr261 err = stopErr
261 }262 }
M internal/infrastructure/audio/pipewire/output_test.go +139 -494
@@ -1,11 +1,11 @@
1-// Package pipewire provides a PipeWire-based audio output implementation.
2-// This wraps purego-pipewire's high-level Player API.
31 package pipewire
42
53 import (
64 "context"
75 "errors"
6+ "sync"
87 "testing"
8+ "time"
99
1010 "github.com/bnema/dumber/internal/application/port"
1111 pipewiremocks "github.com/bnema/dumber/internal/infrastructure/audio/pipewire/mocks"
@@ -16,548 +16,193 @@ import (
1616
1717 type playerRecorder struct {
1818 *pipewiremocks.MockPlayer
19-
20- started bool
21- paused bool
22- stopped bool
23- closedFlag bool
24- state pwpipewire.PlayerState
25-
26- startErr error
27- stopErr error
28- closeErr error
19+ started, stopped, closedFlag bool
20+ startErr, stopErr, closeErr error
2921 }
3022
3123 func newPlayerRecorder(t *testing.T) *playerRecorder {
3224 t.Helper()
33-
34- recorder := &playerRecorder{MockPlayer: pipewiremocks.NewMockPlayer(t)}
35- recorder.EXPECT().Start().RunAndReturn(recorder.start).Maybe()
36- recorder.EXPECT().Pause().RunAndReturn(recorder.pause).Maybe()
37- recorder.EXPECT().Stop().RunAndReturn(recorder.stop).Maybe()
38- recorder.EXPECT().Close().RunAndReturn(recorder.close).Maybe()
39- recorder.EXPECT().State().RunAndReturn(recorder.currentState).Maybe()
40- return recorder
41-}
42-
43-func (p *playerRecorder) start() error {
44- if p.startErr != nil {
45- return p.startErr
46- }
47- p.started = true
48- p.state = pwpipewire.PlayerStatePlaying
49- return nil
50-}
51-
52-func (p *playerRecorder) pause() error {
53- p.paused = true
54- p.state = pwpipewire.PlayerStatePaused
55- return nil
56-}
57-
58-func (p *playerRecorder) stop() error {
59- if p.stopErr != nil {
60- return p.stopErr
61- }
62- p.stopped = true
63- p.state = pwpipewire.PlayerStateStopped
64- return nil
25+ p := &playerRecorder{MockPlayer: pipewiremocks.NewMockPlayer(t)}
26+ p.EXPECT().Start().RunAndReturn(func() error {
27+ if p.startErr != nil {
28+ return p.startErr
29+ }
30+ p.started = true
31+ return nil
32+ }).Maybe()
33+ p.EXPECT().Stop().RunAndReturn(func() error {
34+ if p.stopErr != nil {
35+ return p.stopErr
36+ }
37+ p.stopped = true
38+ return nil
39+ }).Maybe()
40+ p.EXPECT().Close().RunAndReturn(func() error {
41+ if p.closeErr != nil {
42+ return p.closeErr
43+ }
44+ p.closedFlag = true
45+ return nil
46+ }).Maybe()
47+ return p
6548 }
6649
67-func (p *playerRecorder) close() error {
68- if p.closeErr != nil {
69- return p.closeErr
70- }
71- p.closedFlag = true
72- p.state = pwpipewire.PlayerStateClosed
73- return nil
50+func testAdapter(frames int) *playerStreamAdapter {
51+ return newPlayerStreamAdapter(context.Background(), 2, frames)
7452 }
7553
76-func (p *playerRecorder) currentState() pwpipewire.PlayerState {
77- return p.state
54+func stereoPCM(frames int) *pwpipewire.PCMBuffer {
55+ return &pwpipewire.PCMBuffer{Frames: frames, Channels: 2, Samples: [][]float32{make([]float32, frames), make([]float32, frames)}}
7856 }
7957
80-// --- Factory tests ---
81-
82-func TestFactory_NewStream_ValidFormat_CreatesStream(t *testing.T) {
83- var capturedConfig pwpipewire.PlayerConfig
58+func TestFactory_NewStreamValidatesAndStartsPlayer(t *testing.T) {
8459 player := newPlayerRecorder(t)
85-
86- factory := &Factory{
87- createPlayer: func(config pwpipewire.PlayerConfig, _ pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
88- capturedConfig = config
89- return player, nil
90- },
91- }
92-
93- format := port.AudioStreamFormat{
94- SampleRate: 48000,
95- ChannelCount: 2,
96- FramesPerBuffer: 512,
97- }
98-
99- stream, err := factory.NewStream(context.Background(), format)
60+ factory := &Factory{createPlayer: func(config pwpipewire.PlayerConfig, callbacks pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
61+ assert.Equal(t, 48000, config.SampleRate)
62+ assert.Equal(t, 2, config.Channels)
63+ assert.NotNil(t, callbacks.Fill)
64+ return player, nil
65+ }}
66+ stream, err := factory.NewStream(context.Background(), port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 512})
10067 require.NoError(t, err)
101- require.NotNil(t, stream)
102-
103- // Verify config mapping
104- assert.Equal(t, 48000, capturedConfig.SampleRate)
105- assert.Equal(t, 2, capturedConfig.Channels)
106- assert.Equal(t, 512, capturedConfig.FramesPerBuffer)
107- assert.Equal(t, pwpipewire.SampleFormatF32, capturedConfig.SampleFormat)
108- assert.Equal(t, pwpipewire.UnderrunFillSilence, capturedConfig.UnderrunPolicy)
109-
110- // Verify player was started
11168 assert.True(t, player.started)
112-
113- // Clean up
11469 require.NoError(t, stream.Close())
11570 }
11671
117-func TestFactory_NewStream_InvalidFormat_ReturnsError(t *testing.T) {
118- factory := &Factory{
119- createPlayer: func(_ pwpipewire.PlayerConfig, _ pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
120- t.Fatal("createPlayer should not be called for invalid format")
121- return nil, nil
122- },
123- }
124-
125- tests := []struct {
126- name string
127- format port.AudioStreamFormat
128- }{
129- {
130- name: "zero sample rate",
131- format: port.AudioStreamFormat{SampleRate: 0, ChannelCount: 2, FramesPerBuffer: 512},
132- },
133- {
134- name: "zero channels",
135- format: port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 0, FramesPerBuffer: 512},
136- },
137- {
138- name: "negative sample rate",
139- format: port.AudioStreamFormat{SampleRate: -48000, ChannelCount: 2, FramesPerBuffer: 512},
140- },
141- {
142- name: "negative channels",
143- format: port.AudioStreamFormat{SampleRate: 48000, ChannelCount: -2, FramesPerBuffer: 512},
144- },
145- {
146- name: "excessive sample rate",
147- format: port.AudioStreamFormat{SampleRate: 1000000, ChannelCount: 2, FramesPerBuffer: 512},
148- },
149- {
150- name: "excessive channels",
151- format: port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 1025, FramesPerBuffer: 512},
152- },
153- {
154- name: "zero frames per buffer",
155- format: port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 0},
156- },
157- {
158- name: "excessive frames per buffer",
159- format: port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 10000},
160- },
161- }
162-
163- for _, tt := range tests {
164- t.Run(tt.name, func(t *testing.T) {
165- stream, err := factory.NewStream(context.Background(), tt.format)
166- require.Error(t, err)
167- assert.Nil(t, stream)
168- assert.ErrorIs(t, err, ErrInvalidFormat)
169- })
170- }
171-}
172-
173-func TestFactory_NewStream_PlayerCreationFails_ReturnsError(t *testing.T) {
174- expectedErr := errors.New("pipewire not available")
175- factory := &Factory{
176- createPlayer: func(_ pwpipewire.PlayerConfig, _ pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
177- return nil, expectedErr
178- },
72+func TestFactory_NewStreamRejectsInvalidFormat(t *testing.T) {
73+ factory := &Factory{createPlayer: func(pwpipewire.PlayerConfig, pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
74+ t.Fatal("unexpected player")
75+ return nil, nil
76+ }}
77+ for _, format := range []port.AudioStreamFormat{
78+ {SampleRate: 0, ChannelCount: 2, FramesPerBuffer: 512},
79+ {SampleRate: 48000, ChannelCount: 0, FramesPerBuffer: 512},
80+ {SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 0},
81+ } {
82+ stream, err := factory.NewStream(context.Background(), format)
83+ require.ErrorIs(t, err, ErrInvalidFormat)
84+ assert.Nil(t, stream)
17985 }
180-
181- format := port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 512}
182- stream, err := factory.NewStream(context.Background(), format)
183-
184- require.Error(t, err)
185- assert.Nil(t, stream)
186- assert.ErrorIs(t, err, expectedErr)
18786 }
18887
189-func TestFactory_NewStream_PlayerStartFails_ClosesAndReturnsError(t *testing.T) {
88+func TestFactory_NewStreamClosesPlayerWhenStartFails(t *testing.T) {
19089 player := newPlayerRecorder(t)
19190 player.startErr = errors.New("start failed")
192- factory := &Factory{
193- createPlayer: func(_ pwpipewire.PlayerConfig, _ pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
194- return player, nil
195- },
196- }
197-
198- format := port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 512}
199- stream, err := factory.NewStream(context.Background(), format)
200-
201- require.Error(t, err)
202- assert.Nil(t, stream)
203- assert.True(t, player.closedFlag, "player should be closed after start failure")
204-}
205-
206-func TestFactory_NewStream_ContextCancelled_ReturnsError(t *testing.T) {
207- factory := &Factory{
208- createPlayer: func(_ pwpipewire.PlayerConfig, _ pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
209- t.Fatal("createPlayer should not be called when context is canceled")
210- return nil, nil
211- },
212- }
213-
214- ctx, cancel := context.WithCancel(context.Background())
215- cancel()
216-
217- format := port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 512}
218- stream, err := factory.NewStream(ctx, format)
219-
91+ factory := &Factory{createPlayer: func(pwpipewire.PlayerConfig, pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
92+ return player, nil
93+ }}
94+ stream, err := factory.NewStream(context.Background(), port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 512})
22095 require.Error(t, err)
22196 assert.Nil(t, stream)
222-}
223-
224-func TestFactory_NewStream_MultipleFormats_Accepted(t *testing.T) {
225- formats := []port.AudioStreamFormat{
226- {SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 512},
227- {SampleRate: 44100, ChannelCount: 2, FramesPerBuffer: 512},
228- {SampleRate: 48000, ChannelCount: 1, FramesPerBuffer: 256},
229- {SampleRate: 96000, ChannelCount: 2, FramesPerBuffer: 1024},
230- {SampleRate: 48000, ChannelCount: 6, FramesPerBuffer: 512},
231- }
232-
233- for _, fmt := range formats {
234- t.Run("", func(t *testing.T) {
235- factory := &Factory{
236- createPlayer: func(config pwpipewire.PlayerConfig, _ pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
237- assert.Equal(t, fmt.SampleRate, config.SampleRate)
238- assert.Equal(t, fmt.ChannelCount, config.Channels)
239- assert.Equal(t, fmt.FramesPerBuffer, config.FramesPerBuffer)
240- return newPlayerRecorder(t), nil
241- },
242- }
243-
244- stream, err := factory.NewStream(context.Background(), fmt)
245- require.NoError(t, err)
246- assert.NotNil(t, stream)
247-
248- _ = stream.Close()
249- })
250- }
251-}
252-
253-func TestFactory_ImplementsPortInterface(_ *testing.T) {
254- var _ port.AudioOutputFactory = (*Factory)(nil)
255-}
256-
257-// --- Stream adapter tests ---
258-
259-func TestPlayerStreamAdapter_Write_SendsToBuffer(t *testing.T) {
260- player := newPlayerRecorder(t)
261- adapter := &playerStreamAdapter{
262- player: player,
263- buf: make(chan [][]float32, 4),
264- closed: make(chan struct{}),
265- }
266-
267- samples := [][]float32{
268- {0.1, 0.2},
269- {0.3, 0.4},
270- }
271-
272- err := adapter.Write(samples)
273- require.NoError(t, err)
274-
275- // Verify data is in the channel
276- received := <-adapter.buf
277- assert.Equal(t, samples, received)
278-}
279-
280-func TestPlayerStreamAdapter_Write_AfterClose_ReturnsError(t *testing.T) {
281- player := newPlayerRecorder(t)
282- adapter := &playerStreamAdapter{
283- player: player,
284- buf: make(chan [][]float32, 4),
285- closed: make(chan struct{}),
286- }
287-
288- require.NoError(t, adapter.Close())
289-
290- err := adapter.Write([][]float32{{0.1}})
291- assert.ErrorIs(t, err, ErrStreamClosed)
292-}
293-
294-func TestPlayerStreamAdapter_Close_StopsAndClosesPlayer(t *testing.T) {
295- player := newPlayerRecorder(t)
296- adapter := &playerStreamAdapter{
297- player: player,
298- buf: make(chan [][]float32, 4),
299- closed: make(chan struct{}),
300- }
301-
302- err := adapter.Close()
303- require.NoError(t, err)
304-
305- assert.True(t, player.stopped)
30697 assert.True(t, player.closedFlag)
30798 }
30899
309-func TestPlayerStreamAdapter_Close_IsIdempotent(t *testing.T) {
310- player := newPlayerRecorder(t)
311- adapter := &playerStreamAdapter{
312- player: player,
313- buf: make(chan [][]float32, 4),
314- closed: make(chan struct{}),
315- }
316-
317- require.NoError(t, adapter.Close())
318- require.NoError(t, adapter.Close())
319-}
320-
321-func TestPlayerStreamAdapter_Close_PropagatesStopError(t *testing.T) {
322- player := newPlayerRecorder(t)
323- player.stopErr = errors.New("stop error")
324- adapter := &playerStreamAdapter{
325- player: player,
326- buf: make(chan [][]float32, 4),
327- closed: make(chan struct{}),
328- }
329-
330- err := adapter.Close()
331- require.Error(t, err)
332- assert.Contains(t, err.Error(), "stop error")
333-}
334-
335-func TestPlayerStreamAdapter_Close_PropagatesCloseError(t *testing.T) {
336- player := newPlayerRecorder(t)
337- player.closeErr = errors.New("close error")
338- adapter := &playerStreamAdapter{
339- player: player,
340- buf: make(chan [][]float32, 4),
341- closed: make(chan struct{}),
342- }
343-
344- err := adapter.Close()
345- require.Error(t, err)
346- assert.Contains(t, err.Error(), "close error")
347-}
348-
349-// --- Fill callback tests ---
350-
351-func TestFill_PullsFromBuffer(t *testing.T) {
352- adapter := &playerStreamAdapter{
353- buf: make(chan [][]float32, 4),
354- }
355-
356- // Push samples
357- samples := [][]float32{
358- {0.1, 0.2, 0.3},
359- {0.4, 0.5, 0.6},
360- }
361- adapter.buf <- samples
362-
363- // Simulate Fill call
364- pcm := &pwpipewire.PCMBuffer{
365- Frames: 3,
366- Channels: 2,
367- Stride: 1,
368- Samples: [][]float32{
369- make([]float32, 3),
370- make([]float32, 3),
371- },
372- }
373-
100+func TestPlayerStreamAdapter_FillGetsOwnedPacketAfterCallbackMutation(t *testing.T) {
101+ adapter := testAdapter(3)
102+ samples := [][]float32{{1, 2, 3}, {4, 5, 6}}
103+ require.NoError(t, adapter.Write(samples))
104+ samples[0][0], samples[1][0] = 99, 99
105+ pcm := stereoPCM(3)
374106 frames, err := adapter.fill(pcm)
375107 require.NoError(t, err)
376108 assert.Equal(t, 3, frames)
377- assert.InEpsilon(t, 0.1, pcm.Samples[0][0], 0.0001)
378- assert.InEpsilon(t, 0.4, pcm.Samples[1][0], 0.0001)
379-}
380-
381-func TestFill_ReturnsZeroWhenEmpty(t *testing.T) {
382- adapter := &playerStreamAdapter{
383- buf: make(chan [][]float32, 4),
384- }
385-
386- pcm := &pwpipewire.PCMBuffer{
387- Frames: 256,
388- Channels: 2,
389- Stride: 1,
390- Samples: [][]float32{
391- make([]float32, 256),
392- make([]float32, 256),
393- },
394- }
395-
396- frames, err := adapter.fill(pcm)
397- require.NoError(t, err)
398- assert.Equal(t, 0, frames)
399-}
400-
401-func TestCopyToPCM_HandlesChannelMismatch(t *testing.T) {
402- // More source channels than PCM channels — truncates
403- pcm := &pwpipewire.PCMBuffer{
404- Frames: 2,
405- Channels: 1,
406- Samples: [][]float32{make([]float32, 2)},
407- }
408- samples := [][]float32{
409- {1.0, 2.0},
410- {3.0, 4.0}, // should be ignored
411- }
412- frames := copyToPCM(pcm, samples)
413- assert.Equal(t, 2, frames)
414- assert.InEpsilon(t, 1.0, pcm.Samples[0][0], 0.0001)
109+ assert.InEpsilon(t, 1, pcm.Samples[0][0], 0.0001)
110+ assert.InEpsilon(t, 4, pcm.Samples[1][0], 0.0001)
415111 }
416112
417-func TestCopyToPCM_HandlesFrameMismatch(t *testing.T) {
418- // Fewer source frames than PCM expects
419- pcm := &pwpipewire.PCMBuffer{
420- Frames: 4,
421- Channels: 1,
422- Samples: [][]float32{make([]float32, 4)},
113+func TestPlayerStreamAdapter_SaturationDropsWithoutBlocking(t *testing.T) {
114+ adapter := testAdapter(2)
115+ for i := 0; i < packetQueueCapacity; i++ {
116+ require.NoError(t, adapter.Write([][]float32{{float32(i)}, {float32(i)}}))
423117 }
424- samples := [][]float32{
425- {1.0, 2.0}, // only 2 frames
118+ done := make(chan error, 1)
119+ go func() { done <- adapter.Write([][]float32{{99}, {99}}) }()
120+ select {
121+ case err := <-done:
122+ require.NoError(t, err)
123+ case <-time.After(time.Second):
124+ t.Fatal("Write blocked while the bounded queue was saturated")
426125 }
427- frames := copyToPCM(pcm, samples)
428- assert.Equal(t, 2, frames) // reports partial fill
429-}
430-
431-func TestCopyToPCM_NilPCM_ReturnsZero(t *testing.T) {
432- assert.Equal(t, 0, copyToPCM(nil, [][]float32{{1.0}}))
126+ assert.Equal(t, uint64(1), adapter.dropCount.Load())
127+ assert.Len(t, adapter.ready, packetQueueCapacity)
433128 }
434129
435-func TestCopyToPCM_EmptySamples_ReturnsZero(t *testing.T) {
436- pcm := &pwpipewire.PCMBuffer{Frames: 1, Channels: 1, Samples: [][]float32{make([]float32, 1)}}
437- assert.Equal(t, 0, copyToPCM(pcm, nil))
438- assert.Equal(t, 0, copyToPCM(pcm, [][]float32{}))
130+func TestPlayerStreamAdapter_EmptyFillCountsUnderrun(t *testing.T) {
131+ adapter := testAdapter(2)
132+ frames, err := adapter.fill(stereoPCM(2))
133+ require.NoError(t, err)
134+ assert.Zero(t, frames)
135+ assert.Equal(t, uint64(1), adapter.underrunCount.Load())
439136 }
440137
441-// --- Non-blocking Write tests ---
442-
443-func TestPlayerStreamAdapter_Write_DropsWhenBufferFull(t *testing.T) {
444- adapter := &playerStreamAdapter{
445- player: newPlayerRecorder(t),
446- buf: make(chan [][]float32, 4),
447- closed: make(chan struct{}),
448- }
449-
450- // Fill the buffer to capacity
451- for i := range 4 {
452- require.NoError(t, adapter.Write([][]float32{{float32(i)}}))
453- }
454-
455- // The 5th write must NOT block — it should drop the packet and return nil.
456- err := adapter.Write([][]float32{{99.0}})
457- require.NoError(t, err, "Write must not return an error when dropping due to full buffer")
458-
459- // Verify the drop was counted
460- assert.Equal(t, uint64(1), adapter.dropCount.Load(), "one packet should have been dropped")
461-
462- // Verify buffer still contains the original 4 packets (not the dropped one)
463- assert.Len(t, adapter.buf, 4)
138+func TestPlayerStreamAdapter_CopyToPCMHandlesMismatches(t *testing.T) {
139+ pcm := &pwpipewire.PCMBuffer{Frames: 4, Channels: 1, Samples: [][]float32{make([]float32, 4)}}
140+ assert.Equal(t, 2, copyToPCM(pcm, [][]float32{{1, 2}, {3, 4}}, 2))
141+ assert.InEpsilon(t, 1, pcm.Samples[0][0], 0.0001)
142+ assert.Zero(t, copyToPCM(nil, [][]float32{{1}}, 1))
464143 }
465144
466-func TestPlayerStreamAdapter_Write_DropsLogOnlyOnFirstAndPowerOfTwo(t *testing.T) {
467- adapter := &playerStreamAdapter{
468- player: newPlayerRecorder(t),
469- buf: make(chan [][]float32, 1), // tiny buffer to force drops
470- closed: make(chan struct{}),
471- }
472-
473- // Fill buffer
474- require.NoError(t, adapter.Write([][]float32{{0.0}}))
475-
476- // Drop 10 packets
477- for i := range 10 {
478- _ = adapter.Write([][]float32{{float32(i)}})
479- }
480-
481- // Verify all drops were counted
482- assert.Equal(t, uint64(10), adapter.dropCount.Load())
145+func TestPlayerStreamAdapter_CloseVsWrite(t *testing.T) {
146+ adapter := testAdapter(2)
147+ player := newPlayerRecorder(t)
148+ adapter.player = player
149+ require.NoError(t, adapter.Close())
150+ require.ErrorIs(t, adapter.Write([][]float32{{1}, {2}}), ErrStreamClosed)
151+ assert.True(t, player.stopped)
152+ assert.True(t, player.closedFlag)
483153 }
484154
485-// --- Concurrent Write / Close tests ---
486-
487155 func TestPlayerStreamAdapter_ConcurrentWriteClose_NoRace(t *testing.T) {
488- player := newPlayerRecorder(t)
489- adapter := &playerStreamAdapter{
490- player: player,
491- buf: make(chan [][]float32, 4),
492- closed: make(chan struct{}),
493- }
494-
495- done := make(chan struct{})
496- go func() {
497- defer close(done)
498- for i := range 100 {
499- err := adapter.Write([][]float32{{float32(i)}})
500- if err != nil {
501- assert.ErrorIs(t, err, ErrStreamClosed)
502- return
156+ adapter := testAdapter(4)
157+ var writers sync.WaitGroup
158+ for range 4 {
159+ writers.Add(1)
160+ go func() {
161+ defer writers.Done()
162+ for range 1000 {
163+ err := adapter.Write([][]float32{{1, 2}, {3, 4}})
164+ if err != nil {
165+ assert.ErrorIs(t, err, ErrStreamClosed)
166+ return
167+ }
503168 }
504- }
505- }()
506-
507- // Let a few writes land, then close.
169+ }()
170+ }
508171 adapter.Close()
509- <-done
510-}
511-
512-// --- NewFactory tests ---
513-
514-func TestNewFactory_Succeeds(t *testing.T) {
515- factory, err := NewFactory()
516- require.NoError(t, err)
517- require.NotNil(t, factory)
518- assert.NotNil(t, factory.createPlayer)
172+ writers.Wait()
519173 }
520174
521-// --- Integration-style test: full write-through ---
522-
523175 func TestStream_WriteFlowsThroughToFill(t *testing.T) {
524- var capturedCallbacks pwpipewire.PlayerCallbacks
176+ var callbacks pwpipewire.PlayerCallbacks
525177 player := newPlayerRecorder(t)
526-
527- factory := &Factory{
528- createPlayer: func(_ pwpipewire.PlayerConfig, callbacks pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
529- capturedCallbacks = callbacks
530- return player, nil
531- },
532- }
533-
534- format := port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 4}
535- stream, err := factory.NewStream(context.Background(), format)
178+ factory := &Factory{createPlayer: func(_ pwpipewire.PlayerConfig, got pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
179+ callbacks = got
180+ return player, nil
181+ }}
182+ stream, err := factory.NewStream(context.Background(), port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 4})
536183 require.NoError(t, err)
537-
538- // Write samples through the stream
539- samples := [][]float32{
540- {0.1, 0.2, 0.3, 0.4},
541- {0.5, 0.6, 0.7, 0.8},
542- }
543- require.NoError(t, stream.Write(samples))
544-
545- // Simulate the Fill callback being called by the player
546- pcm := &pwpipewire.PCMBuffer{
547- Frames: 4,
548- Channels: 2,
549- Stride: 1,
550- Samples: [][]float32{
551- make([]float32, 4),
552- make([]float32, 4),
553- },
554- }
555-
556- frames, err := capturedCallbacks.Fill(pcm)
184+ require.NoError(t, stream.Write([][]float32{{.1, .2, .3, .4}, {.5, .6, .7, .8}}))
185+ pcm := stereoPCM(4)
186+ frames, err := callbacks.Fill(pcm)
557187 require.NoError(t, err)
558188 assert.Equal(t, 4, frames)
559- assert.InEpsilon(t, 0.1, pcm.Samples[0][0], 0.0001)
560- assert.InEpsilon(t, 0.5, pcm.Samples[1][0], 0.0001)
561-
189+ assert.InEpsilon(t, .1, pcm.Samples[0][0], .0001)
190+ assert.InEpsilon(t, .5, pcm.Samples[1][0], .0001)
562191 require.NoError(t, stream.Close())
563192 }
193+
194+func BenchmarkPlayerStreamAdapterStereo(b *testing.B) {
195+ adapter := testAdapter(512)
196+ samples := [][]float32{make([]float32, 512), make([]float32, 512)}
197+ pcm := stereoPCM(512)
198+ b.ReportAllocs()
199+ b.SetBytes(2 * 512 * 4)
200+ for i := 0; i < b.N; i++ {
201+ if err := adapter.Write(samples); err != nil {
202+ b.Fatal(err)
203+ }
204+ if _, err := adapter.fill(pcm); err != nil {
205+ b.Fatal(err)
206+ }
207+ }
208+}
@@ -1,11 +1,11 @@
1-// Package pipewire provides a PipeWire-based audio output implementation.
2-// This wraps purego-pipewire's high-level Player API.
3 package pipewire1 package pipewire
4 2
5 import (3 import (
6 "context"4 "context"
7 "errors"5 "errors"
6+ "sync"
8 "testing"7 "testing"
8+ "time"
9 9
10 "github.com/bnema/dumber/internal/application/port"10 "github.com/bnema/dumber/internal/application/port"
11 pipewiremocks "github.com/bnema/dumber/internal/infrastructure/audio/pipewire/mocks"11 pipewiremocks "github.com/bnema/dumber/internal/infrastructure/audio/pipewire/mocks"
@@ -16,548 +16,193 @@ import (
16 16
17 type playerRecorder struct {17 type playerRecorder struct {
18 *pipewiremocks.MockPlayer18 *pipewiremocks.MockPlayer
19-19+ started, stopped, closedFlag bool
20- started bool20+ startErr, stopErr, closeErr error
21- paused bool
22- stopped bool
23- closedFlag bool
24- state pwpipewire.PlayerState
25-
26- startErr error
27- stopErr error
28- closeErr error
29 }21 }
30 22
31 func newPlayerRecorder(t *testing.T) *playerRecorder {23 func newPlayerRecorder(t *testing.T) *playerRecorder {
32 t.Helper()24 t.Helper()
33-25+ p := &playerRecorder{MockPlayer: pipewiremocks.NewMockPlayer(t)}
34- recorder := &playerRecorder{MockPlayer: pipewiremocks.NewMockPlayer(t)}26+ p.EXPECT().Start().RunAndReturn(func() error {
35- recorder.EXPECT().Start().RunAndReturn(recorder.start).Maybe()27+ if p.startErr != nil {
36- recorder.EXPECT().Pause().RunAndReturn(recorder.pause).Maybe()28+ return p.startErr
37- recorder.EXPECT().Stop().RunAndReturn(recorder.stop).Maybe()29+ }
38- recorder.EXPECT().Close().RunAndReturn(recorder.close).Maybe()30+ p.started = true
39- recorder.EXPECT().State().RunAndReturn(recorder.currentState).Maybe()31+ return nil
40- return recorder32+ }).Maybe()
41-}33+ p.EXPECT().Stop().RunAndReturn(func() error {
42-34+ if p.stopErr != nil {
43-func (p *playerRecorder) start() error {35+ return p.stopErr
44- if p.startErr != nil {36+ }
45- return p.startErr37+ p.stopped = true
46- }38+ return nil
47- p.started = true39+ }).Maybe()
48- p.state = pwpipewire.PlayerStatePlaying40+ p.EXPECT().Close().RunAndReturn(func() error {
49- return nil41+ if p.closeErr != nil {
50-}42+ return p.closeErr
51-43+ }
52-func (p *playerRecorder) pause() error {44+ p.closedFlag = true
53- p.paused = true45+ return nil
54- p.state = pwpipewire.PlayerStatePaused46+ }).Maybe()
55- return nil47+ return p
56-}
57-
58-func (p *playerRecorder) stop() error {
59- if p.stopErr != nil {
60- return p.stopErr
61- }
62- p.stopped = true
63- p.state = pwpipewire.PlayerStateStopped
64- return nil
65 }48 }
66 49
67-func (p *playerRecorder) close() error {50+func testAdapter(frames int) *playerStreamAdapter {
68- if p.closeErr != nil {51+ return newPlayerStreamAdapter(context.Background(), 2, frames)
69- return p.closeErr
70- }
71- p.closedFlag = true
72- p.state = pwpipewire.PlayerStateClosed
73- return nil
74 }52 }
75 53
76-func (p *playerRecorder) currentState() pwpipewire.PlayerState {54+func stereoPCM(frames int) *pwpipewire.PCMBuffer {
77- return p.state55+ return &pwpipewire.PCMBuffer{Frames: frames, Channels: 2, Samples: [][]float32{make([]float32, frames), make([]float32, frames)}}
78 }56 }
79 57
80-// --- Factory tests ---58+func TestFactory_NewStreamValidatesAndStartsPlayer(t *testing.T) {
81-
82-func TestFactory_NewStream_ValidFormat_CreatesStream(t *testing.T) {
83- var capturedConfig pwpipewire.PlayerConfig
84 player := newPlayerRecorder(t)59 player := newPlayerRecorder(t)
85-60+ factory := &Factory{createPlayer: func(config pwpipewire.PlayerConfig, callbacks pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
86- factory := &Factory{61+ assert.Equal(t, 48000, config.SampleRate)
87- createPlayer: func(config pwpipewire.PlayerConfig, _ pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {62+ assert.Equal(t, 2, config.Channels)
88- capturedConfig = config63+ assert.NotNil(t, callbacks.Fill)
89- return player, nil64+ return player, nil
90- },65+ }}
91- }66+ stream, err := factory.NewStream(context.Background(), port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 512})
92-
93- format := port.AudioStreamFormat{
94- SampleRate: 48000,
95- ChannelCount: 2,
96- FramesPerBuffer: 512,
97- }
98-
99- stream, err := factory.NewStream(context.Background(), format)
100 require.NoError(t, err)67 require.NoError(t, err)
101- require.NotNil(t, stream)
102-
103- // Verify config mapping
104- assert.Equal(t, 48000, capturedConfig.SampleRate)
105- assert.Equal(t, 2, capturedConfig.Channels)
106- assert.Equal(t, 512, capturedConfig.FramesPerBuffer)
107- assert.Equal(t, pwpipewire.SampleFormatF32, capturedConfig.SampleFormat)
108- assert.Equal(t, pwpipewire.UnderrunFillSilence, capturedConfig.UnderrunPolicy)
109-
110- // Verify player was started
111 assert.True(t, player.started)68 assert.True(t, player.started)
112-
113- // Clean up
114 require.NoError(t, stream.Close())69 require.NoError(t, stream.Close())
115 }70 }
116 71
117-func TestFactory_NewStream_InvalidFormat_ReturnsError(t *testing.T) {72+func TestFactory_NewStreamRejectsInvalidFormat(t *testing.T) {
118- factory := &Factory{73+ factory := &Factory{createPlayer: func(pwpipewire.PlayerConfig, pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
119- createPlayer: func(_ pwpipewire.PlayerConfig, _ pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {74+ t.Fatal("unexpected player")
120- t.Fatal("createPlayer should not be called for invalid format")75+ return nil, nil
121- return nil, nil76+ }}
122- },77+ for _, format := range []port.AudioStreamFormat{
123- }78+ {SampleRate: 0, ChannelCount: 2, FramesPerBuffer: 512},
124-79+ {SampleRate: 48000, ChannelCount: 0, FramesPerBuffer: 512},
125- tests := []struct {80+ {SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 0},
126- name string81+ } {
127- format port.AudioStreamFormat82+ stream, err := factory.NewStream(context.Background(), format)
128- }{83+ require.ErrorIs(t, err, ErrInvalidFormat)
129- {84+ assert.Nil(t, stream)
130- name: "zero sample rate",
131- format: port.AudioStreamFormat{SampleRate: 0, ChannelCount: 2, FramesPerBuffer: 512},
132- },
133- {
134- name: "zero channels",
135- format: port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 0, FramesPerBuffer: 512},
136- },
137- {
138- name: "negative sample rate",
139- format: port.AudioStreamFormat{SampleRate: -48000, ChannelCount: 2, FramesPerBuffer: 512},
140- },
141- {
142- name: "negative channels",
143- format: port.AudioStreamFormat{SampleRate: 48000, ChannelCount: -2, FramesPerBuffer: 512},
144- },
145- {
146- name: "excessive sample rate",
147- format: port.AudioStreamFormat{SampleRate: 1000000, ChannelCount: 2, FramesPerBuffer: 512},
148- },
149- {
150- name: "excessive channels",
151- format: port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 1025, FramesPerBuffer: 512},
152- },
153- {
154- name: "zero frames per buffer",
155- format: port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 0},
156- },
157- {
158- name: "excessive frames per buffer",
159- format: port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 10000},
160- },
161- }
162-
163- for _, tt := range tests {
164- t.Run(tt.name, func(t *testing.T) {
165- stream, err := factory.NewStream(context.Background(), tt.format)
166- require.Error(t, err)
167- assert.Nil(t, stream)
168- assert.ErrorIs(t, err, ErrInvalidFormat)
169- })
170- }
171-}
172-
173-func TestFactory_NewStream_PlayerCreationFails_ReturnsError(t *testing.T) {
174- expectedErr := errors.New("pipewire not available")
175- factory := &Factory{
176- createPlayer: func(_ pwpipewire.PlayerConfig, _ pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
177- return nil, expectedErr
178- },
179 }85 }
180-
181- format := port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 512}
182- stream, err := factory.NewStream(context.Background(), format)
183-
184- require.Error(t, err)
185- assert.Nil(t, stream)
186- assert.ErrorIs(t, err, expectedErr)
187 }86 }
188 87
189-func TestFactory_NewStream_PlayerStartFails_ClosesAndReturnsError(t *testing.T) {88+func TestFactory_NewStreamClosesPlayerWhenStartFails(t *testing.T) {
190 player := newPlayerRecorder(t)89 player := newPlayerRecorder(t)
191 player.startErr = errors.New("start failed")90 player.startErr = errors.New("start failed")
192- factory := &Factory{91+ factory := &Factory{createPlayer: func(pwpipewire.PlayerConfig, pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
193- createPlayer: func(_ pwpipewire.PlayerConfig, _ pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {92+ return player, nil
194- return player, nil93+ }}
195- },94+ stream, err := factory.NewStream(context.Background(), port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 512})
196- }
197-
198- format := port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 512}
199- stream, err := factory.NewStream(context.Background(), format)
200-
201- require.Error(t, err)
202- assert.Nil(t, stream)
203- assert.True(t, player.closedFlag, "player should be closed after start failure")
204-}
205-
206-func TestFactory_NewStream_ContextCancelled_ReturnsError(t *testing.T) {
207- factory := &Factory{
208- createPlayer: func(_ pwpipewire.PlayerConfig, _ pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
209- t.Fatal("createPlayer should not be called when context is canceled")
210- return nil, nil
211- },
212- }
213-
214- ctx, cancel := context.WithCancel(context.Background())
215- cancel()
216-
217- format := port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 512}
218- stream, err := factory.NewStream(ctx, format)
219-
220 require.Error(t, err)95 require.Error(t, err)
221 assert.Nil(t, stream)96 assert.Nil(t, stream)
222-}
223-
224-func TestFactory_NewStream_MultipleFormats_Accepted(t *testing.T) {
225- formats := []port.AudioStreamFormat{
226- {SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 512},
227- {SampleRate: 44100, ChannelCount: 2, FramesPerBuffer: 512},
228- {SampleRate: 48000, ChannelCount: 1, FramesPerBuffer: 256},
229- {SampleRate: 96000, ChannelCount: 2, FramesPerBuffer: 1024},
230- {SampleRate: 48000, ChannelCount: 6, FramesPerBuffer: 512},
231- }
232-
233- for _, fmt := range formats {
234- t.Run("", func(t *testing.T) {
235- factory := &Factory{
236- createPlayer: func(config pwpipewire.PlayerConfig, _ pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
237- assert.Equal(t, fmt.SampleRate, config.SampleRate)
238- assert.Equal(t, fmt.ChannelCount, config.Channels)
239- assert.Equal(t, fmt.FramesPerBuffer, config.FramesPerBuffer)
240- return newPlayerRecorder(t), nil
241- },
242- }
243-
244- stream, err := factory.NewStream(context.Background(), fmt)
245- require.NoError(t, err)
246- assert.NotNil(t, stream)
247-
248- _ = stream.Close()
249- })
250- }
251-}
252-
253-func TestFactory_ImplementsPortInterface(_ *testing.T) {
254- var _ port.AudioOutputFactory = (*Factory)(nil)
255-}
256-
257-// --- Stream adapter tests ---
258-
259-func TestPlayerStreamAdapter_Write_SendsToBuffer(t *testing.T) {
260- player := newPlayerRecorder(t)
261- adapter := &playerStreamAdapter{
262- player: player,
263- buf: make(chan [][]float32, 4),
264- closed: make(chan struct{}),
265- }
266-
267- samples := [][]float32{
268- {0.1, 0.2},
269- {0.3, 0.4},
270- }
271-
272- err := adapter.Write(samples)
273- require.NoError(t, err)
274-
275- // Verify data is in the channel
276- received := <-adapter.buf
277- assert.Equal(t, samples, received)
278-}
279-
280-func TestPlayerStreamAdapter_Write_AfterClose_ReturnsError(t *testing.T) {
281- player := newPlayerRecorder(t)
282- adapter := &playerStreamAdapter{
283- player: player,
284- buf: make(chan [][]float32, 4),
285- closed: make(chan struct{}),
286- }
287-
288- require.NoError(t, adapter.Close())
289-
290- err := adapter.Write([][]float32{{0.1}})
291- assert.ErrorIs(t, err, ErrStreamClosed)
292-}
293-
294-func TestPlayerStreamAdapter_Close_StopsAndClosesPlayer(t *testing.T) {
295- player := newPlayerRecorder(t)
296- adapter := &playerStreamAdapter{
297- player: player,
298- buf: make(chan [][]float32, 4),
299- closed: make(chan struct{}),
300- }
301-
302- err := adapter.Close()
303- require.NoError(t, err)
304-
305- assert.True(t, player.stopped)
306 assert.True(t, player.closedFlag)97 assert.True(t, player.closedFlag)
307 }98 }
308 99
309-func TestPlayerStreamAdapter_Close_IsIdempotent(t *testing.T) {100+func TestPlayerStreamAdapter_FillGetsOwnedPacketAfterCallbackMutation(t *testing.T) {
310- player := newPlayerRecorder(t)101+ adapter := testAdapter(3)
311- adapter := &playerStreamAdapter{102+ samples := [][]float32{{1, 2, 3}, {4, 5, 6}}
312- player: player,103+ require.NoError(t, adapter.Write(samples))
313- buf: make(chan [][]float32, 4),104+ samples[0][0], samples[1][0] = 99, 99
314- closed: make(chan struct{}),105+ pcm := stereoPCM(3)
315- }
316-
317- require.NoError(t, adapter.Close())
318- require.NoError(t, adapter.Close())
319-}
320-
321-func TestPlayerStreamAdapter_Close_PropagatesStopError(t *testing.T) {
322- player := newPlayerRecorder(t)
323- player.stopErr = errors.New("stop error")
324- adapter := &playerStreamAdapter{
325- player: player,
326- buf: make(chan [][]float32, 4),
327- closed: make(chan struct{}),
328- }
329-
330- err := adapter.Close()
331- require.Error(t, err)
332- assert.Contains(t, err.Error(), "stop error")
333-}
334-
335-func TestPlayerStreamAdapter_Close_PropagatesCloseError(t *testing.T) {
336- player := newPlayerRecorder(t)
337- player.closeErr = errors.New("close error")
338- adapter := &playerStreamAdapter{
339- player: player,
340- buf: make(chan [][]float32, 4),
341- closed: make(chan struct{}),
342- }
343-
344- err := adapter.Close()
345- require.Error(t, err)
346- assert.Contains(t, err.Error(), "close error")
347-}
348-
349-// --- Fill callback tests ---
350-
351-func TestFill_PullsFromBuffer(t *testing.T) {
352- adapter := &playerStreamAdapter{
353- buf: make(chan [][]float32, 4),
354- }
355-
356- // Push samples
357- samples := [][]float32{
358- {0.1, 0.2, 0.3},
359- {0.4, 0.5, 0.6},
360- }
361- adapter.buf <- samples
362-
363- // Simulate Fill call
364- pcm := &pwpipewire.PCMBuffer{
365- Frames: 3,
366- Channels: 2,
367- Stride: 1,
368- Samples: [][]float32{
369- make([]float32, 3),
370- make([]float32, 3),
371- },
372- }
373-
374 frames, err := adapter.fill(pcm)106 frames, err := adapter.fill(pcm)
375 require.NoError(t, err)107 require.NoError(t, err)
376 assert.Equal(t, 3, frames)108 assert.Equal(t, 3, frames)
377- assert.InEpsilon(t, 0.1, pcm.Samples[0][0], 0.0001)109+ assert.InEpsilon(t, 1, pcm.Samples[0][0], 0.0001)
378- assert.InEpsilon(t, 0.4, pcm.Samples[1][0], 0.0001)110+ assert.InEpsilon(t, 4, pcm.Samples[1][0], 0.0001)
379-}
380-
381-func TestFill_ReturnsZeroWhenEmpty(t *testing.T) {
382- adapter := &playerStreamAdapter{
383- buf: make(chan [][]float32, 4),
384- }
385-
386- pcm := &pwpipewire.PCMBuffer{
387- Frames: 256,
388- Channels: 2,
389- Stride: 1,
390- Samples: [][]float32{
391- make([]float32, 256),
392- make([]float32, 256),
393- },
394- }
395-
396- frames, err := adapter.fill(pcm)
397- require.NoError(t, err)
398- assert.Equal(t, 0, frames)
399-}
400-
401-func TestCopyToPCM_HandlesChannelMismatch(t *testing.T) {
402- // More source channels than PCM channels — truncates
403- pcm := &pwpipewire.PCMBuffer{
404- Frames: 2,
405- Channels: 1,
406- Samples: [][]float32{make([]float32, 2)},
407- }
408- samples := [][]float32{
409- {1.0, 2.0},
410- {3.0, 4.0}, // should be ignored
411- }
412- frames := copyToPCM(pcm, samples)
413- assert.Equal(t, 2, frames)
414- assert.InEpsilon(t, 1.0, pcm.Samples[0][0], 0.0001)
415 }111 }
416 112
417-func TestCopyToPCM_HandlesFrameMismatch(t *testing.T) {113+func TestPlayerStreamAdapter_SaturationDropsWithoutBlocking(t *testing.T) {
418- // Fewer source frames than PCM expects114+ adapter := testAdapter(2)
419- pcm := &pwpipewire.PCMBuffer{115+ for i := 0; i < packetQueueCapacity; i++ {
420- Frames: 4,116+ require.NoError(t, adapter.Write([][]float32{{float32(i)}, {float32(i)}}))
421- Channels: 1,
422- Samples: [][]float32{make([]float32, 4)},
423 }117 }
424- samples := [][]float32{118+ done := make(chan error, 1)
425- {1.0, 2.0}, // only 2 frames119+ go func() { done <- adapter.Write([][]float32{{99}, {99}}) }()
120+ select {
121+ case err := <-done:
122+ require.NoError(t, err)
123+ case <-time.After(time.Second):
124+ t.Fatal("Write blocked while the bounded queue was saturated")
426 }125 }
427- frames := copyToPCM(pcm, samples)126+ assert.Equal(t, uint64(1), adapter.dropCount.Load())
428- assert.Equal(t, 2, frames) // reports partial fill127+ assert.Len(t, adapter.ready, packetQueueCapacity)
429-}
430-
431-func TestCopyToPCM_NilPCM_ReturnsZero(t *testing.T) {
432- assert.Equal(t, 0, copyToPCM(nil, [][]float32{{1.0}}))
433 }128 }
434 129
435-func TestCopyToPCM_EmptySamples_ReturnsZero(t *testing.T) {130+func TestPlayerStreamAdapter_EmptyFillCountsUnderrun(t *testing.T) {
436- pcm := &pwpipewire.PCMBuffer{Frames: 1, Channels: 1, Samples: [][]float32{make([]float32, 1)}}131+ adapter := testAdapter(2)
437- assert.Equal(t, 0, copyToPCM(pcm, nil))132+ frames, err := adapter.fill(stereoPCM(2))
438- assert.Equal(t, 0, copyToPCM(pcm, [][]float32{}))133+ require.NoError(t, err)
134+ assert.Zero(t, frames)
135+ assert.Equal(t, uint64(1), adapter.underrunCount.Load())
439 }136 }
440 137
441-// --- Non-blocking Write tests ---138+func TestPlayerStreamAdapter_CopyToPCMHandlesMismatches(t *testing.T) {
442-139+ pcm := &pwpipewire.PCMBuffer{Frames: 4, Channels: 1, Samples: [][]float32{make([]float32, 4)}}
443-func TestPlayerStreamAdapter_Write_DropsWhenBufferFull(t *testing.T) {140+ assert.Equal(t, 2, copyToPCM(pcm, [][]float32{{1, 2}, {3, 4}}, 2))
444- adapter := &playerStreamAdapter{141+ assert.InEpsilon(t, 1, pcm.Samples[0][0], 0.0001)
445- player: newPlayerRecorder(t),142+ assert.Zero(t, copyToPCM(nil, [][]float32{{1}}, 1))
446- buf: make(chan [][]float32, 4),
447- closed: make(chan struct{}),
448- }
449-
450- // Fill the buffer to capacity
451- for i := range 4 {
452- require.NoError(t, adapter.Write([][]float32{{float32(i)}}))
453- }
454-
455- // The 5th write must NOT block — it should drop the packet and return nil.
456- err := adapter.Write([][]float32{{99.0}})
457- require.NoError(t, err, "Write must not return an error when dropping due to full buffer")
458-
459- // Verify the drop was counted
460- assert.Equal(t, uint64(1), adapter.dropCount.Load(), "one packet should have been dropped")
461-
462- // Verify buffer still contains the original 4 packets (not the dropped one)
463- assert.Len(t, adapter.buf, 4)
464 }143 }
465 144
466-func TestPlayerStreamAdapter_Write_DropsLogOnlyOnFirstAndPowerOfTwo(t *testing.T) {145+func TestPlayerStreamAdapter_CloseVsWrite(t *testing.T) {
467- adapter := &playerStreamAdapter{146+ adapter := testAdapter(2)
468- player: newPlayerRecorder(t),147+ player := newPlayerRecorder(t)
469- buf: make(chan [][]float32, 1), // tiny buffer to force drops148+ adapter.player = player
470- closed: make(chan struct{}),149+ require.NoError(t, adapter.Close())
471- }150+ require.ErrorIs(t, adapter.Write([][]float32{{1}, {2}}), ErrStreamClosed)
472-151+ assert.True(t, player.stopped)
473- // Fill buffer152+ assert.True(t, player.closedFlag)
474- require.NoError(t, adapter.Write([][]float32{{0.0}}))
475-
476- // Drop 10 packets
477- for i := range 10 {
478- _ = adapter.Write([][]float32{{float32(i)}})
479- }
480-
481- // Verify all drops were counted
482- assert.Equal(t, uint64(10), adapter.dropCount.Load())
483 }153 }
484 154
485-// --- Concurrent Write / Close tests ---
486-
487 func TestPlayerStreamAdapter_ConcurrentWriteClose_NoRace(t *testing.T) {155 func TestPlayerStreamAdapter_ConcurrentWriteClose_NoRace(t *testing.T) {
488- player := newPlayerRecorder(t)156+ adapter := testAdapter(4)
489- adapter := &playerStreamAdapter{157+ var writers sync.WaitGroup
490- player: player,158+ for range 4 {
491- buf: make(chan [][]float32, 4),159+ writers.Add(1)
492- closed: make(chan struct{}),160+ go func() {
493- }161+ defer writers.Done()
494-162+ for range 1000 {
495- done := make(chan struct{})163+ err := adapter.Write([][]float32{{1, 2}, {3, 4}})
496- go func() {164+ if err != nil {
497- defer close(done)165+ assert.ErrorIs(t, err, ErrStreamClosed)
498- for i := range 100 {166+ return
499- err := adapter.Write([][]float32{{float32(i)}})167+ }
500- if err != nil {
501- assert.ErrorIs(t, err, ErrStreamClosed)
502- return
503 }168 }
504- }169+ }()
505- }()170+ }
506-
507- // Let a few writes land, then close.
508 adapter.Close()171 adapter.Close()
509- <-done172+ writers.Wait()
510-}
511-
512-// --- NewFactory tests ---
513-
514-func TestNewFactory_Succeeds(t *testing.T) {
515- factory, err := NewFactory()
516- require.NoError(t, err)
517- require.NotNil(t, factory)
518- assert.NotNil(t, factory.createPlayer)
519 }173 }
520 174
521-// --- Integration-style test: full write-through ---
522-
523 func TestStream_WriteFlowsThroughToFill(t *testing.T) {175 func TestStream_WriteFlowsThroughToFill(t *testing.T) {
524- var capturedCallbacks pwpipewire.PlayerCallbacks176+ var callbacks pwpipewire.PlayerCallbacks
525 player := newPlayerRecorder(t)177 player := newPlayerRecorder(t)
526-178+ factory := &Factory{createPlayer: func(_ pwpipewire.PlayerConfig, got pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
527- factory := &Factory{179+ callbacks = got
528- createPlayer: func(_ pwpipewire.PlayerConfig, callbacks pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {180+ return player, nil
529- capturedCallbacks = callbacks181+ }}
530- return player, nil182+ stream, err := factory.NewStream(context.Background(), port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 4})
531- },
532- }
533-
534- format := port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 4}
535- stream, err := factory.NewStream(context.Background(), format)
536 require.NoError(t, err)183 require.NoError(t, err)
537-184+ require.NoError(t, stream.Write([][]float32{{.1, .2, .3, .4}, {.5, .6, .7, .8}}))
538- // Write samples through the stream185+ pcm := stereoPCM(4)
539- samples := [][]float32{186+ frames, err := callbacks.Fill(pcm)
540- {0.1, 0.2, 0.3, 0.4},
541- {0.5, 0.6, 0.7, 0.8},
542- }
543- require.NoError(t, stream.Write(samples))
544-
545- // Simulate the Fill callback being called by the player
546- pcm := &pwpipewire.PCMBuffer{
547- Frames: 4,
548- Channels: 2,
549- Stride: 1,
550- Samples: [][]float32{
551- make([]float32, 4),
552- make([]float32, 4),
553- },
554- }
555-
556- frames, err := capturedCallbacks.Fill(pcm)
557 require.NoError(t, err)187 require.NoError(t, err)
558 assert.Equal(t, 4, frames)188 assert.Equal(t, 4, frames)
559- assert.InEpsilon(t, 0.1, pcm.Samples[0][0], 0.0001)189+ assert.InEpsilon(t, .1, pcm.Samples[0][0], .0001)
560- assert.InEpsilon(t, 0.5, pcm.Samples[1][0], 0.0001)190+ assert.InEpsilon(t, .5, pcm.Samples[1][0], .0001)
561-
562 require.NoError(t, stream.Close())191 require.NoError(t, stream.Close())
563 }192 }
193+
194+func BenchmarkPlayerStreamAdapterStereo(b *testing.B) {
195+ adapter := testAdapter(512)
196+ samples := [][]float32{make([]float32, 512), make([]float32, 512)}
197+ pcm := stereoPCM(512)
198+ b.ReportAllocs()
199+ b.SetBytes(2 * 512 * 4)
200+ for i := 0; i < b.N; i++ {
201+ if err := adapter.Write(samples); err != nil {
202+ b.Fatal(err)
203+ }
204+ if _, err := adapter.fill(pcm); err != nil {
205+ b.Fatal(err)
206+ }
207+ }
208+}
M internal/infrastructure/cef/webview.go +9 -11
@@ -2331,20 +2331,18 @@ func (wv *WebView) takePendingCreate() *pendingBrowserCreate {
23312331 return pc
23322332 }
23332333
2334-// closeAudioStream closes and clears the active audio output stream.
2335-// This is safe to call even if no stream is active.
2334+// closeAudioStream detaches then closes the active output stream. Detaching
2335+// under the lock keeps packet snapshots consistent without holding the lock
2336+// while PipeWire shutdown runs; the port guarantees Close/Write safety.
23362337 func (wv *WebView) closeAudioStream() {
23372338 wv.audioStreamMu.Lock()
2338- defer wv.audioStreamMu.Unlock()
2339+ stream := wv.activeAudioStream
2340+ wv.activeAudioStream = nil
2341+ wv.audioStreamMu.Unlock()
23392342
2340- if wv.activeAudioStream != nil {
2341- if err := wv.activeAudioStream.Close(); err != nil {
2342- if wv.ctx != nil {
2343- logging.FromContext(wv.ctx).Debug().
2344- Err(err).
2345- Msg("cef: error closing audio stream")
2346- }
2343+ if stream != nil {
2344+ if err := stream.Close(); err != nil && wv.ctx != nil {
2345+ logging.FromContext(wv.ctx).Debug().Err(err).Msg("cef: error closing audio stream")
23472346 }
2348- wv.activeAudioStream = nil
23492347 }
23502348 }
@@ -2331,20 +2331,18 @@ func (wv *WebView) takePendingCreate() *pendingBrowserCreate {
2331 return pc2331 return pc
2332 }2332 }
2333 2333
2334-// closeAudioStream closes and clears the active audio output stream.2334+// closeAudioStream detaches then closes the active output stream. Detaching
2335-// This is safe to call even if no stream is active.2335+// under the lock keeps packet snapshots consistent without holding the lock
2336+// while PipeWire shutdown runs; the port guarantees Close/Write safety.
2336 func (wv *WebView) closeAudioStream() {2337 func (wv *WebView) closeAudioStream() {
2337 wv.audioStreamMu.Lock()2338 wv.audioStreamMu.Lock()
2338- defer wv.audioStreamMu.Unlock()2339+ stream := wv.activeAudioStream
2340+ wv.activeAudioStream = nil
2341+ wv.audioStreamMu.Unlock()
2339 2342
2340- if wv.activeAudioStream != nil {2343+ if stream != nil {
2341- if err := wv.activeAudioStream.Close(); err != nil {2344+ if err := stream.Close(); err != nil && wv.ctx != nil {
2342- if wv.ctx != nil {2345+ logging.FromContext(wv.ctx).Debug().Err(err).Msg("cef: error closing audio stream")
2343- logging.FromContext(wv.ctx).Debug().
2344- Err(err).
2345- Msg("cef: error closing audio stream")
2346- }
2347 }2346 }
2348- wv.activeAudioStream = nil
2349 }2347 }
2350 }2348 }
M internal/infrastructure/cef/webview_audio_test.go +24 -26
@@ -258,9 +258,9 @@ func TestHandlerSet_OnAudioStreamStarted_FactoryError_HandlesGracefully(t *testi
258258 // Task 7: PCM Packet Forwarding Tests
259259 // ============================================================================
260260
261-// TestHandlerSet_OnAudioStreamPacket_CopiesAndWrites verifies that
262-// OnAudioStreamPacket copies samples before writing to avoid data races.
263-func TestHandlerSet_OnAudioStreamPacket_CopiesAndWrites(t *testing.T) {
261+// TestHandlerSet_OnAudioStreamPacket_ForwardsCallbackData verifies that CEF
262+// does not allocate a staging copy; the audio output port owns that handoff.
263+func TestHandlerSet_OnAudioStreamPacket_ForwardsCallbackData(t *testing.T) {
264264 ctx := context.Background()
265265 mockStream := &mockAudioStream{}
266266
@@ -287,12 +287,9 @@ func TestHandlerSet_OnAudioStreamPacket_CopiesAndWrites(t *testing.T) {
287287 require.Len(t, mockStream.writeSamples, 2)
288288 require.Len(t, mockStream.writeSamples[0], 512)
289289
290- // Verify data was copied (values match)
291290 assert.InDelta(t, 0.5, mockStream.writeSamples[0][0], 0.000001)
292291 assert.InDelta(t, -0.5, mockStream.writeSamples[1][0], 0.000001)
293-
294- // Verify it's a copy (different underlying array)
295- assert.NotSame(t, &originalData[0][0], &mockStream.writeSamples[0][0])
292+ assert.Same(t, &originalData[0][0], &mockStream.writeSamples[0][0])
296293 }
297294
298295 // TestHandlerSet_OnAudioStreamPacket_NoActiveStream_DoesNothing verifies
@@ -333,6 +330,17 @@ func TestHandlerSet_OnAudioStreamPacket_NilData_DoesNotPanic(t *testing.T) {
333330 })
334331 }
335332
333+func BenchmarkHandlerSet_OnAudioStreamPacketStereo(b *testing.B) {
334+ stream := &mockAudioStream{}
335+ handlers := &handlerSet{wv: &WebView{ctx: context.Background(), activeAudioStream: stream}}
336+ data := [][]float32{make([]float32, 512), make([]float32, 512)}
337+ b.ReportAllocs()
338+ b.SetBytes(2 * 512 * 4)
339+ for i := 0; i < b.N; i++ {
340+ handlers.OnAudioStreamPacket(nil, data, 512, 0)
341+ }
342+}
343+
336344 // ============================================================================
337345 // Task 8: Stream Cleanup Tests
338346 // ============================================================================
@@ -562,15 +570,13 @@ func TestOnAudioStreamStarted_CorrectParameterMapping(t *testing.T) {
562570 // Race-window tests (write/close interleave)
563571 // ============================================================================
564572
565-// TestOnAudioStreamPacket_WriteHoldsLock verifies that closeAudioStream cannot
566-// interleave between the stream snapshot and Write. If the lock is NOT held
567-// across Write, the goroutine calling closeAudioStream will Close the stream
568-// while Write is in-flight, producing a write-after-close.
569-func TestOnAudioStreamPacket_WriteHoldsLock(t *testing.T) {
573+// TestOnAudioStreamPacket_CloseDoesNotWaitForWrite verifies that lifecycle
574+// shutdown is not serialized behind a potentially slow callback write. The
575+// port contract makes Close/Write concurrency safe.
576+func TestOnAudioStreamPacket_CloseDoesNotWaitForWrite(t *testing.T) {
570577 ctx := context.Background()
571578
572- // racyStream blocks inside Write until we signal, giving closeAudioStream
573- // a window to race if the mutex isn't held.
579+ // racyStream blocks inside Write until we signal.
574580 stream := &racyMockAudioStream{
575581 writeCh: make(chan struct{}),
576582 doneCh: make(chan struct{}),
@@ -590,30 +596,22 @@ func TestOnAudioStreamPacket_WriteHoldsLock(t *testing.T) {
590596 // Wait for Write to be entered.
591597 <-stream.writeCh
592598
593- // Now try to close. If the implementation holds the lock across Write,
594- // this will block until Write returns. If not, Close races with Write.
595599 closeDone := make(chan struct{})
596600 go func() {
597601 wv.closeAudioStream()
598602 close(closeDone)
599603 }()
600604
601- // Close should NOT have completed yet because Write holds the lock.
602- // Give the goroutine time to acquire the lock if it can.
603- time.Sleep(50 * time.Millisecond)
604605 select {
605606 case <-closeDone:
606- t.Fatal("closeAudioStream completed while Write was in-flight — lock not held across Write")
607- default:
608- // expected: closeAudioStream is blocked
607+ // expected: only the short stream snapshot takes the mutex
608+ case <-time.After(time.Second):
609+ t.Fatal("closeAudioStream waited for an in-flight Write")
609610 }
610611
611- // Unblock Write.
612+ // The write can finish after shutdown without holding the lifecycle mutex.
612613 close(stream.doneCh)
613614
614- // Now closeAudioStream should complete.
615- <-closeDone
616-
617615 assert.True(t, stream.closeCalled)
618616 assert.Nil(t, wv.activeAudioStream)
619617 }
@@ -258,9 +258,9 @@ func TestHandlerSet_OnAudioStreamStarted_FactoryError_HandlesGracefully(t *testi
258 // Task 7: PCM Packet Forwarding Tests258 // Task 7: PCM Packet Forwarding Tests
259 // ============================================================================259 // ============================================================================
260 260
261-// TestHandlerSet_OnAudioStreamPacket_CopiesAndWrites verifies that261+// TestHandlerSet_OnAudioStreamPacket_ForwardsCallbackData verifies that CEF
262-// OnAudioStreamPacket copies samples before writing to avoid data races.262+// does not allocate a staging copy; the audio output port owns that handoff.
263-func TestHandlerSet_OnAudioStreamPacket_CopiesAndWrites(t *testing.T) {263+func TestHandlerSet_OnAudioStreamPacket_ForwardsCallbackData(t *testing.T) {
264 ctx := context.Background()264 ctx := context.Background()
265 mockStream := &mockAudioStream{}265 mockStream := &mockAudioStream{}
266 266
@@ -287,12 +287,9 @@ func TestHandlerSet_OnAudioStreamPacket_CopiesAndWrites(t *testing.T) {
287 require.Len(t, mockStream.writeSamples, 2)287 require.Len(t, mockStream.writeSamples, 2)
288 require.Len(t, mockStream.writeSamples[0], 512)288 require.Len(t, mockStream.writeSamples[0], 512)
289 289
290- // Verify data was copied (values match)
291 assert.InDelta(t, 0.5, mockStream.writeSamples[0][0], 0.000001)290 assert.InDelta(t, 0.5, mockStream.writeSamples[0][0], 0.000001)
292 assert.InDelta(t, -0.5, mockStream.writeSamples[1][0], 0.000001)291 assert.InDelta(t, -0.5, mockStream.writeSamples[1][0], 0.000001)
293-292+ assert.Same(t, &originalData[0][0], &mockStream.writeSamples[0][0])
294- // Verify it's a copy (different underlying array)
295- assert.NotSame(t, &originalData[0][0], &mockStream.writeSamples[0][0])
296 }293 }
297 294
298 // TestHandlerSet_OnAudioStreamPacket_NoActiveStream_DoesNothing verifies295 // TestHandlerSet_OnAudioStreamPacket_NoActiveStream_DoesNothing verifies
@@ -333,6 +330,17 @@ func TestHandlerSet_OnAudioStreamPacket_NilData_DoesNotPanic(t *testing.T) {
333 })330 })
334 }331 }
335 332
333+func BenchmarkHandlerSet_OnAudioStreamPacketStereo(b *testing.B) {
334+ stream := &mockAudioStream{}
335+ handlers := &handlerSet{wv: &WebView{ctx: context.Background(), activeAudioStream: stream}}
336+ data := [][]float32{make([]float32, 512), make([]float32, 512)}
337+ b.ReportAllocs()
338+ b.SetBytes(2 * 512 * 4)
339+ for i := 0; i < b.N; i++ {
340+ handlers.OnAudioStreamPacket(nil, data, 512, 0)
341+ }
342+}
343+
336 // ============================================================================344 // ============================================================================
337 // Task 8: Stream Cleanup Tests345 // Task 8: Stream Cleanup Tests
338 // ============================================================================346 // ============================================================================
@@ -562,15 +570,13 @@ func TestOnAudioStreamStarted_CorrectParameterMapping(t *testing.T) {
562 // Race-window tests (write/close interleave)570 // Race-window tests (write/close interleave)
563 // ============================================================================571 // ============================================================================
564 572
565-// TestOnAudioStreamPacket_WriteHoldsLock verifies that closeAudioStream cannot573+// TestOnAudioStreamPacket_CloseDoesNotWaitForWrite verifies that lifecycle
566-// interleave between the stream snapshot and Write. If the lock is NOT held574+// shutdown is not serialized behind a potentially slow callback write. The
567-// across Write, the goroutine calling closeAudioStream will Close the stream575+// port contract makes Close/Write concurrency safe.
568-// while Write is in-flight, producing a write-after-close.576+func TestOnAudioStreamPacket_CloseDoesNotWaitForWrite(t *testing.T) {
569-func TestOnAudioStreamPacket_WriteHoldsLock(t *testing.T) {
570 ctx := context.Background()577 ctx := context.Background()
571 578
572- // racyStream blocks inside Write until we signal, giving closeAudioStream579+ // racyStream blocks inside Write until we signal.
573- // a window to race if the mutex isn't held.
574 stream := &racyMockAudioStream{580 stream := &racyMockAudioStream{
575 writeCh: make(chan struct{}),581 writeCh: make(chan struct{}),
576 doneCh: make(chan struct{}),582 doneCh: make(chan struct{}),
@@ -590,30 +596,22 @@ func TestOnAudioStreamPacket_WriteHoldsLock(t *testing.T) {
590 // Wait for Write to be entered.596 // Wait for Write to be entered.
591 <-stream.writeCh597 <-stream.writeCh
592 598
593- // Now try to close. If the implementation holds the lock across Write,
594- // this will block until Write returns. If not, Close races with Write.
595 closeDone := make(chan struct{})599 closeDone := make(chan struct{})
596 go func() {600 go func() {
597 wv.closeAudioStream()601 wv.closeAudioStream()
598 close(closeDone)602 close(closeDone)
599 }()603 }()
600 604
601- // Close should NOT have completed yet because Write holds the lock.
602- // Give the goroutine time to acquire the lock if it can.
603- time.Sleep(50 * time.Millisecond)
604 select {605 select {
605 case <-closeDone:606 case <-closeDone:
606- t.Fatal("closeAudioStream completed while Write was in-flight — lock not held across Write")607+ // expected: only the short stream snapshot takes the mutex
607- default:608+ case <-time.After(time.Second):
608- // expected: closeAudioStream is blocked609+ t.Fatal("closeAudioStream waited for an in-flight Write")
609 }610 }
610 611
611- // Unblock Write.612+ // The write can finish after shutdown without holding the lifecycle mutex.
612 close(stream.doneCh)613 close(stream.doneCh)
613 614
614- // Now closeAudioStream should complete.
615- <-closeDone
616-
617 assert.True(t, stream.closeCalled)615 assert.True(t, stream.closeCalled)
618 assert.Nil(t, wv.activeAudioStream)616 assert.Nil(t, wv.activeAudioStream)
619 }617 }
M internal/infrastructure/cef/webview_handlers.go +6 -19
@@ -1189,31 +1189,19 @@ func (h *handlerSet) buildAudioStreamFormat(params *purecef.AudioParameters, cha
11891189 }
11901190 }
11911191
1192-// OnAudioStreamPacket receives audio packets and forwards them to the output stream.
1193-// The mutex is held across both the stream snapshot and Write to prevent
1194-// closeAudioStream from closing the stream mid-write.
1192+// OnAudioStreamPacket hands CEF's callback-owned packet to the active output.
1193+// AudioOutputStream takes its one owned copy synchronously. The stream lock is
1194+// only held for the snapshot, so neither copying nor the callback is serialized
1195+// behind lifecycle work; Write and Close are a port-level concurrent contract.
11951196 func (h *handlerSet) OnAudioStreamPacket(_ purecef.Browser, data [][]float32, frames int32, pts int64) {
11961197 if len(data) == 0 || frames <= 0 {
11971198 return
11981199 }
11991200
1200- // Copy the data before acquiring the stream lock because CEF can reuse the
1201- // buffer as soon as this callback returns.
1202- copiedData := make([][]float32, len(data))
1203- for i, channel := range data {
1204- if len(channel) < int(frames) {
1205- continue
1206- }
1207- copiedData[i] = make([]float32, frames)
1208- copy(copiedData[i], channel[:frames])
1209- }
1210-
1211- // Hold the lock across the stream read and Write so closeAudioStream
1212- // cannot close the stream between snapshot and write.
12131201 h.wv.audioStreamMu.Lock()
12141202 stream := h.wv.activeAudioStream
1203+ h.wv.audioStreamMu.Unlock()
12151204 if stream == nil {
1216- h.wv.audioStreamMu.Unlock()
12171205 return
12181206 }
12191207
@@ -1226,8 +1214,7 @@ func (h *handlerSet) OnAudioStreamPacket(_ purecef.Browser, data [][]float32, fr
12261214 Msg("cef: first audio packet received")
12271215 }
12281216
1229- err := stream.Write(copiedData)
1230- h.wv.audioStreamMu.Unlock()
1217+ err := stream.Write(data)
12311218
12321219 if err != nil {
12331220 if h.wv.ctx != nil {
@@ -1189,31 +1189,19 @@ func (h *handlerSet) buildAudioStreamFormat(params *purecef.AudioParameters, cha
1189 }1189 }
1190 }1190 }
1191 1191
1192-// OnAudioStreamPacket receives audio packets and forwards them to the output stream.1192+// OnAudioStreamPacket hands CEF's callback-owned packet to the active output.
1193-// The mutex is held across both the stream snapshot and Write to prevent1193+// AudioOutputStream takes its one owned copy synchronously. The stream lock is
1194-// closeAudioStream from closing the stream mid-write.1194+// only held for the snapshot, so neither copying nor the callback is serialized
1195+// behind lifecycle work; Write and Close are a port-level concurrent contract.
1195 func (h *handlerSet) OnAudioStreamPacket(_ purecef.Browser, data [][]float32, frames int32, pts int64) {1196 func (h *handlerSet) OnAudioStreamPacket(_ purecef.Browser, data [][]float32, frames int32, pts int64) {
1196 if len(data) == 0 || frames <= 0 {1197 if len(data) == 0 || frames <= 0 {
1197 return1198 return
1198 }1199 }
1199 1200
1200- // Copy the data before acquiring the stream lock because CEF can reuse the
1201- // buffer as soon as this callback returns.
1202- copiedData := make([][]float32, len(data))
1203- for i, channel := range data {
1204- if len(channel) < int(frames) {
1205- continue
1206- }
1207- copiedData[i] = make([]float32, frames)
1208- copy(copiedData[i], channel[:frames])
1209- }
1210-
1211- // Hold the lock across the stream read and Write so closeAudioStream
1212- // cannot close the stream between snapshot and write.
1213 h.wv.audioStreamMu.Lock()1201 h.wv.audioStreamMu.Lock()
1214 stream := h.wv.activeAudioStream1202 stream := h.wv.activeAudioStream
1203+ h.wv.audioStreamMu.Unlock()
1215 if stream == nil {1204 if stream == nil {
1216- h.wv.audioStreamMu.Unlock()
1217 return1205 return
1218 }1206 }
1219 1207
@@ -1226,8 +1214,7 @@ func (h *handlerSet) OnAudioStreamPacket(_ purecef.Browser, data [][]float32, fr
1226 Msg("cef: first audio packet received")1214 Msg("cef: first audio packet received")
1227 }1215 }
1228 1216
1229- err := stream.Write(copiedData)1217+ err := stream.Write(data)
1230- h.wv.audioStreamMu.Unlock()
1231 1218
1232 if err != nil {1219 if err != nil {
1233 if h.wv.ctx != nil {1220 if h.wv.ctx != nil {

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 8

Tags 42

↑↓ moveenter openesc close