From e9cbde6d426b590a8bdc1a2ee1de32cc5021fd2f Mon Sep 17 00:00:00 2001 From: spacedouut Date: Wed, 26 Aug 2026 12:05:24 -0400 Subject: [PATCH 1/5] spike: implement linux KMS/GBM capture pipeline via libdrm/libgbm (purego) - pipelines/linux/kms.go: libdrm purego bindings, real ListDisplays (DRM connector/mode enumeration) + best-effort KMS framebuffer readback via drmModeGetFB2 + prime handle -> dma-buf -> mmap. - pipelines/linux/gbm.go: libgbm purego bindings + linear XRGB8888 BO (dma-buf + mmap) helper, the stepping stone for DMA-BUF -> nvh264enc. - pipelines/linux/x11.go: libX11 source selector (default screen size). - pipelines/linux/pipeline.go: Pipeline impl, --source kms|x11 selector, BGRA frame stream, synthetic fallback when real readback is unavailable. - mmap_linux.go / stub_other.go: cross-platform (linux-tagged) build. - main.go: add --source flag, import + select linux pipeline. Verified: go vet ./... clean, go build ./..., unit test for synth BGRA, socket smoke test exercises kms (perm-error) and x11 ($DISPLAY) paths. Not pushed to origin. Encode remains in agent ffmpeg per plan 2B. --- go.mod | 2 +- main.go | 4 + pipelines/linux/gbm.go | 141 +++++++++++ pipelines/linux/kms.go | 453 ++++++++++++++++++++++++++++++++++ pipelines/linux/mmap_linux.go | 21 ++ pipelines/linux/pipeline.go | 219 +++++++++++++++- pipelines/linux/stub_other.go | 30 +++ pipelines/linux/synth_test.go | 40 +++ pipelines/linux/x11.go | 134 ++++++++++ 9 files changed, 1042 insertions(+), 2 deletions(-) create mode 100644 pipelines/linux/gbm.go create mode 100644 pipelines/linux/kms.go create mode 100644 pipelines/linux/mmap_linux.go create mode 100644 pipelines/linux/stub_other.go create mode 100644 pipelines/linux/synth_test.go create mode 100644 pipelines/linux/x11.go diff --git a/go.mod b/go.mod index 2fc6566..2a0ff78 100644 --- a/go.mod +++ b/go.mod @@ -4,5 +4,5 @@ go 1.26.3 require ( github.com/LocalKinAI/sckit-go v0.3.1 // indirect - github.com/ebitengine/purego v0.8.0 // indirect + github.com/ebitengine/purego v0.8.0 ) diff --git a/main.go b/main.go index e99eebc..16d7a5c 100644 --- a/main.go +++ b/main.go @@ -14,6 +14,7 @@ import ( "sync" "distancedesktop/captured/pipelines" + "distancedesktop/captured/pipelines/linux" "distancedesktop/captured/pipelines/macos" ) @@ -21,11 +22,14 @@ var pipeline pipelines.Pipeline func main() { listen := flag.String("listen", "", "TCP address for remote control (e.g. :9090)") + source := flag.String("source", "kms", "capture source on linux: kms|x11") flag.Parse() switch runtime.GOOS { case "darwin": pipeline = macos.New() + case "linux": + pipeline = linux.New(*source) default: log.Fatalf("unsupported platform: %s", runtime.GOOS) } diff --git a/pipelines/linux/gbm.go b/pipelines/linux/gbm.go new file mode 100644 index 0000000..4cf6b8e --- /dev/null +++ b/pipelines/linux/gbm.go @@ -0,0 +1,141 @@ +//go:build linux + +package linux + +import ( + "fmt" + "os" + "sync" + + "github.com/ebitengine/purego" +) + +// --------------------------------------------------------------------------- +// libgbm bindings. Used by the synthetic source to allocate a real linear +// GBM BO, export it as a dma-buf and mmap it — the same primitive the future +// DMA-BUF -> nvh264enc encode path will reuse. +// --------------------------------------------------------------------------- + +const gbmLibName = "libgbm.so.1" + +const ( + gbmFormatXRGB8888 = 0x34325258 + gbmBoUseRendering = 1 << 2 + gbmBoUseWrite = 1 << 3 + gbmBoUseLinear = 1 << 4 +) + +var ( + gbmOnce sync.Once + gbmLib uintptr + + gbmCreateDevice func(fd int) uintptr + gbmDeviceDestroy func(dev uintptr) + gbmBoCreate func(dev uintptr, w, h, format, usage uint32) uintptr + gbmBoDestroy func(bo uintptr) + gbmBoGetFD func(bo uintptr) int + gbmBoGetStride func(bo uintptr) uint32 +) + +func loadGBM() { + gbmOnce.Do(func() { + h, err := purego.Dlopen(gbmLibName, purego.RTLD_NOW|purego.RTLD_GLOBAL) + if err != nil { + gbmLib = 0 + return + } + gbmLib = h + purego.RegisterLibFunc(&gbmCreateDevice, gbmLib, "gbm_create_device") + purego.RegisterLibFunc(&gbmDeviceDestroy, gbmLib, "gbm_device_destroy") + purego.RegisterLibFunc(&gbmBoCreate, gbmLib, "gbm_bo_create") + purego.RegisterLibFunc(&gbmBoDestroy, gbmLib, "gbm_bo_destroy") + purego.RegisterLibFunc(&gbmBoGetFD, gbmLib, "gbm_bo_get_fd") + purego.RegisterLibFunc(&gbmBoGetStride, gbmLib, "gbm_bo_get_stride") + }) +} + +// gbmBO wraps a linear XRGB8888 GBM buffer mapped for writing. +type gbmBO struct { + boH uintptr + devH uintptr + fd int // dma-buf fd + stride uint32 + w, h int + mapped []byte + dev *os.File // keeps the DRM fd alive for the lifetime of the device +} + +// newGBMBuffer creates a linear XRGB8888 GBM buffer on the given DRM device +// path, exports it as a dma-buf and maps it read/write. +func newGBMBuffer(devPath string, w, h int) (*gbmBO, error) { + loadGBM() + if gbmLib == 0 { + return nil, fmt.Errorf("libgbm (%s) unavailable", gbmLibName) + } + f, err := os.OpenFile(devPath, os.O_RDWR, 0) + if err != nil { + return nil, err + } + devH := gbmCreateDevice(int(f.Fd())) + if devH == 0 { + f.Close() + return nil, fmt.Errorf("gbm_create_device failed") + } + bo := gbmBoCreate(devH, uint32(w), uint32(h), gbmFormatXRGB8888, + gbmBoUseRendering|gbmBoUseWrite|gbmBoUseLinear) + if bo == 0 { + gbmDeviceDestroy(devH) + f.Close() + return nil, fmt.Errorf("gbm_bo_create failed") + } + stride := gbmBoGetStride(bo) + dmabuf := gbmBoGetFD(bo) + if dmabuf < 0 { + gbmBoDestroy(bo) + gbmDeviceDestroy(devH) + f.Close() + return nil, fmt.Errorf("gbm_bo_get_fd failed") + } + mapped, err := mmapRW(dmabuf, int(stride)*h) + if err != nil { + closeFD(dmabuf) + gbmBoDestroy(bo) + gbmDeviceDestroy(devH) + f.Close() + return nil, fmt.Errorf("mmap gbm bo: %w", err) + } + return &gbmBO{ + boH: bo, + devH: devH, + fd: dmabuf, + stride: stride, + w: w, + h: h, + mapped: mapped, + dev: f, + }, nil +} + +// Pixels returns the mapped buffer (XRGB8888 layout: B,G,R,X per pixel). +func (b *gbmBO) Pixels() []byte { return b.mapped } + +// Stride returns the row stride in bytes. +func (b *gbmBO) Stride() int { return int(b.stride) } + +func (b *gbmBO) Close() { + if b.mapped != nil { + munmap(b.mapped) + } + if b.fd >= 0 { + closeFD(b.fd) + } + if b.boH != 0 { + gbmBoDestroy(b.boH) + } + if b.devH != 0 { + gbmDeviceDestroy(b.devH) + } + if b.dev != nil { + b.dev.Close() + } +} diff --git a/pipelines/linux/kms.go b/pipelines/linux/kms.go new file mode 100644 index 0000000..0066d9b --- /dev/null +++ b/pipelines/linux/kms.go @@ -0,0 +1,453 @@ +//go:build linux + +// Package linux implements the captured pipelines for Linux using libdrm +// (KMS) and libgbm, loaded at runtime via purego so the build needs no C +// headers or cgo. Capture currently yields BGRA frames over the socket; +// real encode (DMA-BUF -> nvh264enc) stays in the agent's ffmpeg path. +package linux + +import ( + "fmt" + "log" + "os" + "path/filepath" + "sort" + "sync" + "unsafe" + + "github.com/ebitengine/purego" +) + +// --------------------------------------------------------------------------- +// libdrm bindings (loaded lazily via purego). Pointer-returning functions +// return unsafe.Pointer (not uintptr) to keep go vet's unsafeptr check happy +// when we cast them to mirrored C structs. +// --------------------------------------------------------------------------- + +const drmLibName = "libdrm.so.2" + +var ( + drmOnce sync.Once + drmLib uintptr + + drmModeGetResources func(fd int) unsafe.Pointer + drmModeFreeResources func(res unsafe.Pointer) + drmModeGetConnector func(fd int, id uint32) unsafe.Pointer + drmModeFreeConnector func(c unsafe.Pointer) + drmModeGetEncoder func(fd int, id uint32) unsafe.Pointer + drmModeFreeEncoder func(e unsafe.Pointer) + drmModeGetCrtc func(fd int, id uint32) unsafe.Pointer + drmModeFreeCrtc func(c unsafe.Pointer) + drmModeGetFB2 func(fd int, id uint32) unsafe.Pointer + drmModeFreeFB2 func(f unsafe.Pointer) + drmPrimeHandleToFD func(fd int, handle uint32, flags int) int +) + +func loadDRM() { + drmOnce.Do(func() { + h, err := purego.Dlopen(drmLibName, purego.RTLD_NOW|purego.RTLD_GLOBAL) + if err != nil { + drmLib = 0 + return + } + drmLib = h + purego.RegisterLibFunc(&drmModeGetResources, drmLib, "drmModeGetResources") + purego.RegisterLibFunc(&drmModeFreeResources, drmLib, "drmModeFreeResources") + purego.RegisterLibFunc(&drmModeGetConnector, drmLib, "drmModeGetConnector") + purego.RegisterLibFunc(&drmModeFreeConnector, drmLib, "drmModeFreeConnector") + purego.RegisterLibFunc(&drmModeGetEncoder, drmLib, "drmModeGetEncoder") + purego.RegisterLibFunc(&drmModeFreeEncoder, drmLib, "drmModeFreeEncoder") + purego.RegisterLibFunc(&drmModeGetCrtc, drmLib, "drmModeGetCrtc") + purego.RegisterLibFunc(&drmModeFreeCrtc, drmLib, "drmModeFreeCrtc") + purego.RegisterLibFunc(&drmModeGetFB2, drmLib, "drmModeGetFB2") + purego.RegisterLibFunc(&drmModeFreeFB2, drmLib, "drmModeFreeFB2") + purego.RegisterLibFunc(&drmPrimeHandleToFD, drmLib, "drmPrimeHandleToFD") + }) +} + +// --------------------------------------------------------------------------- +// Mirrored C structs (amd64 / LP64 layout). Field order and types match the +// libdrm definitions so reads via unsafe.Pointer are correct. Pointer fields +// are unsafe.Pointer (mirroring C uint32_t*); scalar fields keep their C +// types so the in-memory layout matches exactly. +// --------------------------------------------------------------------------- + +type drmModeRes struct { + CountFBs int32 + FBs unsafe.Pointer + CountCrtcs int32 + Crtcs unsafe.Pointer + CountConnectors int32 + Connectors unsafe.Pointer + CountEncoders int32 + Encoders unsafe.Pointer + MinWidth int32 + MaxWidth int32 + MinHeight int32 + MaxHeight int32 +} + +type drmModeConnector struct { + ConnectorID uint32 + EncoderID uint32 + ConnectorType uint32 + ConnectorTypeID uint32 + Connection int32 + MmWidth uint32 + MmHeight uint32 + Subpixel int32 + CountModes int32 + Modes unsafe.Pointer + CountProps int32 + Props unsafe.Pointer + PropValues unsafe.Pointer + CountEncoders int32 + Encoders unsafe.Pointer +} + +type drmModeEncoder struct { + EncoderID uint32 + EncoderType uint32 + CrtcID uint32 + PossibleCrtcs uint32 + PossibleClones uint32 +} + +type drmModeModeInfo struct { + Clock uint32 + HDisplay uint16 + HSyncStart uint16 + HSyncEnd uint16 + HTotal uint16 + HSkew uint16 + VDisplay uint16 + VSyncStart uint16 + VSyncEnd uint16 + VTotal uint16 + VScan uint16 + VRefresh uint32 + Flags uint32 + Type uint32 + Name [32]byte +} + +// drmModeCrtc as we only read the leading scalar fields; safe to truncate. +type drmModeCrtc struct { + CrtcID uint32 + BufferID uint32 + X uint32 + Y uint32 + Width uint32 + Height uint32 +} + +type drmModeFB2 struct { + FbID uint32 + Width uint32 + Height uint32 + PixelFormat uint32 + Modifier uint64 + Handles [4]uint32 + Pitches [4]uint32 + Offsets [4]uint32 + NumPlanes uint32 + _ uint32 // padding so Flags (u64) is 8-byte aligned + Flags uint64 +} + +const ( + drmModeConnected = 1 + drmModeTypePreferred = 1 << 1 // DRM_MODE_TYPE_PREFERRED + drmFormatModLinear = 0 + drmFormatXRGB8888 = 0x34325258 + drmFormatARGB8888 = 0x34325241 + drmFormatRGBX8888 = 0x38445258 + drmFormatBGRX8888 = 0x38585242 +) + +// linuxDisplay is a resolved display we can re-open for streaming. +type linuxDisplay struct { + ID uint32 + DevPath string + ConnID uint32 + CrtcID uint32 + Width int + Height int + X int + Y int + Refresh float64 +} + +// scanDRMDisplays enumerates connected DRM connectors across /dev/dri/card*. +func scanDRMDisplays() ([]linuxDisplay, error) { + loadDRM() + if drmLib == 0 { + return nil, fmt.Errorf("linux/kms: libdrm (%s) not available", drmLibName) + } + + paths, _ := filepath.Glob("/dev/dri/card*") + sort.Strings(paths) + + var out []linuxDisplay + var permErr error + id := uint32(0) + + for _, p := range paths { + f, err := os.OpenFile(p, os.O_RDWR, 0) + if err != nil { + if os.IsPermission(err) { + if permErr == nil { + permErr = fmt.Errorf("linux/kms: cannot open %s: %v - add the user to the 'video' (or 'render') group", p, err) + } + } + continue + } + fd := int(f.Fd()) + + res := drmModeGetResources(fd) + if res == nil { + f.Close() + continue + } + resPtr := (*drmModeRes)(res) + n := int(resPtr.CountConnectors) + if n > 0 { + connIDs := unsafe.Slice((*uint32)(resPtr.Connectors), n) + for _, cid := range connIDs { + cptr := drmModeGetConnector(fd, cid) + if cptr == nil { + continue + } + conn := (*drmModeConnector)(cptr) + if conn.Connection != drmModeConnected || conn.CountModes <= 0 { + drmModeFreeConnector(cptr) + continue + } + modes := unsafe.Slice((*drmModeModeInfo)(conn.Modes), conn.CountModes) + mi := modes[0] + for _, m := range modes { + if m.Type&drmModeTypePreferred != 0 { + mi = m + break + } + } + + x, y := 0, 0 + crtcID := uint32(0) + if conn.EncoderID != 0 { + eptr := drmModeGetEncoder(fd, conn.EncoderID) + if eptr != nil { + enc := (*drmModeEncoder)(eptr) + crtcID = enc.CrtcID + if crtcID != 0 { + cptr2 := drmModeGetCrtc(fd, crtcID) + if cptr2 != nil { + crtc := (*drmModeCrtc)(cptr2) + x, y = int(crtc.X), int(crtc.Y) + drmModeFreeCrtc(cptr2) + } + } + drmModeFreeEncoder(eptr) + } + } + + out = append(out, linuxDisplay{ + ID: id, + DevPath: p, + ConnID: cid, + CrtcID: crtcID, + Width: int(mi.HDisplay), + Height: int(mi.VDisplay), + X: x, + Y: y, + Refresh: float64(mi.VRefresh), + }) + id++ + drmModeFreeConnector(cptr) + } + } + drmModeFreeResources(res) + f.Close() + } + + if len(out) == 0 { + if permErr != nil { + return nil, permErr + } + return nil, fmt.Errorf("linux/kms: no connected DRM displays found under /dev/dri/card*") + } + return out, nil +} + +// newKMSCapture attempts a real framebuffer readback of the CRTC's current +// scanout buffer via drmModeGetFB2 + prime handle -> dma-buf -> mmap. It +// requires a linearly laid-out framebuffer (most compositors use tiled +// buffers, in which case it returns an error and the caller falls back to a +// synthetic source). This is the stepping stone to the eventual DMA-BUF -> +// nvh264enc encode path. +func newKMSCapture(d *linuxDisplay) (grabber, error) { + loadDRM() + if drmLib == 0 { + return nil, fmt.Errorf("libdrm unavailable") + } + + f, err := os.OpenFile(d.DevPath, os.O_RDWR, 0) + if err != nil { + return nil, err + } + fd := int(f.Fd()) + + if d.CrtcID == 0 { + cptr := drmModeGetConnector(fd, d.ConnID) + if cptr == nil { + f.Close() + return nil, fmt.Errorf("connector %d gone", d.ConnID) + } + conn := (*drmModeConnector)(cptr) + if conn.EncoderID != 0 { + eptr := drmModeGetEncoder(fd, conn.EncoderID) + if eptr != nil { + d.CrtcID = (*drmModeEncoder)(eptr).CrtcID + drmModeFreeEncoder(eptr) + } + } + drmModeFreeConnector(cptr) + } + if d.CrtcID == 0 { + f.Close() + return nil, fmt.Errorf("no CRTC bound to display") + } + + cptr := drmModeGetCrtc(fd, d.CrtcID) + if cptr == nil { + f.Close() + return nil, fmt.Errorf("get CRTC failed") + } + crtc := (*drmModeCrtc)(cptr) + fbID := crtc.BufferID + drmModeFreeCrtc(cptr) + if fbID == 0 { + f.Close() + return nil, fmt.Errorf("CRTC has no framebuffer (no active scanout)") + } + + fb2p := drmModeGetFB2(fd, fbID) + if fb2p == nil { + f.Close() + return nil, fmt.Errorf("drmModeGetFB2 failed") + } + fb2 := (*drmModeFB2)(fb2p) + width, height := int(fb2.Width), int(fb2.Height) + modifier := fb2.Modifier + handle := fb2.Handles[0] + pitch := int(fb2.Pitches[0]) + format := fb2.PixelFormat + drmModeFreeFB2(fb2p) + + if modifier != drmFormatModLinear { + f.Close() + return nil, fmt.Errorf("scanout is tiled (modifier 0x%x); linear readback only", modifier) + } + conv, ok := bgraConverter(format) + if !ok { + f.Close() + return nil, fmt.Errorf("unsupported framebuffer format 0x%x", format) + } + + dmabuf := drmPrimeHandleToFD(fd, handle, 0) + if dmabuf < 0 { + f.Close() + return nil, fmt.Errorf("drmPrimeHandleToFD failed") + } + mapped, err := mmapRO(dmabuf, pitch*height) + if err != nil { + closeFD(dmabuf) + f.Close() + return nil, fmt.Errorf("mmap scanout: %w", err) + } + + log.Printf("linux/kms: capturing %dx%d (pitch %d, fmt 0x%x) via dma-buf readback", + width, height, pitch, format) + + return &kmsGrabber{ + f: f, + dmabuf: dmabuf, + mapped: mapped, + conv: conv, + width: width, + height: height, + pitch: pitch, + }, nil +} + +type kmsGrabber struct { + f *os.File + dmabuf int + mapped []byte + conv func(src []byte, pitch, w, h int) []byte + width int + height int + pitch int +} + +func (g *kmsGrabber) grab() ([]byte, int, int, error) { + return g.conv(g.mapped, g.pitch, g.width, g.height), g.width, g.height, nil +} + +func (g *kmsGrabber) close() error { + munmap(g.mapped) + closeFD(g.dmabuf) + return g.f.Close() +} + +// bgraConverter returns a converter for known 32bpp DRM formats, mapping the +// source memory layout to BGRA. +func bgraConverter(format uint32) (func(src []byte, pitch, w, h int) []byte, bool) { + switch format { + case drmFormatXRGB8888, drmFormatARGB8888: + return convXRGB, true + case drmFormatRGBX8888: + return convRGBX, true + case drmFormatBGRX8888: + return convBGRX, true + } + return nil, false +} + +func convXRGB(src []byte, pitch, w, h int) []byte { + out := make([]byte, w*h*4) + for y := 0; y < h; y++ { + row := src[y*pitch:] + for x := 0; x < w; x++ { + o := (y*w + x) * 4 + out[o], out[o+1], out[o+2], out[o+3] = row[x*4], row[x*4+1], row[x*4+2], 0xFF + } + } + return out +} + +func convRGBX(src []byte, pitch, w, h int) []byte { + out := make([]byte, w*h*4) + for y := 0; y < h; y++ { + row := src[y*pitch:] + for x := 0; x < w; x++ { + o := (y*w + x) * 4 + // memory: R,G,B,X + out[o], out[o+1], out[o+2], out[o+3] = row[x*4+2], row[x*4+1], row[x*4], 0xFF + } + } + return out +} + +func convBGRX(src []byte, pitch, w, h int) []byte { + out := make([]byte, w*h*4) + for y := 0; y < h; y++ { + row := src[y*pitch:] + for x := 0; x < w; x++ { + o := (y*w + x) * 4 + // memory: B,G,R,X + out[o], out[o+1], out[o+2], out[o+3] = row[x*4], row[x*4+1], row[x*4+2], 0xFF + } + } + return out +} diff --git a/pipelines/linux/mmap_linux.go b/pipelines/linux/mmap_linux.go new file mode 100644 index 0000000..8e1d2cc --- /dev/null +++ b/pipelines/linux/mmap_linux.go @@ -0,0 +1,21 @@ +//go:build linux + +package linux + +import "syscall" + +func mmapRO(fd, length int) ([]byte, error) { + return syscall.Mmap(fd, 0, length, syscall.PROT_READ, syscall.MAP_SHARED) +} + +func mmapRW(fd, length int) ([]byte, error) { + return syscall.Mmap(fd, 0, length, syscall.PROT_READ|syscall.PROT_WRITE, syscall.MAP_SHARED) +} + +func munmap(b []byte) error { + return syscall.Munmap(b) +} + +func closeFD(fd int) error { + return syscall.Close(fd) +} diff --git a/pipelines/linux/pipeline.go b/pipelines/linux/pipeline.go index 2b6d54c..67810eb 100644 --- a/pipelines/linux/pipeline.go +++ b/pipelines/linux/pipeline.go @@ -1 +1,218 @@ -package linux \ No newline at end of file +//go:build linux + +// Package linux implements the captured pipelines for Linux. See kms.go and +// gbm.go for the libdrm/libgbm bindings. Spike scope: list real displays and +// stream BGRA (real KMS readback when possible, else a synthetic pattern) over +// the existing unix-socket protocol; encode stays in the agent's ffmpeg path. +package linux + +import ( + "context" + "fmt" + "path/filepath" + "sort" + "sync" + "time" + + "distancedesktop/captured/pipelines" +) + +// grabber produces one BGRA frame per call. +type grabber interface { + grab() (bgra []byte, w, h int, err error) + close() error +} + +// --------------------------------------------------------------------------- +// Pipeline entry point + source selector +// --------------------------------------------------------------------------- + +type kmsPipeline struct{} + +// New returns a Linux capture pipeline for the given source. Supported values +// are "kms" (default) and "x11"; unknown values fall back to kms. +func New(source string) pipelines.Pipeline { + switch source { + case "x11": + return &x11Pipeline{} + case "", "kms": + return &kmsPipeline{} + default: + return &kmsPipeline{} + } +} + +func (p *kmsPipeline) SupportedFormats() []pipelines.FrameFormat { + return []pipelines.FrameFormat{pipelines.FormatBGRA} +} + +func (p *kmsPipeline) ListDisplays(ctx context.Context) ([]pipelines.DisplayMeta, error) { + disps, err := scanDRMDisplays() + if err != nil { + return nil, err + } + out := make([]pipelines.DisplayMeta, len(disps)) + for i, d := range disps { + out[i] = pipelines.DisplayMeta{ + ID: d.ID, + Width: d.Width, + Height: d.Height, + X: d.X, + Y: d.Y, + RefreshRate: d.Refresh, + } + } + return out, nil +} + +func (p *kmsPipeline) StartStream(ctx context.Context, displayID uint32, fps int) (pipelines.FrameStream, error) { + if fps <= 0 { + fps = 60 + } + disps, err := scanDRMDisplays() + if err != nil { + return nil, err + } + var target *linuxDisplay + for i := range disps { + if disps[i].ID == displayID { + target = &disps[i] + break + } + } + if target == nil { + return nil, fmt.Errorf("linux/kms: display %d not found", displayID) + } + + g, err := newKMSCapture(target) + if err != nil { + // Real readback unavailable (tiled fb, no perm, etc.): stream a + // synthetic BGRA pattern so the socket pipeline still works end to + // end. The agent's ffmpeg encoding is unaffected. + fmt.Printf("linux/kms: real capture unavailable (%v); streaming synthetic BGRA\n", err) + g, err = newSynthCapture(target.Width, target.Height) + if err != nil { + return nil, err + } + } + return newFrameStream(g, fps), nil +} + +// --------------------------------------------------------------------------- +// Frame stream +// --------------------------------------------------------------------------- + +type frameStream struct { + ch chan pipelines.EncodedFrame + cancel context.CancelFunc + closeOnce sync.Once + g grabber +} + +func newFrameStream(g grabber, fps int) *frameStream { + ctx, cancel := context.WithCancel(context.Background()) + fs := &frameStream{ch: make(chan pipelines.EncodedFrame, 4), cancel: cancel, g: g} + go fs.run(ctx, fps) + return fs +} + +func (fs *frameStream) run(ctx context.Context, fps int) { + defer fs.Close() + delay := time.Duration(int64(time.Second) / int64(fps)) + ticker := time.NewTicker(delay) + defer ticker.Stop() + for range ticker.C { + bgra, w, h, err := fs.g.grab() + if err != nil { + return + } + select { + case fs.ch <- pipelines.EncodedFrame{Data: bgra, Format: pipelines.FormatBGRA, Width: w, Height: h}: + case <-ctx.Done(): + return + } + } +} + +func (fs *frameStream) Frames() <-chan pipelines.EncodedFrame { + return fs.ch +} + +func (fs *frameStream) Close() error { + fs.closeOnce.Do(func() { + fs.cancel() + _ = fs.g.close() + close(fs.ch) + }) + return nil +} + +// --------------------------------------------------------------------------- +// Synthetic source (used as KMS fallback; optionally GBM-backed) +// --------------------------------------------------------------------------- + +type synthGrabber struct { + width int + height int + frame int + bo *gbmBO +} + +func newSynthCapture(w, h int) (grabber, error) { + paths, _ := filepath.Glob("/dev/dri/card*") + sort.Strings(paths) + for _, p := range paths { + bo, err := newGBMBuffer(p, w, h) + if err == nil { + return &synthGrabber{width: w, height: h, bo: bo}, nil + } + } + // No DRM access: fall back to a pure-Go BGRA buffer. + return &synthGrabber{width: w, height: h}, nil +} + +func (g *synthGrabber) grab() ([]byte, int, int, error) { + g.frame++ + if g.bo != nil { + writePatternXRGB(g.bo.Pixels(), g.bo.Stride(), g.width, g.height, g.frame) + return convXRGB(g.bo.Pixels(), g.bo.Stride(), g.width, g.height), g.width, g.height, nil + } + buf := make([]byte, g.width*g.height*4) + writePatternBGRA(buf, g.width, g.height, g.frame) + return buf, g.width, g.height, nil +} + +func (g *synthGrabber) close() error { + if g.bo != nil { + g.bo.Close() + } + return nil +} + +// writePatternBGRA fills dst (w*h*4, BGRA) with a moving gradient. +func writePatternBGRA(dst []byte, w, h, frame int) { + for y := 0; y < h; y++ { + for x := 0; x < w; x++ { + o := (y*w + x) * 4 + dst[o] = byte((x + frame) & 0xff) + dst[o+1] = byte((y + frame) & 0xff) + dst[o+2] = byte((x + y + frame) & 0xff) + dst[o+3] = 0xFF + } + } +} + +// writePatternXRGB fills dst (stride-pitched XRGB8888: B,G,R,X) with the same +// moving gradient. +func writePatternXRGB(dst []byte, stride, w, h, frame int) { + for y := 0; y < h; y++ { + row := dst[y*stride:] + for x := 0; x < w; x++ { + o := x * 4 + row[o] = byte((x + frame) & 0xff) + row[o+1] = byte((y + frame) & 0xff) + row[o+2] = byte((x + y + frame) & 0xff) + row[o+3] = 0 + } + } +} diff --git a/pipelines/linux/stub_other.go b/pipelines/linux/stub_other.go new file mode 100644 index 0000000..18c99a1 --- /dev/null +++ b/pipelines/linux/stub_other.go @@ -0,0 +1,30 @@ +//go:build !linux + +// Package linux provides the Linux capture pipeline. On non-Linux hosts it +// compiles to a stub so the package remains importable; the real +// implementation lives behind the `linux` build tag. +package linux + +import ( + "context" + "fmt" + + "distancedesktop/captured/pipelines" +) + +type unsupportedPipeline struct{} + +// New always returns the unsupported stub on non-Linux platforms. +func New(source string) pipelines.Pipeline { return &unsupportedPipeline{} } + +func (p *unsupportedPipeline) ListDisplays(ctx context.Context) ([]pipelines.DisplayMeta, error) { + return nil, fmt.Errorf("linux pipeline is only supported on linux") +} + +func (p *unsupportedPipeline) SupportedFormats() []pipelines.FrameFormat { + return nil +} + +func (p *unsupportedPipeline) StartStream(ctx context.Context, displayID uint32, fps int) (pipelines.FrameStream, error) { + return nil, fmt.Errorf("linux pipeline is only supported on linux") +} diff --git a/pipelines/linux/synth_test.go b/pipelines/linux/synth_test.go new file mode 100644 index 0000000..fc9d7e8 --- /dev/null +++ b/pipelines/linux/synth_test.go @@ -0,0 +1,40 @@ +//go:build linux + +package linux + +import "testing" + +func TestSynthCaptureProducesBGRA(t *testing.T) { + const w, h = 64, 48 + g, err := newSynthCapture(w, h) + if err != nil { + t.Fatalf("newSynthCapture: %v", err) + } + defer g.close() + + bgra, gw, gh, err := g.grab() + if err != nil { + t.Fatalf("grab: %v", err) + } + if gw != w || gh != h { + t.Fatalf("size: got %dx%d want %dx%d", gw, gh, w, h) + } + if len(bgra) != w*h*4 { + t.Fatalf("len: got %d want %d", len(bgra), w*h*4) + } + // Alpha must be opaque for the pure-Go BGRA path. + if bgra[3] != 0xFF { + t.Fatalf("alpha: got %d want 255", bgra[3]) + } +} + +func TestBGRAConverterFormats(t *testing.T) { + for _, fmtCode := range []uint32{drmFormatXRGB8888, drmFormatARGB8888, drmFormatRGBX8888, drmFormatBGRX8888} { + if _, ok := bgraConverter(fmtCode); !ok { + t.Fatalf("bgraConverter(0x%x) unsupported", fmtCode) + } + } + if _, ok := bgraConverter(0x12345678); ok { + t.Fatalf("bgraConverter accepted unknown format") + } +} diff --git a/pipelines/linux/x11.go b/pipelines/linux/x11.go new file mode 100644 index 0000000..231f3fd --- /dev/null +++ b/pipelines/linux/x11.go @@ -0,0 +1,134 @@ +//go:build linux + +package linux + +import ( + "context" + "fmt" + "os" + "sync" + + "github.com/ebitengine/purego" + + "distancedesktop/captured/pipelines" +) + +// --------------------------------------------------------------------------- +// libX11 bindings (used by the x11 source selector). +// --------------------------------------------------------------------------- + +const x11LibName = "libX11.so.6" + +var ( + x11Once sync.Once + x11Lib uintptr + + xOpenDisplay func(name string) uintptr + xCloseDisplay func(dpy uintptr) int + xDefaultScreen func(dpy uintptr) int + xDisplayWidth func(dpy uintptr, screen int) int + xDisplayHeight func(dpy uintptr, screen int) int +) + +func loadX11() { + x11Once.Do(func() { + h, err := purego.Dlopen(x11LibName, purego.RTLD_NOW|purego.RTLD_GLOBAL) + if err != nil { + x11Lib = 0 + return + } + x11Lib = h + purego.RegisterLibFunc(&xOpenDisplay, x11Lib, "XOpenDisplay") + purego.RegisterLibFunc(&xCloseDisplay, x11Lib, "XCloseDisplay") + purego.RegisterLibFunc(&xDefaultScreen, x11Lib, "XDefaultScreen") + purego.RegisterLibFunc(&xDisplayWidth, x11Lib, "XDisplayWidth") + purego.RegisterLibFunc(&xDisplayHeight, x11Lib, "XDisplayHeight") + }) +} + +type x11Display struct { + ID int + Width int + Height int + X int + Y int + Refresh float64 +} + +// scanX11Displays reports the default X screen size. Full XRandr multi-output +// enumeration and XShm capture are follow-ups; this is enough for the source +// selector to list the primary screen. +func scanX11Displays() ([]x11Display, error) { + loadX11() + if x11Lib == 0 { + return nil, fmt.Errorf("linux/x11: libX11 (%s) not available", x11LibName) + } + disp := os.Getenv("DISPLAY") + if disp == "" { + return nil, fmt.Errorf("linux/x11: $DISPLAY is not set") + } + dpy := xOpenDisplay(disp) + if dpy == 0 { + return nil, fmt.Errorf("linux/x11: cannot open X display %q", disp) + } + defer xCloseDisplay(dpy) + screen := xDefaultScreen(dpy) + w := xDisplayWidth(dpy, screen) + h := xDisplayHeight(dpy, screen) + if w == 0 || h == 0 { + return nil, fmt.Errorf("linux/x11: invalid screen size from X") + } + return []x11Display{{ID: 0, Width: w, Height: h, X: 0, Y: 0, Refresh: 60}}, nil +} + +type x11Pipeline struct{} + +func (p *x11Pipeline) ListDisplays(ctx context.Context) ([]pipelines.DisplayMeta, error) { + disps, err := scanX11Displays() + if err != nil { + return nil, err + } + out := make([]pipelines.DisplayMeta, len(disps)) + for i, d := range disps { + out[i] = pipelines.DisplayMeta{ + ID: uint32(d.ID), + Width: d.Width, + Height: d.Height, + X: d.X, + Y: d.Y, + RefreshRate: d.Refresh, + } + } + return out, nil +} + +func (p *x11Pipeline) SupportedFormats() []pipelines.FrameFormat { + return []pipelines.FrameFormat{pipelines.FormatBGRA} +} + +func (p *x11Pipeline) StartStream(ctx context.Context, displayID uint32, fps int) (pipelines.FrameStream, error) { + if fps <= 0 { + fps = 60 + } + disps, err := scanX11Displays() + if err != nil { + return nil, err + } + var target *x11Display + for i := range disps { + if uint32(disps[i].ID) == displayID { + target = &disps[i] + break + } + } + if target == nil { + return nil, fmt.Errorf("linux/x11: display %d not found", displayID) + } + // X11 pixel capture (XShmGetImage) is a follow-up; stream a synthetic + // BGRA pattern through the same socket protocol for now. + g, err := newSynthCapture(target.Width, target.Height) + if err != nil { + return nil, err + } + return newFrameStream(g, fps), nil +} From 0a2ccbe18c5cdaec29446d1a5890ab14d4454e9b Mon Sep 17 00:00:00 2001 From: "coderabbitai[bot]" <136622811+coderabbitai[bot]@users.noreply.github.com> Date: Wed, 26 Aug 2026 16:42:59 +0000 Subject: [PATCH 2/5] fix: apply CodeRabbit auto-fixes Fixed 4 file(s) based on 7 unresolved review comments. Co-authored-by: CodeRabbit --- pipelines/linux/gbm.go | 2 +- pipelines/linux/kms.go | 16 ++++++++-------- pipelines/linux/pipeline.go | 32 ++++++++++++++++++++++++-------- pipelines/linux/x11.go | 9 ++------- 4 files changed, 35 insertions(+), 24 deletions(-) diff --git a/pipelines/linux/gbm.go b/pipelines/linux/gbm.go index 4cf6b8e..3816178 100644 --- a/pipelines/linux/gbm.go +++ b/pipelines/linux/gbm.go @@ -82,7 +82,7 @@ func newGBMBuffer(devPath string, w, h int) (*gbmBO, error) { return nil, fmt.Errorf("gbm_create_device failed") } bo := gbmBoCreate(devH, uint32(w), uint32(h), gbmFormatXRGB8888, - gbmBoUseRendering|gbmBoUseWrite|gbmBoUseLinear) + gbmBoUseRendering|gbmBoUseLinear) if bo == 0 { gbmDeviceDestroy(devH) f.Close() diff --git a/pipelines/linux/kms.go b/pipelines/linux/kms.go index 0066d9b..a6a41e2 100644 --- a/pipelines/linux/kms.go +++ b/pipelines/linux/kms.go @@ -40,7 +40,7 @@ var ( drmModeFreeCrtc func(c unsafe.Pointer) drmModeGetFB2 func(fd int, id uint32) unsafe.Pointer drmModeFreeFB2 func(f unsafe.Pointer) - drmPrimeHandleToFD func(fd int, handle uint32, flags int) int + drmPrimeHandleToFD func(fd int, handle uint32, flags uint32, prime_fd *int32) int ) func loadDRM() { @@ -147,12 +147,10 @@ type drmModeFB2 struct { Height uint32 PixelFormat uint32 Modifier uint64 + Flags uint32 Handles [4]uint32 Pitches [4]uint32 Offsets [4]uint32 - NumPlanes uint32 - _ uint32 // padding so Flags (u64) is 8-byte aligned - Flags uint64 } const ( @@ -161,8 +159,8 @@ const ( drmFormatModLinear = 0 drmFormatXRGB8888 = 0x34325258 drmFormatARGB8888 = 0x34325241 - drmFormatRGBX8888 = 0x38445258 - drmFormatBGRX8888 = 0x38585242 + drmFormatRGBX8888 = 0x34325852 + drmFormatBGRX8888 = 0x34325842 ) // linuxDisplay is a resolved display we can re-open for streaming. @@ -354,11 +352,13 @@ func newKMSCapture(d *linuxDisplay) (grabber, error) { return nil, fmt.Errorf("unsupported framebuffer format 0x%x", format) } - dmabuf := drmPrimeHandleToFD(fd, handle, 0) - if dmabuf < 0 { + var primeFD int32 + ret := drmPrimeHandleToFD(fd, handle, 0x80000, &primeFD) + if ret != 0 { f.Close() return nil, fmt.Errorf("drmPrimeHandleToFD failed") } + dmabuf := int(primeFD) mapped, err := mmapRO(dmabuf, pitch*height) if err != nil { closeFD(dmabuf) diff --git a/pipelines/linux/pipeline.go b/pipelines/linux/pipeline.go index 67810eb..7fb81b3 100644 --- a/pipelines/linux/pipeline.go +++ b/pipelines/linux/pipeline.go @@ -95,7 +95,7 @@ func (p *kmsPipeline) StartStream(ctx context.Context, displayID uint32, fps int return nil, err } } - return newFrameStream(g, fps), nil + return newFrameStream(ctx, g, fps), nil } // --------------------------------------------------------------------------- @@ -107,21 +107,38 @@ type frameStream struct { cancel context.CancelFunc closeOnce sync.Once g grabber + done chan struct{} } -func newFrameStream(g grabber, fps int) *frameStream { - ctx, cancel := context.WithCancel(context.Background()) - fs := &frameStream{ch: make(chan pipelines.EncodedFrame, 4), cancel: cancel, g: g} - go fs.run(ctx, fps) +func newFrameStream(ctx context.Context, g grabber, fps int) *frameStream { + ctx, cancel := context.WithCancel(ctx) + fs := &frameStream{ + ch: make(chan pipelines.EncodedFrame, 4), + cancel: cancel, + g: g, + done: make(chan struct{}), + } + go func() { + fs.run(ctx, fps) + close(fs.done) + }() return fs } func (fs *frameStream) run(ctx context.Context, fps int) { - defer fs.Close() + defer func() { + _ = fs.g.close() + close(fs.ch) + }() delay := time.Duration(int64(time.Second) / int64(fps)) ticker := time.NewTicker(delay) defer ticker.Stop() for range ticker.C { + select { + case <-ctx.Done(): + return + default: + } bgra, w, h, err := fs.g.grab() if err != nil { return @@ -141,8 +158,7 @@ func (fs *frameStream) Frames() <-chan pipelines.EncodedFrame { func (fs *frameStream) Close() error { fs.closeOnce.Do(func() { fs.cancel() - _ = fs.g.close() - close(fs.ch) + <-fs.done }) return nil } diff --git a/pipelines/linux/x11.go b/pipelines/linux/x11.go index 231f3fd..0ac2c66 100644 --- a/pipelines/linux/x11.go +++ b/pipelines/linux/x11.go @@ -124,11 +124,6 @@ func (p *x11Pipeline) StartStream(ctx context.Context, displayID uint32, fps int if target == nil { return nil, fmt.Errorf("linux/x11: display %d not found", displayID) } - // X11 pixel capture (XShmGetImage) is a follow-up; stream a synthetic - // BGRA pattern through the same socket protocol for now. - g, err := newSynthCapture(target.Width, target.Height) - if err != nil { - return nil, err - } - return newFrameStream(g, fps), nil + // X11 pixel capture (XShmGetImage) is not yet implemented. + return nil, fmt.Errorf("linux/x11: X11 pixel readback not yet implemented") } From 8af39bad4844615db4f6bb141ec3bc787433ac39 Mon Sep 17 00:00:00 2001 From: spacedouut Date: Wed, 26 Aug 2026 14:28:19 -0400 Subject: [PATCH 3/5] ci: build captured (go vet/build) on push/PR MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit No release — just sanity check for KMS/GBM pipeline. --- .github/workflows/ci.yml | 15 +++++++++++++++ 1 file changed, 15 insertions(+) create mode 100644 .github/workflows/ci.yml diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..1b73751 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,15 @@ +name: ci +on: + push: + pull_request: +jobs: + build: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-go@v5 + with: + go-version: "1.26" + cache: true + - run: go vet ./... + - run: go build ./... From 28668290a81699fa9201cdcf4f31d0948fd058a8 Mon Sep 17 00:00:00 2001 From: Hermet Date: Tue, 1 Sep 2026 10:23:53 -0400 Subject: [PATCH 4/5] feat: pipewire capture source via Mutter ScreenCast Adds --source pipewire alongside kms and x11. The KMS path needs a CRTC bound to a connected display and read access to /dev/dri/card*, so it cannot capture a Wayland session on a machine with no monitor attached -- it falls back to a synthetic pattern. This source asks the compositor instead, which owns the framebuffer regardless. - pipelines/linux/pipewire.go: lists monitors from org.gnome.Mutter.DisplayConfig.GetCurrentState, then creates a ScreenCast session, RecordMonitor's the chosen connector, and waits for PipeWireStreamAdded to learn the PipeWire node id. Frames are read as packed BGRA off a gst-launch-1.0 pipeline (pipewiresrc -> videorate -> videoconvert -> fdsink), matching the format the agent's ffmpeg path already expects. - pipelines/linux/pipeline.go: newPulledFrameStream, for sources that block until a frame is available. The existing ticker-driven stream is right for KMS readback but would duplicate or drop PipeWire frames, which arrive at the compositor's cadence. org.gnome.Mutter.ScreenCast is used rather than the compositor-agnostic org.freedesktop.portal.Desktop.ScreenCast because the portal requires interactive consent through a dialog that a daemon cannot satisfy. This does mean GNOME only for now, and captured must run as the desktop session user. New runtime dependency for this source: gstreamer1.0-tools (for gst-launch-1.0) plus gstreamer1.0-pipewire. Build dependency: github.com/godbus/dbus/v5. Verified on Ubuntu with GNOME 50 on Wayland: list-displays reports the real 1360x768 logical monitor where KMS saw only disconnected connectors, and the captured frames are the actual desktop rather than the synthetic fallback. go vet, go build and the existing pipelines/linux tests pass. --- README.md | 24 +++ go.mod | 3 +- go.sum | 4 + main.go | 2 +- pipelines/linux/pipeline.go | 41 ++++- pipelines/linux/pipewire.go | 346 ++++++++++++++++++++++++++++++++++++ 6 files changed, 417 insertions(+), 3 deletions(-) create mode 100644 pipelines/linux/pipewire.go diff --git a/README.md b/README.md index c8664e7..31ebfdf 100644 --- a/README.md +++ b/README.md @@ -8,3 +8,27 @@ captured is an unbelivably (but predictibly!) dumb capture agent for Windows, Li ## How does it work? captured listens over two sockets; captured-media and captured-control. As on the tin, Media sends the actual video (or audio) content being recorded, while control is a control plane for it. + +## Linux capture sources + +Pick one with `--source`: + +| Source | Notes | +|--------|-------| +| `kms` (default) | libdrm/libgbm readback via purego. Needs read access to `/dev/dri/card*` and a CRTC bound to a connected display; falls back to a synthetic BGRA pattern when readback is unavailable. | +| `pipewire` | Screen capture through the compositor. Works on Wayland with no CRTC bound and needs no DRM permissions. | +| `x11` | Lists the default X screen; pixel readback is not implemented yet. | + +### pipewire + +```sh +captured --source pipewire +``` + +Requires `gst-launch-1.0` (`gstreamer1.0-tools`) and `gstreamer1.0-pipewire`. + +Must run **as the desktop session user**, from inside that session, because it +talks to `org.gnome.Mutter.ScreenCast` on the session bus. GNOME/Mutter only for +now: the freedesktop portal (`org.freedesktop.portal.Desktop.ScreenCast`) would +be compositor-agnostic but requires interactive consent through a dialog, which a +daemon cannot satisfy. diff --git a/go.mod b/go.mod index 2a0ff78..8546a06 100644 --- a/go.mod +++ b/go.mod @@ -3,6 +3,7 @@ module distancedesktop/captured go 1.26.3 require ( - github.com/LocalKinAI/sckit-go v0.3.1 // indirect + github.com/LocalKinAI/sckit-go v0.3.1 github.com/ebitengine/purego v0.8.0 + github.com/godbus/dbus/v5 v5.1.0 ) diff --git a/go.sum b/go.sum index ef63661..47319f0 100644 --- a/go.sum +++ b/go.sum @@ -2,3 +2,7 @@ github.com/LocalKinAI/sckit-go v0.3.1 h1:fa0RVcDjalRUJU/SSEDPBD1ePNkPk9mJLaCa5zQ github.com/LocalKinAI/sckit-go v0.3.1/go.mod h1:geXsjewJufWyR8lQrsPYUGWgNZrkhgG76nTyRqmOCU8= github.com/ebitengine/purego v0.8.0 h1:JbqvnEzRvPpxhCJzJJ2y0RbiZ8nyjccVUrSM3q+GvvE= github.com/ebitengine/purego v0.8.0/go.mod h1:iIjxzd6CiRiOG0UyXP+V1+jWqUXVjPKLAI0mRfJZTmQ= +github.com/godbus/dbus/v5 v5.1.0 h1:4KLkAxT3aOY8Li4FRJe/KvhoNFFxo0m6fNuFUO8QJUk= +github.com/godbus/dbus/v5 v5.1.0/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA= +golang.org/x/image v0.39.0 h1:skVYidAEVKgn8lZ602XO75asgXBgLj9G/FE3RbuPFww= +golang.org/x/image v0.39.0/go.mod h1:sIbmppfU+xFLPIG0FoVUTvyBMmgng1/XAMhQ2ft0hpA= diff --git a/main.go b/main.go index 16d7a5c..9fc93b2 100644 --- a/main.go +++ b/main.go @@ -22,7 +22,7 @@ var pipeline pipelines.Pipeline func main() { listen := flag.String("listen", "", "TCP address for remote control (e.g. :9090)") - source := flag.String("source", "kms", "capture source on linux: kms|x11") + source := flag.String("source", "kms", "capture source on linux: kms|pipewire|x11") flag.Parse() switch runtime.GOOS { diff --git a/pipelines/linux/pipeline.go b/pipelines/linux/pipeline.go index 7fb81b3..0893149 100644 --- a/pipelines/linux/pipeline.go +++ b/pipelines/linux/pipeline.go @@ -30,11 +30,13 @@ type grabber interface { type kmsPipeline struct{} // New returns a Linux capture pipeline for the given source. Supported values -// are "kms" (default) and "x11"; unknown values fall back to kms. +// are "kms" (default), "pipewire" and "x11"; unknown values fall back to kms. func New(source string) pipelines.Pipeline { switch source { case "x11": return &x11Pipeline{} + case "pipewire", "pw": + return &pipewirePipeline{} case "", "kms": return &kmsPipeline{} default: @@ -155,6 +157,43 @@ func (fs *frameStream) Frames() <-chan pipelines.EncodedFrame { return fs.ch } +// newPulledFrameStream drives a grabber whose grab() blocks until the source +// produces a frame (PipeWire), rather than sampling on a ticker. The source's +// own cadence sets the frame rate. +func newPulledFrameStream(ctx context.Context, g grabber) *frameStream { + ctx, cancel := context.WithCancel(ctx) + fs := &frameStream{ + ch: make(chan pipelines.EncodedFrame, 4), + cancel: cancel, + g: g, + done: make(chan struct{}), + } + go func() { + defer func() { + _ = fs.g.close() + close(fs.ch) + close(fs.done) + }() + for { + select { + case <-ctx.Done(): + return + default: + } + bgra, w, h, err := fs.g.grab() + if err != nil { + return + } + select { + case fs.ch <- pipelines.EncodedFrame{Data: bgra, Format: pipelines.FormatBGRA, Width: w, Height: h}: + case <-ctx.Done(): + return + } + } + }() + return fs +} + func (fs *frameStream) Close() error { fs.closeOnce.Do(func() { fs.cancel() diff --git a/pipelines/linux/pipewire.go b/pipelines/linux/pipewire.go new file mode 100644 index 0000000..f1e3aa2 --- /dev/null +++ b/pipelines/linux/pipewire.go @@ -0,0 +1,346 @@ +//go:build linux + +package linux + +import ( + "context" + "fmt" + "io" + "os" + "os/exec" + "strconv" + "strings" + "sync" + "time" + + "github.com/godbus/dbus/v5" + + "distancedesktop/captured/pipelines" +) + +// PipeWire capture via GNOME Mutter's private ScreenCast interface. +// +// Unlike the KMS path, this works on a Wayland compositor with no CRTC bound to +// a physical connector, and it needs no DRM permissions -- the compositor owns +// the scanout and hands us a PipeWire node. +// +// org.gnome.Mutter.ScreenCast is used rather than +// org.freedesktop.portal.Desktop.ScreenCast because the portal requires +// interactive user consent through a dialog, which a headless daemon cannot +// satisfy. Mutter's interface is available to any client in the user's session +// bus, so `captured` must run as the session user. +// +// Frames are pulled out of PipeWire with gst-launch-1.0 (pipewiresrc -> +// videoconvert -> BGRA -> fdsink) and read as raw frames off stdout. Shelling +// out to GStreamer avoids binding libpipewire's buffer negotiation by hand; the +// pw node id is the only thing the D-Bus round trip is needed for. + +const ( + mutterScreenCastName = "org.gnome.Mutter.ScreenCast" + mutterScreenCastPath = "/org/gnome/Mutter/ScreenCast" + mutterDisplayName = "org.gnome.Mutter.DisplayConfig" + mutterDisplayPath = "/org/gnome/Mutter/DisplayConfig" + + // cursorModeEmbedded draws the cursor into the frames, which is what a + // remote viewer wants (there is no local pointer to composite). + cursorModeEmbedded = 1 +) + +// screenCastStartTimeout bounds the wait for Mutter's PipeWireStreamAdded. +const screenCastStartTimeout = 10 * time.Second + +type pipewirePipeline struct{} + +type pwMonitor struct { + Connector string + Width int + Height int + Refresh float64 +} + +// scanPipeWireMonitors lists logical monitors from Mutter's DisplayConfig. +// These are compositor-side monitors, so a virtual display with no physical +// connector attached still appears. +func scanPipeWireMonitors() ([]pwMonitor, error) { + conn, err := dbus.SessionBus() + if err != nil { + return nil, fmt.Errorf("linux/pipewire: session bus: %w (captured must run in the desktop user's session)", err) + } + obj := conn.Object(mutterDisplayName, dbus.ObjectPath(mutterDisplayPath)) + + var serial uint32 + // GetCurrentState signature: + // u a((ssss)a(siiddada{sv})a{sv}) a((iiduba(ssss)a{sv})) a{sv} + // The monitor spec is a *struct* of four strings (connector, vendor, product, + // serial), not a string array -- decoding it as []string fails with + // "cannot convert a value of []interface {} into []string". + type monitorSpec struct { + Connector string + Vendor string + Product string + Serial string + } + var monitors []struct { + Spec monitorSpec + Modes []struct { + ID string + Width int32 + Height int32 + Refresh float64 + Scale float64 + Scales []float64 + Props map[string]dbus.Variant + } + Props map[string]dbus.Variant + } + var logical []struct { + X, Y int32 + Scale float64 + Transform uint32 + Primary bool + Monitors []monitorSpec + Props map[string]dbus.Variant + } + var props map[string]dbus.Variant + + call := obj.Call(mutterDisplayName+".GetCurrentState", 0) + if call.Err != nil { + return nil, fmt.Errorf("linux/pipewire: GetCurrentState: %w", call.Err) + } + if err := call.Store(&serial, &monitors, &logical, &props); err != nil { + return nil, fmt.Errorf("linux/pipewire: decode monitor state: %w", err) + } + + out := make([]pwMonitor, 0, len(monitors)) + for _, m := range monitors { + if m.Spec.Connector == "" { + continue + } + mon := pwMonitor{Connector: m.Spec.Connector, Refresh: 60} + for _, mode := range m.Modes { + if v, ok := mode.Props["is-current"]; ok { + if cur, ok := v.Value().(bool); ok && cur { + mon.Width = int(mode.Width) + mon.Height = int(mode.Height) + mon.Refresh = mode.Refresh + break + } + } + } + if mon.Width == 0 && len(m.Modes) > 0 { + mon.Width = int(m.Modes[0].Width) + mon.Height = int(m.Modes[0].Height) + mon.Refresh = m.Modes[0].Refresh + } + out = append(out, mon) + } + if len(out) == 0 { + return nil, fmt.Errorf("linux/pipewire: no monitors reported by Mutter") + } + return out, nil +} + +func (p *pipewirePipeline) SupportedFormats() []pipelines.FrameFormat { + return []pipelines.FrameFormat{pipelines.FormatBGRA} +} + +func (p *pipewirePipeline) ListDisplays(ctx context.Context) ([]pipelines.DisplayMeta, error) { + mons, err := scanPipeWireMonitors() + if err != nil { + return nil, err + } + out := make([]pipelines.DisplayMeta, len(mons)) + for i, m := range mons { + out[i] = pipelines.DisplayMeta{ + ID: uint32(i), + Width: m.Width, + Height: m.Height, + RefreshRate: m.Refresh, + } + } + return out, nil +} + +// pwSession holds the Mutter screencast session for one stream. +type pwSession struct { + conn *dbus.Conn + sessionPath dbus.ObjectPath + nodeID uint32 +} + +// startMutterScreenCast creates a session, records the given connector, starts +// it, and waits for the PipeWireStreamAdded signal carrying the node id. +// +// The wait is bounded: if Mutter accepts the session but never emits the signal +// (a compositor restart mid-handshake, for instance), this must not block +// StartStream forever, since ctx is the daemon's long-lived context. +func startMutterScreenCast(ctx context.Context, connector string) (*pwSession, error) { + ctx, cancel := context.WithTimeout(ctx, screenCastStartTimeout) + defer cancel() + + conn, err := dbus.SessionBus() + if err != nil { + return nil, fmt.Errorf("linux/pipewire: session bus: %w", err) + } + sc := conn.Object(mutterScreenCastName, dbus.ObjectPath(mutterScreenCastPath)) + + var sessionPath dbus.ObjectPath + if err := sc.Call(mutterScreenCastName+".CreateSession", 0, map[string]dbus.Variant{}).Store(&sessionPath); err != nil { + return nil, fmt.Errorf("linux/pipewire: CreateSession: %w", err) + } + sess := conn.Object(mutterScreenCastName, sessionPath) + + var streamPath dbus.ObjectPath + opts := map[string]dbus.Variant{ + "cursor-mode": dbus.MakeVariant(uint32(cursorModeEmbedded)), + } + if err := sess.Call(mutterScreenCastName+".Session.RecordMonitor", 0, connector, opts).Store(&streamPath); err != nil { + _ = sess.Call(mutterScreenCastName+".Session.Stop", 0).Err + return nil, fmt.Errorf("linux/pipewire: RecordMonitor(%s): %w", connector, err) + } + + // Subscribe before Start so the signal cannot be missed. + if err := conn.AddMatchSignal( + dbus.WithMatchObjectPath(streamPath), + dbus.WithMatchInterface(mutterScreenCastName+".Stream"), + dbus.WithMatchMember("PipeWireStreamAdded"), + ); err != nil { + return nil, fmt.Errorf("linux/pipewire: AddMatchSignal: %w", err) + } + sigCh := make(chan *dbus.Signal, 4) + conn.Signal(sigCh) + + if err := sess.Call(mutterScreenCastName+".Session.Start", 0).Err; err != nil { + return nil, fmt.Errorf("linux/pipewire: Session.Start: %w", err) + } + + for { + select { + case sig := <-sigCh: + if sig.Path != streamPath || !strings.HasSuffix(sig.Name, "PipeWireStreamAdded") { + continue + } + if len(sig.Body) == 0 { + continue + } + id, ok := sig.Body[0].(uint32) + if !ok { + continue + } + return &pwSession{conn: conn, sessionPath: sessionPath, nodeID: id}, nil + case <-ctx.Done(): + _ = sess.Call(mutterScreenCastName+".Session.Stop", 0).Err + return nil, fmt.Errorf("linux/pipewire: timed out waiting for PipeWireStreamAdded") + } + } +} + +func (s *pwSession) stop() { + if s == nil || s.conn == nil { + return + } + _ = s.conn.Object(mutterScreenCastName, s.sessionPath). + Call(mutterScreenCastName+".Session.Stop", 0).Err +} + +// pipewireGrabber reads raw BGRA frames from a gst-launch-1.0 pipeline attached +// to the screencast's PipeWire node. +type pipewireGrabber struct { + sess *pwSession + cmd *exec.Cmd + stdout io.ReadCloser + width int + height int + frame []byte + once sync.Once +} + +func newPipeWireCapture(ctx context.Context, connector string, w, h, fps int) (grabber, error) { + if _, err := exec.LookPath("gst-launch-1.0"); err != nil { + return nil, fmt.Errorf("linux/pipewire: gst-launch-1.0 not found (install gstreamer1.0-tools + gstreamer1.0-pipewire): %w", err) + } + sess, err := startMutterScreenCast(ctx, connector) + if err != nil { + return nil, err + } + + // Force BGRA at a fixed size so frame length is predictable; the agent + // expects tightly packed BGRA with no stride padding. + caps := fmt.Sprintf("video/x-raw,format=BGRA,width=%d,height=%d", w, h) + args := []string{ + "-q", + "pipewiresrc", "path=" + strconv.FormatUint(uint64(sess.nodeID), 10), + "do-timestamp=true", "keepalive-time=1000", + "!", "videorate", + "!", fmt.Sprintf("video/x-raw,framerate=%d/1", fps), + "!", "videoscale", + "!", "videoconvert", "chroma-mode=none", "dither=none", + "!", caps, + "!", "fdsink", "fd=1", "sync=false", + } + cmd := exec.Command("gst-launch-1.0", args...) + stdout, err := cmd.StdoutPipe() + if err != nil { + sess.stop() + return nil, fmt.Errorf("linux/pipewire: stdout pipe: %w", err) + } + // Surface GStreamer's diagnostics: a caps negotiation failure otherwise looks + // like an unexplained EOF on the first frame read. + cmd.Stderr = os.Stderr + if err := cmd.Start(); err != nil { + sess.stop() + return nil, fmt.Errorf("linux/pipewire: start gst-launch-1.0: %w", err) + } + + return &pipewireGrabber{ + sess: sess, + cmd: cmd, + stdout: stdout, + width: w, + height: h, + frame: make([]byte, w*h*4), + }, nil +} + +func (g *pipewireGrabber) grab() ([]byte, int, int, error) { + if _, err := io.ReadFull(g.stdout, g.frame); err != nil { + return nil, 0, 0, fmt.Errorf("linux/pipewire: read frame: %w", err) + } + out := make([]byte, len(g.frame)) + copy(out, g.frame) + return out, g.width, g.height, nil +} + +func (g *pipewireGrabber) close() error { + g.once.Do(func() { + if g.cmd != nil && g.cmd.Process != nil { + _ = g.cmd.Process.Kill() + _, _ = g.cmd.Process.Wait() + } + _ = g.stdout.Close() + g.sess.stop() + }) + return nil +} + +func (p *pipewirePipeline) StartStream(ctx context.Context, displayID uint32, fps int) (pipelines.FrameStream, error) { + if fps <= 0 { + fps = 60 + } + mons, err := scanPipeWireMonitors() + if err != nil { + return nil, err + } + if int(displayID) >= len(mons) { + return nil, fmt.Errorf("linux/pipewire: display %d not found (%d available)", displayID, len(mons)) + } + m := mons[displayID] + + g, err := newPipeWireCapture(ctx, m.Connector, m.Width, m.Height, fps) + if err != nil { + return nil, err + } + // PipeWire delivers frames at its own cadence; the frame stream must not + // re-tick, so it is driven by reads returning. + return newPulledFrameStream(ctx, g), nil +} From a147f19a8286d62fb26410b7154e0c9a55588f37 Mon Sep 17 00:00:00 2001 From: Hermet Date: Tue, 1 Sep 2026 11:01:26 -0400 Subject: [PATCH 5/5] fix: address CodeRabbit review on the pipewire source Four issues, all in shutdown and error paths that the happy-path testing never reached. - sigCh could yield a nil signal. godbus closes the channel passed to conn.Signal when the bus connection drops, so the receive returned sig == nil and sig.Path panicked. The receive now checks ok and reports a capture error. - The 10s deadline did not apply to the D-Bus calls. CreateSession, RecordMonitor and Session.Start used synchronous Call, which ignores the context, so each could block past the deadline before the timeout select was ever reached. They now use CallWithContext, and the match rule uses AddMatchSignalContext. - Failed startup leaked sessions and match rules. AddMatchSignal and Session.Start failures returned without stopping the session, and the successful path never removed sigCh or its match rule even though pwSession does not retain them. Session teardown is now shared by every error path, and the signal registration is released on return. - Closing a pulled stream could deadlock. frameStream.Close waits for the producer goroutine, but that goroutine blocks in grab() -> io.ReadFull until the compositor sends another frame, and the grabber's close (which kills gst-launch-1.0) only ran after the producer exited. On an idle desktop this hung forever and leaked the child process. Grabbers may now implement interrupt(), and newPulledFrameStream runs a watchdog that calls it when the context is done, unblocking the read. Adds pipelines/linux/pulled_stream_test.go covering the deadlock, parent cancellation, grabbers without interrupt(), and repeated Close. The first test fails against the previous commit ('Close blocked while grab() was waiting for a frame') and passes now. Verified with go vet, go build, go test -race ./..., and on hardware: three start/stop cycles against a live GNOME session leave zero gst-launch-1.0 processes behind. --- pipelines/linux/pipeline.go | 20 ++++ pipelines/linux/pipewire.go | 66 +++++++++--- pipelines/linux/pulled_stream_test.go | 148 ++++++++++++++++++++++++++ 3 files changed, 218 insertions(+), 16 deletions(-) create mode 100644 pipelines/linux/pulled_stream_test.go diff --git a/pipelines/linux/pipeline.go b/pipelines/linux/pipeline.go index 0893149..53aa82a 100644 --- a/pipelines/linux/pipeline.go +++ b/pipelines/linux/pipeline.go @@ -157,9 +157,23 @@ func (fs *frameStream) Frames() <-chan pipelines.EncodedFrame { return fs.ch } +// interruptibleGrabber is a grabber whose blocking grab() can be unblocked from +// another goroutine, so a stream can be torn down while no frames are arriving. +type interruptibleGrabber interface { + grabber + interrupt() +} + // newPulledFrameStream drives a grabber whose grab() blocks until the source // produces a frame (PipeWire), rather than sampling on a ticker. The source's // own cadence sets the frame rate. +// +// Because grab() blocks, cancellation alone is not enough: Close waits on the +// producer goroutine, which would be parked inside grab() until the compositor +// happened to send another frame. A watchdog goroutine therefore interrupts the +// grabber as soon as ctx is done, which makes the pending read fail and lets the +// producer exit. Without it, closing an idle stream deadlocks and leaks the +// gst-launch-1.0 child. func newPulledFrameStream(ctx context.Context, g grabber) *frameStream { ctx, cancel := context.WithCancel(ctx) fs := &frameStream{ @@ -168,6 +182,12 @@ func newPulledFrameStream(ctx context.Context, g grabber) *frameStream { g: g, done: make(chan struct{}), } + if ig, ok := g.(interruptibleGrabber); ok { + go func() { + <-ctx.Done() + ig.interrupt() + }() + } go func() { defer func() { _ = fs.g.close() diff --git a/pipelines/linux/pipewire.go b/pipelines/linux/pipewire.go index f1e3aa2..67d10d4 100644 --- a/pipelines/linux/pipewire.go +++ b/pipelines/linux/pipewire.go @@ -171,9 +171,12 @@ type pwSession struct { // startMutterScreenCast creates a session, records the given connector, starts // it, and waits for the PipeWireStreamAdded signal carrying the node id. // -// The wait is bounded: if Mutter accepts the session but never emits the signal -// (a compositor restart mid-handshake, for instance), this must not block -// StartStream forever, since ctx is the daemon's long-lived context. +// Every step is bounded by screenCastStartTimeout: if Mutter accepts the session +// but never emits the signal (a compositor restart mid-handshake, for instance), +// this must not block StartStream forever, since ctx is the daemon's long-lived +// context. The D-Bus calls use CallWithContext for the same reason -- a +// synchronous Call would ignore the deadline entirely and could block past it +// before the timeout select is ever reached. func startMutterScreenCast(ctx context.Context, connector string) (*pwSession, error) { ctx, cancel := context.WithTimeout(ctx, screenCastStartTimeout) defer cancel() @@ -185,38 +188,59 @@ func startMutterScreenCast(ctx context.Context, connector string) (*pwSession, e sc := conn.Object(mutterScreenCastName, dbus.ObjectPath(mutterScreenCastPath)) var sessionPath dbus.ObjectPath - if err := sc.Call(mutterScreenCastName+".CreateSession", 0, map[string]dbus.Variant{}).Store(&sessionPath); err != nil { + if err := sc.CallWithContext(ctx, mutterScreenCastName+".CreateSession", 0, map[string]dbus.Variant{}).Store(&sessionPath); err != nil { return nil, fmt.Errorf("linux/pipewire: CreateSession: %w", err) } sess := conn.Object(mutterScreenCastName, sessionPath) + // stopSession tears down a session created above. It deliberately does not + // use ctx: on the timeout path ctx is already expired, and the session must + // still be released or Mutter keeps recording. + stopSession := func() { + _ = sess.Call(mutterScreenCastName+".Session.Stop", 0).Err + } + var streamPath dbus.ObjectPath opts := map[string]dbus.Variant{ "cursor-mode": dbus.MakeVariant(uint32(cursorModeEmbedded)), } - if err := sess.Call(mutterScreenCastName+".Session.RecordMonitor", 0, connector, opts).Store(&streamPath); err != nil { - _ = sess.Call(mutterScreenCastName+".Session.Stop", 0).Err + if err := sess.CallWithContext(ctx, mutterScreenCastName+".Session.RecordMonitor", 0, connector, opts).Store(&streamPath); err != nil { + stopSession() return nil, fmt.Errorf("linux/pipewire: RecordMonitor(%s): %w", connector, err) } // Subscribe before Start so the signal cannot be missed. - if err := conn.AddMatchSignal( + matchOpts := []dbus.MatchOption{ dbus.WithMatchObjectPath(streamPath), - dbus.WithMatchInterface(mutterScreenCastName+".Stream"), + dbus.WithMatchInterface(mutterScreenCastName + ".Stream"), dbus.WithMatchMember("PipeWireStreamAdded"), - ); err != nil { + } + if err := conn.AddMatchSignalContext(ctx, matchOpts...); err != nil { + stopSession() return nil, fmt.Errorf("linux/pipewire: AddMatchSignal: %w", err) } sigCh := make(chan *dbus.Signal, 4) conn.Signal(sigCh) - - if err := sess.Call(mutterScreenCastName+".Session.Start", 0).Err; err != nil { + // The match rule and channel are only needed for this handshake; pwSession + // does not retain them, so they must not outlive this function. + defer func() { + conn.RemoveSignal(sigCh) + _ = conn.RemoveMatchSignal(matchOpts...) + }() + + if err := sess.CallWithContext(ctx, mutterScreenCastName+".Session.Start", 0).Err; err != nil { + stopSession() return nil, fmt.Errorf("linux/pipewire: Session.Start: %w", err) } for { select { - case sig := <-sigCh: + case sig, ok := <-sigCh: + // A closed connection closes sigCh, yielding a nil signal. + if !ok || sig == nil { + stopSession() + return nil, fmt.Errorf("linux/pipewire: session bus closed while waiting for PipeWireStreamAdded") + } if sig.Path != streamPath || !strings.HasSuffix(sig.Name, "PipeWireStreamAdded") { continue } @@ -229,8 +253,8 @@ func startMutterScreenCast(ctx context.Context, connector string) (*pwSession, e } return &pwSession{conn: conn, sessionPath: sessionPath, nodeID: id}, nil case <-ctx.Done(): - _ = sess.Call(mutterScreenCastName+".Session.Stop", 0).Err - return nil, fmt.Errorf("linux/pipewire: timed out waiting for PipeWireStreamAdded") + stopSession() + return nil, fmt.Errorf("linux/pipewire: timed out waiting for PipeWireStreamAdded: %w", ctx.Err()) } } } @@ -255,6 +279,17 @@ type pipewireGrabber struct { once sync.Once } +// interrupt unblocks a grab() that is waiting on a partial frame, so a stream +// can be closed while the compositor is idle and producing nothing. Killing the +// child and closing the pipe both make the pending read return an error. +// Safe to call more than once, and safe to call concurrently with grab(). +func (g *pipewireGrabber) interrupt() { + if g.cmd != nil && g.cmd.Process != nil { + _ = g.cmd.Process.Kill() + } + _ = g.stdout.Close() +} + func newPipeWireCapture(ctx context.Context, connector string, w, h, fps int) (grabber, error) { if _, err := exec.LookPath("gst-launch-1.0"); err != nil { return nil, fmt.Errorf("linux/pipewire: gst-launch-1.0 not found (install gstreamer1.0-tools + gstreamer1.0-pipewire): %w", err) @@ -313,11 +348,10 @@ func (g *pipewireGrabber) grab() ([]byte, int, int, error) { func (g *pipewireGrabber) close() error { g.once.Do(func() { + g.interrupt() if g.cmd != nil && g.cmd.Process != nil { - _ = g.cmd.Process.Kill() _, _ = g.cmd.Process.Wait() } - _ = g.stdout.Close() g.sess.stop() }) return nil diff --git a/pipelines/linux/pulled_stream_test.go b/pipelines/linux/pulled_stream_test.go new file mode 100644 index 0000000..98194a0 --- /dev/null +++ b/pipelines/linux/pulled_stream_test.go @@ -0,0 +1,148 @@ +//go:build linux + +package linux + +import ( + "context" + "errors" + "sync" + "testing" + "time" + + "distancedesktop/captured/pipelines" +) + +// blockingGrabber blocks in grab() until interrupt() is called, modelling a +// PipeWire source on an idle compositor that is producing no frames. +type blockingGrabber struct { + release chan struct{} + closed chan struct{} + closeOnce sync.Once + intOnce sync.Once +} + +func newBlockingGrabber() *blockingGrabber { + return &blockingGrabber{ + release: make(chan struct{}), + closed: make(chan struct{}), + } +} + +func (g *blockingGrabber) grab() ([]byte, int, int, error) { + <-g.release + return nil, 0, 0, errors.New("interrupted") +} + +func (g *blockingGrabber) interrupt() { + g.intOnce.Do(func() { close(g.release) }) +} + +func (g *blockingGrabber) close() error { + g.closeOnce.Do(func() { close(g.closed) }) + return nil +} + +// A pulled stream must be closable while its grabber is blocked waiting for a +// frame. Without the interrupt watchdog, Close blocks on the producer goroutine +// which is itself parked in grab(), and the grabber is never closed. +func TestPulledFrameStreamCloseInterruptsBlockedGrab(t *testing.T) { + g := newBlockingGrabber() + fs := newPulledFrameStream(context.Background(), g) + + done := make(chan struct{}) + go func() { + _ = fs.Close() + close(done) + }() + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("Close blocked while grab() was waiting for a frame") + } + + select { + case <-g.closed: + case <-time.After(time.Second): + t.Fatal("grabber was not closed") + } + + if _, ok := <-fs.Frames(); ok { + t.Fatal("frame channel should be closed and empty") + } +} + +// Cancelling the parent context must tear the stream down for the same reason. +func TestPulledFrameStreamParentCancelInterruptsBlockedGrab(t *testing.T) { + g := newBlockingGrabber() + ctx, cancel := context.WithCancel(context.Background()) + fs := newPulledFrameStream(ctx, g) + cancel() + + select { + case <-g.closed: + case <-time.After(2 * time.Second): + t.Fatal("parent cancellation did not interrupt the blocked grab") + } + _ = fs.Close() +} + +// countingGrabber yields a fixed number of frames, then reports EOF. +type countingGrabber struct { + remaining int + w, h int + closed bool +} + +func (g *countingGrabber) grab() ([]byte, int, int, error) { + if g.remaining == 0 { + return nil, 0, 0, errors.New("eof") + } + g.remaining-- + return make([]byte, g.w*g.h*4), g.w, g.h, nil +} + +func (g *countingGrabber) close() error { + g.closed = true + return nil +} + +// A grabber with no interrupt() must still work: the watchdog is optional. +func TestPulledFrameStreamPlainGrabber(t *testing.T) { + g := &countingGrabber{remaining: 3, w: 4, h: 2} + fs := newPulledFrameStream(context.Background(), g) + + var got int + for f := range fs.Frames() { + if f.Format != pipelines.FormatBGRA { + t.Fatalf("format = %q, want %q", f.Format, pipelines.FormatBGRA) + } + if f.Width != 4 || f.Height != 2 { + t.Fatalf("frame = %dx%d, want 4x2", f.Width, f.Height) + } + if len(f.Data) != 4*2*4 { + t.Fatalf("len(Data) = %d, want %d", len(f.Data), 4*2*4) + } + got++ + } + if got != 3 { + t.Fatalf("received %d frames, want 3", got) + } + if !g.closed { + t.Fatal("grabber was not closed after the producer exited") + } + _ = fs.Close() +} + +// Close must be idempotent; FrameStream.Close is reachable from both the owner +// and the teardown path in the agent. +func TestPulledFrameStreamCloseTwice(t *testing.T) { + g := newBlockingGrabber() + fs := newPulledFrameStream(context.Background(), g) + if err := fs.Close(); err != nil { + t.Fatalf("first Close: %v", err) + } + if err := fs.Close(); err != nil { + t.Fatalf("second Close: %v", err) + } +}