detached 8b5fe8b

fix: harden PipeWire player runtime and bindings

commit
8b5fe8bf7179f426595b1367201798e4d5acb2ef
author
brice · · 2026-04-11 06:09 UTC
parent
8f9cb67
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
15 files · +434 -157
Unified
UnifiedSplit
M cmd/pipewiregen/internal/emitter/emitter.go +32 -43
@@ -6,6 +6,7 @@ import (
66 "fmt"
77 "os"
88 "path/filepath"
9+ "regexp"
910 "strings"
1011 "text/template"
1112 "unicode"
@@ -14,28 +15,22 @@ import (
1415 )
1516
1617 var (
17- capiTmpl *template.Template
18- portTmpl *template.Template
19- adapterTmpl *template.Template
20- compositeTmpl *template.Template
18+ callbackFieldPattern = regexp.MustCompile(` \*func\([^)]*\)`)
19+ capiTmpl *template.Template
20+ portTmpl *template.Template
21+ adapterTmpl *template.Template
22+ compositeTmpl *template.Template
2123 )
2224
2325 func init() {
24- capiTmpl = template.Must(template.New("capi").Funcs(template.FuncMap{
25- "title": titleCase,
26- }).Parse(capiTemplate))
27-
28- portTmpl = template.Must(template.New("port").Funcs(template.FuncMap{
29- "title": titleCase,
30- }).Parse(portTemplate))
31-
32- adapterTmpl = template.Must(template.New("adapter").Funcs(template.FuncMap{
33- "title": titleCase,
34- }).Parse(adapterTemplate))
26+ capiTmpl = mustTemplate("capi", capiTemplate)
27+ portTmpl = mustTemplate("port", portTemplate)
28+ adapterTmpl = mustTemplate("adapter", adapterTemplate)
29+ compositeTmpl = mustTemplate("composite", compositeTemplate)
30+}
3531
36- compositeTmpl = template.Must(template.New("composite").Funcs(template.FuncMap{
37- "title": titleCase,
38- }).Parse(compositeTemplate))
32+func mustTemplate(name, text string) *template.Template {
33+ return template.Must(template.New(name).Funcs(template.FuncMap{"title": titleCase}).Parse(text))
3934 }
4035
4136 // titleCase converts a string to title case (first letter uppercase, rest lowercase).
@@ -112,17 +107,13 @@ func renderCAPI(group model.Group, symbols []model.Symbol, libraries []model.Lib
112107
113108 // Filter callbacks and event structs for this group
114109 var groupCallbacks []model.Callback
115- var groupEventStructs []model.EventStruct
116110 for _, cb := range callbacks {
117111 if cb.Group == group.Name {
112+ cb.Signature = sanitizeCallbackSignature(cb.Signature)
118113 groupCallbacks = append(groupCallbacks, cb)
119114 }
120115 }
121- for _, es := range eventStructs {
122- if es.Group == group.Name {
123- groupEventStructs = append(groupEventStructs, es)
124- }
125- }
116+ groupEventStructs := filterEventStructsByGroup(eventStructs, group.Name)
126117 // Check if callbacks need unsafe
127118 for _, cb := range groupCallbacks {
128119 if strings.Contains(cb.Signature, "unsafe.Pointer") {
@@ -156,6 +147,10 @@ func renderCAPI(group model.Group, symbols []model.Symbol, libraries []model.Lib
156147 return buf.Bytes(), nil
157148 }
158149
150+func sanitizeCallbackSignature(sig string) string {
151+ return callbackFieldPattern.ReplaceAllString(sig, " uintptr")
152+}
153+
159154 // renderPort generates the outbound interface definitions.
160155 func renderPort(group model.Group, symbols []model.Symbol, eventStructs []model.EventStruct) ([]byte, error) {
161156 groupSymbols := filterSymbolsByGroup(symbols, group.Name)
@@ -186,13 +181,7 @@ func renderPort(group model.Group, symbols []model.Symbol, eventStructs []model.
186181 }
187182 }
188183
189- // Filter event structs for this group
190- var groupEventStructs []model.EventStruct
191- for _, es := range eventStructs {
192- if es.Group == group.Name {
193- groupEventStructs = append(groupEventStructs, es)
194- }
195- }
184+ groupEventStructs := filterEventStructsByGroup(eventStructs, group.Name)
196185
197186 data := struct {
198187 Group model.Group
@@ -304,23 +293,23 @@ func filterSymbolsByGroup(symbols []model.Symbol, groupName string) []model.Symb
304293 return result
305294 }
306295
296+func filterEventStructsByGroup(eventStructs []model.EventStruct, groupName string) []model.EventStruct {
297+ var result []model.EventStruct
298+ for _, es := range eventStructs {
299+ if es.Group == groupName {
300+ result = append(result, es)
301+ }
302+ }
303+ return result
304+}
305+
307306 // toGoName converts a C function name like pw_init to a Go name like PWInit.
308307 func toGoName(cName string) string {
309308 // Handle pw_ prefix specially
310309 if strings.HasPrefix(cName, "pw_") {
311- name := cName[3:]
312- parts := strings.Split(name, "_")
313- for i, p := range parts {
314- parts[i] = titleCase(p)
315- }
316- return "PW" + strings.Join(parts, "")
310+ return "PW" + toCamelCase(cName[3:])
317311 }
318- // Generic conversion
319- parts := strings.Split(cName, "_")
320- for i, p := range parts {
321- parts[i] = titleCase(p)
322- }
323- return strings.Join(parts, "")
312+ return toCamelCase(cName)
324313 }
325314
326315 // toCamelCase converts a snake_case name like stream_playback to StreamPlayback.
@@ -6,6 +6,7 @@ import (
6 "fmt"6 "fmt"
7 "os"7 "os"
8 "path/filepath"8 "path/filepath"
9+ "regexp"
9 "strings"10 "strings"
10 "text/template"11 "text/template"
11 "unicode"12 "unicode"
@@ -14,28 +15,22 @@ import (
14 )15 )
15 16
16 var (17 var (
17- capiTmpl *template.Template18+ callbackFieldPattern = regexp.MustCompile(` \*func\([^)]*\)`)
18- portTmpl *template.Template19+ capiTmpl *template.Template
19- adapterTmpl *template.Template20+ portTmpl *template.Template
20- compositeTmpl *template.Template21+ adapterTmpl *template.Template
22+ compositeTmpl *template.Template
21 )23 )
22 24
23 func init() {25 func init() {
24- capiTmpl = template.Must(template.New("capi").Funcs(template.FuncMap{26+ capiTmpl = mustTemplate("capi", capiTemplate)
25- "title": titleCase,27+ portTmpl = mustTemplate("port", portTemplate)
26- }).Parse(capiTemplate))28+ adapterTmpl = mustTemplate("adapter", adapterTemplate)
27-29+ compositeTmpl = mustTemplate("composite", compositeTemplate)
28- portTmpl = template.Must(template.New("port").Funcs(template.FuncMap{30+}
29- "title": titleCase,
30- }).Parse(portTemplate))
31-
32- adapterTmpl = template.Must(template.New("adapter").Funcs(template.FuncMap{
33- "title": titleCase,
34- }).Parse(adapterTemplate))
35 31
36- compositeTmpl = template.Must(template.New("composite").Funcs(template.FuncMap{32+func mustTemplate(name, text string) *template.Template {
37- "title": titleCase,33+ return template.Must(template.New(name).Funcs(template.FuncMap{"title": titleCase}).Parse(text))
38- }).Parse(compositeTemplate))
39 }34 }
40 35
41 // titleCase converts a string to title case (first letter uppercase, rest lowercase).36 // titleCase converts a string to title case (first letter uppercase, rest lowercase).
@@ -112,17 +107,13 @@ func renderCAPI(group model.Group, symbols []model.Symbol, libraries []model.Lib
112 107
113 // Filter callbacks and event structs for this group108 // Filter callbacks and event structs for this group
114 var groupCallbacks []model.Callback109 var groupCallbacks []model.Callback
115- var groupEventStructs []model.EventStruct
116 for _, cb := range callbacks {110 for _, cb := range callbacks {
117 if cb.Group == group.Name {111 if cb.Group == group.Name {
112+ cb.Signature = sanitizeCallbackSignature(cb.Signature)
118 groupCallbacks = append(groupCallbacks, cb)113 groupCallbacks = append(groupCallbacks, cb)
119 }114 }
120 }115 }
121- for _, es := range eventStructs {116+ groupEventStructs := filterEventStructsByGroup(eventStructs, group.Name)
122- if es.Group == group.Name {
123- groupEventStructs = append(groupEventStructs, es)
124- }
125- }
126 // Check if callbacks need unsafe117 // Check if callbacks need unsafe
127 for _, cb := range groupCallbacks {118 for _, cb := range groupCallbacks {
128 if strings.Contains(cb.Signature, "unsafe.Pointer") {119 if strings.Contains(cb.Signature, "unsafe.Pointer") {
@@ -156,6 +147,10 @@ func renderCAPI(group model.Group, symbols []model.Symbol, libraries []model.Lib
156 return buf.Bytes(), nil147 return buf.Bytes(), nil
157 }148 }
158 149
150+func sanitizeCallbackSignature(sig string) string {
151+ return callbackFieldPattern.ReplaceAllString(sig, " uintptr")
152+}
153+
159 // renderPort generates the outbound interface definitions.154 // renderPort generates the outbound interface definitions.
160 func renderPort(group model.Group, symbols []model.Symbol, eventStructs []model.EventStruct) ([]byte, error) {155 func renderPort(group model.Group, symbols []model.Symbol, eventStructs []model.EventStruct) ([]byte, error) {
161 groupSymbols := filterSymbolsByGroup(symbols, group.Name)156 groupSymbols := filterSymbolsByGroup(symbols, group.Name)
@@ -186,13 +181,7 @@ func renderPort(group model.Group, symbols []model.Symbol, eventStructs []model.
186 }181 }
187 }182 }
188 183
189- // Filter event structs for this group184+ groupEventStructs := filterEventStructsByGroup(eventStructs, group.Name)
190- var groupEventStructs []model.EventStruct
191- for _, es := range eventStructs {
192- if es.Group == group.Name {
193- groupEventStructs = append(groupEventStructs, es)
194- }
195- }
196 185
197 data := struct {186 data := struct {
198 Group model.Group187 Group model.Group
@@ -304,23 +293,23 @@ func filterSymbolsByGroup(symbols []model.Symbol, groupName string) []model.Symb
304 return result293 return result
305 }294 }
306 295
296+func filterEventStructsByGroup(eventStructs []model.EventStruct, groupName string) []model.EventStruct {
297+ var result []model.EventStruct
298+ for _, es := range eventStructs {
299+ if es.Group == groupName {
300+ result = append(result, es)
301+ }
302+ }
303+ return result
304+}
305+
307 // toGoName converts a C function name like pw_init to a Go name like PWInit.306 // toGoName converts a C function name like pw_init to a Go name like PWInit.
308 func toGoName(cName string) string {307 func toGoName(cName string) string {
309 // Handle pw_ prefix specially308 // Handle pw_ prefix specially
310 if strings.HasPrefix(cName, "pw_") {309 if strings.HasPrefix(cName, "pw_") {
311- name := cName[3:]310+ return "PW" + toCamelCase(cName[3:])
312- parts := strings.Split(name, "_")
313- for i, p := range parts {
314- parts[i] = titleCase(p)
315- }
316- return "PW" + strings.Join(parts, "")
317 }311 }
318- // Generic conversion312+ return toCamelCase(cName)
319- parts := strings.Split(cName, "_")
320- for i, p := range parts {
321- parts[i] = titleCase(p)
322- }
323- return strings.Join(parts, "")
324 }313 }
325 314
326 // toCamelCase converts a snake_case name like stream_playback to StreamPlayback.315 // toCamelCase converts a snake_case name like stream_playback to StreamPlayback.
M cmd/pipewiregen/internal/emitter/emitter_test.go +16 -12
@@ -114,17 +114,17 @@ func TestPWStreamEventsABILayout(t *testing.T) {
114114 // The correct C ABI field order from <pipewire/stream.h>
115115 expectedFields := []string{
116116 "version uint32",
117- "destroy *func(",
118- "state_changed *func(",
119- "control_info *func(",
120- "io_changed *func(",
121- "param_changed *func(",
122- "add_buffer *func(",
123- "remove_buffer *func(",
124- "process *func(",
125- "drained *func(",
126- "command *func(",
127- "trigger_done *func(",
117+ "destroy uintptr",
118+ "state_changed uintptr",
119+ "control_info uintptr",
120+ "io_changed uintptr",
121+ "param_changed uintptr",
122+ "add_buffer uintptr",
123+ "remove_buffer uintptr",
124+ "process uintptr",
125+ "drained uintptr",
126+ "command uintptr",
127+ "trigger_done uintptr",
128128 }
129129
130130 // Verify all expected fields are present
@@ -135,10 +135,14 @@ func TestPWStreamEventsABILayout(t *testing.T) {
135135 }
136136
137137 // Verify the spurious "flush" field is NOT present
138- if strings.Contains(capiContent, "flush *func(") {
138+ if strings.Contains(capiContent, "flush uintptr") {
139139 t.Error("pw_stream_events contains spurious 'flush' field — this does not exist in the PipeWire C ABI")
140140 }
141141
142+ if strings.Contains(capiContent, "*func(") {
143+ t.Error("pw_stream_events callback fields must be emitted as uintptr function pointers, not Go func pointers")
144+ }
145+
142146 // Verify field order: each field must appear after the previous one
143147 lastIdx := 0
144148 for _, field := range expectedFields {
@@ -114,17 +114,17 @@ func TestPWStreamEventsABILayout(t *testing.T) {
114 // The correct C ABI field order from <pipewire/stream.h>114 // The correct C ABI field order from <pipewire/stream.h>
115 expectedFields := []string{115 expectedFields := []string{
116 "version uint32",116 "version uint32",
117- "destroy *func(",117+ "destroy uintptr",
118- "state_changed *func(",118+ "state_changed uintptr",
119- "control_info *func(",119+ "control_info uintptr",
120- "io_changed *func(",120+ "io_changed uintptr",
121- "param_changed *func(",121+ "param_changed uintptr",
122- "add_buffer *func(",122+ "add_buffer uintptr",
123- "remove_buffer *func(",123+ "remove_buffer uintptr",
124- "process *func(",124+ "process uintptr",
125- "drained *func(",125+ "drained uintptr",
126- "command *func(",126+ "command uintptr",
127- "trigger_done *func(",127+ "trigger_done uintptr",
128 }128 }
129 129
130 // Verify all expected fields are present130 // Verify all expected fields are present
@@ -135,10 +135,14 @@ func TestPWStreamEventsABILayout(t *testing.T) {
135 }135 }
136 136
137 // Verify the spurious "flush" field is NOT present137 // Verify the spurious "flush" field is NOT present
138- if strings.Contains(capiContent, "flush *func(") {138+ if strings.Contains(capiContent, "flush uintptr") {
139 t.Error("pw_stream_events contains spurious 'flush' field — this does not exist in the PipeWire C ABI")139 t.Error("pw_stream_events contains spurious 'flush' field — this does not exist in the PipeWire C ABI")
140 }140 }
141 141
142+ if strings.Contains(capiContent, "*func(") {
143+ t.Error("pw_stream_events callback fields must be emitted as uintptr function pointers, not Go func pointers")
144+ }
145+
142 // Verify field order: each field must appear after the previous one146 // Verify field order: each field must appear after the previous one
143 lastIdx := 0147 lastIdx := 0
144 for _, field := range expectedFields {148 for _, field := range expectedFields {
M cmd/pipewiregen/internal/parser/testdata/callback_unknown_group.json +1 -1
@@ -11,4 +11,4 @@
1111 "callbacks": [
1212 {"name": "pw_stream_events", "signature": "struct{}", "group": "nonexistent"}
1313 ]
14-}⏎̸
14+}
@@ -11,4 +11,4 @@
11 "callbacks": [11 "callbacks": [
12 {"name": "pw_stream_events", "signature": "struct{}", "group": "nonexistent"}12 {"name": "pw_stream_events", "signature": "struct{}", "group": "nonexistent"}
13 ]13 ]
14-}⏎̸14+}
M internal/capi/stream_ops.go +6 -12
@@ -6,6 +6,8 @@ import (
66 "sync"
77 "unsafe"
88
9+ "github.com/ebitengine/purego"
10+
911 portout "github.com/bnema/purego-pipewire/internal/ports/out"
1012 )
1113
@@ -43,7 +45,7 @@ const (
4345 // Without this, the stack-allocated values in CreatePlaybackStream
4446 // would be collected while PipeWire still holds pointers to them.
4547 type streamCallbackStorage struct {
46- processCb func(unsafe.Pointer)
48+ processCb uintptr
4749 events pw_stream_events
4850 }
4951
@@ -78,12 +80,12 @@ func (s *streamOpsImpl) CreatePlaybackStream(loopPtr unsafe.Pointer, name string
7880
7981 // Build persistent callback storage that outlives this function.
8082 storage := &streamCallbackStorage{}
81- storage.processCb = func(_ unsafe.Pointer) {
83+ storage.processCb = purego.NewCallback(func(_ unsafe.Pointer) {
8284 onProcess()
83- }
85+ })
8486 storage.events = pw_stream_events{
8587 version: pwVersionStreamEvents,
86- process: &storage.processCb,
88+ process: storage.processCb,
8789 }
8890
8991 ptr := pw_stream_new_simple(
@@ -109,14 +111,6 @@ func (s *streamOpsImpl) CreatePlaybackStream(loopPtr unsafe.Pointer, name string
109111 }
110112
111113 func (s *streamOpsImpl) ConnectPlaybackStream(streamPtr unsafe.Pointer, format portout.PlaybackFormat) error {
112- // Validate format before making the C call.
113- // FramesPerBuffer is not used by buildRawAudioParams but is still required
114- // for a well-formed playback setup.
115- if format.SampleRate <= 0 || format.Channels <= 0 || format.FramesPerBuffer <= 0 {
116- return fmt.Errorf("%w: SampleRate=%d, Channels=%d, FramesPerBuffer=%d",
117- ErrInvalidPlaybackFormat, format.SampleRate, format.Channels, format.FramesPerBuffer)
118- }
119-
120114 // Build the SPA POD params that describe the stream's audio format.
121115 cp, err := buildRawAudioParams(format)
122116 if err != nil {
@@ -6,6 +6,8 @@ import (
6 "sync"6 "sync"
7 "unsafe"7 "unsafe"
8 8
9+ "github.com/ebitengine/purego"
10+
9 portout "github.com/bnema/purego-pipewire/internal/ports/out"11 portout "github.com/bnema/purego-pipewire/internal/ports/out"
10 )12 )
11 13
@@ -43,7 +45,7 @@ const (
43 // Without this, the stack-allocated values in CreatePlaybackStream45 // Without this, the stack-allocated values in CreatePlaybackStream
44 // would be collected while PipeWire still holds pointers to them.46 // would be collected while PipeWire still holds pointers to them.
45 type streamCallbackStorage struct {47 type streamCallbackStorage struct {
46- processCb func(unsafe.Pointer)48+ processCb uintptr
47 events pw_stream_events49 events pw_stream_events
48 }50 }
49 51
@@ -78,12 +80,12 @@ func (s *streamOpsImpl) CreatePlaybackStream(loopPtr unsafe.Pointer, name string
78 80
79 // Build persistent callback storage that outlives this function.81 // Build persistent callback storage that outlives this function.
80 storage := &streamCallbackStorage{}82 storage := &streamCallbackStorage{}
81- storage.processCb = func(_ unsafe.Pointer) {83+ storage.processCb = purego.NewCallback(func(_ unsafe.Pointer) {
82 onProcess()84 onProcess()
83- }85+ })
84 storage.events = pw_stream_events{86 storage.events = pw_stream_events{
85 version: pwVersionStreamEvents,87 version: pwVersionStreamEvents,
86- process: &storage.processCb,88+ process: storage.processCb,
87 }89 }
88 90
89 ptr := pw_stream_new_simple(91 ptr := pw_stream_new_simple(
@@ -109,14 +111,6 @@ func (s *streamOpsImpl) CreatePlaybackStream(loopPtr unsafe.Pointer, name string
109 }111 }
110 112
111 func (s *streamOpsImpl) ConnectPlaybackStream(streamPtr unsafe.Pointer, format portout.PlaybackFormat) error {113 func (s *streamOpsImpl) ConnectPlaybackStream(streamPtr unsafe.Pointer, format portout.PlaybackFormat) error {
112- // Validate format before making the C call.
113- // FramesPerBuffer is not used by buildRawAudioParams but is still required
114- // for a well-formed playback setup.
115- if format.SampleRate <= 0 || format.Channels <= 0 || format.FramesPerBuffer <= 0 {
116- return fmt.Errorf("%w: SampleRate=%d, Channels=%d, FramesPerBuffer=%d",
117- ErrInvalidPlaybackFormat, format.SampleRate, format.Channels, format.FramesPerBuffer)
118- }
119-
120 // Build the SPA POD params that describe the stream's audio format.114 // Build the SPA POD params that describe the stream's audio format.
121 cp, err := buildRawAudioParams(format)115 cp, err := buildRawAudioParams(format)
122 if err != nil {116 if err != nil {
M internal/capi/stream_ops_test.go +79 -11
@@ -2,6 +2,8 @@ package capi
22
33 import (
44 "errors"
5+ "reflect"
6+ "strings"
57 "testing"
68 "unsafe"
79
@@ -44,6 +46,18 @@ func TestDestroyStreamCallsPwStreamDestroy(t *testing.T) {
4446 }
4547 }
4648
49+// TestPWStreamEventsUsesFunctionPointerABI verifies that the generated callback
50+// fields use raw function-pointer storage, not Go func pointers.
51+func TestPWStreamEventsUsesFunctionPointerABI(t *testing.T) {
52+ field, ok := reflect.TypeOf(pw_stream_events{}).FieldByName("process")
53+ if !ok {
54+ t.Fatal("pw_stream_events.process field missing")
55+ }
56+ if field.Type.Kind() != reflect.Uintptr {
57+ t.Fatalf("pw_stream_events.process has kind %s, want uintptr", field.Type.Kind())
58+ }
59+}
60+
4761 // TestDestroyStreamDoubleDestroySkipsSecondCall verifies idempotency:
4862 // a second DestroyStream on the same pointer does NOT call pw_stream_destroy again.
4963 func TestDestroyStreamDoubleDestroySkipsSecondCall(t *testing.T) {
@@ -267,32 +281,86 @@ func TestDisconnectStreamReturnsNilOnSuccess(t *testing.T) {
267281 }
268282 }
269283
270-// TestConnectPlaybackStreamRejectsMissingFormat verifies that ConnectPlaybackStream
271-// rejects a zero-valued PlaybackFormat at the StreamOps layer by returning
272-// ErrInvalidPlaybackFormat. This is a temporary guard; once SPA params are
273-// wired, the rejection will also cover missing/invalid SPA param construction.
274-func TestConnectPlaybackStreamRejectsMissingFormat(t *testing.T) {
284+// TestConnectPlaybackStreamDelegatesFramesPerBufferValidation verifies that
285+// ConnectPlaybackStream rejects invalid FramesPerBuffer values via the helper
286+// and does not reach pw_stream_connect.
287+func TestConnectPlaybackStreamDelegatesFramesPerBufferValidation(t *testing.T) {
275288 origConnect := pw_stream_connect
276289 t.Cleanup(func() { pw_stream_connect = origConnect })
277290
278- // Even if pw_stream_connect succeeds, a zero PlaybackFormat should be rejected
279- // at the StreamOps layer before the C call is made.
291+ var called bool
280292 pw_stream_connect = func(stream unsafe.Pointer, direction int32, id uint32, flags uint32, ports unsafe.Pointer, n_ports uint32) int32 {
293+ called = true
281294 return 0
282295 }
283296
284297 ops := &streamOpsImpl{}
285298 fakePtr := unsafe.Pointer(uintptr(0x1234))
286299
287- // Zero-valued PlaybackFormat must be rejected.
288- zeroFmt := portout.PlaybackFormat{}
289- err := ops.ConnectPlaybackStream(fakePtr, zeroFmt)
300+ fmt := portout.PlaybackFormat{SampleRate: 48000, Channels: 2, FramesPerBuffer: 0}
301+ err := ops.ConnectPlaybackStream(fakePtr, fmt)
290302 if err == nil {
291- t.Fatal("expected error for zero-valued PlaybackFormat, got nil")
303+ t.Fatal("expected error for missing FramesPerBuffer, got nil")
292304 }
293305 if !errors.Is(err, ErrInvalidPlaybackFormat) {
294306 t.Fatalf("expected error wrapping ErrInvalidPlaybackFormat, got: %v", err)
295307 }
308+ if !strings.Contains(err.Error(), "frames per buffer") {
309+ t.Fatalf("expected helper FramesPerBuffer validation, got: %v", err)
310+ }
311+ if called {
312+ t.Fatal("pw_stream_connect should not be called for invalid FramesPerBuffer")
313+ }
314+}
315+
316+// TestConnectPlaybackStreamDelegatesSampleRateAndChannelsValidation verifies
317+// that invalid sample rate and channel values are rejected by buildRawAudioParams.
318+func TestConnectPlaybackStreamDelegatesSampleRateAndChannelsValidation(t *testing.T) {
319+ origConnect := pw_stream_connect
320+ t.Cleanup(func() { pw_stream_connect = origConnect })
321+
322+ pw_stream_connect = func(stream unsafe.Pointer, direction int32, id uint32, flags uint32, ports unsafe.Pointer, n_ports uint32) int32 {
323+ t.Fatal("pw_stream_connect should not be called for invalid sample rate or channels")
324+ return 0
325+ }
326+
327+ ops := &streamOpsImpl{}
328+ fakePtr := unsafe.Pointer(uintptr(0x1234))
329+
330+ tests := []struct {
331+ name string
332+ fmt portout.PlaybackFormat
333+ wantMsg string
334+ }{
335+ {
336+ name: "zero_sample_rate",
337+ fmt: portout.PlaybackFormat{SampleRate: 0, Channels: 2, FramesPerBuffer: 1024},
338+ wantMsg: "sample rate must be positive",
339+ },
340+ {
341+ name: "zero_channels",
342+ fmt: portout.PlaybackFormat{SampleRate: 48000, Channels: 0, FramesPerBuffer: 1024},
343+ wantMsg: "channels must be positive",
344+ },
345+ }
346+
347+ for _, tt := range tests {
348+ t.Run(tt.name, func(t *testing.T) {
349+ err := ops.ConnectPlaybackStream(fakePtr, tt.fmt)
350+ if err == nil {
351+ t.Fatalf("expected error for %s, got nil", tt.name)
352+ }
353+ if !errors.Is(err, ErrInvalidPlaybackFormat) {
354+ t.Fatalf("expected error wrapping ErrInvalidPlaybackFormat, got: %v", err)
355+ }
356+ if !strings.Contains(err.Error(), tt.wantMsg) {
357+ t.Fatalf("expected error to contain %q, got: %v", tt.wantMsg, err)
358+ }
359+ if strings.Contains(err.Error(), "FramesPerBuffer") {
360+ t.Fatalf("expected helper validation to report only sample rate/channels, got: %v", err)
361+ }
362+ })
363+ }
296364 }
297365
298366 // TestConnectPlaybackStreamAcceptsValidFormat verifies that ConnectPlaybackStream
@@ -2,6 +2,8 @@ package capi
2 2
3 import (3 import (
4 "errors"4 "errors"
5+ "reflect"
6+ "strings"
5 "testing"7 "testing"
6 "unsafe"8 "unsafe"
7 9
@@ -44,6 +46,18 @@ func TestDestroyStreamCallsPwStreamDestroy(t *testing.T) {
44 }46 }
45 }47 }
46 48
49+// TestPWStreamEventsUsesFunctionPointerABI verifies that the generated callback
50+// fields use raw function-pointer storage, not Go func pointers.
51+func TestPWStreamEventsUsesFunctionPointerABI(t *testing.T) {
52+ field, ok := reflect.TypeOf(pw_stream_events{}).FieldByName("process")
53+ if !ok {
54+ t.Fatal("pw_stream_events.process field missing")
55+ }
56+ if field.Type.Kind() != reflect.Uintptr {
57+ t.Fatalf("pw_stream_events.process has kind %s, want uintptr", field.Type.Kind())
58+ }
59+}
60+
47 // TestDestroyStreamDoubleDestroySkipsSecondCall verifies idempotency:61 // TestDestroyStreamDoubleDestroySkipsSecondCall verifies idempotency:
48 // a second DestroyStream on the same pointer does NOT call pw_stream_destroy again.62 // a second DestroyStream on the same pointer does NOT call pw_stream_destroy again.
49 func TestDestroyStreamDoubleDestroySkipsSecondCall(t *testing.T) {63 func TestDestroyStreamDoubleDestroySkipsSecondCall(t *testing.T) {
@@ -267,32 +281,86 @@ func TestDisconnectStreamReturnsNilOnSuccess(t *testing.T) {
267 }281 }
268 }282 }
269 283
270-// TestConnectPlaybackStreamRejectsMissingFormat verifies that ConnectPlaybackStream284+// TestConnectPlaybackStreamDelegatesFramesPerBufferValidation verifies that
271-// rejects a zero-valued PlaybackFormat at the StreamOps layer by returning285+// ConnectPlaybackStream rejects invalid FramesPerBuffer values via the helper
272-// ErrInvalidPlaybackFormat. This is a temporary guard; once SPA params are286+// and does not reach pw_stream_connect.
273-// wired, the rejection will also cover missing/invalid SPA param construction.287+func TestConnectPlaybackStreamDelegatesFramesPerBufferValidation(t *testing.T) {
274-func TestConnectPlaybackStreamRejectsMissingFormat(t *testing.T) {
275 origConnect := pw_stream_connect288 origConnect := pw_stream_connect
276 t.Cleanup(func() { pw_stream_connect = origConnect })289 t.Cleanup(func() { pw_stream_connect = origConnect })
277 290
278- // Even if pw_stream_connect succeeds, a zero PlaybackFormat should be rejected291+ var called bool
279- // at the StreamOps layer before the C call is made.
280 pw_stream_connect = func(stream unsafe.Pointer, direction int32, id uint32, flags uint32, ports unsafe.Pointer, n_ports uint32) int32 {292 pw_stream_connect = func(stream unsafe.Pointer, direction int32, id uint32, flags uint32, ports unsafe.Pointer, n_ports uint32) int32 {
293+ called = true
281 return 0294 return 0
282 }295 }
283 296
284 ops := &streamOpsImpl{}297 ops := &streamOpsImpl{}
285 fakePtr := unsafe.Pointer(uintptr(0x1234))298 fakePtr := unsafe.Pointer(uintptr(0x1234))
286 299
287- // Zero-valued PlaybackFormat must be rejected.300+ fmt := portout.PlaybackFormat{SampleRate: 48000, Channels: 2, FramesPerBuffer: 0}
288- zeroFmt := portout.PlaybackFormat{}301+ err := ops.ConnectPlaybackStream(fakePtr, fmt)
289- err := ops.ConnectPlaybackStream(fakePtr, zeroFmt)
290 if err == nil {302 if err == nil {
291- t.Fatal("expected error for zero-valued PlaybackFormat, got nil")303+ t.Fatal("expected error for missing FramesPerBuffer, got nil")
292 }304 }
293 if !errors.Is(err, ErrInvalidPlaybackFormat) {305 if !errors.Is(err, ErrInvalidPlaybackFormat) {
294 t.Fatalf("expected error wrapping ErrInvalidPlaybackFormat, got: %v", err)306 t.Fatalf("expected error wrapping ErrInvalidPlaybackFormat, got: %v", err)
295 }307 }
308+ if !strings.Contains(err.Error(), "frames per buffer") {
309+ t.Fatalf("expected helper FramesPerBuffer validation, got: %v", err)
310+ }
311+ if called {
312+ t.Fatal("pw_stream_connect should not be called for invalid FramesPerBuffer")
313+ }
314+}
315+
316+// TestConnectPlaybackStreamDelegatesSampleRateAndChannelsValidation verifies
317+// that invalid sample rate and channel values are rejected by buildRawAudioParams.
318+func TestConnectPlaybackStreamDelegatesSampleRateAndChannelsValidation(t *testing.T) {
319+ origConnect := pw_stream_connect
320+ t.Cleanup(func() { pw_stream_connect = origConnect })
321+
322+ pw_stream_connect = func(stream unsafe.Pointer, direction int32, id uint32, flags uint32, ports unsafe.Pointer, n_ports uint32) int32 {
323+ t.Fatal("pw_stream_connect should not be called for invalid sample rate or channels")
324+ return 0
325+ }
326+
327+ ops := &streamOpsImpl{}
328+ fakePtr := unsafe.Pointer(uintptr(0x1234))
329+
330+ tests := []struct {
331+ name string
332+ fmt portout.PlaybackFormat
333+ wantMsg string
334+ }{
335+ {
336+ name: "zero_sample_rate",
337+ fmt: portout.PlaybackFormat{SampleRate: 0, Channels: 2, FramesPerBuffer: 1024},
338+ wantMsg: "sample rate must be positive",
339+ },
340+ {
341+ name: "zero_channels",
342+ fmt: portout.PlaybackFormat{SampleRate: 48000, Channels: 0, FramesPerBuffer: 1024},
343+ wantMsg: "channels must be positive",
344+ },
345+ }
346+
347+ for _, tt := range tests {
348+ t.Run(tt.name, func(t *testing.T) {
349+ err := ops.ConnectPlaybackStream(fakePtr, tt.fmt)
350+ if err == nil {
351+ t.Fatalf("expected error for %s, got nil", tt.name)
352+ }
353+ if !errors.Is(err, ErrInvalidPlaybackFormat) {
354+ t.Fatalf("expected error wrapping ErrInvalidPlaybackFormat, got: %v", err)
355+ }
356+ if !strings.Contains(err.Error(), tt.wantMsg) {
357+ t.Fatalf("expected error to contain %q, got: %v", tt.wantMsg, err)
358+ }
359+ if strings.Contains(err.Error(), "FramesPerBuffer") {
360+ t.Fatalf("expected helper validation to report only sample rate/channels, got: %v", err)
361+ }
362+ })
363+ }
296 }364 }
297 365
298 // TestConnectPlaybackStreamAcceptsValidFormat verifies that ConnectPlaybackStream366 // TestConnectPlaybackStreamAcceptsValidFormat verifies that ConnectPlaybackStream
M internal/capi/stream_params.go +5 -3
@@ -10,7 +10,7 @@ import (
1010
1111 // SPA POD type identifiers (from spa/utils/type.h).
1212 const (
13- spaTypeNone uint32 = iota + 1 // 1
13+ spaTypeNone uint32 = iota + 1 // 1; reserved so iota matches SPA type numbering
1414 spaTypeBool // 2
1515 spaTypeId // 3
1616 spaTypeInt // 4
@@ -44,7 +44,6 @@ const (
4444
4545 // Audio format keys (SPA_FORMAT_START_Audio = 0x10000).
4646 spaFormatAudioFormat uint32 = 0x10001 // SPA_FORMAT_AUDIO_format
47- spaFormatAudioFlags uint32 = 0x10002 // SPA_FORMAT_AUDIO_flags
4847 spaFormatAudioRate uint32 = 0x10003 // SPA_FORMAT_AUDIO_rate
4948 spaFormatAudioChannels uint32 = 0x10004 // SPA_FORMAT_AUDIO_channels
5049 )
@@ -53,7 +52,6 @@ const (
5352 const (
5453 spaMediaTypeAudio uint32 = 1 // SPA_MEDIA_TYPE_audio
5554 spaMediaSubtypeRaw uint32 = 1 // SPA_MEDIA_SUBTYPE_raw
56- spaMediaSubtypeDSP uint32 = 2 // SPA_MEDIA_SUBTYPE_dsp
5755 )
5856
5957 // SPA audio format identifiers (from spa/param/audio/raw.h).
@@ -79,6 +77,10 @@ func buildRawAudioParams(format portout.PlaybackFormat) (*connectParams, error)
7977 return nil, fmt.Errorf("%w: channels must be positive, got %d",
8078 ErrInvalidPlaybackFormat, format.Channels)
8179 }
80+ if format.FramesPerBuffer <= 0 {
81+ return nil, fmt.Errorf("%w: frames per buffer must be positive, got %d",
82+ ErrInvalidPlaybackFormat, format.FramesPerBuffer)
83+ }
8284
8385 // Layout of an SPA POD Object for EnumFormat with 5 properties:
8486 //
@@ -10,7 +10,7 @@ import (
10 10
11 // SPA POD type identifiers (from spa/utils/type.h).11 // SPA POD type identifiers (from spa/utils/type.h).
12 const (12 const (
13- spaTypeNone uint32 = iota + 1 // 113+ spaTypeNone uint32 = iota + 1 // 1; reserved so iota matches SPA type numbering
14 spaTypeBool // 214 spaTypeBool // 2
15 spaTypeId // 315 spaTypeId // 3
16 spaTypeInt // 416 spaTypeInt // 4
@@ -44,7 +44,6 @@ const (
44 44
45 // Audio format keys (SPA_FORMAT_START_Audio = 0x10000).45 // Audio format keys (SPA_FORMAT_START_Audio = 0x10000).
46 spaFormatAudioFormat uint32 = 0x10001 // SPA_FORMAT_AUDIO_format46 spaFormatAudioFormat uint32 = 0x10001 // SPA_FORMAT_AUDIO_format
47- spaFormatAudioFlags uint32 = 0x10002 // SPA_FORMAT_AUDIO_flags
48 spaFormatAudioRate uint32 = 0x10003 // SPA_FORMAT_AUDIO_rate47 spaFormatAudioRate uint32 = 0x10003 // SPA_FORMAT_AUDIO_rate
49 spaFormatAudioChannels uint32 = 0x10004 // SPA_FORMAT_AUDIO_channels48 spaFormatAudioChannels uint32 = 0x10004 // SPA_FORMAT_AUDIO_channels
50 )49 )
@@ -53,7 +52,6 @@ const (
53 const (52 const (
54 spaMediaTypeAudio uint32 = 1 // SPA_MEDIA_TYPE_audio53 spaMediaTypeAudio uint32 = 1 // SPA_MEDIA_TYPE_audio
55 spaMediaSubtypeRaw uint32 = 1 // SPA_MEDIA_SUBTYPE_raw54 spaMediaSubtypeRaw uint32 = 1 // SPA_MEDIA_SUBTYPE_raw
56- spaMediaSubtypeDSP uint32 = 2 // SPA_MEDIA_SUBTYPE_dsp
57 )55 )
58 56
59 // SPA audio format identifiers (from spa/param/audio/raw.h).57 // SPA audio format identifiers (from spa/param/audio/raw.h).
@@ -79,6 +77,10 @@ func buildRawAudioParams(format portout.PlaybackFormat) (*connectParams, error)
79 return nil, fmt.Errorf("%w: channels must be positive, got %d",77 return nil, fmt.Errorf("%w: channels must be positive, got %d",
80 ErrInvalidPlaybackFormat, format.Channels)78 ErrInvalidPlaybackFormat, format.Channels)
81 }79 }
80+ if format.FramesPerBuffer <= 0 {
81+ return nil, fmt.Errorf("%w: frames per buffer must be positive, got %d",
82+ ErrInvalidPlaybackFormat, format.FramesPerBuffer)
83+ }
82 84
83 // Layout of an SPA POD Object for EnumFormat with 5 properties:85 // Layout of an SPA POD Object for EnumFormat with 5 properties:
84 //86 //
M internal/capi/stream_params_test.go +13 -3
@@ -49,10 +49,10 @@ func TestBuildRawAudioParamsProducesOneConnectParam(t *testing.T) {
4949 objType := binary.LittleEndian.Uint32(cp.storage[8:12])
5050 objID := binary.LittleEndian.Uint32(cp.storage[12:16])
5151
52- const spaTypeObjectFORMAT uint32 = 0x40003
52+ const wantSpaTypeObjectFormat uint32 = 0x40003
5353 const spaParamEnumFormat uint32 = 3
54- if objType != spaTypeObjectFormat {
55- t.Fatalf("expected object type SPA_TYPE_OBJECT_Format (0x%x), got 0x%x", spaTypeObjectFormat, objType)
54+ if objType != wantSpaTypeObjectFormat {
55+ t.Fatalf("expected object type SPA_TYPE_OBJECT_Format (0x%x), got 0x%x", wantSpaTypeObjectFormat, objType)
5656 }
5757 if objID != spaParamEnumFormat {
5858 t.Fatalf("expected object id SPA_PARAM_EnumFormat (%d), got %d", spaParamEnumFormat, objID)
@@ -114,6 +114,16 @@ func TestBuildRawAudioParamsRejectsInvalidFormat(t *testing.T) {
114114 fmt: portout.PlaybackFormat{SampleRate: 48000, Channels: -1, FramesPerBuffer: 1024},
115115 errMsg: "channels",
116116 },
117+ {
118+ name: "zero_frames_per_buffer",
119+ fmt: portout.PlaybackFormat{SampleRate: 48000, Channels: 2, FramesPerBuffer: 0},
120+ errMsg: "frames per buffer",
121+ },
122+ {
123+ name: "negative_frames_per_buffer",
124+ fmt: portout.PlaybackFormat{SampleRate: 48000, Channels: 2, FramesPerBuffer: -1},
125+ errMsg: "frames per buffer",
126+ },
117127 }
118128
119129 for _, tt := range tests {
@@ -49,10 +49,10 @@ func TestBuildRawAudioParamsProducesOneConnectParam(t *testing.T) {
49 objType := binary.LittleEndian.Uint32(cp.storage[8:12])49 objType := binary.LittleEndian.Uint32(cp.storage[8:12])
50 objID := binary.LittleEndian.Uint32(cp.storage[12:16])50 objID := binary.LittleEndian.Uint32(cp.storage[12:16])
51 51
52- const spaTypeObjectFORMAT uint32 = 0x4000352+ const wantSpaTypeObjectFormat uint32 = 0x40003
53 const spaParamEnumFormat uint32 = 353 const spaParamEnumFormat uint32 = 3
54- if objType != spaTypeObjectFormat {54+ if objType != wantSpaTypeObjectFormat {
55- t.Fatalf("expected object type SPA_TYPE_OBJECT_Format (0x%x), got 0x%x", spaTypeObjectFormat, objType)55+ t.Fatalf("expected object type SPA_TYPE_OBJECT_Format (0x%x), got 0x%x", wantSpaTypeObjectFormat, objType)
56 }56 }
57 if objID != spaParamEnumFormat {57 if objID != spaParamEnumFormat {
58 t.Fatalf("expected object id SPA_PARAM_EnumFormat (%d), got %d", spaParamEnumFormat, objID)58 t.Fatalf("expected object id SPA_PARAM_EnumFormat (%d), got %d", spaParamEnumFormat, objID)
@@ -114,6 +114,16 @@ func TestBuildRawAudioParamsRejectsInvalidFormat(t *testing.T) {
114 fmt: portout.PlaybackFormat{SampleRate: 48000, Channels: -1, FramesPerBuffer: 1024},114 fmt: portout.PlaybackFormat{SampleRate: 48000, Channels: -1, FramesPerBuffer: 1024},
115 errMsg: "channels",115 errMsg: "channels",
116 },116 },
117+ {
118+ name: "zero_frames_per_buffer",
119+ fmt: portout.PlaybackFormat{SampleRate: 48000, Channels: 2, FramesPerBuffer: 0},
120+ errMsg: "frames per buffer",
121+ },
122+ {
123+ name: "negative_frames_per_buffer",
124+ fmt: portout.PlaybackFormat{SampleRate: 48000, Channels: 2, FramesPerBuffer: -1},
125+ errMsg: "frames per buffer",
126+ },
117 }127 }
118 128
119 for _, tt := range tests {129 for _, tt := range tests {
M internal/capi/stream_playback_gen.go +1 -1 generated file: collapsed
load diff
M internal/core/player.go +15 -11
@@ -129,7 +129,7 @@ func (p *player) Start() error {
129129
130130 // Check if we already have PipeWire resources (restart after Stop)
131131 p.mu.Lock()
132- hasResources := p.loopPtr != nil && p.streamPtr != nil
132+ hasResources := p.loopPtr != nil && atomic.LoadPointer((*unsafe.Pointer)(unsafe.Pointer(&p.streamPtr))) != nil
133133 p.mu.Unlock()
134134
135135 if !hasResources {
@@ -139,9 +139,7 @@ func (p *player) Start() error {
139139 }
140140 } else {
141141 // Restart — just reactivate existing stream
142- p.mu.Lock()
143- streamPtr := p.streamPtr
144- p.mu.Unlock()
142+ streamPtr := atomic.LoadPointer((*unsafe.Pointer)(unsafe.Pointer(&p.streamPtr)))
145143
146144 if err := p.streamOps.SetStreamActive(streamPtr, true); err != nil {
147145 // Activation failed — clean up owned resources so stale pointers
@@ -205,7 +203,7 @@ func (p *player) startCreateResourcesAndActivate() error {
205203 // All steps succeeded — store owned resources
206204 p.mu.Lock()
207205 p.loopPtr = loopPtr
208- p.streamPtr = streamPtr
206+ atomic.StorePointer((*unsafe.Pointer)(unsafe.Pointer(&p.streamPtr)), streamPtr)
209207 p.mu.Unlock()
210208
211209 // Start main loop in an internal goroutine
@@ -276,7 +274,7 @@ func (p *player) Close() error {
276274 }
277275 }
278276
279- // Teardown resources (placeholder)
277+ // Teardown the player-owned stream and main-loop resources.
280278 p.teardown()
281279
282280 // Transition to Closed
@@ -285,8 +283,9 @@ func (p *player) Close() error {
285283
286284 // ensureRuntime ensures the player has a usable StreamOps implementation.
287285 // If streamOps is already set (e.g., by tests), it is used as-is.
288-// Otherwise, PipeWire is registered via capi.Register() and the default
289-// StreamOps implementation is obtained.
286+// Otherwise, PipeWire is registered via capi.Register(), the default CAPI
287+// instance is retrieved, and the shared process-level init guard is used
288+// before obtaining the default StreamOps implementation.
290289 func (p *player) ensureRuntime() error {
291290 if p.streamOps != nil {
292291 return nil
@@ -294,6 +293,11 @@ func (p *player) ensureRuntime() error {
294293 if err := capi.Register(); err != nil {
295294 return fmt.Errorf("register pipewire: %w", err)
296295 }
296+ capiInstance := capi.Default()
297+ if capiInstance == nil {
298+ return errors.New("failed to get default CAPI instance")
299+ }
300+ initPipeWire(capiInstance)
297301 ops := capi.DefaultStreamOps()
298302 if ops == nil {
299303 return errors.New("failed to obtain default stream operations")
@@ -312,7 +316,7 @@ func (p *player) setPaused(paused bool) {
312316 func (p *player) deactivateStream() error {
313317 p.mu.Lock()
314318 streamOps := p.streamOps
315- streamPtr := p.streamPtr
319+ streamPtr := atomic.LoadPointer((*unsafe.Pointer)(unsafe.Pointer(&p.streamPtr)))
316320 p.mu.Unlock()
317321
318322 if streamOps == nil || streamPtr == nil {
@@ -328,9 +332,9 @@ func (p *player) deactivateStream() error {
328332 func (p *player) teardown() {
329333 p.mu.Lock()
330334 streamOps := p.streamOps
331- streamPtr := p.streamPtr
335+ streamPtr := atomic.LoadPointer((*unsafe.Pointer)(unsafe.Pointer(&p.streamPtr)))
332336 loopPtr := p.loopPtr
333- p.streamPtr = nil
337+ atomic.StorePointer((*unsafe.Pointer)(unsafe.Pointer(&p.streamPtr)), nil)
334338 p.loopPtr = nil
335339 p.mu.Unlock()
336340
@@ -129,7 +129,7 @@ func (p *player) Start() error {
129 129
130 // Check if we already have PipeWire resources (restart after Stop)130 // Check if we already have PipeWire resources (restart after Stop)
131 p.mu.Lock()131 p.mu.Lock()
132- hasResources := p.loopPtr != nil && p.streamPtr != nil132+ hasResources := p.loopPtr != nil && atomic.LoadPointer((*unsafe.Pointer)(unsafe.Pointer(&p.streamPtr))) != nil
133 p.mu.Unlock()133 p.mu.Unlock()
134 134
135 if !hasResources {135 if !hasResources {
@@ -139,9 +139,7 @@ func (p *player) Start() error {
139 }139 }
140 } else {140 } else {
141 // Restart — just reactivate existing stream141 // Restart — just reactivate existing stream
142- p.mu.Lock()142+ streamPtr := atomic.LoadPointer((*unsafe.Pointer)(unsafe.Pointer(&p.streamPtr)))
143- streamPtr := p.streamPtr
144- p.mu.Unlock()
145 143
146 if err := p.streamOps.SetStreamActive(streamPtr, true); err != nil {144 if err := p.streamOps.SetStreamActive(streamPtr, true); err != nil {
147 // Activation failed — clean up owned resources so stale pointers145 // Activation failed — clean up owned resources so stale pointers
@@ -205,7 +203,7 @@ func (p *player) startCreateResourcesAndActivate() error {
205 // All steps succeeded — store owned resources203 // All steps succeeded — store owned resources
206 p.mu.Lock()204 p.mu.Lock()
207 p.loopPtr = loopPtr205 p.loopPtr = loopPtr
208- p.streamPtr = streamPtr206+ atomic.StorePointer((*unsafe.Pointer)(unsafe.Pointer(&p.streamPtr)), streamPtr)
209 p.mu.Unlock()207 p.mu.Unlock()
210 208
211 // Start main loop in an internal goroutine209 // Start main loop in an internal goroutine
@@ -276,7 +274,7 @@ func (p *player) Close() error {
276 }274 }
277 }275 }
278 276
279- // Teardown resources (placeholder)277+ // Teardown the player-owned stream and main-loop resources.
280 p.teardown()278 p.teardown()
281 279
282 // Transition to Closed280 // Transition to Closed
@@ -285,8 +283,9 @@ func (p *player) Close() error {
285 283
286 // ensureRuntime ensures the player has a usable StreamOps implementation.284 // ensureRuntime ensures the player has a usable StreamOps implementation.
287 // If streamOps is already set (e.g., by tests), it is used as-is.285 // If streamOps is already set (e.g., by tests), it is used as-is.
288-// Otherwise, PipeWire is registered via capi.Register() and the default286+// Otherwise, PipeWire is registered via capi.Register(), the default CAPI
289-// StreamOps implementation is obtained.287+// instance is retrieved, and the shared process-level init guard is used
288+// before obtaining the default StreamOps implementation.
290 func (p *player) ensureRuntime() error {289 func (p *player) ensureRuntime() error {
291 if p.streamOps != nil {290 if p.streamOps != nil {
292 return nil291 return nil
@@ -294,6 +293,11 @@ func (p *player) ensureRuntime() error {
294 if err := capi.Register(); err != nil {293 if err := capi.Register(); err != nil {
295 return fmt.Errorf("register pipewire: %w", err)294 return fmt.Errorf("register pipewire: %w", err)
296 }295 }
296+ capiInstance := capi.Default()
297+ if capiInstance == nil {
298+ return errors.New("failed to get default CAPI instance")
299+ }
300+ initPipeWire(capiInstance)
297 ops := capi.DefaultStreamOps()301 ops := capi.DefaultStreamOps()
298 if ops == nil {302 if ops == nil {
299 return errors.New("failed to obtain default stream operations")303 return errors.New("failed to obtain default stream operations")
@@ -312,7 +316,7 @@ func (p *player) setPaused(paused bool) {
312 func (p *player) deactivateStream() error {316 func (p *player) deactivateStream() error {
313 p.mu.Lock()317 p.mu.Lock()
314 streamOps := p.streamOps318 streamOps := p.streamOps
315- streamPtr := p.streamPtr319+ streamPtr := atomic.LoadPointer((*unsafe.Pointer)(unsafe.Pointer(&p.streamPtr)))
316 p.mu.Unlock()320 p.mu.Unlock()
317 321
318 if streamOps == nil || streamPtr == nil {322 if streamOps == nil || streamPtr == nil {
@@ -328,9 +332,9 @@ func (p *player) deactivateStream() error {
328 func (p *player) teardown() {332 func (p *player) teardown() {
329 p.mu.Lock()333 p.mu.Lock()
330 streamOps := p.streamOps334 streamOps := p.streamOps
331- streamPtr := p.streamPtr335+ streamPtr := atomic.LoadPointer((*unsafe.Pointer)(unsafe.Pointer(&p.streamPtr)))
332 loopPtr := p.loopPtr336 loopPtr := p.loopPtr
333- p.streamPtr = nil337+ atomic.StorePointer((*unsafe.Pointer)(unsafe.Pointer(&p.streamPtr)), nil)
334 p.loopPtr = nil338 p.loopPtr = nil
335 p.mu.Unlock()339 p.mu.Unlock()
336 340
M internal/core/player_rt.go +20 -14
@@ -4,6 +4,8 @@ import (
44 "errors"
55 "fmt"
66 "io"
7+ "sync/atomic"
8+ "unsafe"
79 )
810
911 // UnderrunPolicy determines how to handle underruns
@@ -37,9 +39,7 @@ func (b *PCMBuffer) allocate() {
3739
3840 // PlayerConfig holds configuration for the player
3941 type PlayerConfig struct {
40- // SampleRate is carried here for upcoming runtime format negotiation.
41- // It is not yet consumed by the core player but will be used when the
42- // StreamOps layer builds SPA params for PipeWire.
42+ // SampleRate is the playback sample rate used when opening the stream.
4343 SampleRate int
4444 FramesPerBuffer int
4545 Channels int
@@ -161,12 +161,13 @@ func (p *player) fail(err error) {
161161 // queues the buffer back. If any step fails, it routes the error
162162 // through p.fail.
163163 func (p *player) onProcess() {
164- p.mu.Lock()
165164 ops := p.streamOps
166- sp := p.streamPtr
167- p.mu.Unlock()
165+ if ops == nil {
166+ return
167+ }
168168
169- if ops == nil || sp == nil {
169+ sp := atomic.LoadPointer((*unsafe.Pointer)(unsafe.Pointer(&p.streamPtr)))
170+ if sp == nil {
170171 return
171172 }
172173
@@ -175,21 +176,26 @@ func (p *player) onProcess() {
175176 return
176177 }
177178
178- view, err := newPWBufferView(bufPtr, p.config.Channels, p.config.FramesPerBuffer)
179+ queueOnError := true
180+ defer func() {
181+ if err := ops.QueueBuffer(sp, bufPtr); err != nil && queueOnError {
182+ p.fail(fmt.Errorf("queue buffer: %w", err))
183+ }
184+ }()
185+
186+ frames := p.config.FramesPerBuffer
187+ view, err := newPWBufferView(bufPtr, p.config.Channels, frames)
179188 if err != nil {
189+ queueOnError = false
180190 p.fail(fmt.Errorf("buffer view: %w", err))
181191 return
182192 }
183193
184- frames, err := p.processPCM(view.PCM())
194+ frames, err = p.processPCM(view.PCM())
185195 if err != nil {
186- // processPCM error: do not queue the buffer; let existing
187- // state/error behavior stand.
196+ queueOnError = false
188197 return
189198 }
190199
191200 view.Commit(frames)
192- if err := ops.QueueBuffer(sp, bufPtr); err != nil {
193- p.fail(fmt.Errorf("queue buffer: %w", err))
194- }
195201 }
@@ -4,6 +4,8 @@ import (
4 "errors"4 "errors"
5 "fmt"5 "fmt"
6 "io"6 "io"
7+ "sync/atomic"
8+ "unsafe"
7 )9 )
8 10
9 // UnderrunPolicy determines how to handle underruns11 // UnderrunPolicy determines how to handle underruns
@@ -37,9 +39,7 @@ func (b *PCMBuffer) allocate() {
37 39
38 // PlayerConfig holds configuration for the player40 // PlayerConfig holds configuration for the player
39 type PlayerConfig struct {41 type PlayerConfig struct {
40- // SampleRate is carried here for upcoming runtime format negotiation.42+ // SampleRate is the playback sample rate used when opening the stream.
41- // It is not yet consumed by the core player but will be used when the
42- // StreamOps layer builds SPA params for PipeWire.
43 SampleRate int43 SampleRate int
44 FramesPerBuffer int44 FramesPerBuffer int
45 Channels int45 Channels int
@@ -161,12 +161,13 @@ func (p *player) fail(err error) {
161 // queues the buffer back. If any step fails, it routes the error161 // queues the buffer back. If any step fails, it routes the error
162 // through p.fail.162 // through p.fail.
163 func (p *player) onProcess() {163 func (p *player) onProcess() {
164- p.mu.Lock()
165 ops := p.streamOps164 ops := p.streamOps
166- sp := p.streamPtr165+ if ops == nil {
167- p.mu.Unlock()166+ return
167+ }
168 168
169- if ops == nil || sp == nil {169+ sp := atomic.LoadPointer((*unsafe.Pointer)(unsafe.Pointer(&p.streamPtr)))
170+ if sp == nil {
170 return171 return
171 }172 }
172 173
@@ -175,21 +176,26 @@ func (p *player) onProcess() {
175 return176 return
176 }177 }
177 178
178- view, err := newPWBufferView(bufPtr, p.config.Channels, p.config.FramesPerBuffer)179+ queueOnError := true
180+ defer func() {
181+ if err := ops.QueueBuffer(sp, bufPtr); err != nil && queueOnError {
182+ p.fail(fmt.Errorf("queue buffer: %w", err))
183+ }
184+ }()
185+
186+ frames := p.config.FramesPerBuffer
187+ view, err := newPWBufferView(bufPtr, p.config.Channels, frames)
179 if err != nil {188 if err != nil {
189+ queueOnError = false
180 p.fail(fmt.Errorf("buffer view: %w", err))190 p.fail(fmt.Errorf("buffer view: %w", err))
181 return191 return
182 }192 }
183 193
184- frames, err := p.processPCM(view.PCM())194+ frames, err = p.processPCM(view.PCM())
185 if err != nil {195 if err != nil {
186- // processPCM error: do not queue the buffer; let existing196+ queueOnError = false
187- // state/error behavior stand.
188 return197 return
189 }198 }
190 199
191 view.Commit(frames)200 view.Commit(frames)
192- if err := ops.QueueBuffer(sp, bufPtr); err != nil {
193- p.fail(fmt.Errorf("queue buffer: %w", err))
194- }
195 }201 }
M internal/core/player_rt_test.go +108 -9
@@ -747,7 +747,7 @@ func TestProcessCallbackQueueBufferFailsRoutesThroughFail(t *testing.T) {
747747
748748 // TestProcessCallbackBufferViewFailsRoutesThroughFail verifies that when
749749 // newPWBufferView fails, onProcess routes the error through p.fail()
750-// and does not attempt to queue the buffer.
750+// and still re-queues the buffer.
751751 func TestProcessCallbackBufferViewFailsRoutesThroughFail(t *testing.T) {
752752 mockOps := mocks.NewMockStreamOps(t)
753753 fakeLoop := unsafe.Pointer(uintptr(0xCAFE))
@@ -796,9 +796,10 @@ func TestProcessCallbackBufferViewFailsRoutesThroughFail(t *testing.T) {
796796 }
797797 waitOnMainLoop(t, wg)
798798
799- // DequeueBuffer returns a buffer that will fail newPWBufferView
800- // QueueBuffer should NOT be called
799+ // DequeueBuffer returns a buffer that will fail newPWBufferView.
800+ // The buffer should still be re-queued.
801801 mockOps.EXPECT().DequeueBuffer(fakeStream).Return(bufPtr)
802+ mockOps.EXPECT().QueueBuffer(fakeStream, bufPtr).Return(nil)
802803
803804 p.onProcess()
804805
@@ -816,10 +817,10 @@ func TestProcessCallbackBufferViewFailsRoutesThroughFail(t *testing.T) {
816817 }
817818 }
818819
819-// TestProcessCallbackProcessCMErrorSkipsQueue verifies that when processPCM
820-// returns an error, onProcess does not queue the buffer and lets the existing
820+// TestProcessCallbackProcessCMErrorRequeuesBuffer verifies that when processPCM
821+// returns an error, onProcess still re-queues the buffer and lets the existing
821822 // error behavior stand.
822-func TestProcessCallbackProcessCMErrorSkipsQueue(t *testing.T) {
823+func TestProcessCallbackProcessCMErrorRequeuesBuffer(t *testing.T) {
823824 mockOps := mocks.NewMockStreamOps(t)
824825 fakeLoop := unsafe.Pointer(uintptr(0xCAFE))
825826 fakeStream := unsafe.Pointer(uintptr(0xBEEF))
@@ -880,10 +881,10 @@ func TestProcessCallbackProcessCMErrorSkipsQueue(t *testing.T) {
880881 }
881882 waitOnMainLoop(t, wg)
882883
883- // DequeueBuffer returns a buffer; QueueBuffer should NOT be called
884- // because processPCM (via Fill) returns an error
884+ // DequeueBuffer returns a buffer; the buffer should still be re-queued
885+ // even though processPCM (via Fill) returns an error.
885886 mockOps.EXPECT().DequeueBuffer(fakeStream).Return(bufPtr)
886- // Note: no QueueBuffer expectation
887+ mockOps.EXPECT().QueueBuffer(fakeStream, bufPtr).Return(nil)
887888
888889 p.onProcess()
889890
@@ -1049,3 +1050,101 @@ func TestPlayerStartFailedCreateStreamDestroysLoop(t *testing.T) {
10491050 t.Error("expected streamPtr to be nil after failed start")
10501051 }
10511052 }
1053+
1054+// TestProcessCallbackUsesConfigFramesPerBuffer verifies that onProcess uses
1055+// config.FramesPerBuffer even when pwBuffer.Requested is smaller.
1056+func TestProcessCallbackUsesConfigFramesPerBuffer(t *testing.T) {
1057+ mockOps := mocks.NewMockStreamOps(t)
1058+ fakeLoop := unsafe.Pointer(uintptr(0xCAFE))
1059+ fakeStream := unsafe.Pointer(uintptr(0xBEEF))
1060+
1061+ // Config requests 1024 frames per buffer, but PipeWire will request
1062+ // fewer (941) due to rate conversion.
1063+ const configFrames = 1024
1064+ const requestedFrames = 941
1065+ const channels = 2
1066+
1067+ cfg := PlayerConfig{
1068+ SampleRate: 48000,
1069+ Channels: channels,
1070+ FramesPerBuffer: configFrames,
1071+ }
1072+
1073+ // Build a real pwBuffer with backing storage for the full configFrames
1074+ // so that a view up to configFrames would succeed — but set
1075+ // pwBuf.Requested smaller to ensure onProcess still uses configFrames.
1076+ channelData := make([][]float32, channels)
1077+ floatPtrs := make([]unsafe.Pointer, channels)
1078+ for ch := 0; ch < channels; ch++ {
1079+ channelData[ch] = make([]float32, configFrames)
1080+ floatPtrs[ch] = unsafe.Pointer(&channelData[ch][0])
1081+ }
1082+ chunkSlice := make([]spaChunk, channels)
1083+ dataSlice := make([]spaData, channels)
1084+ for ch := 0; ch < channels; ch++ {
1085+ dataSlice[ch] = spaData{
1086+ Type: spaDataTypeMemPtr,
1087+ Flags: spaDataFlagRW,
1088+ Maxsize: uint32(configFrames * 4),
1089+ Data: floatPtrs[ch],
1090+ Chunk: &chunkSlice[ch],
1091+ }
1092+ }
1093+ spaBuf := spaBuffer{
1094+ NDatas: uint32(channels),
1095+ Datas: unsafe.Pointer(&dataSlice[0]),
1096+ }
1097+ pwBuf := pwBuffer{
1098+ Buffer: &spaBuf,
1099+ Requested: uint64(requestedFrames), // PipeWire asks for 941, not 1024
1100+ }
1101+ bufPtr := unsafe.Pointer(&pwBuf)
1102+
1103+ wg := expectStartWithSync(mockOps, fakeLoop, fakeStream, cfg)
1104+ mockOps.EXPECT().DequeueBuffer(fakeStream).Return(bufPtr)
1105+ mockOps.EXPECT().QueueBuffer(fakeStream, bufPtr).Return(nil)
1106+
1107+ var receivedFrames int
1108+ p := newPlayer(cfg, PlayerCallbacks{
1109+ Fill: func(buf *PCMBuffer) (int, error) {
1110+ receivedFrames = buf.Frames
1111+ // Write data into the buffer to verify propagation
1112+ for ch := 0; ch < buf.Channels; ch++ {
1113+ for f := 0; f < buf.Frames; f++ {
1114+ buf.Samples[ch][f] = float32(f)
1115+ }
1116+ }
1117+ return buf.Frames, nil
1118+ },
1119+ })
1120+ p.streamOps = mockOps
1121+
1122+ if err := p.Start(); err != nil {
1123+ t.Fatalf("Start failed: %v", err)
1124+ }
1125+
1126+ p.onProcess()
1127+
1128+ // Fill must have been called with the configured frame count.
1129+ if receivedFrames != configFrames {
1130+ t.Errorf("Fill received %d frames, want %d (config.FramesPerBuffer)", receivedFrames, configFrames)
1131+ }
1132+
1133+ // Commit must set chunk sizes and pwBuf.Size to reflect configFrames.
1134+ for ch := 0; ch < channels; ch++ {
1135+ wantChunkSize := uint32(configFrames * 4)
1136+ if chunkSlice[ch].Size != wantChunkSize {
1137+ t.Errorf("chunk[%d].Size = %d, want %d (configFrames*4)", ch, chunkSlice[ch].Size, wantChunkSize)
1138+ }
1139+ if chunkSlice[ch].Stride != 4 {
1140+ t.Errorf("chunk[%d].Stride = %d, want 4", ch, chunkSlice[ch].Stride)
1141+ }
1142+ }
1143+
1144+ if pwBuf.Size != uint64(configFrames) {
1145+ t.Errorf("pwBuf.Size = %d, want %d (configFrames)", pwBuf.Size, configFrames)
1146+ }
1147+
1148+ _ = channelData // prevent escape
1149+ waitOnMainLoop(t, wg)
1150+}
@@ -747,7 +747,7 @@ func TestProcessCallbackQueueBufferFailsRoutesThroughFail(t *testing.T) {
747 747
748 // TestProcessCallbackBufferViewFailsRoutesThroughFail verifies that when748 // TestProcessCallbackBufferViewFailsRoutesThroughFail verifies that when
749 // newPWBufferView fails, onProcess routes the error through p.fail()749 // newPWBufferView fails, onProcess routes the error through p.fail()
750-// and does not attempt to queue the buffer.750+// and still re-queues the buffer.
751 func TestProcessCallbackBufferViewFailsRoutesThroughFail(t *testing.T) {751 func TestProcessCallbackBufferViewFailsRoutesThroughFail(t *testing.T) {
752 mockOps := mocks.NewMockStreamOps(t)752 mockOps := mocks.NewMockStreamOps(t)
753 fakeLoop := unsafe.Pointer(uintptr(0xCAFE))753 fakeLoop := unsafe.Pointer(uintptr(0xCAFE))
@@ -796,9 +796,10 @@ func TestProcessCallbackBufferViewFailsRoutesThroughFail(t *testing.T) {
796 }796 }
797 waitOnMainLoop(t, wg)797 waitOnMainLoop(t, wg)
798 798
799- // DequeueBuffer returns a buffer that will fail newPWBufferView799+ // DequeueBuffer returns a buffer that will fail newPWBufferView.
800- // QueueBuffer should NOT be called800+ // The buffer should still be re-queued.
801 mockOps.EXPECT().DequeueBuffer(fakeStream).Return(bufPtr)801 mockOps.EXPECT().DequeueBuffer(fakeStream).Return(bufPtr)
802+ mockOps.EXPECT().QueueBuffer(fakeStream, bufPtr).Return(nil)
802 803
803 p.onProcess()804 p.onProcess()
804 805
@@ -816,10 +817,10 @@ func TestProcessCallbackBufferViewFailsRoutesThroughFail(t *testing.T) {
816 }817 }
817 }818 }
818 819
819-// TestProcessCallbackProcessCMErrorSkipsQueue verifies that when processPCM820+// TestProcessCallbackProcessCMErrorRequeuesBuffer verifies that when processPCM
820-// returns an error, onProcess does not queue the buffer and lets the existing821+// returns an error, onProcess still re-queues the buffer and lets the existing
821 // error behavior stand.822 // error behavior stand.
822-func TestProcessCallbackProcessCMErrorSkipsQueue(t *testing.T) {823+func TestProcessCallbackProcessCMErrorRequeuesBuffer(t *testing.T) {
823 mockOps := mocks.NewMockStreamOps(t)824 mockOps := mocks.NewMockStreamOps(t)
824 fakeLoop := unsafe.Pointer(uintptr(0xCAFE))825 fakeLoop := unsafe.Pointer(uintptr(0xCAFE))
825 fakeStream := unsafe.Pointer(uintptr(0xBEEF))826 fakeStream := unsafe.Pointer(uintptr(0xBEEF))
@@ -880,10 +881,10 @@ func TestProcessCallbackProcessCMErrorSkipsQueue(t *testing.T) {
880 }881 }
881 waitOnMainLoop(t, wg)882 waitOnMainLoop(t, wg)
882 883
883- // DequeueBuffer returns a buffer; QueueBuffer should NOT be called884+ // DequeueBuffer returns a buffer; the buffer should still be re-queued
884- // because processPCM (via Fill) returns an error885+ // even though processPCM (via Fill) returns an error.
885 mockOps.EXPECT().DequeueBuffer(fakeStream).Return(bufPtr)886 mockOps.EXPECT().DequeueBuffer(fakeStream).Return(bufPtr)
886- // Note: no QueueBuffer expectation887+ mockOps.EXPECT().QueueBuffer(fakeStream, bufPtr).Return(nil)
887 888
888 p.onProcess()889 p.onProcess()
889 890
@@ -1049,3 +1050,101 @@ func TestPlayerStartFailedCreateStreamDestroysLoop(t *testing.T) {
1049 t.Error("expected streamPtr to be nil after failed start")1050 t.Error("expected streamPtr to be nil after failed start")
1050 }1051 }
1051 }1052 }
1053+
1054+// TestProcessCallbackUsesConfigFramesPerBuffer verifies that onProcess uses
1055+// config.FramesPerBuffer even when pwBuffer.Requested is smaller.
1056+func TestProcessCallbackUsesConfigFramesPerBuffer(t *testing.T) {
1057+ mockOps := mocks.NewMockStreamOps(t)
1058+ fakeLoop := unsafe.Pointer(uintptr(0xCAFE))
1059+ fakeStream := unsafe.Pointer(uintptr(0xBEEF))
1060+
1061+ // Config requests 1024 frames per buffer, but PipeWire will request
1062+ // fewer (941) due to rate conversion.
1063+ const configFrames = 1024
1064+ const requestedFrames = 941
1065+ const channels = 2
1066+
1067+ cfg := PlayerConfig{
1068+ SampleRate: 48000,
1069+ Channels: channels,
1070+ FramesPerBuffer: configFrames,
1071+ }
1072+
1073+ // Build a real pwBuffer with backing storage for the full configFrames
1074+ // so that a view up to configFrames would succeed — but set
1075+ // pwBuf.Requested smaller to ensure onProcess still uses configFrames.
1076+ channelData := make([][]float32, channels)
1077+ floatPtrs := make([]unsafe.Pointer, channels)
1078+ for ch := 0; ch < channels; ch++ {
1079+ channelData[ch] = make([]float32, configFrames)
1080+ floatPtrs[ch] = unsafe.Pointer(&channelData[ch][0])
1081+ }
1082+ chunkSlice := make([]spaChunk, channels)
1083+ dataSlice := make([]spaData, channels)
1084+ for ch := 0; ch < channels; ch++ {
1085+ dataSlice[ch] = spaData{
1086+ Type: spaDataTypeMemPtr,
1087+ Flags: spaDataFlagRW,
1088+ Maxsize: uint32(configFrames * 4),
1089+ Data: floatPtrs[ch],
1090+ Chunk: &chunkSlice[ch],
1091+ }
1092+ }
1093+ spaBuf := spaBuffer{
1094+ NDatas: uint32(channels),
1095+ Datas: unsafe.Pointer(&dataSlice[0]),
1096+ }
1097+ pwBuf := pwBuffer{
1098+ Buffer: &spaBuf,
1099+ Requested: uint64(requestedFrames), // PipeWire asks for 941, not 1024
1100+ }
1101+ bufPtr := unsafe.Pointer(&pwBuf)
1102+
1103+ wg := expectStartWithSync(mockOps, fakeLoop, fakeStream, cfg)
1104+ mockOps.EXPECT().DequeueBuffer(fakeStream).Return(bufPtr)
1105+ mockOps.EXPECT().QueueBuffer(fakeStream, bufPtr).Return(nil)
1106+
1107+ var receivedFrames int
1108+ p := newPlayer(cfg, PlayerCallbacks{
1109+ Fill: func(buf *PCMBuffer) (int, error) {
1110+ receivedFrames = buf.Frames
1111+ // Write data into the buffer to verify propagation
1112+ for ch := 0; ch < buf.Channels; ch++ {
1113+ for f := 0; f < buf.Frames; f++ {
1114+ buf.Samples[ch][f] = float32(f)
1115+ }
1116+ }
1117+ return buf.Frames, nil
1118+ },
1119+ })
1120+ p.streamOps = mockOps
1121+
1122+ if err := p.Start(); err != nil {
1123+ t.Fatalf("Start failed: %v", err)
1124+ }
1125+
1126+ p.onProcess()
1127+
1128+ // Fill must have been called with the configured frame count.
1129+ if receivedFrames != configFrames {
1130+ t.Errorf("Fill received %d frames, want %d (config.FramesPerBuffer)", receivedFrames, configFrames)
1131+ }
1132+
1133+ // Commit must set chunk sizes and pwBuf.Size to reflect configFrames.
1134+ for ch := 0; ch < channels; ch++ {
1135+ wantChunkSize := uint32(configFrames * 4)
1136+ if chunkSlice[ch].Size != wantChunkSize {
1137+ t.Errorf("chunk[%d].Size = %d, want %d (configFrames*4)", ch, chunkSlice[ch].Size, wantChunkSize)
1138+ }
1139+ if chunkSlice[ch].Stride != 4 {
1140+ t.Errorf("chunk[%d].Stride = %d, want 4", ch, chunkSlice[ch].Stride)
1141+ }
1142+ }
1143+
1144+ if pwBuf.Size != uint64(configFrames) {
1145+ t.Errorf("pwBuf.Size = %d, want %d (configFrames)", pwBuf.Size, configFrames)
1146+ }
1147+
1148+ _ = channelData // prevent escape
1149+ waitOnMainLoop(t, wg)
1150+}
M internal/core/pwbuffer.go +17 -12
@@ -40,9 +40,11 @@ type spaBuffer struct {
4040
4141 // pwBuffer mirrors struct pw_buffer from <pipewire/stream.h>.
4242 type pwBuffer struct {
43- Buffer *spaBuffer
44- UserData unsafe.Pointer
45- Size uint64
43+ Buffer *spaBuffer
44+ UserData unsafe.Pointer
45+ Size uint64
46+ Requested uint64
47+ Time uint64
4648 }
4749
4850 // pwBufferView provides a safe Go view over a dequeue'd PipeWire buffer,
@@ -52,6 +54,8 @@ type pwBufferView struct {
5254 buf *pwBuffer
5355 channels int
5456 frames int
57+ data []spaData
58+ pcm PCMBuffer
5559 samples [][]float32
5660 }
5761
@@ -102,7 +106,14 @@ func newPWBufferView(bufPtr unsafe.Pointer, channels, frames int) (*pwBufferView
102106 buf: pw,
103107 channels: channels,
104108 frames: frames,
105- samples: samples,
109+ data: dataSlice,
110+ pcm: PCMBuffer{
111+ Frames: frames,
112+ Channels: channels,
113+ Stride: 4,
114+ Samples: samples,
115+ },
116+ samples: samples,
106117 }, nil
107118 }
108119
@@ -110,12 +121,7 @@ func newPWBufferView(bufPtr unsafe.Pointer, channels, frames int) (*pwBufferView
110121 // The caller can write into Samples and then call Commit.
111122 // Stride is set to 4 (bytes per float32 sample) for planar float32 layout.
112123 func (v *pwBufferView) PCM() *PCMBuffer {
113- return &PCMBuffer{
114- Frames: v.frames,
115- Channels: v.channels,
116- Stride: 4, // sizeof(float32) — planar float32 byte stride per channel
117- Samples: v.samples,
118- }
124+ return &v.pcm
119125 }
120126
121127 // Commit updates each data chunk's size and stride to reflect the number of
@@ -123,9 +129,8 @@ func (v *pwBufferView) PCM() *PCMBuffer {
123129 // audio data into the PCM buffer.
124130 func (v *pwBufferView) Commit(frames int) {
125131 v.buf.Size = uint64(frames)
126- dataSlice := unsafe.Slice((*spaData)(v.buf.Buffer.Datas), int(v.buf.Buffer.NDatas))
127132 for i := 0; i < v.channels; i++ {
128- data := &dataSlice[i]
133+ data := &v.data[i]
129134 if data.Chunk != nil {
130135 data.Chunk.Size = uint32(frames * 4) // float32 = 4 bytes
131136 data.Chunk.Stride = 4 // stride for planar float32
@@ -40,9 +40,11 @@ type spaBuffer struct {
40 40
41 // pwBuffer mirrors struct pw_buffer from <pipewire/stream.h>.41 // pwBuffer mirrors struct pw_buffer from <pipewire/stream.h>.
42 type pwBuffer struct {42 type pwBuffer struct {
43- Buffer *spaBuffer43+ Buffer *spaBuffer
44- UserData unsafe.Pointer44+ UserData unsafe.Pointer
45- Size uint6445+ Size uint64
46+ Requested uint64
47+ Time uint64
46 }48 }
47 49
48 // pwBufferView provides a safe Go view over a dequeue'd PipeWire buffer,50 // pwBufferView provides a safe Go view over a dequeue'd PipeWire buffer,
@@ -52,6 +54,8 @@ type pwBufferView struct {
52 buf *pwBuffer54 buf *pwBuffer
53 channels int55 channels int
54 frames int56 frames int
57+ data []spaData
58+ pcm PCMBuffer
55 samples [][]float3259 samples [][]float32
56 }60 }
57 61
@@ -102,7 +106,14 @@ func newPWBufferView(bufPtr unsafe.Pointer, channels, frames int) (*pwBufferView
102 buf: pw,106 buf: pw,
103 channels: channels,107 channels: channels,
104 frames: frames,108 frames: frames,
105- samples: samples,109+ data: dataSlice,
110+ pcm: PCMBuffer{
111+ Frames: frames,
112+ Channels: channels,
113+ Stride: 4,
114+ Samples: samples,
115+ },
116+ samples: samples,
106 }, nil117 }, nil
107 }118 }
108 119
@@ -110,12 +121,7 @@ func newPWBufferView(bufPtr unsafe.Pointer, channels, frames int) (*pwBufferView
110 // The caller can write into Samples and then call Commit.121 // The caller can write into Samples and then call Commit.
111 // Stride is set to 4 (bytes per float32 sample) for planar float32 layout.122 // Stride is set to 4 (bytes per float32 sample) for planar float32 layout.
112 func (v *pwBufferView) PCM() *PCMBuffer {123 func (v *pwBufferView) PCM() *PCMBuffer {
113- return &PCMBuffer{124+ return &v.pcm
114- Frames: v.frames,
115- Channels: v.channels,
116- Stride: 4, // sizeof(float32) — planar float32 byte stride per channel
117- Samples: v.samples,
118- }
119 }125 }
120 126
121 // Commit updates each data chunk's size and stride to reflect the number of127 // Commit updates each data chunk's size and stride to reflect the number of
@@ -123,9 +129,8 @@ func (v *pwBufferView) PCM() *PCMBuffer {
123 // audio data into the PCM buffer.129 // audio data into the PCM buffer.
124 func (v *pwBufferView) Commit(frames int) {130 func (v *pwBufferView) Commit(frames int) {
125 v.buf.Size = uint64(frames)131 v.buf.Size = uint64(frames)
126- dataSlice := unsafe.Slice((*spaData)(v.buf.Buffer.Datas), int(v.buf.Buffer.NDatas))
127 for i := 0; i < v.channels; i++ {132 for i := 0; i < v.channels; i++ {
128- data := &dataSlice[i]133+ data := &v.data[i]
129 if data.Chunk != nil {134 if data.Chunk != nil {
130 data.Chunk.Size = uint32(frames * 4) // float32 = 4 bytes135 data.Chunk.Size = uint32(frames * 4) // float32 = 4 bytes
131 data.Chunk.Stride = 4 // stride for planar float32136 data.Chunk.Stride = 4 // stride for planar float32
M internal/core/runtime.go +61 -6
@@ -1,27 +1,82 @@
11 package core
22
33 import (
4+ "fmt"
45 "sync"
56
67 portout "github.com/bnema/purego-pipewire/internal/ports/out"
78 )
89
910 type Runtime struct {
10- capi portout.CAPI
11- once sync.Once
11+ mu sync.Mutex
12+ capi portout.CAPI
13+ initialized bool
1214 }
1315
16+var (
17+ pipewireMu sync.Mutex
18+ pipewireRefCnt int
19+ pipewireCAPI portout.CAPI
20+)
21+
1422 func NewRuntime(capi portout.CAPI) *Runtime {
1523 return &Runtime{capi: capi}
1624 }
1725
1826 func (r *Runtime) Init() error {
19- r.once.Do(func() {
20- r.capi.PWInit(nil, nil)
21- })
27+ r.mu.Lock()
28+ defer r.mu.Unlock()
29+
30+ if r.initialized {
31+ return nil
32+ }
33+ if err := initPipeWire(r.capi); err != nil {
34+ return err
35+ }
36+ r.initialized = true
37+ return nil
38+}
39+
40+// initPipeWire initializes the process-level PipeWire API once.
41+func initPipeWire(capi portout.CAPI) error {
42+ pipewireMu.Lock()
43+ defer pipewireMu.Unlock()
44+
45+ if capi == nil {
46+ return fmt.Errorf("runtime requires a non-nil CAPI implementation")
47+ }
48+
49+ if pipewireRefCnt > 0 && pipewireCAPI != nil && capi != pipewireCAPI {
50+ return fmt.Errorf("runtime already initialized with a different CAPI implementation")
51+ }
52+
53+ if pipewireRefCnt == 0 {
54+ pipewireCAPI = capi
55+ capi.PWInit(nil, nil)
56+ }
57+ pipewireRefCnt++
2258 return nil
2359 }
2460
2561 func (r *Runtime) Deinit() {
26- r.capi.PWDeinit()
62+ r.mu.Lock()
63+ defer r.mu.Unlock()
64+
65+ if !r.initialized {
66+ return
67+ }
68+ r.initialized = false
69+
70+ pipewireMu.Lock()
71+ defer pipewireMu.Unlock()
72+
73+ if pipewireRefCnt == 0 {
74+ return
75+ }
76+
77+ pipewireRefCnt--
78+ if pipewireRefCnt == 0 {
79+ pipewireCAPI.PWDeinit()
80+ pipewireCAPI = nil
81+ }
2782 }
@@ -1,27 +1,82 @@
1 package core1 package core
2 2
3 import (3 import (
4+ "fmt"
4 "sync"5 "sync"
5 6
6 portout "github.com/bnema/purego-pipewire/internal/ports/out"7 portout "github.com/bnema/purego-pipewire/internal/ports/out"
7 )8 )
8 9
9 type Runtime struct {10 type Runtime struct {
10- capi portout.CAPI11+ mu sync.Mutex
11- once sync.Once12+ capi portout.CAPI
13+ initialized bool
12 }14 }
13 15
16+var (
17+ pipewireMu sync.Mutex
18+ pipewireRefCnt int
19+ pipewireCAPI portout.CAPI
20+)
21+
14 func NewRuntime(capi portout.CAPI) *Runtime {22 func NewRuntime(capi portout.CAPI) *Runtime {
15 return &Runtime{capi: capi}23 return &Runtime{capi: capi}
16 }24 }
17 25
18 func (r *Runtime) Init() error {26 func (r *Runtime) Init() error {
19- r.once.Do(func() {27+ r.mu.Lock()
20- r.capi.PWInit(nil, nil)28+ defer r.mu.Unlock()
21- })29+
30+ if r.initialized {
31+ return nil
32+ }
33+ if err := initPipeWire(r.capi); err != nil {
34+ return err
35+ }
36+ r.initialized = true
37+ return nil
38+}
39+
40+// initPipeWire initializes the process-level PipeWire API once.
41+func initPipeWire(capi portout.CAPI) error {
42+ pipewireMu.Lock()
43+ defer pipewireMu.Unlock()
44+
45+ if capi == nil {
46+ return fmt.Errorf("runtime requires a non-nil CAPI implementation")
47+ }
48+
49+ if pipewireRefCnt > 0 && pipewireCAPI != nil && capi != pipewireCAPI {
50+ return fmt.Errorf("runtime already initialized with a different CAPI implementation")
51+ }
52+
53+ if pipewireRefCnt == 0 {
54+ pipewireCAPI = capi
55+ capi.PWInit(nil, nil)
56+ }
57+ pipewireRefCnt++
22 return nil58 return nil
23 }59 }
24 60
25 func (r *Runtime) Deinit() {61 func (r *Runtime) Deinit() {
26- r.capi.PWDeinit()62+ r.mu.Lock()
63+ defer r.mu.Unlock()
64+
65+ if !r.initialized {
66+ return
67+ }
68+ r.initialized = false
69+
70+ pipewireMu.Lock()
71+ defer pipewireMu.Unlock()
72+
73+ if pipewireRefCnt == 0 {
74+ return
75+ }
76+
77+ pipewireRefCnt--
78+ if pipewireRefCnt == 0 {
79+ pipewireCAPI.PWDeinit()
80+ pipewireCAPI = nil
81+ }
27 }82 }
M internal/core/runtime_test.go +56 -11
@@ -6,23 +6,68 @@ import (
66 portmocks "github.com/bnema/purego-pipewire/internal/ports/out/mocks"
77 )
88
9-func TestRuntimeInitCallsPWInitOnce(t *testing.T) {
10- api := portmocks.NewMockCAPI(t)
11- api.EXPECT().PWInit((*int32)(nil), (***byte)(nil)).Once()
9+func TestRuntimeInitRejectsDifferentCAPIAfterProcessWideInit(t *testing.T) {
10+ api1 := portmocks.NewMockCAPI(t)
11+ api2 := portmocks.NewMockCAPI(t)
1212
13- r := NewRuntime(api)
14- if err := r.Init(); err != nil {
13+ api1.EXPECT().PWDeinit().Once()
14+ api1.EXPECT().PWInit((*int32)(nil), (***byte)(nil)).Once()
15+
16+ r1 := NewRuntime(api1)
17+ if err := r1.Init(); err != nil {
1518 t.Fatalf("Init returned error: %v", err)
1619 }
17- if err := r.Init(); err != nil {
18- t.Fatalf("second Init returned error: %v", err)
20+ r2 := NewRuntime(api2)
21+ if err := r2.Init(); err == nil {
22+ t.Fatal("second Init returned nil error, want mismatch rejection")
1923 }
24+
25+ r2.Deinit()
26+ r1.Deinit()
27+
28+ api2.AssertNotCalled(t, "PWInit", (*int32)(nil), (***byte)(nil))
29+ api2.AssertNotCalled(t, "PWDeinit")
2030 }
2131
22-func TestRuntimeDeinitCallsPWDeinit(t *testing.T) {
23- api := portmocks.NewMockCAPI(t)
24- api.EXPECT().PWDeinit().Once()
32+func TestRuntimeInitRejectsNilCAPI(t *testing.T) {
33+ r := NewRuntime(nil)
34+ if err := r.Init(); err == nil {
35+ t.Fatal("Init returned nil error, want nil CAPI rejection")
36+ }
2537
26- r := NewRuntime(api)
2738 r.Deinit()
2839 }
40+
41+func TestRuntimeDeinitWaitsForLastInitializedRuntime(t *testing.T) {
42+ api1 := portmocks.NewMockCAPI(t)
43+
44+ deinitCalls := 0
45+ api1.EXPECT().PWInit((*int32)(nil), (***byte)(nil)).Once()
46+ api1.EXPECT().PWDeinit().Run(func() {
47+ deinitCalls++
48+ }).Once()
49+
50+ r1 := NewRuntime(api1)
51+ if err := r1.Init(); err != nil {
52+ t.Fatalf("first Init returned error: %v", err)
53+ }
54+ r2 := NewRuntime(api1)
55+ if err := r2.Init(); err != nil {
56+ t.Fatalf("second Init returned error: %v", err)
57+ }
58+
59+ r1.Deinit()
60+ if deinitCalls != 0 {
61+ t.Fatalf("PWDeinit called after first Deinit, want 0 calls before last runtime deinitializes")
62+ }
63+
64+ NewRuntime(api1).Deinit()
65+ if deinitCalls != 0 {
66+ t.Fatalf("PWDeinit called after non-initialized runtime Deinit, want 0 calls before last runtime deinitializes")
67+ }
68+
69+ r2.Deinit()
70+ if deinitCalls != 1 {
71+ t.Fatalf("PWDeinit called %d times, want 1", deinitCalls)
72+ }
73+}
@@ -6,23 +6,68 @@ import (
6 portmocks "github.com/bnema/purego-pipewire/internal/ports/out/mocks"6 portmocks "github.com/bnema/purego-pipewire/internal/ports/out/mocks"
7 )7 )
8 8
9-func TestRuntimeInitCallsPWInitOnce(t *testing.T) {9+func TestRuntimeInitRejectsDifferentCAPIAfterProcessWideInit(t *testing.T) {
10- api := portmocks.NewMockCAPI(t)10+ api1 := portmocks.NewMockCAPI(t)
11- api.EXPECT().PWInit((*int32)(nil), (***byte)(nil)).Once()11+ api2 := portmocks.NewMockCAPI(t)
12 12
13- r := NewRuntime(api)13+ api1.EXPECT().PWDeinit().Once()
14- if err := r.Init(); err != nil {14+ api1.EXPECT().PWInit((*int32)(nil), (***byte)(nil)).Once()
15+
16+ r1 := NewRuntime(api1)
17+ if err := r1.Init(); err != nil {
15 t.Fatalf("Init returned error: %v", err)18 t.Fatalf("Init returned error: %v", err)
16 }19 }
17- if err := r.Init(); err != nil {20+ r2 := NewRuntime(api2)
18- t.Fatalf("second Init returned error: %v", err)21+ if err := r2.Init(); err == nil {
22+ t.Fatal("second Init returned nil error, want mismatch rejection")
19 }23 }
24+
25+ r2.Deinit()
26+ r1.Deinit()
27+
28+ api2.AssertNotCalled(t, "PWInit", (*int32)(nil), (***byte)(nil))
29+ api2.AssertNotCalled(t, "PWDeinit")
20 }30 }
21 31
22-func TestRuntimeDeinitCallsPWDeinit(t *testing.T) {32+func TestRuntimeInitRejectsNilCAPI(t *testing.T) {
23- api := portmocks.NewMockCAPI(t)33+ r := NewRuntime(nil)
24- api.EXPECT().PWDeinit().Once()34+ if err := r.Init(); err == nil {
35+ t.Fatal("Init returned nil error, want nil CAPI rejection")
36+ }
25 37
26- r := NewRuntime(api)
27 r.Deinit()38 r.Deinit()
28 }39 }
40+
41+func TestRuntimeDeinitWaitsForLastInitializedRuntime(t *testing.T) {
42+ api1 := portmocks.NewMockCAPI(t)
43+
44+ deinitCalls := 0
45+ api1.EXPECT().PWInit((*int32)(nil), (***byte)(nil)).Once()
46+ api1.EXPECT().PWDeinit().Run(func() {
47+ deinitCalls++
48+ }).Once()
49+
50+ r1 := NewRuntime(api1)
51+ if err := r1.Init(); err != nil {
52+ t.Fatalf("first Init returned error: %v", err)
53+ }
54+ r2 := NewRuntime(api1)
55+ if err := r2.Init(); err != nil {
56+ t.Fatalf("second Init returned error: %v", err)
57+ }
58+
59+ r1.Deinit()
60+ if deinitCalls != 0 {
61+ t.Fatalf("PWDeinit called after first Deinit, want 0 calls before last runtime deinitializes")
62+ }
63+
64+ NewRuntime(api1).Deinit()
65+ if deinitCalls != 0 {
66+ t.Fatalf("PWDeinit called after non-initialized runtime Deinit, want 0 calls before last runtime deinitializes")
67+ }
68+
69+ r2.Deinit()
70+ if deinitCalls != 1 {
71+ t.Fatalf("PWDeinit called %d times, want 1", deinitCalls)
72+ }
73+}
M internal/ports/out/stream_ops.go +4 -8
@@ -2,11 +2,8 @@ package out
22
33 import "unsafe"
44
5-// PlaybackFormat describes the audio format for a playback stream.
6-// The fields are carried through the public API and core config now so the
7-// plumbing is in place for upcoming runtime format negotiation. The values
8-// are not yet forwarded to PipeWire; that will happen when SPA param
9-// building is wired in a subsequent task.
5+// PlaybackFormat describes the audio format used when connecting a playback
6+// stream.
107 type PlaybackFormat struct {
118 SampleRate int
129 Channels int
@@ -22,9 +19,8 @@ type StreamOps interface {
2219 // stream needs more audio data. Returns the stream pointer or an error.
2320 CreatePlaybackStream(loopPtr unsafe.Pointer, name string, onProcess func()) (streamPtr unsafe.Pointer, err error)
2421
25- // ConnectPlaybackStream connects the stream for playback output.
26- // The format is validated but not yet forwarded to PipeWire as SPA params;
27- // that wiring will happen in a subsequent task.
22+ // ConnectPlaybackStream connects the stream for playback output using the
23+ // supplied format.
2824 ConnectPlaybackStream(streamPtr unsafe.Pointer, format PlaybackFormat) error
2925
3026 // SetStreamActive activates or deactivates the stream.
@@ -2,11 +2,8 @@ package out
2 2
3 import "unsafe"3 import "unsafe"
4 4
5-// PlaybackFormat describes the audio format for a playback stream.5+// PlaybackFormat describes the audio format used when connecting a playback
6-// The fields are carried through the public API and core config now so the6+// stream.
7-// plumbing is in place for upcoming runtime format negotiation. The values
8-// are not yet forwarded to PipeWire; that will happen when SPA param
9-// building is wired in a subsequent task.
10 type PlaybackFormat struct {7 type PlaybackFormat struct {
11 SampleRate int8 SampleRate int
12 Channels int9 Channels int
@@ -22,9 +19,8 @@ type StreamOps interface {
22 // stream needs more audio data. Returns the stream pointer or an error.19 // stream needs more audio data. Returns the stream pointer or an error.
23 CreatePlaybackStream(loopPtr unsafe.Pointer, name string, onProcess func()) (streamPtr unsafe.Pointer, err error)20 CreatePlaybackStream(loopPtr unsafe.Pointer, name string, onProcess func()) (streamPtr unsafe.Pointer, err error)
24 21
25- // ConnectPlaybackStream connects the stream for playback output.22+ // ConnectPlaybackStream connects the stream for playback output using the
26- // The format is validated but not yet forwarded to PipeWire as SPA params;23+ // supplied format.
27- // that wiring will happen in a subsequent task.
28 ConnectPlaybackStream(streamPtr unsafe.Pointer, format PlaybackFormat) error24 ConnectPlaybackStream(streamPtr unsafe.Pointer, format PlaybackFormat) error
29 25
30 // SetStreamActive activates or deactivates the stream.26 // SetStreamActive activates or deactivates the stream.

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 1

Tags 8

↑↓ moveenter openesc close