test(integration): add video source tests and usage docs

commit: fcbefd12b21182a478deb90ea925d08d130e7522
author: brice · 8d ago · 2026-10-14 19:28 UTC
parent: e43eec4
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

5 files · +500 -1

M README.md +71 -1


@@ -23,7 +23,7 @@ The binding layer is generated from the checked-in model in `gen/pipewire.js
23	23		- Inbound port interfaces for the handwritten core layer
24	24		- Public API wrappers in `pipewire/` and SPA constants/helpers in `spa/`
25	25
26		-	A small hand-written core implements runtime init/deinit, player lifecycle, stream setup, and PCM buffer handoff for playback.
	26	+	A small hand-written core implements runtime init/deinit, player lifecycle, stream setup, and PCM buffer handoff for playback, plus the zero-allocation `VideoSource` (loop, context, core, memfd video buffers). The per-frame C calls of the video path use `purego.Syscall6` instead of `RegisterLibFunc` wrappers, which allocate on every call.
27	27
28	28		Mocks for public interfaces and outbound ports are generated by [Mockery v3](https://vektra.github.io/mockery/) and shipped in `pipewire/mocks/` and `internal/ports/out/mocks/`.
29	29

@@ -86,6 +86,76 @@ if err := player.Start(); err != nil {
86	86		}
87	87		```
88	88
	89	+	### Video source (screen-cast style producer)
	90	+
	91	+	`VideoSource` publishes a PipeWire `Video/Source` node whose frames you write
	92	+	yourself into shared memory (memfd). It drives the graph, so there is no
	93	+	process callback to wait for, and a steady-state frame performs no heap
	94	+	allocation. DMA-BUF is not supported; buffers are always memfd.
	95	+
	96	+	A `Loop`, the `Context` and `Core` created on it, and the `VideoSource` must be
	97	+	used from a single goroutine locked to one OS thread. You own the loop: poll
	98	+	`Loop.Fd()` together with your other file descriptors and call `Loop.Iterate()`
	99	+	when it is readable. Stream callbacks (`VideoHandler`) run inside `Iterate`.
	100	+
	101	+	```go
	102	+	import (
	103	+	"runtime"
	104	+	"time"
	105	+
	106	+	"github.com/bnema/purego-pipewire/pipewire"
	107	+	)
	108	+
	109	+	runtime.LockOSThread()
	110	+	defer runtime.UnlockOSThread()
	111	+
	112	+	rt, err := pipewire.Init()
	113	+	if err != nil {
	114	+	return err
	115	+	}
	116	+	defer rt.Deinit()
	117	+
	118	+	loop, _ := rt.NewLoop()
	119	+	defer loop.Destroy()
	120	+	ctx, _ := rt.NewContext(loop)
	121	+	defer ctx.Destroy()
	122	+	core, err := ctx.Connect()
	123	+	if err != nil {
	124	+	return err
	125	+	}
	126	+	defer core.Disconnect()
	127	+
	128	+	src, err := core.NewVideoSource(pipewire.VideoConfig{
	129	+	Name: "my-screen",
	130	+	Width: 1920,
	131	+	Height: 1080,
	132	+	MaxFPS: 60,
	133	+	Formats: []pipewire.VideoFormat{pipewire.VideoFormatBGRx, pipewire.VideoFormatBGRA},
	134	+	}, handler) // handler implements pipewire.VideoHandler
	135	+	if err != nil {
	136	+	return err
	137	+	}
	138	+	defer src.Destroy()
	139	+
	140	+	// handler.StateChanged reports the node id once the stream is PAUSED or
	141	+	// STREAMING; hand it to the consumer (for example pipewiresrc path=<id>).
	142	+
	143	+	// One frame: Dequeue, fill, Queue, then Trigger, in that order.
	144	+	if b := src.Dequeue(); b != nil { // nil: no buffer is free, skip the frame
	145	+	// b.Data is a mapped memfd, b.Stride bytes per row.
	146	+	copy(b.Data, frame)
	147	+	src.Queue(b, seq, time.Since(start), false) // true flags the frame corrupted
	148	+	src.Trigger()
	149	+	}
	150	+	```
	151	+
	152	+	Buffers that PipeWire removes while you still hold them (renegotiation,
	153	+	`Resize`, `Destroy`, the consumer leaving) stay mapped until you hand them back
	154	+	with `Queue` or `Release`. `Resize` and `Destroy` are best called when no buffer
	155	+	is borrowed. A consumer that is not a plain `pipewiresrc` may need to declare
	156	+	`media.type=Video` (for example `pipewiresrc stream-properties=props,media.type=Video`)
	157	+	for WirePlumber to link it.
	158	+
89	159		### Testing with mocks
90	160
91	161		The package ships pre-generated [Mockery v3](https://vektra.github.io/mockery/) mocks for the public interfaces in `github.com/bnema/purego-pipewire/pipewire/mocks`.


@@ -23,7 +23,7 @@ The binding layer is generated from the checked-in model in `gen/pipewire.js
23		- Inbound port interfaces for the handwritten core layer	23		- Inbound port interfaces for the handwritten core layer
24		- Public API wrappers in `pipewire/` and SPA constants/helpers in `spa/`	24		- Public API wrappers in `pipewire/` and SPA constants/helpers in `spa/`
25			25
26	-	A small hand-written core implements runtime init/deinit, player lifecycle, stream setup, and PCM buffer handoff for playback.	26	+	A small hand-written core implements runtime init/deinit, player lifecycle, stream setup, and PCM buffer handoff for playback, plus the zero-allocation `VideoSource` (loop, context, core, memfd video buffers). The per-frame C calls of the video path use `purego.Syscall6` instead of `RegisterLibFunc` wrappers, which allocate on every call.
27			27
28		Mocks for public interfaces and outbound ports are generated by [Mockery v3](https://vektra.github.io/mockery/) and shipped in `pipewire/mocks/` and `internal/ports/out/mocks/`.	28		Mocks for public interfaces and outbound ports are generated by [Mockery v3](https://vektra.github.io/mockery/) and shipped in `pipewire/mocks/` and `internal/ports/out/mocks/`.
29			29

@@ -86,6 +86,76 @@ if err := player.Start(); err != nil {
86		}	86		}
87		```	87		```
88			88
			89	+	### Video source (screen-cast style producer)
			90	+
			91	+	`VideoSource` publishes a PipeWire `Video/Source` node whose frames you write
			92	+	yourself into shared memory (memfd). It drives the graph, so there is no
			93	+	process callback to wait for, and a steady-state frame performs no heap
			94	+	allocation. DMA-BUF is not supported; buffers are always memfd.
			95	+
			96	+	A `Loop`, the `Context` and `Core` created on it, and the `VideoSource` must be
			97	+	used from a single goroutine locked to one OS thread. You own the loop: poll
			98	+	`Loop.Fd()` together with your other file descriptors and call `Loop.Iterate()`
			99	+	when it is readable. Stream callbacks (`VideoHandler`) run inside `Iterate`.
			100	+
			101	+	```go
			102	+	import (
			103	+	"runtime"
			104	+	"time"
			105	+
			106	+	"github.com/bnema/purego-pipewire/pipewire"
			107	+	)
			108	+
			109	+	runtime.LockOSThread()
			110	+	defer runtime.UnlockOSThread()
			111	+
			112	+	rt, err := pipewire.Init()
			113	+	if err != nil {
			114	+	return err
			115	+	}
			116	+	defer rt.Deinit()
			117	+
			118	+	loop, _ := rt.NewLoop()
			119	+	defer loop.Destroy()
			120	+	ctx, _ := rt.NewContext(loop)
			121	+	defer ctx.Destroy()
			122	+	core, err := ctx.Connect()
			123	+	if err != nil {
			124	+	return err
			125	+	}
			126	+	defer core.Disconnect()
			127	+
			128	+	src, err := core.NewVideoSource(pipewire.VideoConfig{
			129	+	Name: "my-screen",
			130	+	Width: 1920,
			131	+	Height: 1080,
			132	+	MaxFPS: 60,
			133	+	Formats: []pipewire.VideoFormat{pipewire.VideoFormatBGRx, pipewire.VideoFormatBGRA},
			134	+	}, handler) // handler implements pipewire.VideoHandler
			135	+	if err != nil {
			136	+	return err
			137	+	}
			138	+	defer src.Destroy()
			139	+
			140	+	// handler.StateChanged reports the node id once the stream is PAUSED or
			141	+	// STREAMING; hand it to the consumer (for example pipewiresrc path=<id>).
			142	+
			143	+	// One frame: Dequeue, fill, Queue, then Trigger, in that order.
			144	+	if b := src.Dequeue(); b != nil { // nil: no buffer is free, skip the frame
			145	+	// b.Data is a mapped memfd, b.Stride bytes per row.
			146	+	copy(b.Data, frame)
			147	+	src.Queue(b, seq, time.Since(start), false) // true flags the frame corrupted
			148	+	src.Trigger()
			149	+	}
			150	+	```
			151	+
			152	+	Buffers that PipeWire removes while you still hold them (renegotiation,
			153	+	`Resize`, `Destroy`, the consumer leaving) stay mapped until you hand them back
			154	+	with `Queue` or `Release`. `Resize` and `Destroy` are best called when no buffer
			155	+	is borrowed. A consumer that is not a plain `pipewiresrc` may need to declare
			156	+	`media.type=Video` (for example `pipewiresrc stream-properties=props,media.type=Video`)
			157	+	for WirePlumber to link it.
			158	+
89		### Testing with mocks	159		### Testing with mocks
90			160
91		The package ships pre-generated [Mockery v3](https://vektra.github.io/mockery/) mocks for the public interfaces in `github.com/bnema/purego-pipewire/pipewire/mocks`.	161		The package ships pre-generated [Mockery v3](https://vektra.github.io/mockery/) mocks for the public interfaces in `github.com/bnema/purego-pipewire/pipewire/mocks`.

A integration/helpers_test.go +53 -0

@@ -0,0 +1,53 @@
	1	+	//go:build integration
	2	+
	3	+	package integration
	4	+
	5	+	import (
	6	+	"fmt"
	7	+	"os"
	8	+	"strings"
	9	+	"syscall"
	10	+	"time"
	11	+	"unsafe"
	12	+	)
	13	+
	14	+	type pollFd struct {
	15	+	fd int32
	16	+	events int16
	17	+	revents int16
	18	+	}
	19	+
	20	+	const pollIn = 0x1
	21	+
	22	+	// pollPFD and pollTS are package-level because syscall.Syscall6 forces
	23	+	// pointer arguments passed through uintptr(unsafe.Pointer(&x)) to the heap.
	24	+	// Only the test goroutine (locked to its thread) uses them.
	25	+	var (
	26	+	pollPFD pollFd
	27	+	pollTS syscall.Timespec
	28	+	)
	29	+
	30	+	// waitReadable waits up to timeout for fd to become readable (or for a
	31	+	// signal). It does not allocate, so it can sit inside testing.AllocsPerRun.
	32	+	func waitReadable(fd int, timeout time.Duration) {
	33	+	pollPFD = pollFd{fd: int32(fd), events: pollIn}
	34	+	pollTS = syscall.Timespec{Sec: int64(timeout / time.Second), Nsec: int64(timeout % time.Second)}
	35	+	_, _, _ = syscall.Syscall6(syscall.SYS_PPOLL, uintptr(unsafe.Pointer(&pollPFD)), 1, uintptr(unsafe.Pointer(&pollTS)), 0, 0, 0)
	36	+	}
	37	+
	38	+	// isMapped reports whether the address of p lies inside a mapping listed in
	39	+	// /proc/self/maps.
	40	+	func isMapped(p *byte) bool {
	41	+	addr := uint64(uintptr(unsafe.Pointer(p)))
	42	+	b, err := os.ReadFile("/proc/self/maps")
	43	+	if err != nil {
	44	+	return true
	45	+	}
	46	+	for _, line := range strings.Split(string(b), "\n") {
	47	+	var lo, hi uint64
	48	+	if _, err := fmt.Sscanf(line, "%x-%x", &lo, &hi); err == nil && addr >= lo && addr < hi {
	49	+	return true
	50	+	}
	51	+	}
	52	+	return false
	53	+	}

@@ -0,0 +1,53 @@
	1	+	//go:build integration
	2	+
	3	+	package integration
	4	+
	5	+	import (
	6	+	"fmt"
	7	+	"os"
	8	+	"strings"
	9	+	"syscall"
	10	+	"time"
	11	+	"unsafe"
	12	+	)
	13	+
	14	+	type pollFd struct {
	15	+	fd int32
	16	+	events int16
	17	+	revents int16
	18	+	}
	19	+
	20	+	const pollIn = 0x1
	21	+
	22	+	// pollPFD and pollTS are package-level because syscall.Syscall6 forces
	23	+	// pointer arguments passed through uintptr(unsafe.Pointer(&x)) to the heap.
	24	+	// Only the test goroutine (locked to its thread) uses them.
	25	+	var (
	26	+	pollPFD pollFd
	27	+	pollTS syscall.Timespec
	28	+	)
	29	+
	30	+	// waitReadable waits up to timeout for fd to become readable (or for a
	31	+	// signal). It does not allocate, so it can sit inside testing.AllocsPerRun.
	32	+	func waitReadable(fd int, timeout time.Duration) {
	33	+	pollPFD = pollFd{fd: int32(fd), events: pollIn}
	34	+	pollTS = syscall.Timespec{Sec: int64(timeout / time.Second), Nsec: int64(timeout % time.Second)}
	35	+	_, _, _ = syscall.Syscall6(syscall.SYS_PPOLL, uintptr(unsafe.Pointer(&pollPFD)), 1, uintptr(unsafe.Pointer(&pollTS)), 0, 0, 0)
	36	+	}
	37	+
	38	+	// isMapped reports whether the address of p lies inside a mapping listed in
	39	+	// /proc/self/maps.
	40	+	func isMapped(p *byte) bool {
	41	+	addr := uint64(uintptr(unsafe.Pointer(p)))
	42	+	b, err := os.ReadFile("/proc/self/maps")
	43	+	if err != nil {
	44	+	return true
	45	+	}
	46	+	for _, line := range strings.Split(string(b), "\n") {
	47	+	var lo, hi uint64
	48	+	if _, err := fmt.Sscanf(line, "%x-%x", &lo, &hi); err == nil && addr >= lo && addr < hi {
	49	+	return true
	50	+	}
	51	+	}
	52	+	return false
	53	+	}

A integration/race_off_test.go +5 -0

@@ -0,0 +1,5 @@
	1	+	//go:build integration && !race
	2	+
	3	+	package integration
	4	+
	5	+	const raceEnabled = false

@@ -0,0 +1,5 @@
	1	+	//go:build integration && !race
	2	+
	3	+	package integration
	4	+
	5	+	const raceEnabled = false

A integration/race_on_test.go +5 -0

@@ -0,0 +1,5 @@
	1	+	//go:build integration && race
	2	+
	3	+	package integration
	4	+
	5	+	const raceEnabled = true

@@ -0,0 +1,5 @@
	1	+	//go:build integration && race
	2	+
	3	+	package integration
	4	+
	5	+	const raceEnabled = true

A integration/video_test.go +366 -0

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

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