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 ./... 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 2fc6566..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/ebitengine/purego v0.8.0 // 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 e99eebc..9fc93b2 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|pipewire|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..3816178 --- /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|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..a6a41e2 --- /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 uint32, prime_fd *int32) 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 + Flags uint32 + Handles [4]uint32 + Pitches [4]uint32 + Offsets [4]uint32 +} + +const ( + drmModeConnected = 1 + drmModeTypePreferred = 1 << 1 // DRM_MODE_TYPE_PREFERRED + drmFormatModLinear = 0 + drmFormatXRGB8888 = 0x34325258 + drmFormatARGB8888 = 0x34325241 + drmFormatRGBX8888 = 0x34325852 + drmFormatBGRX8888 = 0x34325842 +) + +// 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) + } + + 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) + 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..53aa82a 100644 --- a/pipelines/linux/pipeline.go +++ b/pipelines/linux/pipeline.go @@ -1 +1,293 @@ -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), "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: + 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(ctx, g, fps), nil +} + +// --------------------------------------------------------------------------- +// Frame stream +// --------------------------------------------------------------------------- + +type frameStream struct { + ch chan pipelines.EncodedFrame + cancel context.CancelFunc + closeOnce sync.Once + g grabber + done chan struct{} +} + +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 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 + } + 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 +} + +// 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{ + ch: make(chan pipelines.EncodedFrame, 4), + cancel: cancel, + 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() + 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() + <-fs.done + }) + 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/pipewire.go b/pipelines/linux/pipewire.go new file mode 100644 index 0000000..67d10d4 --- /dev/null +++ b/pipelines/linux/pipewire.go @@ -0,0 +1,380 @@ +//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. +// +// 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() + + 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.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.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. + matchOpts := []dbus.MatchOption{ + dbus.WithMatchObjectPath(streamPath), + dbus.WithMatchInterface(mutterScreenCastName + ".Stream"), + dbus.WithMatchMember("PipeWireStreamAdded"), + } + 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) + // 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, 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 + } + 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(): + stopSession() + return nil, fmt.Errorf("linux/pipewire: timed out waiting for PipeWireStreamAdded: %w", ctx.Err()) + } + } +} + +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 +} + +// 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) + } + 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() { + g.interrupt() + if g.cmd != nil && g.cmd.Process != nil { + _, _ = g.cmd.Process.Wait() + } + 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 +} 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) + } +} 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..0ac2c66 --- /dev/null +++ b/pipelines/linux/x11.go @@ -0,0 +1,129 @@ +//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 not yet implemented. + return nil, fmt.Errorf("linux/x11: X11 pixel readback not yet implemented") +}