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