pecord/tether
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

core: make the engine multi-transport (Transport trait + inbound dedup)

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
Patrick Ecord
2026-06-20 19:48:01 -05:00
parent 2c375632a5
commit 19f5eaccdf
1 changed files with 195 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files

189
core/src/lib.rs
View File

@@ -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<dyn Fn(Message) + Send + Sync>;
type Status = Arc<dyn Fn(bool) + Send + Sync>;
/// 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<Self>, rt: Handle, running: Arc<AtomicBool>, 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<tokio::runtime::Runtime>,
running: Arc<AtomicBool>,
transports: Vec<Arc<dyn Transport>>,
dedup: Arc<Mutex<Dedup>>,
}
#[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<Self> {
let rt = tokio::runtime::Builder::new_multi_thread()
@@ -82,60 +98,118 @@ 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<Arc<dyn Transport>> = 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<dyn MessageHandler>) {
if self.running.swap(true, Ordering::SeqCst) {
return;
}
let handler: Arc<dyn MessageHandler> = 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;
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(),
);
}
}
/// Publish clipboard text to the room. Fire-and-forget.
/// Publish clipboard text to the room across all transports.
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 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 streaming. The reconnect loop exits at its next checkpoint.
/// Stop all transports at their next checkpoint.
pub fn stop(&self) {
self.running.store(false, Ordering::SeqCst);
}
}
async fn run_loop(
http: reqwest::Client,
/// 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)>,
}
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
}
}
// ---------------------------------------------------------------------------
// SSE transport — the reliable floor: server relay over text/event-stream.
// ---------------------------------------------------------------------------
struct SseTransport {
cfg: Config,
running: Arc<AtomicBool>,
handler: Arc<dyn MessageHandler>,
) {
http: reqwest::Client,
}
impl Transport for SseTransport {
fn name(&self) -> &'static str {
"sse"
}
fn start(self: Arc<Self>, rt: Handle, running: Arc<AtomicBool>, 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 connect_stream(&http, &cfg, &running, &handler).await {
match this.stream_once(&running, &sink, &status).await {
Ok(()) => backoff = min, // clean EOF — reconnect promptly
Err(e) => {
handler.on_status(false);
eprintln!("tethercore: stream error: {e}");
status(false);
eprintln!("tethercore[{}]: {e}", this.name());
}
}
if running.load(Ordering::SeqCst) {
@@ -143,32 +217,47 @@ async fn run_loop(
backoff = (backoff * 2).min(max);
}
}
handler.on_status(false);
status(false);
});
}
async fn connect_stream(
http: &reqwest::Client,
cfg: &Config,
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,
handler: &Arc<dyn MessageHandler>,
sink: &Sink,
status: &Status,
) -> Result<(), String> {
let url = format!(
"{}/api/stream?room={}",
cfg.server.trim_end_matches('/'),
cfg.room
self.cfg.server.trim_end_matches('/'),
self.cfg.room
);
let resp = http
let resp = self
.http
.get(&url)
.header("X-Tether-Client", &cfg.from)
.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()));
}
handler.on_status(true);
status(true);
// Byte-buffer the SSE stream; lines split on '\n' (ASCII, so chunk-safe).
// Byte-buffer the stream; lines split on '\n' (ASCII, so chunk-safe).
let mut stream = resp.bytes_stream();
let mut buf: Vec<u8> = Vec::new();
let mut data: Vec<u8> = Vec::new();
@@ -184,11 +273,10 @@ async fn connect_stream(
line.pop();
}
if line.is_empty() {
// Event boundary: flush accumulated data field.
if !data.is_empty() {
if let Ok(m) = serde_json::from_slice::<Message>(&data) {
if m.from != cfg.from {
handler.on_message(m);
if m.from != self.cfg.from {
sink(m); // engine de-dups + forwards to handler
}
}
data.clear();
@@ -202,6 +290,7 @@ async fn connect_stream(
}
Ok(())
}
}
async fn post(http: &reqwest::Client, cfg: &Config, text: &str) -> Result<(), String> {
let url = format!("{}/api/send", cfg.server.trim_end_matches('/'));