Wayland compositor (wlroots)
git clone https://git.lucas.co/cce-compositor.git
feat: window-stream socket — damage-driven frame export per window
/tmp/cce-stream-{WAYLAND_DISPLAY}.sock: a client subscribes with
'window <id|app_id|focused>' and receives raw RGBA frames
('frame <w> <h> <len>' + bytes). Frames are damage-driven at the
source: Window.stream_dirty is set by the commit listener and consumed
by a new wl_event_loop stream timer (33ms while subscribers exist,
idle cadence otherwise) that captures via the screenshot readback path
(capture_window split into capture_window_rgba + PNG wrapper), so idle
windows cost nothing and off-viewport/occluded windows stream fine.
'focused' re-resolves every tick, so a stream follows focus.
Threading mirrors status_server: an accept thread owns the listener;
each subscriber gets a writer thread fed through a bounded(2) channel —
the main thread only try_send()s, so a stalled client skips frames and
can never block the compositor. A 15s keepalive frame lets writers
detect dead clients. Consumer: cce-remote's live window view.
Co-Authored-By: Claude Fable 5 <[email protected]>
src/lib.rs | 2 +
src/server/run_server.rs | 3 +
src/server/screenshot.rs | 15 +++--
src/server/stream_server.rs | 130 +++++++++++++++++++++++++++++++++++++++++++
src/server/window.rs | 6 ++
src/server/window_manager.rs | 88 ++++++++++++++++++++++++++++-
6 files changed, 239 insertions(+), 5 deletions(-)
diff --git a/src/lib.rs b/src/lib.rs
index b622d38..61ea145 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -33,6 +33,8 @@ pub mod ipc_server;
pub mod screenshot;
#[path = "server/status_server.rs"]
pub mod status_server;
+#[path = "server/stream_server.rs"]
+pub mod stream_server;
#[path = "server/scene_node_data.rs"]
pub mod scene_node_data;
#[path = "server/output.rs"]
diff --git a/src/server/run_server.rs b/src/server/run_server.rs
index d54eb15..5dc8fbc 100644
--- a/src/server/run_server.rs
+++ b/src/server/run_server.rs
@@ -180,6 +180,9 @@ pub fn run_server() {
let status_sender = crate::status_server::spawn_status_server(Some(socket_str.clone()));
server.wm.status_sender = Some(status_sender);
+ let stream_hub = crate::stream_server::spawn_stream_server(Some(socket_str.clone()));
+ server.wm.stream_hub = Some(stream_hub);
+
let started = unsafe { ffi::wlr_backend_start(server.backend) };
if !started {
diff --git a/src/server/screenshot.rs b/src/server/screenshot.rs
index 57a05f9..8ab8425 100644
--- a/src/server/screenshot.rs
+++ b/src/server/screenshot.rs
@@ -161,6 +161,16 @@ pub unsafe fn capture_state_buffer(
/// Works for windows outside the visible viewport (their last committed
/// buffers persist), but needs the client to have committed at least once.
pub unsafe fn capture_window(window: *mut crate::window::Window, path: PathBuf) -> Result<String, String> {
+ let (canvas, bw, bh) = capture_window_rgba(window)?;
+ let reply = path.display().to_string();
+ spawn_encode(canvas, bw as u32, bh as u32, path);
+ Ok(reply)
+}
+
+/// The readback+composite half of [`capture_window`], PNG-free: returns the
+/// tightly packed RGBA canvas and its pixel dimensions. Also the frame source
+/// for the window-stream server.
+pub unsafe fn capture_window_rgba(window: *mut crate::window::Window) -> Result<(Vec<u8>, i32, i32), String> {
let root = (*window).root_surface();
if root.is_null() {
return Err("window has no surface".to_string());
@@ -212,10 +222,7 @@ pub unsafe fn capture_window(window: *mut crate::window::Window, path: PathBuf)
if composited == 0 {
return Err("no readable surface content".to_string());
}
-
- let reply = path.display().to_string();
- spawn_encode(canvas, bw as u32, bh as u32, path);
- Ok(reply)
+ Ok((canvas, bw, bh))
}
/// Copy `src` (sw×sh RGBA) into `dst` (dw×dh RGBA) at (dx, dy), clipped.
diff --git a/src/server/stream_server.rs b/src/server/stream_server.rs
new file mode 100644
index 0000000..e168a65
--- /dev/null
+++ b/src/server/stream_server.rs
@@ -0,0 +1,130 @@
+// Window-stream socket server.
+//
+// The compositor-native answer to remote window viewing (cce-remote):
+// a client connects to /tmp/cce-stream-{WAYLAND_DISPLAY}.sock, sends one
+// subscription line — `window <query>` where query is an id, app_id, or
+// the literal `focused` — and then receives raw frames of that window:
+//
+// frame <width> <height> <len>\n
+// <len bytes of tightly packed RGBA>
+//
+// Frames are DAMAGE-DRIVEN at the source: `Window::stream_dirty` is set by
+// the window's commit listener, and the window manager's stream timer (a
+// wlroots event-loop timer, ~30 fps while subscribers exist, idle cadence
+// otherwise) captures only dirty windows via the screenshot readback path —
+// so an idle window costs nothing, occluded/off-viewport windows stream
+// fine, and `focused` re-resolves every tick so the stream follows focus.
+//
+// Threading mirrors status_server: an accept thread owns the listener; each
+// subscriber gets a writer thread fed through a BOUNDED channel — the main
+// thread only try_send()s, so a stalled client skips frames (backpressure =
+// frame dropping) and can never block the compositor.
+
+use std::io::{BufRead, Write};
+use std::os::unix::net::{UnixListener, UnixStream};
+use std::sync::mpsc;
+use std::sync::{Arc, Mutex};
+use std::time::Instant;
+
+/// One captured frame, shared across all subscribers of the same window.
+pub struct Frame {
+ pub width: i32,
+ pub height: i32,
+ pub rgba: Vec<u8>,
+}
+
+/// A subscriber as the MAIN THREAD sees it: the query to resolve each tick
+/// and the bounded sender feeding its writer thread.
+pub struct Sub {
+ pub query: String,
+ pub tx: mpsc::SyncSender<Arc<Frame>>,
+ /// Force a frame regardless of damage (first frame after subscribing,
+ /// and the periodic keepalive that lets writers detect dead clients).
+ pub needs_frame: bool,
+ pub last_sent: Instant,
+}
+
+/// Shared subscriber registry: the accept thread pushes, the window
+/// manager's stream timer drains dead entries and feeds frames.
+#[derive(Clone)]
+pub struct StreamHub {
+ pub subs: Arc<Mutex<Vec<Sub>>>,
+}
+
+pub fn get_stream_socket_path(display_socket: Option<&str>) -> String {
+ match display_socket {
+ Some(display) => format!("/tmp/cce-stream-{}.sock", display),
+ None => "/tmp/cce-stream.sock".to_string(),
+ }
+}
+
+/// Spawn the accept thread; returns the hub for the main loop's timer.
+pub fn spawn_stream_server(display_socket: Option<String>) -> StreamHub {
+ let hub = StreamHub { subs: Arc::new(Mutex::new(Vec::new())) };
+ let accept_hub = hub.clone();
+ std::thread::Builder::new()
+ .name("cce-stream-server".into())
+ .spawn(move || accept_loop(accept_hub, display_socket))
+ .expect("failed to spawn stream server thread");
+ hub
+}
+
+fn accept_loop(hub: StreamHub, display_socket: Option<String>) {
+ let socket_path = get_stream_socket_path(display_socket.as_deref());
+ let _ = std::fs::remove_file(&socket_path);
+ let listener = match UnixListener::bind(&socket_path) {
+ Ok(l) => l,
+ Err(e) => {
+ log::error!("[stream] failed to bind {}: {}", socket_path, e);
+ return;
+ }
+ };
+ log::info!("[stream] listening on {}", socket_path);
+
+ for stream in listener.incoming() {
+ let Ok(stream) = stream else { continue };
+ // Subscription line, with a timeout so a silent connect can't park.
+ let _ = stream.set_read_timeout(Some(std::time::Duration::from_secs(5)));
+ let mut line = String::new();
+ {
+ let mut reader = std::io::BufReader::new(&stream);
+ if reader.read_line(&mut line).is_err() {
+ continue;
+ }
+ }
+ let Some(query) = line.trim().strip_prefix("window ").map(str::trim) else {
+ continue;
+ };
+ if query.is_empty() || query.len() > 128 {
+ continue;
+ }
+ let _ = stream.set_read_timeout(None);
+
+ // Bounded at 2: the main thread never blocks; a slow client just
+ // gets the freshest frame that fits.
+ let (tx, rx) = mpsc::sync_channel::<Arc<Frame>>(2);
+ let query = query.to_string();
+ log::info!("[stream] subscriber for window '{}'", query);
+ std::thread::Builder::new()
+ .name("cce-stream-writer".into())
+ .spawn(move || writer_loop(stream, rx))
+ .ok();
+ if let Ok(mut subs) = hub.subs.lock() {
+ subs.push(Sub { query, tx, needs_frame: true, last_sent: Instant::now() });
+ }
+ }
+}
+
+/// Blocking writes on a dedicated thread per subscriber. Exits on write
+/// error; the dropped receiver surfaces as Disconnected on the main
+/// thread's next try_send, which prunes the Sub.
+fn writer_loop(mut stream: UnixStream, rx: mpsc::Receiver<Arc<Frame>>) {
+ while let Ok(frame) = rx.recv() {
+ let header = format!("frame {} {} {}\n", frame.width, frame.height, frame.rgba.len());
+ if stream.write_all(header.as_bytes()).is_err()
+ || stream.write_all(&frame.rgba).is_err()
+ {
+ return;
+ }
+ }
+}
diff --git a/src/server/window.rs b/src/server/window.rs
index bf44382..834b0db 100644
--- a/src/server/window.rs
+++ b/src/server/window.rs
@@ -343,6 +343,10 @@ pub struct Window {
/// render-start snapshot (`rendering_sent`), which still holds the previous
/// size and would snap the border back. Cleared once consumed.
pub self_resized: bool,
+ /// Set by the commit listener, cleared by the window-manager stream
+ /// timer after a capture: the damage gate for `stream_server` frames.
+ /// Starts true so a fresh subscriber gets an immediate first frame.
+ pub stream_dirty: bool,
pub commit: ffi::wl_listener,
pub was_fullscreen: bool,
pub saved_width: i32,
@@ -549,6 +553,7 @@ impl Window {
resize_start_h: 0,
resize_edges: None,
self_resized: false,
+ stream_dirty: true,
commit: std::mem::zeroed(),
was_fullscreen: false,
saved_width: 0,
@@ -3673,6 +3678,7 @@ unsafe fn wl_listener_remove_safe(listener: *mut ffi::wl_listener) {
unsafe extern "C" fn handle_window_commit(listener: *mut ffi::wl_listener, _data: *mut std::ffi::c_void) {
let window = crate::container_of!(listener, Window, commit);
+ (*window).stream_dirty = true;
let was_status = (*window).is_status_bar();
(*window).render_finish();
if was_status {
diff --git a/src/server/window_manager.rs b/src/server/window_manager.rs
index 2e298fc..c9ff49f 100644
--- a/src/server/window_manager.rs
+++ b/src/server/window_manager.rs
@@ -89,6 +89,10 @@ pub struct WindowManager {
pub gesture_binds: Vec<crate::config::GestureBind>,
pub ipc_rx: Option<std::sync::mpsc::Receiver<crate::ipc_server::IpcRequest>>,
pub ipc_timer: *mut ffi::wl_event_source,
+ /// Window-stream subscribers (cce-remote's live view); frames are
+ /// produced by `handle_stream_timer` when a subscribed window is dirty.
+ pub stream_hub: Option<crate::stream_server::StreamHub>,
+ pub stream_timer: *mut ffi::wl_event_source,
/// A full-output/region screenshot parked for the next composited frame
/// (`ccectl screenshot`); consumed by `Output::render_and_commit`.
pub pending_screenshot: Option<crate::screenshot::PendingScreenshot>,
@@ -201,6 +205,8 @@ impl WindowManager {
self.gesture_binds = Vec::new();
self.ipc_rx = None;
self.ipc_timer = std::ptr::null_mut();
+ self.stream_hub = None;
+ self.stream_timer = std::ptr::null_mut();
self.startup = Vec::new();
self.startup_pids = Vec::new();
self.status_sender = None;
@@ -248,6 +254,16 @@ impl WindowManager {
}
self.border_fade_running = false;
+ self.stream_timer = ffi::wl_event_loop_add_timer(event_loop, Some(handle_stream_timer), self as *mut WindowManager as *mut _);
+ if self.stream_timer.is_null() {
+ ffi::wl_event_source_remove(self.timeout);
+ ffi::wl_event_source_remove(self.ipc_timer);
+ ffi::wl_event_source_remove(self.clean_exit_timer);
+ ffi::wl_event_source_remove(self.border_fade_timer);
+ return Err("Failed to create stream timer event source");
+ }
+ ffi::wl_event_source_timer_update(self.stream_timer, 200);
+
// Default until the config is parsed (which happens after this init).
self.center_on_spawn = true;
@@ -3442,7 +3458,77 @@ unsafe extern "C" fn handle_ipc_timer(data: *mut std::ffi::c_void) -> std::os::r
if !(*wm).ipc_timer.is_null() {
ffi::wl_event_source_timer_update((*wm).ipc_timer, 10);
}
-
+
+ 0
+}
+
+/// Window-stream tick: resolve each subscription (`focused` re-resolves per
+/// tick, so streams follow focus), capture windows that are dirty (commit
+/// listener set `stream_dirty`) or due a keepalive, and try_send frames to
+/// the writer threads — never blocking the compositor (a full channel means
+/// the client is slow and simply skips the frame). Fast cadence only while
+/// subscribers exist.
+unsafe extern "C" fn handle_stream_timer(data: *mut std::ffi::c_void) -> std::os::raw::c_int {
+ let wm = &mut *(data as *mut WindowManager);
+ let idle_rearm = |wm: &WindowManager, ms: i32| {
+ if !wm.stream_timer.is_null() {
+ ffi::wl_event_source_timer_update(wm.stream_timer, ms);
+ }
+ };
+ let Some(hub) = wm.stream_hub.clone() else {
+ idle_rearm(wm, 500);
+ return 0;
+ };
+ let Ok(mut subs) = hub.subs.lock() else {
+ idle_rearm(wm, 500);
+ return 0;
+ };
+ if subs.is_empty() {
+ idle_rearm(wm, 200);
+ return 0;
+ }
+
+ // Each unique window is captured at most once per tick, shared by Arc.
+ let mut captured: Vec<(*mut Window, std::sync::Arc<crate::stream_server::Frame>)> = Vec::new();
+ let mut dead: Vec<usize> = Vec::new();
+ for i in 0..subs.len() {
+ let win = if subs[i].query == "focused" {
+ wm.focused_window()
+ } else {
+ wm.find_window_by_query(&subs[i].query)
+ };
+ if win.is_null() {
+ continue;
+ }
+ let keepalive = subs[i].last_sent.elapsed().as_secs() >= 15;
+ if !(*win).stream_dirty && !subs[i].needs_frame && !keepalive {
+ continue;
+ }
+ let frame = match captured.iter().find(|(w, _)| *w == win) {
+ Some((_, f)) => f.clone(),
+ None => match crate::screenshot::capture_window_rgba(win) {
+ Ok((rgba, w, h)) => {
+ let f = std::sync::Arc::new(crate::stream_server::Frame { width: w, height: h, rgba });
+ captured.push((win, f.clone()));
+ (*win).stream_dirty = false;
+ f
+ }
+ Err(_) => continue,
+ },
+ };
+ match subs[i].tx.try_send(frame) {
+ Ok(()) => {
+ subs[i].needs_frame = false;
+ subs[i].last_sent = std::time::Instant::now();
+ }
+ Err(std::sync::mpsc::TrySendError::Full(_)) => {} // slow client: drop frame
+ Err(std::sync::mpsc::TrySendError::Disconnected(_)) => dead.push(i),
+ }
+ }
+ for i in dead.into_iter().rev() {
+ subs.remove(i);
+ }
+ idle_rearm(wm, 33);
0
}