remote trackpad and keyboard server
git clone https://git.lucas.co/cce-remote.git
feat: ack-clocked latest-wins live view; adaptive resolution; TCP_NODELAY
The live view degraded into seconds of lag after a short time, and the cause
was structural, not tuning: nothing on this side ever dropped a frame. The
compositor's stream server drops for slow subscribers, but cce-remote read
every frame in order, encoded every frame, and write_all'd every frame into
the TCP socket — whose auto-tuned send buffer holds 10-30 frames. The moment
wifi throughput dipped below the frame rate, that buffer became a standing
queue; MJPEG over TCP cannot skip ahead, so every frame shown was queue-depth
old and the lag never drained.
Delivery is now sender-side flow control (src/stream.rs), the shape VNC/RDP
use:
- A Slot holds only the NEWEST frame; the producer (winstream → screencopy →
grim, unchanged preference) overwrites it. Overwriting is the drop point —
stale frames cease to exist before they cost encode or network.
- The page rides /wstream, a dedicated WebSocket with the same auth gate: send
one frame, wait for the page's ack, send the newest. At most ONE frame in
flight ever, so a degraded link costs frame rate, never accumulating lag.
The ack is sent after drawImage, so the measured send→ack covers network +
decode + paint.
- That measurement drives a resolution/quality ladder: up to 1400px edge (vs
the old fixed 560) when the link is fast, quality dropping before size on
the way down — the view's job is READING the window — downgrades immediate,
upgrades needing sustained headroom. Encoding happens per SENT frame.
- /wstream is a separate socket from the input WS on purpose: 30-150KB frames
would head-of-line-block pointer motion on a slow link.
- Every accepted socket now sets TCP_NODELAY; nothing did before, so Nagle
was batching tiny input events behind delayed ACKs.
The page's <img> becomes a <canvas> (createImageBitmap + drawImage), which
also retires the multipart-img repaint scar tissue; the wrapper-transform
structure stays for tap mapping. /wstream only auto-reconnects if the
connection paired, so a stale PIN cannot retry-loop into the rate limiter and
lock the phone out of the input socket. /stream survives as the curl-friendly
MJPEG debug endpoint, thin over the same slot (verified live: 17fps, 37KB
frames through the winstream source).
Six new tests (26 total): the slot overwrites and never queues (the fix
itself), close wakes blocked waiters, a push wakes a blocked waiter, the
ladder never increases resolution on a downgrade step, adaptation downgrades
immediately but upgrades only after sustained headroom, and both ends clamp.
The ceiling above this is hardware H.264 + WebCodecs; deliberately not taken —
VAAPI/GStreamer deps and Safari codec quirks against JPEG-on-LAN that already
saturates the need. Revisit only if bandwidth-starved in practice.
Co-Authored-By: Claude Fable 5 <[email protected]>
CLAUDE.md | 78 +++++++---
README.md | 10 +-
index.html | 87 +++++++----
src/main.rs | 100 +++++++------
src/screencopy.rs | 104 ++++---------
src/stream.rs | 434 ++++++++++++++++++++++++++++++++++++++++++++++++++++++
src/winstream.rs | 97 ++++++------
7 files changed, 695 insertions(+), 215 deletions(-)
diff --git a/CLAUDE.md b/CLAUDE.md
index 9460b48..ba881fe 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -14,11 +14,14 @@ out — and its entire user interface is one hand-written `index.html` compiled
binary with `include_str!`.
Mirroring a sibling app for structure is therefore the wrong instinct here. There is no
-`Application` trait, no `Message` enum, no widget tree. Three files, ~880 lines:
+`Application` trait, no `Message` enum, no widget tree. Four files:
-- **`src/main.rs`** — PIN auth, the HTTP/WS dispatch, the frame→command translator.
+- **`src/main.rs`** — PIN auth + rate limiting, the HTTP/WS dispatch, the
+ frame→command translator.
+- **`src/stream.rs`** — live-view delivery: the latest-wins `Slot`, the ack-clocked
+ sender and its adaptation ladder, and the producer that picks a frame source.
- **`src/screencopy.rs`** — a persistent `wlr-screencopy` client (frame source #2), and
- `downscale_encode`, shared by both live frame sources.
+ `downscale_encode`, shared by both raw frame sources.
- **`src/winstream.rs`** — consumer of the compositor's window-stream socket (source #1).
## The invariant: the control socket is a full-privilege injection channel
@@ -82,7 +85,7 @@ gates consult it, and an unresolvable peer address is refused rather than exempt
Three properties it must keep, each with a test:
- **Only failures are charged, and a success clears the record.** The page reconnects
- its stream on every hiccup — an `error` event, the 20s watchdog — each time presenting
+ its stream on every hiccup — a WS close, the no-frame watchdog — each time presenting
a correct PIN. If those consumed budget, a working client would throttle itself off.
- **Refill caps at the burst.** Otherwise an idle attacker banks attempts and the limit
is only an average. Note the test asserts this on `refilled()` *directly*: going
@@ -120,10 +123,47 @@ timer so a fast drag becomes ~80 commands/sec, not one per touch event. One thre
spawned per accepted connection, uncapped, and a `/stream` connection holds its thread
for the life of the stream.
-## Three frame sources, in preference order
-
-`handle_stream` tries each in turn and falls through on error. All three are live code —
-the fallbacks exist because the first two have real preconditions.
+## The live view: latest-wins delivery, three frame sources
+
+**Delivery and capture are separate concerns since the 2026-08-22 rework.** The
+original MJPEG path pushed every frame, in order, into a blocking TCP write; nothing on
+this side ever dropped one, so the kernel's send buffer (~10-30 frames) became a queue,
+and the moment wifi throughput dipped below the frame rate the view fell seconds behind
+and never recovered — "fine at first, unusable after a short time".
+
+The delivery design (`stream.rs`) makes that failure structurally impossible:
+
+- A **`Slot`** holds only the newest frame; the producer overwrites it. Overwriting IS
+ the frame-dropping — stale frames cease to exist before they cost encode or network.
+- The page's live view rides **`/wstream`**, a dedicated WebSocket (same `auth <pin>`
+ first-frame gate): the server sends one frame, the page renders it and acks `n`, and
+ only then does the newest frame go out. **At most one frame is ever in flight**, so a
+ degraded link costs frame *rate*, never accumulating latency. The ack is sent after
+ `drawImage`, not on receipt — so the measured send→ack time covers network + decode +
+ paint, which is what the user experiences.
+- That measurement drives an **adaptation ladder** (`LADDER`/`adapt()`): resolution up
+ to 1400px edge when the link is fast, quality degrading before size on the way down,
+ downgrades immediate, upgrades requiring sustained headroom. Encoding happens per
+ *sent* frame at the chosen level.
+- `/wstream` is deliberately a **separate socket from the input WS**: frames are
+ 30-150KB and input events are bytes; one TCP stream would head-of-line-block pointer
+ motion behind every frame.
+- `/stream` (MJPEG over HTTP) survives as the **curl-debuggable endpoint**, thin over
+ the same slot at fixed 560/q60. Without acks its TCP buffer can still hold a few
+ frames — fine for debugging, which is all it is for now.
+
+Every accepted socket gets `TCP_NODELAY` — before the rework nothing set it, so Nagle
+was batching tiny input events behind delayed ACKs.
+
+The ceiling above this design is hardware H.264 + WebCodecs/WebRTC (~5-10× fewer bytes),
+at the cost of VAAPI/GStreamer deps and Safari codec quirks. Ack-clocked adaptive JPEG
+is the right cost/benefit for a single-window view on a LAN; revisit only if it proves
+bandwidth-starved in practice.
+
+### The three frame sources
+
+`spawn_producer` tries each in turn. All three are live code — the fallbacks exist
+because the first two have real preconditions.
1. **`winstream`** — subscribe `window focused` on `/tmp/cce-stream-{WAYLAND_DISPLAY}.sock`
and read `frame <w> <h> <len>` + packed RGBA. Best source: damage is *per window*, it
@@ -137,8 +177,8 @@ the fallbacks exist because the first two have real preconditions.
still wakes on unrelated screen activity.
3. **`grim`** — fork per frame, `-s 0.5 -q 65`. ~2.5 fps. The floor.
-Frames are box-downscaled to `MAX_EDGE` 560 and JPEG'd at quality 60 (~29KB/frame) by the
-shared `downscale_encode` — tuned for wifi latency and phone-side decode, not fidelity.
+Frames are box-downscaled and JPEG'd by the shared `downscale_encode`, at whatever
+(edge, quality) the ladder picked for the link — not a fixed size anymore.
**The cursor differs between sources, and the page compensates for the worst case.**
Compositor window-stream frames are surface textures with no cursor composited, so the
@@ -175,14 +215,18 @@ Four of its non-obvious constructs are scar tissue. Do not "clean them up":
on focus plus a 1s drift-repair timer). iOS never fires `deleteContentBackward` on an
empty field, so without something to delete, backspace silently does nothing.
`beforeinput` is used throughout because iOS `keydown` reports keyCode 229.
-- **The zoom/pan transform lives on `#screenwrap`, never on the `<img>`.** iOS Safari
- stops repainting a GPU-promoted layer when its `multipart/x-mixed-replace` `<img>`
- updates — the live view goes black.
+- **The zoom/pan transform lives on `#screenwrap`, not the frame element.** Uniform
+ ancestor transform means `getBoundingClientRect` reflects it, keeping tap mapping
+ correct while zoomed. (It also used to dodge an iOS bug where a transformed
+ multipart-MJPEG `<img>` stopped repainting; the view is a `<canvas>` since the
+ 2026-08-22 rework, but the structure stays.)
- **The `overflow: hidden` clip lives on `#pad`, the non-transformed ancestor.** A clip
on the transformed element scales with its own content and clips nothing.
-- **The stream `<img>` src carries a nonce, plus an `error` handler and a 20s no-frame
- watchdog.** MJPEG in an `<img>` goes blank when its connection ends, and a browser will
- not re-request an unchanged src, so a network blip left the view dead forever.
+- **The stream self-heals: reconnect on WS close plus a 30s no-frame watchdog** (the
+ frame sources force keepalives ≤20s, so 30s of silence is a dead connection, not an
+ idle window). One guard worth keeping: the page only auto-reconnects `/wstream` if
+ that connection *paired successfully* — retry-looping a stale PIN would feed the
+ rate limiter and lock the phone's address out of the input socket too.
`SCROLL = 0.8`, not the 0.045 it started as: axis values reach clients as surface-px
deltas, so near-unity is the trackpad-like 1:1 feel. A 300px swipe used to scroll one line.
@@ -194,7 +238,7 @@ The awkward part: **there is no WebSocket client on this machine** (no `websocat
be driven from a real phone, or by writing a throwaway client.
The pure functions are the exception, and they are where the crate's invariants are
-actually enforced, so they carry all the tests (`cargo test -p cce-remote`, 15 of them,
+actually enforced, so they carry all the tests (`cargo test -p cce-remote`, 26 of them,
in `main.rs`) — `translate()` for what a paired client may say, and the three PIN gates
for who is paired at all. They cover the accepted shapes and — more to the point —
everything that
diff --git a/README.md b/README.md
index 3f5fc30..8502f7a 100644
--- a/README.md
+++ b/README.md
@@ -30,9 +30,13 @@ Home Screen for a fullscreen app feel.
char→evdev map; iOS `beforeinput` is used, so autocorrect noise is
filtered)
- ☰ — window switcher: tap a window to focus it
-- 🖥 — window view mode: a live MJPEG stream of the focused window
- (damage-driven wlr-screencopy: ~11 fps when the window is active, idle
- throttled to output damage; grim remains as a fallback path).
+- 🖥 — window view mode: a live stream of the focused window, delivered
+ ack-clocked over a WebSocket — at most one frame in flight, so a slow
+ link drops frame rate instead of falling behind — with resolution and
+ quality adapting to the measured link (up to 1400px edge when it's fast).
+ Frames come from the compositor's damage-driven window stream, falling
+ back to wlr-screencopy, then grim; `/stream` remains as a curl-friendly
+ MJPEG debug endpoint.
Tap to click that spot, long-press to right-click; one-finger drag moves
the pointer exactly like the trackpad (a cyan ring marks the cursor —
compositor frames carry none), and two fingers pinch-zoom / pan the view
diff --git a/index.html b/index.html
index 3bb4ed3..3aa1c8f 100644
--- a/index.html
+++ b/index.html
@@ -31,9 +31,10 @@
border-radius: 10px; }
#clicks button:active, #bar button:active { background: #2c5f69; }
#kb { position: absolute; opacity: 0.02; width: 1px; height: 1px; left: -10px; top: -10px; }
- /* Zoom/pan transform lives on the WRAPPER, never the <img>: iOS Safari
- stops repainting a GPU-promoted layer when its multipart-MJPEG <img>
- updates (black square). The img stays in the normal paint path. */
+ /* Zoom/pan transform lives on the WRAPPER: getBoundingClientRect then
+ reflects it, so tap mapping stays correct zoomed. (Historically this also
+ dodged an iOS bug where a transformed multipart-MJPEG <img> stopped
+ repainting; the live view is a canvas now, but the structure stays.) */
#screenwrap { display: none; position: absolute; inset: 0; z-index: 2;
transform-origin: 0 0; border-radius: 12px; overflow: hidden; }
#pad.view #screenwrap { display: block; }
@@ -84,7 +85,7 @@
</div>
<div id="preview"></div>
<div id="pad"><div class="hint">drag = move · tap = click · 2-finger drag = scroll<br>2-finger tap = right click · hold = drag</div>
- <div id="screenwrap"><img id="screen" draggable="false"></div>
+ <div id="screenwrap"><canvas id="screen" width="0" height="0"></canvas></div>
<div id="vcursor"></div>
<div id="switcher"></div>
<div id="menu"></div>
@@ -147,24 +148,57 @@ const ACCEL = 1.6, SCROLL = 0.8;
let touches = new Map(), moved = 0, startT = 0, maxFingers = 0, startX = 0, startY = 0;
let dragging = false, dragTimer = null, pendDx = 0, pendDy = 0, pendSy = 0, flushT = null;
-// ── Window view mode: live MJPEG stream of the focused window; taps
-// click it. The <img> renders /stream (multipart/x-mixed-replace)
-// natively; winRect for tap mapping is kept fresh by polling `wl` over
-// the WS (the stream follows focus server-side).
+// ── Window view mode: live view of the focused window; taps click it.
+// Frames arrive as JPEG binaries on a dedicated /wstream WebSocket, drawn to
+// the canvas; winRect for tap mapping is kept fresh by polling `wl` over the
+// input WS (the stream follows focus server-side).
+//
+// Delivery is ACK-CLOCKED: after rendering a frame we send "n", and only then
+// does the server send its NEWEST frame. At most one frame is ever in flight,
+// so a slow link costs frame rate, never accumulating lag — the old MJPEG
+// path queued frames in the TCP buffer and grew seconds behind. The ack also
+// times the whole pipeline for the server's resolution/quality adaptation,
+// which is why it is sent after drawImage, not on receipt.
let viewMode = false, winRect = null, rectTimer = null, cursorTimer = null;
const screenEl = document.getElementById("screen");
-// MJPEG in an <img> shows blank when its HTTP connection ends, and a browser
-// never re-requests an unchanged src — so reconnect on error, and via a
-// watchdog for silent stalls (a nonce forces a fresh request each time).
-let streamGen = 0, lastFrameAt = 0, watchdog = null;
+const screenCtx = screenEl.getContext("2d");
+let sws = null, lastFrameAt = 0, watchdog = null;
function streamStart() {
+ if (sws) { const old = sws; sws = null; try { old.close(); } catch (_) {} }
const pin = localStorage.getItem("cce-remote-pin") || "";
- streamGen++;
+ const s = new WebSocket("ws://" + location.host + "/wstream");
+ s.binaryType = "arraybuffer";
+ sws = s;
lastFrameAt = Date.now();
- screenEl.src = "/stream?pin=" + encodeURIComponent(pin) + "&g=" + streamGen;
+ let paired = false;
+ s.onopen = () => s.send("auth " + pin);
+ s.onmessage = async e => {
+ if (typeof e.data === "string") {
+ if (e.data === "auth ok") { paired = true; s.send("n"); }
+ return;
+ }
+ lastFrameAt = Date.now();
+ try {
+ const bmp = await createImageBitmap(new Blob([e.data], { type: "image/jpeg" }));
+ if (screenEl.width !== bmp.width || screenEl.height !== bmp.height) {
+ screenEl.width = bmp.width;
+ screenEl.height = bmp.height;
+ }
+ screenCtx.drawImage(bmp, 0, 0);
+ bmp.close();
+ } catch (_) {}
+ if (sws === s && s.readyState === 1) s.send("n");
+ };
+ // Reconnect only if this connection ever paired: a stale PIN would
+ // otherwise retry-loop failures into the server's rate limiter and lock
+ // the phone's address out of the INPUT socket too.
+ s.onclose = () => { if (viewMode && sws === s && paired) setTimeout(streamStart, 600); };
+}
+function streamStop() {
+ if (sws) { const old = sws; sws = null; try { old.close(); } catch (_) {} }
+ screenEl.width = 0;
+ screenEl.height = 0;
}
-screenEl.addEventListener("load", () => { lastFrameAt = Date.now(); });
-screenEl.addEventListener("error", () => { if (viewMode) setTimeout(streamStart, 600); });
function updateRect(wins) {
const f = wins.find(w => w.focused && w.mode !== "Status");
if (f) winRect = { id: f.id, x: f.x, y: f.y, w: f.w, h: f.h };
@@ -191,12 +225,12 @@ function clampPan() {
const vcursor = document.getElementById("vcursor");
let ptr = null; // last known pointer position in layout px (from ploc + prediction)
function placeMarker(lx, ly) {
- if (!viewMode || !winRect || !screenEl.naturalWidth) { vcursor.classList.remove("on"); return; }
+ if (!viewMode || !winRect || !screenEl.width) { vcursor.classList.remove("on"); return; }
const u = (lx - winRect.x) / winRect.w, v = (ly - winRect.y) / winRect.h;
if (!(u >= 0 && u <= 1 && v >= 0 && v <= 1)) { vcursor.classList.remove("on"); return; }
const box = screenEl.getBoundingClientRect();
- const scale = Math.min(box.width / screenEl.naturalWidth, box.height / screenEl.naturalHeight);
- const dw = screenEl.naturalWidth * scale, dh = screenEl.naturalHeight * scale;
+ const scale = Math.min(box.width / screenEl.width, box.height / screenEl.height);
+ const dw = screenEl.width * scale, dh = screenEl.height * scale;
const padBox = pad.getBoundingClientRect();
vcursor.style.left = (box.left - padBox.left + (box.width - dw) / 2 + u * dw) + "px";
vcursor.style.top = (box.top - padBox.top + (box.height - dh) / 2 + v * dh) + "px";
@@ -221,10 +255,10 @@ function predictMarker(dLx, dLy) {
// touch point (viewport px) → layout px inside the focused window,
// accounting for the contain-fit letterbox. null when off the image.
function mapToWindow(cx, cy) {
- if (!winRect || !screenEl.naturalWidth) return null;
+ if (!winRect || !screenEl.width) return null;
const box = screenEl.getBoundingClientRect();
- const scale = Math.min(box.width / screenEl.naturalWidth, box.height / screenEl.naturalHeight);
- const dw = screenEl.naturalWidth * scale, dh = screenEl.naturalHeight * scale;
+ const scale = Math.min(box.width / screenEl.width, box.height / screenEl.height);
+ const dw = screenEl.width * scale, dh = screenEl.height * scale;
const ox = box.left + (box.width - dw) / 2, oy = box.top + (box.height - dh) / 2;
const u = (cx - ox) / dw, v = (cy - oy) / dh;
if (u < 0 || u > 1 || v < 0 || v > 1) return null;
@@ -351,17 +385,18 @@ function setViewMode(on) {
send("wl"); // prime winRect
rectTimer = setInterval(() => send("wl"), 2000);
cursorTimer = setInterval(() => send("pl"), 120);
- // Reconnect if no frame has loaded for a while (the compositor keepalives
- // every ≤15s, so ~20s of silence means the connection is dead, not idle).
+ // Reconnect if no frame arrives for a while (the frame sources force
+ // keepalive frames every ≤20s, so ~30s of silence means the connection
+ // is dead, not idle).
watchdog = setInterval(() => {
- if (viewMode && Date.now() - lastFrameAt > 20000) streamStart();
+ if (viewMode && Date.now() - lastFrameAt > 30000) streamStart();
}, 5000);
} else {
clearInterval(rectTimer);
clearInterval(cursorTimer);
clearInterval(watchdog);
vcursor.classList.remove("on");
- screenEl.src = "data:,"; // closes the stream connection
+ streamStop();
}
}
const menuEl = document.getElementById("menu");
diff --git a/src/main.rs b/src/main.rs
index fd88147..735b4d0 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -27,6 +27,7 @@ use std::os::unix::net::UnixStream;
use std::time::Duration;
mod screencopy;
+mod stream;
mod winstream;
const INDEX_HTML: &str = include_str!("../index.html");
@@ -460,54 +461,53 @@ fn handle_stream(mut stream: TcpStream, request_head: &str, pin: &str, ip: std::
{
return;
}
- // Source preference: the compositor's window stream (per-window damage,
- // follows focus, works off-screen) → screencopy (output damage) → grim.
- match winstream::stream_mjpeg(&mut stream) {
- Ok(()) => return, // client disconnected
- Err(e) => eprintln!("[cce-remote] window stream unavailable ({e}), trying screencopy"),
- }
- let rect = || focused_window().map(|(_, x, y, w, h)| (x as i32, y as i32, w as i32, h as i32));
- match screencopy::stream_mjpeg(&mut stream, rect) {
- Ok(()) => return, // client disconnected
- Err(e) => eprintln!("[cce-remote] screencopy stream failed ({e}), falling back to grim"),
- }
- let mut win = focused_window();
- let mut tick = 0u32;
- loop {
- if tick % 4 == 0 {
- if let Some(w) = focused_window() {
- win = Some(w);
- }
- }
- tick = tick.wrapping_add(1);
- let Some((_, x, y, w, h)) = win else {
- std::thread::sleep(Duration::from_millis(400));
- continue;
- };
- let out = std::process::Command::new("grim")
- .args([
- "-g",
- &format!("{},{} {}x{}", x as i32, y as i32, w as i32, h as i32),
- "-t", "jpeg", "-q", "65", "-s", "0.5", "-",
- ])
- .output();
- match out {
- Ok(o) if o.status.success() && o.stdout.starts_with(&[0xff, 0xd8]) => {
- let part = format!(
- "--frame\r\nContent-Type: image/jpeg\r\nContent-Length: {}\r\n\r\n",
- o.stdout.len()
- );
- if stream.write_all(part.as_bytes()).is_err()
- || stream.write_all(&o.stdout).is_err()
- || stream.write_all(b"\r\n").is_err()
- {
- return; // client gone — the loop (and grim spawning) stops
- }
- }
- _ => std::thread::sleep(Duration::from_millis(400)),
+ // Source selection (winstream → screencopy → grim) lives in the producer;
+ // this endpoint is the curl-debuggable MJPEG view over the same slot the
+ // page's ack-clocked /wstream uses.
+ let slot = stream::Slot::new();
+ let stop = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
+ stream::spawn_producer(std::sync::Arc::clone(&slot), std::sync::Arc::clone(&stop));
+ stream::run_mjpeg_sender(&mut stream, &slot);
+ stop.store(true, std::sync::atomic::Ordering::Relaxed);
+}
+
+/// The page's live view: ack-clocked latest-wins frame delivery over a
+/// dedicated WebSocket. Same first-frame `auth <pin>` gate as the input WS —
+/// and a separate socket on purpose: frames are 30-150KB and input events are
+/// bytes, so sharing one TCP stream would head-of-line-block pointer motion
+/// behind every frame on a slow link.
+fn handle_wstream(stream: TcpStream, pin: &str, ip: std::net::IpAddr, limiter: &RateLimiter) {
+ let peer = stream.peer_addr().map(|a| a.to_string()).unwrap_or_default();
+ let _ = stream.set_read_timeout(Some(Duration::from_secs(10)));
+ let mut ws = match tungstenite::accept(stream) {
+ Ok(ws) => ws,
+ Err(e) => {
+ eprintln!("[cce-remote] wstream handshake failed ({peer}): {e}");
+ return;
}
- std::thread::sleep(Duration::from_millis(40));
+ };
+ if !limiter.allow(ip, std::time::Instant::now()) {
+ let _ = ws.close(None);
+ return;
+ }
+ let authed = matches!(
+ ws.read(),
+ Ok(tungstenite::Message::Text(t)) if auth_frame_ok(&t, pin)
+ );
+ if !authed {
+ limiter.record_failure(ip, std::time::Instant::now());
+ eprintln!("[cce-remote] wstream auth failed: {peer}");
+ let _ = ws.close(None);
+ return;
}
+ limiter.record_success(ip);
+ let _ = ws.send(tungstenite::Message::Text("auth ok".into()));
+ let slot = stream::Slot::new();
+ let stop = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
+ stream::spawn_producer(std::sync::Arc::clone(&slot), std::sync::Arc::clone(&stop));
+ stream::run_ws_sender(&mut ws, &slot);
+ stop.store(true, std::sync::atomic::Ordering::Relaxed);
+ let _ = ws.close(None);
}
fn handle_http(mut stream: TcpStream, request_head: &str, pin: &str, ip: std::net::IpAddr, limiter: &RateLimiter) {
@@ -591,6 +591,9 @@ fn main() {
// held accountable for its guesses, so it is refused rather than
// exempted.
let Ok(ip) = stream.peer_addr().map(|a| a.ip()) else { continue };
+ // Input events and stream acks are tiny and latency-critical; Nagle
+ // would batch them behind delayed ACKs.
+ let _ = stream.set_nodelay(true);
let pin = pin.clone();
let limiter = std::sync::Arc::clone(&limiter);
std::thread::spawn(move || {
@@ -602,7 +605,10 @@ fn main() {
_ => return,
};
let head = String::from_utf8_lossy(&buf[..n]).to_string();
- if head.starts_with("GET /ws") {
+ if head.starts_with("GET /wstream") {
+ // before /ws: "GET /ws" is a prefix of this
+ handle_wstream(stream, &pin, ip, &limiter);
+ } else if head.starts_with("GET /ws") {
handle_ws(stream, &pin, ip, &limiter);
} else {
// Consume the request before replying (keeps curl happy).
diff --git a/src/screencopy.rs b/src/screencopy.rs
index 515675c..a6315ab 100644
--- a/src/screencopy.rs
+++ b/src/screencopy.rs
@@ -13,7 +13,6 @@
//! coordinates, which equals layout coordinates when the (only) output sits
//! at 0,0 — true for this DE's eDP-1 setup, same assumption grim ran under.
-use std::io::Write;
use std::os::fd::AsFd;
use std::time::{Duration, Instant};
@@ -25,10 +24,6 @@ use wayland_protocols_wlr::screencopy::v1::client::{
zwlr_screencopy_manager_v1::ZwlrScreencopyManagerV1,
};
-/// Longest edge of the encoded frame, in px — the downscale factor is chosen
-/// per frame to stay under this.
-const MAX_EDGE: u32 = 560;
-const JPEG_QUALITY: u8 = 60;
#[derive(Default)]
struct CapState {
@@ -210,14 +205,15 @@ impl CaptureSession {
/// Capture one frame of `rect` (output-local logical px). With
/// `use_damage`, blocks until the region changes or `timeout` — a timeout
- /// returns Ok(None) so the caller can force a keepalive frame. The frame
- /// is returned already JPEG-encoded.
+ /// returns Ok(None) so the caller can force a keepalive frame. The pixels
+ /// are returned RAW (copied out of the shm slot, which is reused);
+ /// encoding happens at send time so dropped frames cost nothing.
pub fn next_frame(
&mut self,
rect: (i32, i32, i32, i32),
use_damage: bool,
timeout: Duration,
- ) -> Result<Option<Vec<u8>>, String> {
+ ) -> Result<Option<crate::stream::Payload>, String> {
self.state = CapState::default();
let frame = self.manager.capture_output_region(
1, // overlay the cursor — the remote wants to see it
@@ -257,37 +253,41 @@ impl CaptureSession {
}
frame.destroy();
let slot = self.slot.as_ref().unwrap();
- Ok(Some(encode_jpeg(&slot.map, slot.meta)?))
+ let (w, h, stride, format) = slot.meta;
+ // byte offsets of R,G,B within each little-endian 32-bit pixel
+ let rgb = match format {
+ wl_shm::Format::Xrgb8888 | wl_shm::Format::Argb8888 => (2usize, 1usize, 0usize),
+ _ => (0usize, 1usize, 2usize), // Xbgr8888 / Abgr8888
+ };
+ let size = (stride * h) as usize;
+ Ok(Some(crate::stream::Payload::Raw {
+ data: slot.map[..size].to_vec(),
+ w,
+ h,
+ stride,
+ rgb,
+ }))
}
}
-/// Box-downscale the 32-bit shm pixels to ≤ MAX_EDGE and encode as JPEG.
-fn encode_jpeg(
- map: &memmap2::MmapMut,
- (w, h, stride, format): (u32, u32, u32, wl_shm::Format),
-) -> Result<Vec<u8>, String> {
- // byte offsets of R,G,B within each little-endian 32-bit pixel
- let rgb_at = match format {
- wl_shm::Format::Xrgb8888 | wl_shm::Format::Argb8888 => (2usize, 1usize, 0usize),
- _ => (0usize, 1usize, 2usize), // Xbgr8888 / Abgr8888
- };
- downscale_encode(map, w, h, stride, rgb_at)
-}
-
-/// Shared by both frame sources (screencopy shm and the compositor's
-/// window-stream RGBA): box-downscale 32-bit pixels to ≤ MAX_EDGE and JPEG
-/// them. `rgb_at` gives the byte offsets of R,G,B within each 4-byte pixel.
+/// Shared by both raw frame sources (screencopy shm and the compositor's
+/// window-stream RGBA): box-downscale 32-bit pixels to ≤ `max_edge` and JPEG
+/// them at `quality`. `rgb_at` gives the byte offsets of R,G,B within each
+/// 4-byte pixel. Called per SENT frame, with the (edge, quality) the
+/// adaptation ladder picked for the link.
pub fn downscale_encode(
data: &[u8],
w: u32,
h: u32,
stride: u32,
(ri, gi, bi): (usize, usize, usize),
+ max_edge: u32,
+ quality: u8,
) -> Result<Vec<u8>, String> {
if w == 0 || h == 0 || (stride * h) as usize > data.len() {
return Err("bad frame dimensions".into());
}
- let f = ((w.max(h) + MAX_EDGE - 1) / MAX_EDGE).max(1);
+ let f = ((w.max(h) + max_edge - 1) / max_edge).max(1);
let (ow, oh) = (w / f, h / f);
let mut rgb = Vec::with_capacity((ow * oh * 3) as usize);
let fsq = (f * f) as u32;
@@ -309,59 +309,9 @@ pub fn downscale_encode(
}
}
let mut out = Vec::new();
- let encoder = jpeg_encoder::Encoder::new(&mut out, JPEG_QUALITY);
+ let encoder = jpeg_encoder::Encoder::new(&mut out, quality);
encoder
.encode(&rgb, ow as u16, oh as u16, jpeg_encoder::ColorType::Rgb)
.map_err(|e| e.to_string())?;
Ok(out)
}
-
-/// Drive `session` frames into an MJPEG multipart writer until the client
-/// disconnects. `rect_of_focused` re-resolves the focused window (layout px).
-pub fn stream_mjpeg(
- tcp: &mut std::net::TcpStream,
- rect_of_focused: impl Fn() -> Option<(i32, i32, i32, i32)>,
-) -> Result<(), String> {
- let mut session = CaptureSession::new()?;
- let mut rect = rect_of_focused();
- let mut rect_at = Instant::now();
- let mut force_full = true; // first frame immediately; also after timeouts
- loop {
- if rect_at.elapsed() > Duration::from_millis(500) {
- if let Some(r) = rect_of_focused() {
- if Some(r) != rect {
- force_full = true; // focus moved: don't wait for damage
- }
- rect = Some(r);
- }
- rect_at = Instant::now();
- }
- let Some(r) = rect else {
- std::thread::sleep(Duration::from_millis(400));
- rect = rect_of_focused();
- continue;
- };
- // 20s damage timeout doubles as a keepalive: the forced frame's write
- // is what detects a silently-gone client.
- let timeout = if force_full { Duration::from_secs(5) } else { Duration::from_secs(20) };
- match session.next_frame(r, !force_full, timeout) {
- Ok(Some(jpeg)) => {
- force_full = false;
- let head = format!(
- "--frame\r\nContent-Type: image/jpeg\r\nContent-Length: {}\r\n\r\n",
- jpeg.len()
- );
- if tcp.write_all(head.as_bytes()).is_err()
- || tcp.write_all(&jpeg).is_err()
- || tcp.write_all(b"\r\n").is_err()
- {
- return Ok(()); // client gone
- }
- // cap runaway damage bursts (~30 fps)
- std::thread::sleep(Duration::from_millis(33));
- }
- Ok(None) => force_full = true,
- Err(e) => return Err(e),
- }
- }
-}
diff --git a/src/stream.rs b/src/stream.rs
new file mode 100644
index 0000000..48984ee
--- /dev/null
+++ b/src/stream.rs
@@ -0,0 +1,434 @@
+//! Live-view delivery: latest-frame-wins, clocked by client acks.
+//!
+//! The failure this module exists to prevent: the original MJPEG path pushed
+//! every frame, in order, into a blocking TCP write. Nothing between the
+//! compositor and the phone ever dropped a frame, so the kernel's send buffer
+//! (hundreds of KB, ~10-30 frames) became a queue — the moment wifi throughput
+//! dipped below the frame rate, the queue filled, and every frame the phone
+//! showed was queue-depth old. Latency accumulated and never drained: "fine at
+//! first, unusable after a short time".
+//!
+//! The fix is sender-side flow control, the same shape VNC/RDP use:
+//!
+//! - A `Slot` holds only the NEWEST frame from the source (winstream →
+//! screencopy → grim, tried in that order by `spawn_producer`). Overwriting
+//! is the drop point: stale frames cease to exist before they cost anything.
+//! - `run_ws_sender` sends one frame, then waits for the page's `n` ack before
+//! sending the newest frame available. In-flight is capped at ONE frame, so
+//! degraded wifi costs frame RATE, never growing latency.
+//! - Encoding happens at send time, only for frames actually sent, at a
+//! (max_edge, jpeg_quality) picked by `adapt()` from the measured send→ack
+//! time — readable resolution when the link allows, graceful degradation
+//! when it doesn't.
+//!
+//! `/stream` (MJPEG over HTTP) remains as a curl-debuggable endpoint via
+//! `run_mjpeg_sender`, thin over the same slot; the page no longer uses it.
+
+use std::io::Write;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, Condvar, Mutex};
+use std::time::{Duration, Instant};
+
+/// One captured frame. `Raw` is encoded at send time; `Jpeg` (grim fallback)
+/// is passed through as-is, so adaptation does not apply to it.
+#[derive(Clone)]
+pub enum Payload {
+ Raw { data: Vec<u8>, w: u32, h: u32, stride: u32, rgb: (usize, usize, usize) },
+ Jpeg(Vec<u8>),
+}
+
+/// The latest-wins seam between the frame source and however many senders are
+/// consuming it. `push` overwrites; senders track the last seq they delivered.
+pub struct Slot {
+ inner: Mutex<Inner>,
+ cv: Condvar,
+}
+
+struct Inner {
+ seq: u64,
+ frame: Option<Payload>,
+ alive: bool,
+}
+
+impl Slot {
+ pub fn new() -> Arc<Self> {
+ Arc::new(Self {
+ inner: Mutex::new(Inner { seq: 0, frame: None, alive: true }),
+ cv: Condvar::new(),
+ })
+ }
+
+ pub fn push(&self, p: Payload) {
+ let mut g = self.inner.lock().unwrap();
+ g.seq += 1;
+ g.frame = Some(p);
+ self.cv.notify_all();
+ }
+
+ /// The producer died (source unavailable/broke). Senders return, the
+ /// client reconnects, and the fresh producer re-picks a source.
+ pub fn close(&self) {
+ self.inner.lock().unwrap().alive = false;
+ self.cv.notify_all();
+ }
+
+ /// Newest frame with seq > `last`: Ok(Some) on a new frame, Ok(None) on
+ /// timeout (window idle — sources force keepalive frames ≤20s), Err when
+ /// the producer is gone.
+ pub fn wait_newer(&self, last: u64, timeout: Duration) -> Result<Option<(u64, Payload)>, ()> {
+ let deadline = Instant::now() + timeout;
+ let mut g = self.inner.lock().unwrap();
+ loop {
+ if g.seq > last {
+ return Ok(Some((g.seq, g.frame.clone().unwrap())));
+ }
+ if !g.alive {
+ return Err(());
+ }
+ let now = Instant::now();
+ if now >= deadline {
+ return Ok(None);
+ }
+ let (ng, _) = self.cv.wait_timeout(g, deadline - now).unwrap();
+ g = ng;
+ }
+ }
+}
+
+fn encode(p: &Payload, max_edge: u32, quality: u8) -> Result<Vec<u8>, String> {
+ match p {
+ Payload::Raw { data, w, h, stride, rgb } => {
+ crate::screencopy::downscale_encode(data, *w, *h, *stride, *rgb, max_edge, quality)
+ }
+ Payload::Jpeg(b) => Ok(b.clone()),
+ }
+}
+
+// ---- adaptation ------------------------------------------------------------
+
+/// (max_edge px, jpeg quality), best first. Resolution is held as long as
+/// possible — quality drops before size does — because the point of the live
+/// view is READING the window.
+pub const LADDER: &[(u32, u8)] = &[
+ (1400, 68),
+ (1120, 68),
+ (1120, 55),
+ (840, 58),
+ (840, 46),
+ (560, 48),
+];
+pub const START_LEVEL: usize = 1;
+
+/// Pick the next ladder level from the smoothed send→ack time. Downgrades are
+/// immediate (lag is being felt NOW); upgrades need `acks_since_change` of
+/// stability so the level does not oscillate at a threshold. The dead band
+/// between the two thresholds is the hysteresis.
+pub fn adapt(level: usize, ewma_ms: f64, acks_since_change: u32) -> usize {
+ if ewma_ms > 220.0 {
+ (level + 1).min(LADDER.len() - 1)
+ } else if ewma_ms < 90.0 && acks_since_change >= 10 {
+ level.saturating_sub(1)
+ } else {
+ level
+ }
+}
+
+// ---- the frame source ------------------------------------------------------
+
+/// Source preference is unchanged from the MJPEG design: compositor window
+/// stream (per-window damage, follows focus) → screencopy (output damage) →
+/// grim. The producer owns the source; on source death it closes the slot.
+pub fn spawn_producer(slot: Arc<Slot>, stop: Arc<AtomicBool>) {
+ std::thread::spawn(move || {
+ let result = match crate::winstream::Reader::connect() {
+ Ok(mut r) => produce_winstream(&mut r, &slot, &stop),
+ Err(e) => {
+ eprintln!("[cce-remote] window stream unavailable ({e}), trying screencopy");
+ match crate::screencopy::CaptureSession::new() {
+ Ok(mut s) => produce_screencopy(&mut s, &slot, &stop),
+ Err(e2) => {
+ eprintln!("[cce-remote] screencopy unavailable ({e2}), falling back to grim");
+ produce_grim(&slot, &stop)
+ }
+ }
+ }
+ };
+ if let Err(e) = result {
+ if !stop.load(Ordering::Relaxed) {
+ eprintln!("[cce-remote] frame source ended: {e}");
+ }
+ }
+ slot.close();
+ });
+}
+
+fn produce_winstream(
+ r: &mut crate::winstream::Reader,
+ slot: &Slot,
+ stop: &AtomicBool,
+) -> Result<(), String> {
+ loop {
+ if stop.load(Ordering::Relaxed) {
+ return Ok(());
+ }
+ // Blocks ≤20s: the compositor keepalives every ≤15s, so a stopped
+ // sender's producer lingers at most one keepalive interval.
+ slot.push(r.next()?);
+ }
+}
+
+fn produce_screencopy(
+ sess: &mut crate::screencopy::CaptureSession,
+ slot: &Slot,
+ stop: &AtomicBool,
+) -> Result<(), String> {
+ let mut rect: Option<(i32, i32, i32, i32)> = None;
+ let mut rect_at: Option<Instant> = None;
+ let mut force_full = true; // first frame immediately; also after timeouts
+ loop {
+ if stop.load(Ordering::Relaxed) {
+ return Ok(());
+ }
+ if rect_at.is_none_or(|t| t.elapsed() > Duration::from_millis(500)) {
+ if let Some((_, x, y, w, h)) = crate::focused_window() {
+ let r = (x as i32, y as i32, w as i32, h as i32);
+ if Some(r) != rect {
+ force_full = true; // focus moved: don't wait for damage
+ }
+ rect = Some(r);
+ }
+ rect_at = Some(Instant::now());
+ }
+ let Some(r) = rect else {
+ std::thread::sleep(Duration::from_millis(300));
+ continue;
+ };
+ // The 20s damage timeout doubles as the keepalive cadence.
+ let timeout = if force_full { Duration::from_secs(5) } else { Duration::from_secs(20) };
+ match sess.next_frame(r, !force_full, timeout) {
+ Ok(Some(frame)) => {
+ force_full = false;
+ slot.push(frame);
+ // cap runaway damage bursts (~30 fps)
+ std::thread::sleep(Duration::from_millis(33));
+ }
+ Ok(None) => force_full = true,
+ Err(e) => return Err(e),
+ }
+ }
+}
+
+fn produce_grim(slot: &Slot, stop: &AtomicBool) -> Result<(), String> {
+ loop {
+ if stop.load(Ordering::Relaxed) {
+ return Ok(());
+ }
+ let Some((_, x, y, w, h)) = crate::focused_window() else {
+ std::thread::sleep(Duration::from_millis(400));
+ continue;
+ };
+ let out = std::process::Command::new("grim")
+ .args([
+ "-g",
+ &format!("{},{} {}x{}", x as i32, y as i32, w as i32, h as i32),
+ "-t", "jpeg", "-q", "65", "-s", "0.5", "-",
+ ])
+ .output();
+ match out {
+ Ok(o) if o.status.success() && o.stdout.starts_with(&[0xff, 0xd8]) => {
+ slot.push(Payload::Jpeg(o.stdout));
+ std::thread::sleep(Duration::from_millis(350));
+ }
+ _ => std::thread::sleep(Duration::from_millis(400)),
+ }
+ }
+}
+
+// ---- senders ---------------------------------------------------------------
+
+/// Ack-clocked delivery over the page's stream WebSocket. The page sends `n`
+/// after RENDERING each frame — so the measured send→ack time covers network,
+/// decode and paint, i.e. what the user actually experiences — and only then
+/// does the newest frame go out.
+pub fn run_ws_sender(ws: &mut tungstenite::WebSocket<std::net::TcpStream>, slot: &Slot) {
+ let mut last_seq = 0u64;
+ let mut level = START_LEVEL;
+ let mut ewma_ms = 120.0f64;
+ let mut acks_since_change = 0u32;
+ let mut sent_at: Option<Instant> = None;
+ // Acks can legitimately stop for a long time (iOS suspends the page when
+ // backgrounded); pings distinguish suspended-but-alive from gone. The
+ // browser answers pings in its network stack, JS not required.
+ let _ = ws.get_ref().set_read_timeout(Some(Duration::from_secs(75)));
+ let mut silent = 0u32;
+ loop {
+ // 1. wait for the ack of the previous frame
+ loop {
+ match ws.read() {
+ Ok(tungstenite::Message::Text(t)) if t == "n" => break,
+ Ok(tungstenite::Message::Close(_)) => return,
+ Ok(_) => {
+ silent = 0; // pong: peer alive, keep waiting
+ continue;
+ }
+ Err(tungstenite::Error::Io(e))
+ if e.kind() == std::io::ErrorKind::WouldBlock
+ || e.kind() == std::io::ErrorKind::TimedOut =>
+ {
+ silent += 1;
+ if silent >= 2 {
+ return; // two silent windows with no pong: gone
+ }
+ if ws.send(tungstenite::Message::Ping(Vec::new())).is_err() {
+ return;
+ }
+ continue;
+ }
+ Err(_) => return,
+ }
+ }
+ silent = 0;
+ if let Some(t0) = sent_at.take() {
+ let ms = t0.elapsed().as_secs_f64() * 1000.0;
+ ewma_ms = 0.7 * ewma_ms + 0.3 * ms;
+ acks_since_change += 1;
+ let next = adapt(level, ewma_ms, acks_since_change);
+ if next != level {
+ level = next;
+ acks_since_change = 0;
+ }
+ }
+ // 2. newest frame (blocks while the window is idle; sources force
+ // keepalive frames ≤20s, so this wakes regularly)
+ let (seq, payload) = loop {
+ match slot.wait_newer(last_seq, Duration::from_secs(30)) {
+ Ok(Some(x)) => break x,
+ Ok(None) => {
+ if ws.send(tungstenite::Message::Ping(Vec::new())).is_err() {
+ return;
+ }
+ }
+ Err(()) => return,
+ }
+ };
+ last_seq = seq;
+ let (edge, quality) = LADDER[level];
+ let Ok(jpeg) = encode(&payload, edge, quality) else { continue };
+ sent_at = Some(Instant::now());
+ if ws.send(tungstenite::Message::Binary(jpeg)).is_err() {
+ return;
+ }
+ }
+}
+
+/// MJPEG over the slot, fixed 560/q60 — kept as the curl-debuggable endpoint.
+/// Latest-wins still applies (each iteration encodes only the newest frame),
+/// but without acks the TCP buffer can still queue a few frames; the page no
+/// longer uses this path.
+pub fn run_mjpeg_sender(tcp: &mut std::net::TcpStream, slot: &Slot) {
+ let mut last = 0u64;
+ loop {
+ match slot.wait_newer(last, Duration::from_secs(25)) {
+ Ok(Some((seq, p))) => {
+ last = seq;
+ let Ok(jpeg) = encode(&p, 560, 60) else { continue };
+ let head = format!(
+ "--frame\r\nContent-Type: image/jpeg\r\nContent-Length: {}\r\n\r\n",
+ jpeg.len()
+ );
+ if tcp.write_all(head.as_bytes()).is_err()
+ || tcp.write_all(&jpeg).is_err()
+ || tcp.write_all(b"\r\n").is_err()
+ {
+ return; // client gone
+ }
+ }
+ Ok(None) => continue, // idle; dead clients surface on the next write
+ Err(()) => return,
+ }
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ // The slot IS the fix: if it ever queues instead of overwriting, the
+ // unbounded-lag failure mode comes back silently.
+
+ fn raw(tag: u8) -> Payload {
+ Payload::Raw { data: vec![tag; 4], w: 1, h: 1, stride: 4, rgb: (0, 1, 2) }
+ }
+ fn tag_of(p: &Payload) -> u8 {
+ match p {
+ Payload::Raw { data, .. } => data[0],
+ Payload::Jpeg(b) => b[0],
+ }
+ }
+
+ #[test]
+ fn slot_overwrites_never_queues() {
+ let s = Slot::new();
+ s.push(raw(1));
+ s.push(raw(2));
+ s.push(raw(3));
+ // A consumer that fell behind gets the NEWEST frame, once — frames 1
+ // and 2 are gone, not waiting their turn.
+ let (seq, p) = s.wait_newer(0, Duration::from_millis(10)).unwrap().unwrap();
+ assert_eq!(seq, 3);
+ assert_eq!(tag_of(&p), 3);
+ // Nothing newer: times out rather than re-delivering.
+ assert!(s.wait_newer(seq, Duration::from_millis(10)).unwrap().is_none());
+ }
+
+ #[test]
+ fn slot_close_wakes_and_errs() {
+ let s = Slot::new();
+ s.push(raw(9));
+ let _ = s.wait_newer(0, Duration::from_millis(10)).unwrap().unwrap();
+ s.close();
+ assert!(s.wait_newer(99, Duration::from_secs(5)).is_err(), "close must wake, not time out");
+ }
+
+ #[test]
+ fn slot_wakes_a_blocked_waiter() {
+ let s = Slot::new();
+ let s2 = Arc::clone(&s);
+ let t = std::thread::spawn(move || s2.wait_newer(0, Duration::from_secs(5)));
+ std::thread::sleep(Duration::from_millis(30));
+ s.push(raw(7));
+ let got = t.join().unwrap().unwrap().unwrap();
+ assert_eq!(got.0, 1);
+ assert_eq!(tag_of(&got.1), 7);
+ }
+
+ #[test]
+ fn ladder_prefers_resolution_over_quality() {
+ // The point of the live view is reading the window: stepping down the
+ // ladder must drop quality before it drops size.
+ for pair in LADDER.windows(2) {
+ let ((e1, _), (e2, _)) = (pair[0], pair[1]);
+ assert!(e2 <= e1, "ladder edge must be non-increasing: {pair:?}");
+ }
+ assert!(START_LEVEL < LADDER.len());
+ }
+
+ #[test]
+ fn adapt_downgrades_immediately_upgrades_cautiously() {
+ // Lag is felt now: no stability requirement to step down.
+ assert_eq!(adapt(1, 300.0, 0), 2);
+ // Upgrades need sustained headroom, or the level oscillates at the
+ // threshold.
+ assert_eq!(adapt(2, 50.0, 3), 2);
+ assert_eq!(adapt(2, 50.0, 10), 1);
+ // The dead band holds steady in both directions.
+ assert_eq!(adapt(2, 150.0, 100), 2);
+ }
+
+ #[test]
+ fn adapt_clamps_at_both_ends() {
+ let worst = LADDER.len() - 1;
+ assert_eq!(adapt(worst, 10_000.0, 0), worst);
+ assert_eq!(adapt(0, 1.0, 1000), 0);
+ }
+}
diff --git a/src/winstream.rs b/src/winstream.rs
index 5fb868f..888f528 100644
--- a/src/winstream.rs
+++ b/src/winstream.rs
@@ -2,7 +2,8 @@
//! frame source. The compositor pushes damage-driven RGBA frames of the
//! focused window (`window focused` subscription follows focus server-side,
//! works off-viewport/occluded, and a truly idle window sends nothing but a
-//! 15s keepalive). We downscale + JPEG each frame into the MJPEG response.
+//! ≤15s keepalive). Frames land in the latest-wins slot (`stream::Slot`);
+//! encoding happens at send time, not here.
use std::io::{BufRead, BufReader, Read, Write};
use std::os::unix::net::UnixStream;
@@ -15,52 +16,58 @@ fn stream_socket_path() -> String {
format!("/tmp/cce-stream-{display}.sock")
}
-/// Bridge the compositor stream into `tcp` as MJPEG parts. Ok(()) = the HTTP
-/// client went away; Err = the stream source is unavailable/broke (caller
-/// falls back to screencopy).
-pub fn stream_mjpeg(tcp: &mut std::net::TcpStream) -> Result<(), String> {
- let path = stream_socket_path();
- let sock = UnixStream::connect(&path).map_err(|e| format!("{path}: {e}"))?;
- // The compositor keepalives every ≤15s; a 40s silence means it's gone.
- sock.set_read_timeout(Some(Duration::from_secs(40))).ok();
- let mut sock = BufReader::new(sock);
- sock.get_mut()
- .write_all(b"window focused\n")
- .map_err(|e| e.to_string())?;
+pub struct Reader {
+ sock: BufReader<UnixStream>,
+}
- let mut header = String::new();
- loop {
- header.clear();
- if sock.read_line(&mut header).map_err(|e| e.to_string())? == 0 {
- return Err("stream socket closed".into());
- }
- let mut it = header.split_ascii_whitespace();
- if it.next() != Some("frame") {
- continue;
- }
- let (Some(w), Some(h), Some(len)) = (
- it.next().and_then(|v| v.parse::<u32>().ok()),
- it.next().and_then(|v| v.parse::<u32>().ok()),
- it.next().and_then(|v| v.parse::<u32>().ok()),
- ) else {
- return Err(format!("bad frame header: {header:?}"));
- };
- if len != w.saturating_mul(h).saturating_mul(4) || len > MAX_FRAME_BYTES {
- return Err(format!("implausible frame: {header:?}"));
- }
- let mut rgba = vec![0u8; len as usize];
- sock.read_exact(&mut rgba).map_err(|e| e.to_string())?;
+impl Reader {
+ pub fn connect() -> Result<Self, String> {
+ let path = stream_socket_path();
+ let sock = UnixStream::connect(&path).map_err(|e| format!("{path}: {e}"))?;
+ // The compositor keepalives every ≤15s; 20s of silence means it's
+ // gone. This timeout is also what bounds how long a stopped
+ // producer thread lingers.
+ sock.set_read_timeout(Some(Duration::from_secs(20))).ok();
+ let mut sock = BufReader::new(sock);
+ sock.get_mut()
+ .write_all(b"window focused\n")
+ .map_err(|e| e.to_string())?;
+ Ok(Self { sock })
+ }
- let jpeg = crate::screencopy::downscale_encode(&rgba, w, h, w * 4, (0, 1, 2))?;
- let part = format!(
- "--frame\r\nContent-Type: image/jpeg\r\nContent-Length: {}\r\n\r\n",
- jpeg.len()
- );
- if tcp.write_all(part.as_bytes()).is_err()
- || tcp.write_all(&jpeg).is_err()
- || tcp.write_all(b"\r\n").is_err()
- {
- return Ok(()); // HTTP client gone — unsubscribes by dropping the socket
+ /// The next frame. Err = the source is unavailable/broke (caller falls
+ /// back or gives up) — including a read timeout, which given the
+ /// keepalive cadence means a dead compositor, not an idle window.
+ pub fn next(&mut self) -> Result<crate::stream::Payload, String> {
+ let mut header = String::new();
+ loop {
+ header.clear();
+ if self.sock.read_line(&mut header).map_err(|e| e.to_string())? == 0 {
+ return Err("stream socket closed".into());
+ }
+ let mut it = header.split_ascii_whitespace();
+ if it.next() != Some("frame") {
+ continue;
+ }
+ let (Some(w), Some(h), Some(len)) = (
+ it.next().and_then(|v| v.parse::<u32>().ok()),
+ it.next().and_then(|v| v.parse::<u32>().ok()),
+ it.next().and_then(|v| v.parse::<u32>().ok()),
+ ) else {
+ return Err(format!("bad frame header: {header:?}"));
+ };
+ if len != w.saturating_mul(h).saturating_mul(4) || len > MAX_FRAME_BYTES {
+ return Err(format!("implausible frame: {header:?}"));
+ }
+ let mut rgba = vec![0u8; len as usize];
+ self.sock.read_exact(&mut rgba).map_err(|e| e.to_string())?;
+ return Ok(crate::stream::Payload::Raw {
+ data: rgba,
+ w,
+ h,
+ stride: w * 4,
+ rgb: (0, 1, 2),
+ });
}
}
}