From 19f5eaccdfaea69a4cb0b846812d4ecb7d6a9bc3 Mon Sep 17 00:00:00 2001 From: Patrick Ecord Date: Sat, 20 Jun 2026 19:48:01 -0500 Subject: [PATCH] core: make the engine multi-transport (Transport trait + inbound dedup) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Restructure tethercore so the engine owns a list of pluggable Transports instead of a hardcoded SSE path. SSE moves behind the trait unchanged; RTC (webrtc-rs data channel) and BT (BLE GATT) now have a clean socket to implement the same interface — start(sink, status) / publish(msg). The engine multiplexes outbound across all transports and de-dups inbound (keyed on from+ts+text) so a clipboard echoed over two channels surfaces once. No-op with SSE alone; required the moment RTC lands. FFI surface is unchanged (Engine::new / start / send / stop, Message, MessageHandler) — no binding regen needed. Round-trip example still green. Co-Authored-By: Claude Opus 4.8 --- core/src/lib.rs | 301 +++++++++++++++++++++++++++++++----------------- 1 file changed, 195 insertions(+), 106 deletions(-) diff --git a/core/src/lib.rs b/core/src/lib.rs index 896d991..267b67e 100644 --- a/core/src/lib.rs +++ b/core/src/lib.rs @@ -1,23 +1,26 @@ //! tethercore — the shared sync engine. //! -//! Owns the wire protocol (codec + SSE client + reconnect/backoff + send) and -//! nothing else: no clipboard, no UI. Those stay native on each platform and -//! talk to this engine through a 4-method seam exported via UniFFI: +//! Owns the wire protocol (codec + transports + reconnect/backoff + send) and +//! nothing else: no clipboard, no UI. Exposed via UniFFI behind a 4-method seam: //! //! Engine::new(server, room, from, source) //! engine.start(handler) // inbound clipboard → handler.on_message //! engine.send(text) //! engine.stop() //! -//! Desktop links it directly; iOS binds the .xcframework, Android the .aar. -//! RTC mesh (the `signal` envelope) is intentionally out of scope for v1. +//! Transports are pluggable (see `Transport`). SSE ships today; RTC and BT slot +//! in beside it without touching the engine. The engine multiplexes outbound +//! across all transports and de-duplicates inbound so the same clipboard +//! arriving over two channels surfaces once. +use std::collections::VecDeque; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::Arc; +use std::sync::{Arc, Mutex}; use std::time::Duration; use futures_util::StreamExt; use serde::{Deserialize, Serialize}; +use tokio::runtime::Handle; uniffi::setup_scaffolding!(); @@ -44,13 +47,11 @@ pub struct Message { } /// Implemented on the foreign side (Swift/Kotlin) or natively (desktop agent). -/// All callbacks fire on the engine's runtime thread — marshal to the UI thread +/// Callbacks fire on the engine's runtime thread — marshal to the UI thread /// on the consumer side. #[uniffi::export(callback_interface)] pub trait MessageHandler: Send + Sync { - /// A clipboard message from another device in the room (self-echo filtered). fn on_message(&self, msg: Message); - /// Connection state changed: true once subscribed, false on drop/error. fn on_status(&self, connected: bool); } @@ -62,19 +63,34 @@ struct Config { source: String, } +/// Sink/Status closures a transport calls for inbound traffic and link state. +type Sink = Arc; +type Status = Arc; + +/// A pluggable delivery channel. The engine owns one or more; today just SSE. +/// RTC (WebRTC data channel) and BT (BLE GATT) implement the same trait. +trait Transport: Send + Sync { + fn name(&self) -> &'static str; + /// Spawn the inbound loop on `rt`; deliver received messages via `sink` + /// and link transitions via `status`. Runs until `running` clears. + fn start(self: Arc, rt: Handle, running: Arc, sink: Sink, status: Status); + /// Publish an outbound message. Fire-and-forget. + fn publish(&self, rt: &Handle, msg: Message); +} + #[derive(uniffi::Object)] pub struct Engine { cfg: Config, - http: reqwest::Client, rt: Arc, running: Arc, + transports: Vec>, + dedup: Arc>, } #[uniffi::export] impl Engine { /// `server` is the bus base URL, e.g. "http://192.168.11.233:8765". - /// `from` is this device's stable id (used for echo suppression); - /// `source` is a human label like "macos" / "iphone". + /// `from` is this device's stable id (echo suppression); `source` a label. #[uniffi::constructor] pub fn new(server: String, room: String, from: String, source: String) -> Arc { let rt = tokio::runtime::Builder::new_multi_thread() @@ -82,125 +98,198 @@ impl Engine { .enable_all() .build() .expect("tokio runtime"); - Arc::new(Self { - cfg: Config { server, room, from, source }, + let cfg = Config { server, room, from, source }; + let transports: Vec> = vec![Arc::new(SseTransport { + cfg: cfg.clone(), http: reqwest::Client::new(), + })]; + Arc::new(Self { + cfg, rt: Arc::new(rt), running: Arc::new(AtomicBool::new(false)), + transports, + dedup: Arc::new(Mutex::new(Dedup::default())), }) } - /// Begin streaming. Idempotent: a second call while running is a no-op. + /// Begin streaming on every transport. Idempotent while already running. pub fn start(&self, handler: Box) { if self.running.swap(true, Ordering::SeqCst) { return; } let handler: Arc = Arc::from(handler); - let http = self.http.clone(); - let cfg = self.cfg.clone(); - let running = self.running.clone(); - self.rt.spawn(async move { - run_loop(http, cfg, running, handler).await; - }); - } - - /// Publish clipboard text to the room. Fire-and-forget. - pub fn send(&self, text: String) { - let http = self.http.clone(); - let cfg = self.cfg.clone(); - self.rt.spawn(async move { - if let Err(e) = post(&http, &cfg, &text).await { - eprintln!("tethercore: send failed: {e}"); + let dedup = self.dedup.clone(); + let h_msg = handler.clone(); + let sink: Sink = Arc::new(move |m| { + if dedup.lock().unwrap().seen(&m) { + return; // same clipboard already delivered via another transport } + h_msg.on_message(m); }); + let h_status = handler.clone(); + let status: Status = Arc::new(move |c| h_status.on_status(c)); + + for t in &self.transports { + t.clone().start( + self.rt.handle().clone(), + self.running.clone(), + sink.clone(), + status.clone(), + ); + } } - /// Stop streaming. The reconnect loop exits at its next checkpoint. + /// Publish clipboard text to the room across all transports. + pub fn send(&self, text: String) { + let msg = Message { + kind: "clipboard".into(), + text, + from: self.cfg.from.clone(), + to: String::new(), + role: String::new(), + source: self.cfg.source.clone(), + room: self.cfg.room.clone(), + ts: 0, // server stamps authoritative ts + }; + let rt = self.rt.handle(); + for t in &self.transports { + t.publish(rt, msg.clone()); + } + } + + /// Stop all transports at their next checkpoint. pub fn stop(&self) { self.running.store(false, Ordering::SeqCst); } } -async fn run_loop( - http: reqwest::Client, - cfg: Config, - running: Arc, - handler: Arc, -) { - let min = Duration::from_millis(500); - let max = Duration::from_secs(10); - let mut backoff = min; - while running.load(Ordering::SeqCst) { - match connect_stream(&http, &cfg, &running, &handler).await { - Ok(()) => backoff = min, // clean EOF — reconnect promptly - Err(e) => { - handler.on_status(false); - eprintln!("tethercore: stream error: {e}"); - } - } - if running.load(Ordering::SeqCst) { - tokio::time::sleep(backoff).await; - backoff = (backoff * 2).min(max); - } - } - handler.on_status(false); +/// Short-window de-dup keyed on the fields the server preserves. Once RTC lands, +/// a clipboard echoed over both SSE and the data channel collapses to one. +#[derive(Default)] +struct Dedup { + recent: VecDeque<(String, i64, String)>, } -async fn connect_stream( - http: &reqwest::Client, - cfg: &Config, - running: &AtomicBool, - handler: &Arc, -) -> Result<(), String> { - let url = format!( - "{}/api/stream?room={}", - cfg.server.trim_end_matches('/'), - cfg.room - ); - let resp = http - .get(&url) - .header("X-Tether-Client", &cfg.from) - .send() - .await - .map_err(|e| e.to_string())?; - if !resp.status().is_success() { - return Err(format!("HTTP {}", resp.status())); +impl Dedup { + fn seen(&mut self, m: &Message) -> bool { + let key = (m.from.clone(), m.ts, m.text.clone()); + if self.recent.contains(&key) { + return true; + } + self.recent.push_back(key); + if self.recent.len() > 256 { + self.recent.pop_front(); + } + false } - handler.on_status(true); +} - // Byte-buffer the SSE stream; lines split on '\n' (ASCII, so chunk-safe). - let mut stream = resp.bytes_stream(); - let mut buf: Vec = Vec::new(); - let mut data: Vec = Vec::new(); - while let Some(chunk) = stream.next().await { - if !running.load(Ordering::SeqCst) { - return Ok(()); - } - buf.extend_from_slice(&chunk.map_err(|e| e.to_string())?); - while let Some(nl) = buf.iter().position(|&b| b == b'\n') { - let mut line: Vec = buf.drain(..=nl).collect(); - line.pop(); // '\n' - if line.last() == Some(&b'\r') { - line.pop(); - } - if line.is_empty() { - // Event boundary: flush accumulated data field. - if !data.is_empty() { - if let Ok(m) = serde_json::from_slice::(&data) { - if m.from != cfg.from { - handler.on_message(m); - } - } - data.clear(); - } - } else if let Some(rest) = line.strip_prefix(b"data:") { - let rest = rest.strip_prefix(b" ").unwrap_or(rest); - data.extend_from_slice(rest); - } - // ':' comments (keepalives) and event:/id: lines are ignored. - } +// --------------------------------------------------------------------------- +// SSE transport — the reliable floor: server relay over text/event-stream. +// --------------------------------------------------------------------------- + +struct SseTransport { + cfg: Config, + http: reqwest::Client, +} + +impl Transport for SseTransport { + fn name(&self) -> &'static str { + "sse" + } + + fn start(self: Arc, rt: Handle, running: Arc, sink: Sink, status: Status) { + let this = self; + rt.spawn(async move { + let min = Duration::from_millis(500); + let max = Duration::from_secs(10); + let mut backoff = min; + while running.load(Ordering::SeqCst) { + match this.stream_once(&running, &sink, &status).await { + Ok(()) => backoff = min, // clean EOF — reconnect promptly + Err(e) => { + status(false); + eprintln!("tethercore[{}]: {e}", this.name()); + } + } + if running.load(Ordering::SeqCst) { + tokio::time::sleep(backoff).await; + backoff = (backoff * 2).min(max); + } + } + status(false); + }); + } + + fn publish(&self, rt: &Handle, msg: Message) { + let http = self.http.clone(); + let cfg = self.cfg.clone(); + let name = self.name(); + rt.spawn(async move { + if let Err(e) = post(&http, &cfg, &msg.text).await { + eprintln!("tethercore[{name}]: send failed: {e}"); + } + }); + } +} + +impl SseTransport { + async fn stream_once( + &self, + running: &AtomicBool, + sink: &Sink, + status: &Status, + ) -> Result<(), String> { + let url = format!( + "{}/api/stream?room={}", + self.cfg.server.trim_end_matches('/'), + self.cfg.room + ); + let resp = self + .http + .get(&url) + .header("X-Tether-Client", &self.cfg.from) + .send() + .await + .map_err(|e| e.to_string())?; + if !resp.status().is_success() { + return Err(format!("HTTP {}", resp.status())); + } + status(true); + + // Byte-buffer the stream; lines split on '\n' (ASCII, so chunk-safe). + let mut stream = resp.bytes_stream(); + let mut buf: Vec = Vec::new(); + let mut data: Vec = Vec::new(); + while let Some(chunk) = stream.next().await { + if !running.load(Ordering::SeqCst) { + return Ok(()); + } + buf.extend_from_slice(&chunk.map_err(|e| e.to_string())?); + while let Some(nl) = buf.iter().position(|&b| b == b'\n') { + let mut line: Vec = buf.drain(..=nl).collect(); + line.pop(); // '\n' + if line.last() == Some(&b'\r') { + line.pop(); + } + if line.is_empty() { + if !data.is_empty() { + if let Ok(m) = serde_json::from_slice::(&data) { + if m.from != self.cfg.from { + sink(m); // engine de-dups + forwards to handler + } + } + data.clear(); + } + } else if let Some(rest) = line.strip_prefix(b"data:") { + let rest = rest.strip_prefix(b" ").unwrap_or(rest); + data.extend_from_slice(rest); + } + // ':' comments (keepalives) and event:/id: lines are ignored. + } + } + Ok(()) } - Ok(()) } async fn post(http: &reqwest::Client, cfg: &Config, text: &str) -> Result<(), String> {