feat: wire PipeWire stream creation and connect params

commit: af7a9d7c3889f59dc856e04e3bb66f74d3eb5e97
author: brice · 6mo ago · 2026-04-10 13:37 UTC
parent: 66c3355
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +145 -11

M internal/capi/stream_ops.go +21 -8


@@ -25,8 +25,9 @@ func (e *PWError) Error() string {
25	25		var ErrInvalidPlaybackFormat = errors.New("invalid playback format")
26	26
27	27		var (
28		-	errStreamCreate = &PWError{Func: "pw_stream_new_simple", Code: 0}
29		-	errMainLoopCreate = &PWError{Func: "pw_main_loop_new", Code: 0}
	28	+	errStreamCreate = &PWError{Func: "pw_stream_new_simple", Code: 0}
	29	+	errMainLoopCreate = &PWError{Func: "pw_main_loop_new", Code: 0}
	30	+	errPWLoopFromMainLoop = &PWError{Func: "pw_main_loop_get_loop", Code: 0}
30	31		)
31	32
32	33		const pwVersionStreamEvents = 2

@@ -57,6 +58,13 @@ type streamOpsImpl struct {
57	58		var _ portout.StreamOps = (*streamOpsImpl)(nil)
58	59
59	60		func (s *streamOpsImpl) CreatePlaybackStream(loopPtr unsafe.Pointer, name string, onProcess func()) (unsafe.Pointer, error) {
	61	+	// Obtain the inner pw_loop from the main loop; pw_stream_new_simple
	62	+	// expects a pw_loop, not a pw_main_loop.
	63	+	pwLoop := pw_main_loop_get_loop(loopPtr)
	64	+	if pwLoop == nil {
	65	+	return nil, errPWLoopFromMainLoop
	66	+	}
	67	+
60	68		nameBytes := append([]byte(name), 0)
61	69
62	70		// Build persistent callback storage that outlives this function.

@@ -70,7 +78,7 @@ func (s *streamOpsImpl) CreatePlaybackStream(loopPtr unsafe.Pointer, name string
70	78		}
71	79
72	80		ptr := pw_stream_new_simple(
73		-	loopPtr,
	81	+	pwLoop,
74	82		&nameBytes[0],
75	83		nil, // props
76	84		unsafe.Pointer(&storage.events), // events

@@ -93,17 +101,22 @@ func (s *streamOpsImpl) CreatePlaybackStream(loopPtr unsafe.Pointer, name string
93	101
94	102		func (s *streamOpsImpl) ConnectPlaybackStream(streamPtr unsafe.Pointer, format portout.PlaybackFormat) error {
95	103		// Validate format before making the C call.
96		-	// NOTE: The format fields are not yet forwarded to PipeWire as SPA params.
97		-	// This validation is a temporary guard that will be superseded once SPA
98		-	// param building is wired in a subsequent task.
99		-	// Returns ErrInvalidPlaybackFormat (wrappable via errors.Is) for bad values.
	104	+	// FramesPerBuffer is not used by buildRawAudioParams but is still required
	105	+	// for a well-formed playback setup.
100	106		if format.SampleRate <= 0 \|\| format.Channels <= 0 \|\| format.FramesPerBuffer <= 0 {
101	107		return fmt.Errorf("%w: SampleRate=%d, Channels=%d, FramesPerBuffer=%d",
102	108		ErrInvalidPlaybackFormat, format.SampleRate, format.Channels, format.FramesPerBuffer)
103	109		}
	110	+
	111	+	// Build the SPA POD params that describe the stream's audio format.
	112	+	cp, err := buildRawAudioParams(format)
	113	+	if err != nil {
	114	+	return err
	115	+	}
	116	+
104	117		// PW_DIRECTION_OUTPUT = 1, PW_ID_ANY = 0xffffffff
105	118		// PW_STREAM_FLAG_AUTOCONNECT \| PW_STREAM_FLAG_MAP_BUFFERS = 0x01 \| 0x04
106		-	ret := pw_stream_connect(streamPtr, pwDirectionOutput, pwIDAny, pwStreamFlagAutoConnectMapBuffers, nil, 0)
	119	+	ret := pw_stream_connect(streamPtr, pwDirectionOutput, pwIDAny, pwStreamFlagAutoConnectMapBuffers, cp.Pointer(), cp.Count())
107	120		if ret < 0 {
108	121		return &PWError{Func: "pw_stream_connect", Code: ret}
109	122		}

 var ErrInvalidPlaybackFormat = errors.New("invalid playback format")
 var (
-	errStreamCreate   = &PWError{Func: "pw_stream_new_simple", Code: 0}
+	errStreamCreate       = &PWError{Func: "pw_stream_new_simple", Code: 0}
-	errMainLoopCreate = &PWError{Func: "pw_main_loop_new", Code: 0}
+	errMainLoopCreate     = &PWError{Func: "pw_main_loop_new", Code: 0}
+	errPWLoopFromMainLoop = &PWError{Func: "pw_main_loop_get_loop", Code: 0}
 )
 const pwVersionStreamEvents = 2
 var _ portout.StreamOps = (*streamOpsImpl)(nil)
 func (s *streamOpsImpl) CreatePlaybackStream(loopPtr unsafe.Pointer, name string, onProcess func()) (unsafe.Pointer, error) {
+	// Obtain the inner pw_loop from the main loop; pw_stream_new_simple
+	// expects a pw_loop, not a pw_main_loop.
+	pwLoop := pw_main_loop_get_loop(loopPtr)
+	if pwLoop == nil {
+		return nil, errPWLoopFromMainLoop
+	}
 	nameBytes := append([]byte(name), 0)
 	// Build persistent callback storage that outlives this function.
 	}
 	ptr := pw_stream_new_simple(
-		loopPtr,
+		pwLoop,
 		&nameBytes[0],
 		nil,                             // props
 		unsafe.Pointer(&storage.events), // events
 func (s *streamOpsImpl) ConnectPlaybackStream(streamPtr unsafe.Pointer, format portout.PlaybackFormat) error {
 	// Validate format before making the C call.
-	// NOTE: The format fields are not yet forwarded to PipeWire as SPA params.
+	// FramesPerBuffer is not used by buildRawAudioParams but is still required
-	// This validation is a temporary guard that will be superseded once SPA
+	// for a well-formed playback setup.
-	// param building is wired in a subsequent task.
-	// Returns ErrInvalidPlaybackFormat (wrappable via errors.Is) for bad values.
 	if format.SampleRate <= 0 || format.Channels <= 0 || format.FramesPerBuffer <= 0 {
 		return fmt.Errorf("%w: SampleRate=%d, Channels=%d, FramesPerBuffer=%d",
 			ErrInvalidPlaybackFormat, format.SampleRate, format.Channels, format.FramesPerBuffer)
 	}
+	// Build the SPA POD params that describe the stream's audio format.
+	cp, err := buildRawAudioParams(format)
+	if err != nil {
+		return err
+	}
 	// PW_DIRECTION_OUTPUT = 1, PW_ID_ANY = 0xffffffff
 	// PW_STREAM_FLAG_AUTOCONNECT | PW_STREAM_FLAG_MAP_BUFFERS = 0x01 | 0x04
-	ret := pw_stream_connect(streamPtr, pwDirectionOutput, pwIDAny, pwStreamFlagAutoConnectMapBuffers, nil, 0)
+	ret := pw_stream_connect(streamPtr, pwDirectionOutput, pwIDAny, pwStreamFlagAutoConnectMapBuffers, cp.Pointer(), cp.Count())
 	if ret < 0 {
 		return &PWError{Func: "pw_stream_connect", Code: ret}
 	}

M internal/capi/stream_ops_test.go +124 -3


@@ -327,9 +327,9 @@ func TestConnectPlaybackStreamAcceptsValidFormat(t *testing.T) {
327	327		}
328	328		}
329	329
330		-	// TestRunMainLoopReturnsErrorOnNegativeReturn verifies that RunMainLoop
331		-	// returns a PWError when pw_main_loop_run returns a negative value.
332		-	func TestRunMainLoopReturnsErrorOnNegativeReturn(t *testing.T) {
	330	+	// TestRunMainLoopReturnsPWErrorOnNegativeReturn verifies that RunMainLoop
	331	+	// returns a typed PWError when pw_main_loop_run returns a negative value.
	332	+	func TestRunMainLoopReturnsPWErrorOnNegativeResult(t *testing.T) {
333	333		origRun := pw_main_loop_run
334	334		t.Cleanup(func() { pw_main_loop_run = origRun })
335	335

@@ -375,3 +375,124 @@ func TestRunMainLoopReturnsNilOnSuccess(t *testing.T) {
375	375		t.Fatalf("expected nil on success, got %v", err)
376	376		}
377	377		}
	378	+
	379	+	// TestCreatePlaybackStreamUsesPWLoopFromMainLoop verifies that CreatePlaybackStream
	380	+	// calls pw_main_loop_get_loop to obtain the pw_loop pointer and passes it to
	381	+	// pw_stream_new_simple (not the raw main loop pointer).
	382	+	func TestCreatePlaybackStreamUsesPWLoopFromMainLoop(t *testing.T) {
	383	+	origGetLoop := pw_main_loop_get_loop
	384	+	origNewSimple := pw_stream_new_simple
	385	+	t.Cleanup(func() {
	386	+	pw_main_loop_get_loop = origGetLoop
	387	+	pw_stream_new_simple = origNewSimple
	388	+	})
	389	+
	390	+	fakeMainLoop := unsafe.Pointer(uintptr(0xAAAA))
	391	+	fakePWLoop := unsafe.Pointer(uintptr(0xBBBB))
	392	+	var gotContext unsafe.Pointer
	393	+
	394	+	pw_main_loop_get_loop = func(loop unsafe.Pointer) unsafe.Pointer {
	395	+	if loop != fakeMainLoop {
	396	+	t.Errorf("pw_main_loop_get_loop received %v, want %v", loop, fakeMainLoop)
	397	+	}
	398	+	return fakePWLoop
	399	+	}
	400	+
	401	+	pw_stream_new_simple = func(context unsafe.Pointer, name *byte, props unsafe.Pointer, events unsafe.Pointer, data unsafe.Pointer) unsafe.Pointer {
	402	+	gotContext = context
	403	+	// Return a non-nil fake stream pointer.
	404	+	return unsafe.Pointer(uintptr(0xCCCC))
	405	+	}
	406	+
	407	+	ops := &streamOpsImpl{
	408	+	pinned: make(map[unsafe.Pointer]*streamCallbackStorage),
	409	+	}
	410	+
	411	+	stream, err := ops.CreatePlaybackStream(fakeMainLoop, "test", func() {})
	412	+	if err != nil {
	413	+	t.Fatalf("unexpected error: %v", err)
	414	+	}
	415	+	if stream == nil {
	416	+	t.Fatal("expected non-nil stream pointer")
	417	+	}
	418	+	if gotContext != fakePWLoop {
	419	+	t.Errorf("pw_stream_new_simple received context %v, want %v (the pw_loop from pw_main_loop_get_loop)", gotContext, fakePWLoop)
	420	+	}
	421	+	}
	422	+
	423	+	// TestCreatePlaybackStreamReturnsPWErrorWhenPWLoopIsNull verifies that
	424	+	// CreatePlaybackStream returns a typed PWError when pw_main_loop_get_loop
	425	+	// returns nil.
	426	+	func TestCreatePlaybackStreamReturnsPWErrorWhenPWLoopIsNull(t *testing.T) {
	427	+	origGetLoop := pw_main_loop_get_loop
	428	+	origNewSimple := pw_stream_new_simple
	429	+	t.Cleanup(func() {
	430	+	pw_main_loop_get_loop = origGetLoop
	431	+	pw_stream_new_simple = origNewSimple
	432	+	})
	433	+
	434	+	pw_main_loop_get_loop = func(loop unsafe.Pointer) unsafe.Pointer {
	435	+	return nil
	436	+	}
	437	+
	438	+	// pw_stream_new_simple should NOT be called.
	439	+	pw_stream_new_simple = func(context unsafe.Pointer, name *byte, props unsafe.Pointer, events unsafe.Pointer, data unsafe.Pointer) unsafe.Pointer {
	440	+	t.Fatal("pw_stream_new_simple should not be called when pw_main_loop_get_loop returns nil")
	441	+	return nil
	442	+	}
	443	+
	444	+	ops := &streamOpsImpl{
	445	+	pinned: make(map[unsafe.Pointer]*streamCallbackStorage),
	446	+	}
	447	+
	448	+	fakeMainLoop := unsafe.Pointer(uintptr(0xAAAA))
	449	+	_, err := ops.CreatePlaybackStream(fakeMainLoop, "test", func() {})
	450	+	if err == nil {
	451	+	t.Fatal("expected error when pw_main_loop_get_loop returns nil")
	452	+	}
	453	+
	454	+	var pwErr *PWError
	455	+	if !errors.As(err, &pwErr) {
	456	+	t.Fatalf("expected *PWError, got %T: %v", err, err)
	457	+	}
	458	+	if pwErr.Func != "pw_main_loop_get_loop" {
	459	+	t.Errorf("expected Func='pw_main_loop_get_loop', got '%s'", pwErr.Func)
	460	+	}
	461	+	}
	462	+
	463	+	// TestConnectPlaybackStreamPassesParamsFromHelper verifies that ConnectPlaybackStream
	464	+	// calls buildRawAudioParams and passes the resulting non-nil params and nParams > 0
	465	+	// into pw_stream_connect.
	466	+	func TestConnectPlaybackStreamPassesParamsFromHelper(t *testing.T) {
	467	+	origConnect := pw_stream_connect
	468	+	t.Cleanup(func() { pw_stream_connect = origConnect })
	469	+
	470	+	var gotPorts unsafe.Pointer
	471	+	var gotNPorts uint32
	472	+
	473	+	pw_stream_connect = func(stream unsafe.Pointer, direction int32, id uint32, flags uint32, ports unsafe.Pointer, n_ports uint32) int32 {
	474	+	gotPorts = ports
	475	+	gotNPorts = n_ports
	476	+	return 0
	477	+	}
	478	+
	479	+	ops := &streamOpsImpl{}
	480	+	fakePtr := unsafe.Pointer(uintptr(0x5678))
	481	+
	482	+	validFmt := portout.PlaybackFormat{
	483	+	SampleRate: 48000,
	484	+	Channels: 2,
	485	+	FramesPerBuffer: 1024,
	486	+	}
	487	+	err := ops.ConnectPlaybackStream(fakePtr, validFmt)
	488	+	if err != nil {
	489	+	t.Fatalf("expected nil error for valid format, got %v", err)
	490	+	}
	491	+
	492	+	if gotPorts == nil {
	493	+	t.Fatal("expected non-nil params pointer from buildRawAudioParams, got nil")
	494	+	}
	495	+	if gotNPorts == 0 {
	496	+	t.Fatal("expected nParams > 0 from buildRawAudioParams, got 0")
	497	+	}
	498	+	}


@@ -327,9 +327,9 @@ func TestConnectPlaybackStreamAcceptsValidFormat(t *testing.T) {
327		}	327		}
328		}	328		}
329			329
330	-	// TestRunMainLoopReturnsErrorOnNegativeReturn verifies that RunMainLoop	330	+	// TestRunMainLoopReturnsPWErrorOnNegativeReturn verifies that RunMainLoop
331	-	// returns a PWError when pw_main_loop_run returns a negative value.	331	+	// returns a typed PWError when pw_main_loop_run returns a negative value.
332	-	func TestRunMainLoopReturnsErrorOnNegativeReturn(t *testing.T) {	332	+	func TestRunMainLoopReturnsPWErrorOnNegativeResult(t *testing.T) {
333		origRun := pw_main_loop_run	333		origRun := pw_main_loop_run
334		t.Cleanup(func() { pw_main_loop_run = origRun })	334		t.Cleanup(func() { pw_main_loop_run = origRun })
335			335

@@ -375,3 +375,124 @@ func TestRunMainLoopReturnsNilOnSuccess(t *testing.T) {
375		t.Fatalf("expected nil on success, got %v", err)	375		t.Fatalf("expected nil on success, got %v", err)
376		}	376		}
377		}	377		}
			378	+
			379	+	// TestCreatePlaybackStreamUsesPWLoopFromMainLoop verifies that CreatePlaybackStream
			380	+	// calls pw_main_loop_get_loop to obtain the pw_loop pointer and passes it to
			381	+	// pw_stream_new_simple (not the raw main loop pointer).
			382	+	func TestCreatePlaybackStreamUsesPWLoopFromMainLoop(t *testing.T) {
			383	+	origGetLoop := pw_main_loop_get_loop
			384	+	origNewSimple := pw_stream_new_simple
			385	+	t.Cleanup(func() {
			386	+	pw_main_loop_get_loop = origGetLoop
			387	+	pw_stream_new_simple = origNewSimple
			388	+	})
			389	+
			390	+	fakeMainLoop := unsafe.Pointer(uintptr(0xAAAA))
			391	+	fakePWLoop := unsafe.Pointer(uintptr(0xBBBB))
			392	+	var gotContext unsafe.Pointer
			393	+
			394	+	pw_main_loop_get_loop = func(loop unsafe.Pointer) unsafe.Pointer {
			395	+	if loop != fakeMainLoop {
			396	+	t.Errorf("pw_main_loop_get_loop received %v, want %v", loop, fakeMainLoop)
			397	+	}
			398	+	return fakePWLoop
			399	+	}
			400	+
			401	+	pw_stream_new_simple = func(context unsafe.Pointer, name *byte, props unsafe.Pointer, events unsafe.Pointer, data unsafe.Pointer) unsafe.Pointer {
			402	+	gotContext = context
			403	+	// Return a non-nil fake stream pointer.
			404	+	return unsafe.Pointer(uintptr(0xCCCC))
			405	+	}
			406	+
			407	+	ops := &streamOpsImpl{
			408	+	pinned: make(map[unsafe.Pointer]*streamCallbackStorage),
			409	+	}
			410	+
			411	+	stream, err := ops.CreatePlaybackStream(fakeMainLoop, "test", func() {})
			412	+	if err != nil {
			413	+	t.Fatalf("unexpected error: %v", err)
			414	+	}
			415	+	if stream == nil {
			416	+	t.Fatal("expected non-nil stream pointer")
			417	+	}
			418	+	if gotContext != fakePWLoop {
			419	+	t.Errorf("pw_stream_new_simple received context %v, want %v (the pw_loop from pw_main_loop_get_loop)", gotContext, fakePWLoop)
			420	+	}
			421	+	}
			422	+
			423	+	// TestCreatePlaybackStreamReturnsPWErrorWhenPWLoopIsNull verifies that
			424	+	// CreatePlaybackStream returns a typed PWError when pw_main_loop_get_loop
			425	+	// returns nil.
			426	+	func TestCreatePlaybackStreamReturnsPWErrorWhenPWLoopIsNull(t *testing.T) {
			427	+	origGetLoop := pw_main_loop_get_loop
			428	+	origNewSimple := pw_stream_new_simple
			429	+	t.Cleanup(func() {
			430	+	pw_main_loop_get_loop = origGetLoop
			431	+	pw_stream_new_simple = origNewSimple
			432	+	})
			433	+
			434	+	pw_main_loop_get_loop = func(loop unsafe.Pointer) unsafe.Pointer {
			435	+	return nil
			436	+	}
			437	+
			438	+	// pw_stream_new_simple should NOT be called.
			439	+	pw_stream_new_simple = func(context unsafe.Pointer, name *byte, props unsafe.Pointer, events unsafe.Pointer, data unsafe.Pointer) unsafe.Pointer {
			440	+	t.Fatal("pw_stream_new_simple should not be called when pw_main_loop_get_loop returns nil")
			441	+	return nil
			442	+	}
			443	+
			444	+	ops := &streamOpsImpl{
			445	+	pinned: make(map[unsafe.Pointer]*streamCallbackStorage),
			446	+	}
			447	+
			448	+	fakeMainLoop := unsafe.Pointer(uintptr(0xAAAA))
			449	+	_, err := ops.CreatePlaybackStream(fakeMainLoop, "test", func() {})
			450	+	if err == nil {
			451	+	t.Fatal("expected error when pw_main_loop_get_loop returns nil")
			452	+	}
			453	+
			454	+	var pwErr *PWError
			455	+	if !errors.As(err, &pwErr) {
			456	+	t.Fatalf("expected *PWError, got %T: %v", err, err)
			457	+	}
			458	+	if pwErr.Func != "pw_main_loop_get_loop" {
			459	+	t.Errorf("expected Func='pw_main_loop_get_loop', got '%s'", pwErr.Func)
			460	+	}
			461	+	}
			462	+
			463	+	// TestConnectPlaybackStreamPassesParamsFromHelper verifies that ConnectPlaybackStream
			464	+	// calls buildRawAudioParams and passes the resulting non-nil params and nParams > 0
			465	+	// into pw_stream_connect.
			466	+	func TestConnectPlaybackStreamPassesParamsFromHelper(t *testing.T) {
			467	+	origConnect := pw_stream_connect
			468	+	t.Cleanup(func() { pw_stream_connect = origConnect })
			469	+
			470	+	var gotPorts unsafe.Pointer
			471	+	var gotNPorts uint32
			472	+
			473	+	pw_stream_connect = func(stream unsafe.Pointer, direction int32, id uint32, flags uint32, ports unsafe.Pointer, n_ports uint32) int32 {
			474	+	gotPorts = ports
			475	+	gotNPorts = n_ports
			476	+	return 0
			477	+	}
			478	+
			479	+	ops := &streamOpsImpl{}
			480	+	fakePtr := unsafe.Pointer(uintptr(0x5678))
			481	+
			482	+	validFmt := portout.PlaybackFormat{
			483	+	SampleRate: 48000,
			484	+	Channels: 2,
			485	+	FramesPerBuffer: 1024,
			486	+	}
			487	+	err := ops.ConnectPlaybackStream(fakePtr, validFmt)
			488	+	if err != nil {
			489	+	t.Fatalf("expected nil error for valid format, got %v", err)
			490	+	}
			491	+
			492	+	if gotPorts == nil {
			493	+	t.Fatal("expected non-nil params pointer from buildRawAudioParams, got nil")
			494	+	}
			495	+	if gotNPorts == 0 {
			496	+	t.Fatal("expected nParams > 0 from buildRawAudioParams, got 0")
			497	+	}
			498	+	}