1package capi
2
3import (
4 "errors"
5 "fmt"
6 "runtime"
7 "sync"
8 "unsafe"
9
10 "github.com/bnema/purego"
11
12 portout "github.com/bnema/purego-pipewire/internal/ports/out"
13)
14
15// PWError represents a PipeWire C API error with the function name and return code.
16type PWError struct {
17 Func string
18 Code int32
19}
20
21func (e *PWError) Error() string {
22 return fmt.Sprintf("%s failed: %d", e.Func, e.Code)
23}
24
25// ErrInvalidPlaybackFormat is returned by ConnectPlaybackStream when the
26// provided PlaybackFormat has zero or negative fields. Callers can use
27// errors.Is to check for this condition specifically.
28var ErrInvalidPlaybackFormat = errors.New("invalid playback format")
29
30var (
31 errStreamCreate = &PWError{Func: "pw_stream_new_simple", Code: 0}
32 errMainLoopCreate = &PWError{Func: "pw_main_loop_new", Code: 0}
33 errPWLoopFromMainLoop = &PWError{Func: "pw_main_loop_get_loop", Code: 0}
34)
35
36const pwVersionStreamEvents = 2
37
38const (
39 pwDirectionOutput = 1
40 pwIDAny = 0xffffffff
41 pwStreamFlagAutoConnectMapBuffers = 0x05
42)
43
44// streamCallbackStorage keeps the pw_stream_events struct and its
45// associated Go callback alive for the lifetime of the stream.
46// Without this, the stack-allocated values in CreatePlaybackStream
47// would be collected while PipeWire still holds pointers to them.
48type streamCallbackStorage struct {
49 processCb uintptr
50 events pw_stream_events
51}
52
53// streamOpsImpl implements portout.StreamOps using the real PipeWire C API
54// bindings registered via purego.
55//
56// Ownership/lifetime contract:
57// - pinned holds streamCallbackStorage for each stream created via CreatePlaybackStream.
58// It keeps Go callback closures alive so PipeWire can call them safely.
59// - pinnedParams holds connectParams for each stream connected via ConnectPlaybackStream.
60// PipeWire reads the SPA POD params asynchronously during format negotiation, so the
61// backing byte storage must remain live until DestroyStream releases it.
62// - Both maps are keyed by the stream pointer and must be cleaned up in DestroyStream.
63type streamLoopContext struct {
64 mainLoop unsafe.Pointer
65 pwLoop unsafe.Pointer
66}
67
68type streamOpsImpl struct {
69 mu sync.Mutex
70 pinned map[unsafe.Pointer]*streamCallbackStorage // keyed by stream ptr
71 pinnedParams map[unsafe.Pointer]*connectParams // keyed by stream ptr
72 streamLoops map[unsafe.Pointer]streamLoopContext // keyed by stream ptr
73 runningLoops map[unsafe.Pointer]bool // keyed by main loop ptr
74}
75
76// Verify interface compliance at compile time.
77var _ portout.StreamOps = (*streamOpsImpl)(nil)
78
79func (s *streamOpsImpl) CreatePlaybackStream(loopPtr unsafe.Pointer, name string, onProcess func()) (unsafe.Pointer, error) {
80 // Obtain the inner pw_loop from the main loop; pw_stream_new_simple
81 // expects a pw_loop, not a pw_main_loop.
82 pwLoop := pw_main_loop_get_loop(loopPtr)
83 if pwLoop == nil {
84 return nil, errPWLoopFromMainLoop
85 }
86
87 nameBytes := append([]byte(name), 0)
88
89 // Build persistent callback storage that outlives this function.
90 storage := &streamCallbackStorage{}
91 storage.processCb = purego.NewCallback(func(_ unsafe.Pointer) {
92 onProcess()
93 })
94 storage.events = pw_stream_events{
95 version: pwVersionStreamEvents,
96 process: storage.processCb,
97 }
98
99 ptr := pw_stream_new_simple(
100 pwLoop,
101 &nameBytes[0],
102 nil, // props
103 unsafe.Pointer(&storage.events), // events
104 nil, // data
105 )
106 if ptr == nil {
107 return nil, errStreamCreate
108 }
109
110 // Pin the storage so GC cannot collect it while PipeWire holds references.
111 s.mu.Lock()
112 if s.pinned == nil {
113 s.pinned = make(map[unsafe.Pointer]*streamCallbackStorage)
114 }
115 if s.streamLoops == nil {
116 s.streamLoops = make(map[unsafe.Pointer]streamLoopContext)
117 }
118 s.pinned[ptr] = storage
119 s.streamLoops[ptr] = streamLoopContext{mainLoop: loopPtr, pwLoop: pwLoop}
120 s.mu.Unlock()
121
122 return ptr, nil
123}
124
125func (s *streamOpsImpl) ConnectPlaybackStream(streamPtr unsafe.Pointer, format portout.PlaybackFormat) error {
126 // Build the SPA POD params that describe the stream's audio format.
127 cp, err := buildRawAudioParams(format)
128 if err != nil {
129 return err
130 }
131
132 // PW_DIRECTION_OUTPUT = 1, PW_ID_ANY = 0xffffffff
133 // PW_STREAM_FLAG_AUTOCONNECT | PW_STREAM_FLAG_MAP_BUFFERS = 0x01 | 0x04
134 ret := pw_stream_connect(streamPtr, pwDirectionOutput, pwIDAny, pwStreamFlagAutoConnectMapBuffers, cp.Pointer(), cp.Count())
135 if ret < 0 {
136 return &PWError{Func: "pw_stream_connect", Code: ret}
137 }
138
139 // Pin the connectParams for the lifetime of the stream. PipeWire reads
140 // the SPA POD params asynchronously during format negotiation, so the
141 // backing storage must remain alive until the stream is destroyed.
142 s.mu.Lock()
143 if s.pinnedParams == nil {
144 s.pinnedParams = make(map[unsafe.Pointer]*connectParams)
145 }
146 s.pinnedParams[streamPtr] = cp
147 s.mu.Unlock()
148
149 return nil
150}
151
152func (s *streamOpsImpl) SetStreamActive(streamPtr unsafe.Pointer, active bool) error {
153 ret := s.callStreamWithLoopLock(streamPtr, func() int32 {
154 return pw_stream_set_active(streamPtr, active)
155 })
156 if ret < 0 {
157 return &PWError{Func: "pw_stream_set_active", Code: ret}
158 }
159 return nil
160}
161
162func (s *streamOpsImpl) DequeueBuffer(streamPtr unsafe.Pointer) unsafe.Pointer {
163 return pw_stream_dequeue_buffer(streamPtr)
164}
165
166func (s *streamOpsImpl) QueueBuffer(streamPtr unsafe.Pointer, bufPtr unsafe.Pointer) error {
167 ret := pw_stream_queue_buffer(streamPtr, bufPtr)
168 if ret < 0 {
169 return &PWError{Func: "pw_stream_queue_buffer", Code: ret}
170 }
171 return nil
172}
173
174// DisconnectStream disconnects the stream from its port.
175// Returns a PWError if the operation fails.
176func (s *streamOpsImpl) DisconnectStream(streamPtr unsafe.Pointer) error {
177 ret := s.callStreamWithLoopLock(streamPtr, func() int32 {
178 return pw_stream_disconnect(streamPtr)
179 })
180 if ret < 0 {
181 return &PWError{Func: "pw_stream_disconnect", Code: ret}
182 }
183 return nil
184}
185
186// DestroyStream releases the pw_stream and associated pinned Go storage.
187// Callers must ensure the owning main loop has already stopped, or that they
188// otherwise satisfy PipeWire's required stream context, before invoking this.
189func (s *streamOpsImpl) DestroyStream(streamPtr unsafe.Pointer) {
190 s.mu.Lock()
191 storage, alive := s.pinned[streamPtr]
192 if !alive {
193 // Already destroyed or never tracked — skip.
194 s.mu.Unlock()
195 return
196 }
197 params := s.pinnedParams[streamPtr]
198 delete(s.pinned, streamPtr)
199 delete(s.pinnedParams, streamPtr)
200 delete(s.streamLoops, streamPtr)
201 s.mu.Unlock()
202
203 pw_stream_destroy(streamPtr)
204
205 // Keep callback and format storage alive until after pw_stream_destroy returns.
206 runtime.KeepAlive(storage)
207 runtime.KeepAlive(params)
208}
209
210func (s *streamOpsImpl) CreateMainLoop() (unsafe.Pointer, error) {
211 ptr := pw_main_loop_new(nil)
212 if ptr == nil {
213 return nil, errMainLoopCreate
214 }
215 return ptr, nil
216}
217
218func (s *streamOpsImpl) RunMainLoop(loopPtr unsafe.Pointer) error {
219 s.mu.Lock()
220 if s.runningLoops == nil {
221 s.runningLoops = make(map[unsafe.Pointer]bool)
222 }
223 s.runningLoops[loopPtr] = true
224 s.mu.Unlock()
225 defer func() {
226 s.mu.Lock()
227 delete(s.runningLoops, loopPtr)
228 s.mu.Unlock()
229 }()
230
231 ret := pw_main_loop_run(loopPtr)
232 if ret < 0 {
233 return &PWError{Func: "pw_main_loop_run", Code: ret}
234 }
235 return nil
236}
237
238func (s *streamOpsImpl) QuitMainLoop(loopPtr unsafe.Pointer) {
239 // Use the generated pw_main_loop_quit binding to signal loop exit.
240 pw_main_loop_quit(loopPtr)
241}
242
243func (s *streamOpsImpl) DestroyMainLoop(loopPtr unsafe.Pointer) {
244 pw_main_loop_destroy(loopPtr)
245}
246
247func (s *streamOpsImpl) callStreamWithLoopLock(streamPtr unsafe.Pointer, fn func() int32) int32 {
248 s.mu.Lock()
249 ctx, running := s.streamLoopContextLocked(streamPtr)
250 s.mu.Unlock()
251 if running && ctx.pwLoop != nil {
252 return withLoopLock(ctx.pwLoop, fn)
253 }
254 return fn()
255}
256
257func (s *streamOpsImpl) streamLoopContextLocked(streamPtr unsafe.Pointer) (streamLoopContext, bool) {
258 if s.streamLoops == nil {
259 return streamLoopContext{}, false
260 }
261 ctx, ok := s.streamLoops[streamPtr]
262 if !ok {
263 return streamLoopContext{}, false
264 }
265 return ctx, s.runningLoops != nil && s.runningLoops[ctx.mainLoop]
266}
267
268// DefaultStreamOps returns a StreamOps implementation backed by the real
269// PipeWire C API. Must be called after Register().
270func DefaultStreamOps() portout.StreamOps {
271 return &streamOpsImpl{}
272}