Wayland compositor (wlroots)
git clone https://git.lucas.co/cce-compositor.git
src/server/stream_server.rs (5.3K)
1 // Window-stream socket server.
2 //
3 // The compositor-native answer to remote window viewing (cce-remote):
4 // a client connects to /tmp/cce-stream-{WAYLAND_DISPLAY}.sock, sends one
5 // subscription line — `window <query>` where query is an id, app_id, or
6 // the literal `focused` — and then receives raw frames of that window:
7 //
8 // frame <width> <height> <len>\n
9 // <len bytes of tightly packed RGBA>
10 //
11 // Frames are DAMAGE-DRIVEN at the source: `Window::stream_dirty` is set by
12 // the window's commit listener, and the window manager's stream timer (a
13 // wlroots event-loop timer, ~30 fps while subscribers exist, idle cadence
14 // otherwise) captures only dirty windows via the screenshot readback path —
15 // so an idle window costs nothing, occluded/off-viewport windows stream
16 // fine, and `focused` re-resolves every tick so the stream follows focus.
17 //
18 // Threading mirrors status_server: an accept thread owns the listener; each
19 // subscriber gets a writer thread fed through a BOUNDED channel — the main
20 // thread only try_send()s, so a stalled client skips frames (backpressure =
21 // frame dropping) and can never block the compositor.
22
23 use std::io::{BufRead, Write};
24 use std::os::unix::net::{UnixListener, UnixStream};
25 use std::sync::mpsc;
26 use std::sync::{Arc, Mutex};
27 use std::time::Instant;
28
29 /// One captured frame, shared across all subscribers of the same window.
30 pub struct Frame {
31 pub width: i32,
32 pub height: i32,
33 pub rgba: Vec<u8>,
34 }
35
36 /// A subscriber as the MAIN THREAD sees it: the query to resolve each tick
37 /// and the bounded sender feeding its writer thread.
38 pub struct Sub {
39 pub query: String,
40 pub tx: mpsc::SyncSender<Arc<Frame>>,
41 /// Force a frame regardless of damage (first frame after subscribing,
42 /// and the periodic keepalive that lets writers detect dead clients).
43 pub needs_frame: bool,
44 pub last_sent: Instant,
45 }
46
47 /// Shared subscriber registry: the accept thread pushes, the window
48 /// manager's stream timer drains dead entries and feeds frames.
49 #[derive(Clone)]
50 pub struct StreamHub {
51 pub subs: Arc<Mutex<Vec<Sub>>>,
52 /// Bumped by the accept thread after pushing a subscriber; the window
53 /// manager has it as an event source and arms its frame tick from it.
54 pub wake: Arc<std::os::fd::OwnedFd>,
55 }
56
57 pub fn get_stream_socket_path(display_socket: Option<&str>) -> String {
58 match display_socket {
59 Some(display) => format!("/tmp/cce-stream-{}.sock", display),
60 None => "/tmp/cce-stream.sock".to_string(),
61 }
62 }
63
64 /// Spawn the accept thread; returns the hub for the main loop's timer.
65 pub fn spawn_stream_server(display_socket: Option<String>) -> StreamHub {
66 let hub = StreamHub {
67 subs: Arc::new(Mutex::new(Vec::new())),
68 wake: crate::ipc_server::new_wake_fd().expect("failed to create stream wake eventfd"),
69 };
70 let accept_hub = hub.clone();
71 std::thread::Builder::new()
72 .name("cce-stream-server".into())
73 .spawn(move || accept_loop(accept_hub, display_socket))
74 .expect("failed to spawn stream server thread");
75 hub
76 }
77
78 fn accept_loop(hub: StreamHub, display_socket: Option<String>) {
79 let socket_path = get_stream_socket_path(display_socket.as_deref());
80 let _ = std::fs::remove_file(&socket_path);
81 let listener = match UnixListener::bind(&socket_path) {
82 Ok(l) => l,
83 Err(e) => {
84 log::error!("[stream] failed to bind {}: {}", socket_path, e);
85 return;
86 }
87 };
88 log::info!("[stream] listening on {}", socket_path);
89
90 for stream in listener.incoming() {
91 let Ok(stream) = stream else { continue };
92 // Subscription line, with a timeout so a silent connect can't park.
93 let _ = stream.set_read_timeout(Some(std::time::Duration::from_secs(5)));
94 let mut line = String::new();
95 {
96 let mut reader = std::io::BufReader::new(&stream);
97 if reader.read_line(&mut line).is_err() {
98 continue;
99 }
100 }
101 let Some(query) = line.trim().strip_prefix("window ").map(str::trim) else {
102 continue;
103 };
104 if query.is_empty() || query.len() > 128 {
105 continue;
106 }
107 let _ = stream.set_read_timeout(None);
108
109 // Bounded at 2: the main thread never blocks; a slow client just
110 // gets the freshest frame that fits.
111 let (tx, rx) = mpsc::sync_channel::<Arc<Frame>>(2);
112 let query = query.to_string();
113 log::info!("[stream] subscriber for window '{}'", query);
114 std::thread::Builder::new()
115 .name("cce-stream-writer".into())
116 .spawn(move || writer_loop(stream, rx))
117 .ok();
118 if let Ok(mut subs) = hub.subs.lock() {
119 subs.push(Sub { query, tx, needs_frame: true, last_sent: Instant::now() });
120 }
121 crate::ipc_server::wake_fd(&hub.wake);
122 }
123 }
124
125 /// Blocking writes on a dedicated thread per subscriber. Exits on write
126 /// error; the dropped receiver surfaces as Disconnected on the main
127 /// thread's next try_send, which prunes the Sub.
128 fn writer_loop(mut stream: UnixStream, rx: mpsc::Receiver<Arc<Frame>>) {
129 while let Ok(frame) = rx.recv() {
130 let header = format!("frame {} {} {}\n", frame.width, frame.height, frame.rgba.len());
131 if stream.write_all(header.as_bytes()).is_err()
132 || stream.write_all(&frame.rgba).is_err()
133 {
134 return;
135 }
136 }
137 }