1//go:build integration
2
3package integration
4
5import (
6 "context"
7 "fmt"
8 "os"
9 "os/exec"
10 "path/filepath"
11 "runtime"
12 "sync/atomic"
13 "testing"
14 "time"
15
16 "github.com/bnema/purego-pipewire/internal/ports/in/mocks"
17 "github.com/bnema/purego-pipewire/pipewire"
18 "github.com/bnema/purego-pipewire/spa"
19 "github.com/stretchr/testify/mock"
20)
21
22const (
23 vidW = 320
24 vidH = 240
25)
26
27// videoEvents records what the source reports. Callbacks run on the loop
28// thread; the fields the test reads are atomics.
29type videoEvents struct {
30 state atomic.Int32
31 nodeID atomic.Uint32
32 format atomic.Uint32
33 width atomic.Int32
34 height atomic.Int32
35 stride atomic.Int32
36 buffers atomic.Int32
37 statesCh chan int32
38}
39
40// newVideoHandler returns a generated VideoHandler mock that records every
41// callback into the returned events.
42func newVideoHandler(t *testing.T) (*mocks.MockVideoHandler, *videoEvents) {
43 ev := &videoEvents{statesCh: make(chan int32, 64)}
44 h := mocks.NewMockVideoHandler(t)
45 h.EXPECT().StateChanged(mock.Anything, mock.Anything, mock.Anything).Run(func(state int32, nodeID uint32, _ string) {
46 ev.state.Store(state)
47 if nodeID != 0 {
48 ev.nodeID.Store(nodeID)
49 }
50 select {
51 case ev.statesCh <- state:
52 default:
53 }
54 }).Maybe()
55 h.EXPECT().FormatChanged(mock.Anything, mock.Anything, mock.Anything, mock.Anything).Run(func(f spa.VideoFormat, w, hh, stride int) {
56 ev.format.Store(uint32(f))
57 ev.width.Store(int32(w))
58 ev.height.Store(int32(hh))
59 ev.stride.Store(int32(stride))
60 }).Maybe()
61 h.EXPECT().BuffersChanged().Run(func() { ev.buffers.Add(1) }).Maybe()
62 return h, ev
63}
64
65type videoEnv struct {
66 rt pipewire.Runtime
67 loop pipewire.Loop
68 ctx pipewire.Context
69 core pipewire.Core
70 src *pipewire.VideoSource
71 handler *videoEvents
72}
73
74func (e *videoEnv) close() {
75 if e.src != nil {
76 e.src.Destroy()
77 }
78 if e.core != nil {
79 e.core.Disconnect()
80 }
81 if e.ctx != nil {
82 e.ctx.Destroy()
83 }
84 if e.loop != nil {
85 e.loop.Destroy()
86 }
87 e.rt.Deinit()
88}
89
90// pollIterate waits up to timeout for the loop fd and iterates once.
91func pollIterate(loop pipewire.Loop, timeout time.Duration) {
92 waitReadable(loop.Fd(), timeout)
93 loop.Iterate()
94}
95
96func skipUnlessPipeWire(t *testing.T) {
97 t.Helper()
98 dir := os.Getenv("XDG_RUNTIME_DIR")
99 if dir == "" {
100 t.Skip("XDG_RUNTIME_DIR not set")
101 }
102 if _, err := os.Stat(filepath.Join(dir, "pipewire-0")); err != nil {
103 t.Skipf("no PipeWire socket: %v", err)
104 }
105}
106
107// newVideoEnv must be called from a goroutine locked to its OS thread.
108func newVideoEnv(t *testing.T) *videoEnv {
109 t.Helper()
110 skipUnlessPipeWire(t)
111 rt, err := pipewire.Init()
112 if err != nil {
113 t.Fatalf("Init: %v", err)
114 }
115 h, ev := newVideoHandler(t)
116 e := &videoEnv{rt: rt, handler: ev}
117 if e.loop, err = rt.NewLoop(); err != nil {
118 e.close()
119 t.Fatalf("NewLoop: %v", err)
120 }
121 if e.ctx, err = rt.NewContext(e.loop); err != nil {
122 e.close()
123 t.Fatalf("NewContext: %v", err)
124 }
125 if e.core, err = e.ctx.Connect(); err != nil {
126 e.close()
127 t.Skipf("cannot connect to PipeWire: %v", err)
128 }
129 e.src, err = e.core.NewVideoSource(pipewire.VideoConfig{
130 Name: "purego-pipewire-test", Width: vidW, Height: vidH, MaxFPS: 60,
131 Formats: []pipewire.VideoFormat{pipewire.VideoFormatBGRx},
132 }, h)
133 if err != nil {
134 e.close()
135 t.Fatalf("NewVideoSource: %v", err)
136 }
137 return e
138}
139
140// waitNode iterates until the node id is known.
141func (e *videoEnv) waitNode(t *testing.T) uint32 {
142 t.Helper()
143 deadline := time.Now().Add(5 * time.Second)
144 for e.handler.nodeID.Load() == 0 {
145 if time.Now().After(deadline) {
146 t.Fatal("timed out waiting for the node id")
147 }
148 pollIterate(e.loop, 20*time.Millisecond)
149 }
150 return e.handler.nodeID.Load()
151}
152
153// produce runs one Dequeue -> fill -> Queue -> Trigger cycle. It reports whether a
154// buffer was available.
155func (e *videoEnv) produce(seq uint64, fill byte) bool {
156 b := e.src.Dequeue()
157 if b == nil {
158 return false
159 }
160 // Touch the first and last byte of the frame; filling everything would
161 // dominate the allocation-free path being measured.
162 b.Data[0] = fill
163 b.Data[len(b.Data)-1] = fill
164 e.src.Queue(b, seq, time.Duration(seq)*time.Second/60, false)
165 e.src.Trigger()
166 return true
167}
168
169// gstMediaType makes pipewiresrc declare media.type=Video. Without it the
170// stream registers as Stream/Input/Unknown and WirePlumber refuses to link it
171// to a Video node ("target not found").
172const gstMediaType = "stream-properties=props,media.type=Video"
173
174func startGst(t *testing.T, args ...string) *exec.Cmd {
175 t.Helper()
176 path, err := exec.LookPath("gst-launch-1.0")
177 if err != nil {
178 t.Skip("gst-launch-1.0 not found")
179 }
180 cmd := exec.Command(path, args...)
181 cmd.Stdout, cmd.Stderr = os.Stdout, os.Stderr
182 if err := cmd.Start(); err != nil {
183 t.Fatalf("start gst-launch-1.0: %v", err)
184 }
185 return cmd
186}
187
188func TestVideoSource(t *testing.T) {
189 runtime.LockOSThread()
190 defer runtime.UnlockOSThread()
191 skipUnlessPipeWire(t)
192 if _, err := exec.LookPath("gst-launch-1.0"); err != nil {
193 t.Skip("gst-launch-1.0 not found")
194 }
195 e := newVideoEnv(t)
196 defer e.close()
197 node := e.waitNode(t)
198
199 cmd := startGst(t, "-q", "pipewiresrc", fmt.Sprintf("path=%d", node), gstMediaType, "num-buffers=120", "!",
200 fmt.Sprintf("video/x-raw,format=BGRx,width=%d,height=%d", vidW, vidH), "!", "fakesink")
201 done := make(chan error, 1)
202 go func() { done <- cmd.Wait() }()
203 defer func() { _ = cmd.Process.Kill() }()
204
205 var seq, produced uint64
206 next := time.Now()
207 ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
208 defer cancel()
209 for {
210 select {
211 case err := <-done:
212 if err != nil {
213 t.Fatalf("gst-launch-1.0 failed: %v", err)
214 }
215 if produced < 120 {
216 t.Fatalf("gst finished after only %d produced frames", produced)
217 }
218 if got := e.handler.width.Load(); got != vidW || e.handler.height.Load() != vidH || e.handler.stride.Load() != vidW*4 {
219 t.Fatalf("negotiated %dx%d stride %d", got, e.handler.height.Load(), e.handler.stride.Load())
220 }
221 if e.handler.format.Load() != uint32(pipewire.VideoFormatBGRx) {
222 t.Fatalf("negotiated format %d", e.handler.format.Load())
223 }
224 t.Logf("gst received 120 frames; produced %d", produced)
225 return
226 case <-ctx.Done():
227 t.Fatalf("timed out; produced %d frames", produced)
228 default:
229 }
230 // 60 Hz tick from the loop's poll timeout.
231 wait := time.Until(next)
232 if wait < 0 {
233 wait = 0
234 }
235 pollIterate(e.loop, wait)
236 if !time.Now().Before(next) {
237 next = next.Add(time.Second / 60)
238 if e.handler.state.Load() == pipewire.StreamStateStreaming && e.produce(seq, byte(seq)) {
239 seq++
240 produced++
241 }
242 }
243 }
244}
245
246func TestVideoSourceConsumerLeaves(t *testing.T) {
247 runtime.LockOSThread()
248 defer runtime.UnlockOSThread()
249 skipUnlessPipeWire(t)
250 if _, err := exec.LookPath("gst-launch-1.0"); err != nil {
251 t.Skip("gst-launch-1.0 not found")
252 }
253 e := newVideoEnv(t)
254 defer e.close()
255 node := e.waitNode(t)
256
257 cmd := startGst(t, "-q", "pipewiresrc", fmt.Sprintf("path=%d", node), gstMediaType, "!", "fakesink")
258 defer func() { _ = cmd.Process.Kill(); _ = cmd.Wait() }()
259
260 // Stream until the consumer has started streaming.
261 deadline := time.Now().Add(10 * time.Second)
262 var seq uint64
263 for e.handler.state.Load() != pipewire.StreamStateStreaming {
264 if time.Now().After(deadline) {
265 t.Fatal("stream never reached STREAMING")
266 }
267 pollIterate(e.loop, 16*time.Millisecond)
268 }
269 var borrowed *pipewire.VideoBuffer
270 for borrowed == nil {
271 if time.Now().After(deadline) {
272 t.Fatal("no buffer to borrow")
273 }
274 pollIterate(e.loop, 16*time.Millisecond)
275 e.produce(seq, 1)
276 seq++
277 borrowed = e.src.Dequeue()
278 }
279 data := borrowed.Data
280 buffersBefore := e.handler.buffers.Load()
281
282 // Kill the consumer while holding a buffer.
283 _ = cmd.Process.Kill()
284 _ = cmd.Wait()
285 deadline = time.Now().Add(10 * time.Second)
286 for st := e.handler.state.Load(); st == pipewire.StreamStateStreaming; st = e.handler.state.Load() {
287 if time.Now().After(deadline) {
288 t.Fatal("stream stayed STREAMING after the consumer left")
289 }
290 pollIterate(e.loop, 20*time.Millisecond)
291 e.src.Trigger()
292 }
293 if st := e.handler.state.Load(); st != pipewire.StreamStatePaused && st != pipewire.StreamStateUnconnected {
294 t.Fatalf("state after consumer left = %d, want PAUSED or UNCONNECTED", st)
295 }
296
297 // The borrowed buffer must still be writable (no SIGBUS/SIGSEGV), even if
298 // PipeWire removed it meanwhile.
299 data[0] = 0xEE
300 data[len(data)-1] = 0xEE
301 if !isMapped(&data[0]) {
302 t.Fatal("borrowed buffer was unmapped before Release")
303 }
304
305 // Give the buffer back. If PipeWire removed it while borrowed, Release must
306 // unmap it now; otherwise it goes back to PipeWire and stays mapped.
307 removed := e.handler.buffers.Load() != buffersBefore
308 e.src.Release(borrowed)
309 if removed && isMapped(&data[0]) {
310 t.Fatal("removed buffer still mapped after Release")
311 }
312 t.Logf("consumer left: removed while borrowed = %v", removed)
313 e.src.Release(borrowed) // a second Release is ignored
314}
315
316func TestVideoSourceFrameAllocations(t *testing.T) {
317 runtime.LockOSThread()
318 defer runtime.UnlockOSThread()
319 skipUnlessPipeWire(t)
320 if _, err := exec.LookPath("gst-launch-1.0"); err != nil {
321 t.Skip("gst-launch-1.0 not found")
322 }
323 if raceEnabled {
324 t.Skip("allocation counts are not meaningful under -race")
325 }
326 e := newVideoEnv(t)
327 defer e.close()
328 node := e.waitNode(t)
329
330 cmd := startGst(t, "-q", "pipewiresrc", fmt.Sprintf("path=%d", node), gstMediaType, "!", "fakesink", "sync=false")
331 defer func() { _ = cmd.Process.Kill(); _ = cmd.Wait() }()
332
333 deadline := time.Now().Add(15 * time.Second)
334 for e.handler.state.Load() != pipewire.StreamStateStreaming {
335 if time.Now().After(deadline) {
336 t.Fatal("stream never reached STREAMING")
337 }
338 pollIterate(e.loop, 16*time.Millisecond)
339 }
340 // Warm up so every buffer and every lazily initialised path has been used.
341 var seq uint64
342 for i := 0; i < 60; i++ {
343 pollIterate(e.loop, 16*time.Millisecond)
344 if e.produce(seq, 1) {
345 seq++
346 }
347 }
348 if seq == 0 {
349 t.Fatal("no frame could be produced during warm-up")
350 }
351
352 // Each run waits (without allocating) for the loop fd, so that the consumer
353 // has returned a buffer and most runs queue and trigger a real frame.
354 loopFd := e.loop.Fd() // read once, like a real poll loop does
355 var produced int
356 allocs := testing.AllocsPerRun(200, func() {
357 waitReadable(loopFd, 17*time.Millisecond)
358 e.loop.Iterate()
359 if e.produce(seq, byte(seq)) {
360 seq++
361 produced++
362 }
363 })
364 t.Logf("allocs per Dequeue/fill/Queue/Trigger/Iterate cycle: %v (%d frames queued)", allocs, produced)
365 if allocs != 0 {
366 t.Fatalf("allocs per cycle = %v, want 0", allocs)
367 }
368 if produced < 50 {
369 t.Fatalf("only %d frames were produced inside the measured closure, want >= 50", produced)
370 }
371}