diff --git a/Cargo.toml b/Cargo.toml index 1a029c9..11b62b8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -11,8 +11,15 @@ httparse = "1.9" libc = "0.2" sha1_smol = "1" +# dep #4, ratified 2026-06-12: serde/serde_json behind the opt-in +# "phoenix" feature only — the "channels" core stays dependency-free. +serde = { version = "1", optional = true } +serde_json = { version = "1", optional = true } + [features] smarm-trace = ["smarm/smarm-trace"] +channels = [] +phoenix = ["channels", "dep:serde", "dep:serde_json"] [dev-dependencies] serde = { version = "1", features = ["derive"] } diff --git a/src/channels/mod.rs b/src/channels/mod.rs new file mode 100644 index 0000000..5973e04 --- /dev/null +++ b/src/channels/mod.rs @@ -0,0 +1,1048 @@ +//! Channels — join/leave/event protocol over the WebSocket transport, +//! PubSub underneath (v0.6, design ratified 2026-06-12). +//! +//! Feature `"channels"`: this module — zero new dependencies. The wire +//! format is abstracted behind the [`Encode`]/[`Decode`] codec traits +//! (one method each); the `"phoenix"` feature supplies a phoenix.js- +//! compatible V2 JSON codec via serde, but nothing here knows about +//! JSON. +//! +//! # Topology +//! +//! - The app builds one [`ChannelHub`] (a [`TopicRouter`] + a +//! `PubSub>`). **In-runtime only** — `PubSub::new` +//! spawns the table actor — and the hub must be NON-static (the +//! `Arc>`-in-the-route-closure pattern from v0.5; +//! a `static` hub pins the pubsub table and hangs +//! `serve_with_shutdown`). +//! - [`ChannelHub::upgrade`] turns an HTTP `Conn` into a channel +//! socket: a [`WsHandler`] running in the connection actor that +//! decodes frames and routes them by topic. +//! - Each successful join spawns ONE channel actor per `(socket, +//! topic)`, linked to the connection actor (a channel panic tears +//! the socket down; the client reconnects — divergence from Phoenix, +//! where the socket survives, documented and accepted). The channel +//! actor owns the [`Channel`] impl, its [`ChannelSocket`], and its +//! pubsub subscription (pinned to the channel actor's pid, so the +//! monitor scopes cleanup exactly to the channel's life). +//! - The channel actor parks on `select(&[&inbound, &bus_rx])` — +//! inbound (decoded client frames forwarded by the conn actor) at +//! index 0, broadcasts at 1. When the socket dies the conn-side +//! handler drops, the inbound sender drops, the closed arm wakes the +//! actor, `terminate` runs, the actor exits. Every link in that +//! chain is in-runtime; no `PubSub`-clone keepalive cycle (the v0.5 +//! relay corollary does not bite here because the channel actor's +//! exit is conn-driven, never table-driven). +//! +//! # The join ack is first-class +//! +//! The v0.5 discovery (the 101 races `on_open`) is answered +//! structurally: the machinery subscribes the channel actor to its +//! topic **before** sending the ok reply, and `subscribe` is a call. +//! A client that has seen its join reply is guaranteed to be in the +//! broadcast table. +//! +//! # As-landed deviations from the ratified sketch (veto by diff) +//! +//! - `Channel::join` takes `&mut self`: a no-self method is uncallable +//! on a trait object, so construction moved to [`ChannelFactory`] +//! (which receives the full raw topic, preserving the +//! wildcard-extraction guarantee) and `join` became the +//! first-callback-after-construction. Signature is otherwise the +//! ratified `(topic, payload, socket) -> Result`. +//! - `ChannelSocket::reply(status, payload)` — the ratified text both +//! listed `ref` as a parameter and said the loop threads it +//! invisibly; invisible won (the socket carries the in-flight ref, +//! set by the loop around each `handle_in`). +//! - `ChannelSocket::broadcast` gained an `event` parameter — a push +//! without an event name cannot be encoded; sketch-level omission. +//! - `TopicRouter::route` returns `Option>` +//! (not `Box`): routers hand out shared factories without cloning +//! the factory itself per join. + +use crate::pubsub::PubSub; +use crate::ws::{Message, WsHandler, WsSender}; + +use smarm::{Pid, Receiver, Sender}; + +use std::cell::RefCell; +use std::collections::HashMap; +use std::marker::PhantomData; +use std::sync::Arc; + +#[cfg(feature = "phoenix")] +pub mod phoenix; + +// --------------------------------------------------------------------------- +// Wire-neutral frames +// --------------------------------------------------------------------------- + +/// Reply status — `ok` / `error` on the wire (spelling is the codec's +/// business). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Status { + Ok, + Error, +} + +/// A decoded inbound frame. `join_ref` identifies the join generation a +/// frame belongs to; `reference` correlates a reply to the request that +/// asked for it. Both are opaque client-chosen strings. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ChannelFrame

{ + pub join_ref: Option, + pub reference: Option, + pub topic: String, + pub message: ChannelMessage

, +} + +/// The message half of a decoded frame. Clients only ever send events; +/// `Reply` exists so a codec is symmetric (and is ignored inbound). +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ChannelMessage

{ + Event { event: String, payload: P }, + Reply { status: Status, payload: Option

}, +} + +/// A borrowed outbound frame — what [`Encode`] consumes. Borrowed so a +/// broadcast fan-out encodes straight from the shared `Arc>` +/// without cloning the payload (the zero-copy claim from the roadmap). +/// `payload: None` on an event encodes as the codec's empty payload +/// (`{}` in the phoenix codec) — server-side control events (`phx_close`) +/// have no payload to conjure for an arbitrary `P`. +#[derive(Debug, Clone, Copy)] +pub struct FrameRef<'a, P> { + pub join_ref: Option<&'a str>, + pub reference: Option<&'a str>, + pub topic: &'a str, + pub message: MessageRef<'a, P>, +} + +/// Borrowed message half of [`FrameRef`]. +#[derive(Debug, Clone, Copy)] +pub enum MessageRef<'a, P> { + Event { event: &'a str, payload: Option<&'a P> }, + Reply { status: Status, payload: Option<&'a P> }, +} + +/// Decoding failed; the socket is closed `1002` (a peer speaking the +/// wrong dialect is a protocol error, not a recoverable hiccup). +#[derive(Debug)] +pub struct DecodeError(pub String); + +impl std::fmt::Display for DecodeError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "channel frame decode error: {}", self.0) + } +} +impl std::error::Error for DecodeError {} + +/// Encode one outbound frame into a WebSocket message. One method — the +/// whole wire format, envelope included, lives behind it. +pub trait Encode: Sized { + fn encode(frame: FrameRef<'_, Self>) -> Message; +} + +/// Decode one inbound WebSocket message into a frame. One method. +pub trait Decode: Sized { + fn decode(msg: &Message) -> Result, DecodeError>; +} + +// Protocol event names (phoenix.js vocabulary — these are channel +// PROTOCOL, not JSON; a custom codec still speaks them, just spelled +// however it likes on the wire). +pub const EV_JOIN: &str = "phx_join"; +pub const EV_LEAVE: &str = "phx_leave"; +pub const EV_REPLY: &str = "phx_reply"; +pub const EV_CLOSE: &str = "phx_close"; +pub const EV_HEARTBEAT: &str = "heartbeat"; +/// The transport-control topic heartbeats arrive on. +pub const TOPIC_PHOENIX: &str = "phoenix"; + +// --------------------------------------------------------------------------- +// Channel trait + factory + router +// --------------------------------------------------------------------------- + +/// One joined topic on one socket. Runs in its own actor; callbacks are +/// sequential. A panic in any callback unwinds the channel actor and — +/// via the link — tears down the whole socket. +pub trait Channel: Send + 'static { + /// The client asked to join `topic`. `Ok(reply)` accepts (the reply + /// payload goes back with status ok, and the channel is live); + /// `Err(reply)` rejects (status error, the channel actor exits, + /// `terminate` is NOT called — the channel never existed). + /// + /// Don't call `socket.reply` here; the return value IS the reply. + /// `push`/`broadcast` are legal but note the subscription to this + /// topic only becomes live on `Ok` — a broadcast to your own topic + /// from inside `join` will not reach this socket. + fn join(&mut self, topic: &str, payload: P, socket: &ChannelSocket

) -> Result; + + /// A client event on the joined topic. Reply (correlated to this + /// event's ref) via [`ChannelSocket::reply`]; replying is optional. + fn handle_in(&mut self, event: &str, payload: P, socket: &ChannelSocket

); + + /// The channel is over: explicit leave, socket death, or server + /// shutdown. Not called after a rejected join or a callback panic. + fn terminate(&mut self) {} +} + +/// Builds a fresh [`Channel`] per accepted-routing join. Receives the +/// full raw topic string, so wildcard-segment extraction is always +/// possible (`"room:lobby"` arrives whole; pull `"lobby"` out yourself). +pub trait ChannelFactory: Send + Sync { + fn create(&self, topic: &str) -> Box>; +} + +impl ChannelFactory

for F +where + F: Fn(&str) -> Box> + Send + Sync, +{ + fn create(&self, topic: &str) -> Box> { + self(topic) + } +} + +/// [`ChannelFactory`] via `Default` — for channels whose state is built +/// in `join`. +pub struct DefaultFactory(PhantomData C>); + +impl + Default> ChannelFactory

for DefaultFactory { + fn create(&self, _topic: &str) -> Box> { + Box::new(C::default()) + } +} + +/// Maps a join's topic to the factory that builds its channel. `None` +/// rejects the join (status error). +pub trait TopicRouter: Send + Sync + 'static { + fn route(&self, topic: &str) -> Option>>; +} + +/// The shipped router: exact topics and `head:*` prefix patterns. +/// +/// Registration scans the pattern to `'*'` and discards the rest; +/// lookup is an exact-map probe, then one prefix-map probe on +/// `topic[..=first_colon]`. No regex, no glob machinery — patterns more +/// exotic than `"room:*"` mean implementing [`TopicRouter`] yourself. +pub struct PrefixRouter { + exact: HashMap>>, + prefix: HashMap>>, +} + +impl Default for PrefixRouter

{ + fn default() -> Self { + Self::new() + } +} + +impl PrefixRouter

{ + pub fn new() -> Self { + Self { exact: HashMap::new(), prefix: HashMap::new() } + } + + /// Register `factory` for `pattern` (`"room:*"` or an exact topic). + pub fn channel(mut self, pattern: &str, factory: impl ChannelFactory

+ 'static) -> Self { + match pattern.find('*') { + Some(i) => self.prefix.insert(pattern[..i].to_string(), Arc::new(factory)), + None => self.exact.insert(pattern.to_string(), Arc::new(factory)), + }; + self + } + + /// [`channel`](Self::channel) with a [`Default`]-built impl. + pub fn channel_default + Default>(self, pattern: &str) -> Self + where + P: 'static, + { + self.channel(pattern, DefaultFactory::(PhantomData)) + } +} + +impl TopicRouter

for PrefixRouter

{ + fn route(&self, topic: &str) -> Option>> { + if let Some(f) = self.exact.get(topic) { + return Some(f.clone()); + } + let head = topic.find(':').map(|i| &topic[..=i])?; + self.prefix.get(head).cloned() + } +} + +// --------------------------------------------------------------------------- +// Broadcast bus payload + ChannelSocket +// --------------------------------------------------------------------------- + +/// What rides the hub's pubsub: a named event for everyone joined to a +/// topic. `Arc>` end to end — one allocation per broadcast +/// regardless of subscriber count. +pub struct Broadcast

{ + pub event: String, + pub payload: P, +} + +/// The channel's handle to its transport: a [`WsSender`] for this +/// socket, the hub's pubsub for everyone else's, and the in-flight ref +/// the loop threads around each `handle_in` (invisible to the impl — +/// [`reply`](Self::reply) just works). +pub struct ChannelSocket { + ws: WsSender, + bus: PubSub>, + topic: String, + join_ref: Option, + cur_ref: RefCell>, +} + +impl ChannelSocket

{ + /// The full raw topic this channel is joined to. + pub fn topic(&self) -> &str { + &self.topic + } + + /// Reply to the event currently being handled. Correlated to that + /// event's ref; a refless event gets a refless reply (phoenix.js + /// drops those — harmless). + pub fn reply(&self, status: Status, payload: P) { + self.send_reply(status, Some(&payload)); + } + + /// Push an event to THIS socket only. + pub fn push(&self, event: &str, payload: P) { + let _ = self.ws.send(P::encode(FrameRef { + join_ref: self.join_ref.as_deref(), + reference: None, + topic: &self.topic, + message: MessageRef::Event { event, payload: Some(&payload) }, + })); + } + + /// Broadcast to every socket joined to this channel's topic, + /// including this one. + pub fn broadcast(&self, event: &str, payload: P) { + let _ = self + .bus + .broadcast(self.topic.clone(), Broadcast { event: event.to_string(), payload }); + } + + /// [`broadcast`](Self::broadcast), excluding this socket (the + /// no-echo shape — phoenix's `broadcast_from`). + pub fn broadcast_from(&self, event: &str, payload: P) { + let _ = self.bus.broadcast_from( + smarm::self_pid(), + self.topic.clone(), + Broadcast { event: event.to_string(), payload }, + ); + } + + /// Broadcast to an arbitrary topic (joined sockets there receive + /// it; this channel's own topic membership is irrelevant). + pub fn broadcast_to(&self, topic: &str, event: &str, payload: P) { + let _ = self + .bus + .broadcast(topic.to_string(), Broadcast { event: event.to_string(), payload }); + } + + fn send_reply(&self, status: Status, payload: Option<&P>) { + let r = self.cur_ref.borrow(); + let _ = self.ws.send(P::encode(FrameRef { + join_ref: self.join_ref.as_deref(), + reference: r.as_deref(), + topic: &self.topic, + message: MessageRef::Reply { status, payload }, + })); + } + + fn set_ref(&self, r: Option) { + *self.cur_ref.borrow_mut() = r; + } +} + +// --------------------------------------------------------------------------- +// ChannelHub — the app-facing entry point +// --------------------------------------------------------------------------- + +/// One per app (or per channel namespace): the router plus the pubsub +/// bus every channel broadcasts on. +/// +/// **In-runtime only** (spawns the pubsub table) and **must not live in +/// a `static`** — use the non-static `Arc>>` +/// captured by the route closure, exactly the v0.5 pubsub pattern, or +/// graceful shutdown will hang waiting for the table to exit. +pub struct ChannelHub { + bus: PubSub>, + router: Arc>, +} + +impl Clone for ChannelHub

{ + fn clone(&self) -> Self { + Self { bus: self.bus.clone(), router: self.router.clone() } + } +} + +impl ChannelHub

{ + pub fn new(router: impl TopicRouter

) -> Self { + Self { bus: PubSub::new(), router: Arc::new(router) } + } + + /// Accept the WebSocket upgrade on `conn` and speak channels over + /// it. The HTTP handshake-validation rules of + /// [`Conn::upgrade`](crate::Conn::upgrade) apply unchanged. + pub fn upgrade(&self, conn: crate::Conn) -> crate::Conn { + conn.upgrade(SocketHandler { + bus: self.bus.clone(), + router: self.router.clone(), + joined: HashMap::new(), + }) + } + + /// Server-side broadcast (an HTTP handler, a background actor): + /// every socket joined to `topic` gets the event. + pub fn broadcast(&self, topic: &str, event: &str, payload: P) { + let _ = self + .bus + .broadcast(topic.to_string(), Broadcast { event: event.to_string(), payload }); + } +} + +// --------------------------------------------------------------------------- +// The conn-side socket handler +// --------------------------------------------------------------------------- + +enum In

{ + Event { reference: Option, event: String, payload: P }, + Leave { reference: Option }, +} + +struct Joined

{ + join_ref: Option, + tx: Sender>, +} + +struct SocketHandler { + bus: PubSub>, + router: Arc>, + joined: HashMap>, +} + +impl SocketHandler

{ + fn reply_direct( + sender: &WsSender, + join_ref: Option<&str>, + reference: Option<&str>, + topic: &str, + status: Status, + ) { + let _ = sender.send(P::encode(FrameRef { + join_ref, + reference, + topic, + message: MessageRef::Reply { status, payload: None }, + })); + } + + fn close_event_direct(sender: &WsSender, join_ref: Option<&str>, topic: &str) { + let _ = sender.send(P::encode(FrameRef { + join_ref, + reference: None, + topic, + message: MessageRef::Event { event: EV_CLOSE, payload: None }, + })); + } +} + +impl WsHandler for SocketHandler

{ + fn on_message(&mut self, msg: Message, sender: &WsSender) { + let frame = match P::decode(&msg) { + Ok(f) => f, + Err(_) => { + // Wrong dialect: protocol error, not a per-message skip. + let _ = sender.close(1002, "channel frame decode error"); + return; + } + }; + let ChannelMessage::Event { event, payload } = frame.message else { + return; // clients don't reply; ignore + }; + + // Transport heartbeat — answered here, no channel involved. + if frame.topic == TOPIC_PHOENIX && event == EV_HEARTBEAT { + Self::reply_direct( + sender, + None, + frame.reference.as_deref(), + TOPIC_PHOENIX, + Status::Ok, + ); + return; + } + + match event.as_str() { + EV_JOIN => { + // A re-join replaces: the old actor's inbound sender + // drops (it exits via the closed arm, terminate runs) + // and the old join generation is told it's over. + if let Some(old) = self.joined.remove(&frame.topic) { + drop(old.tx); + Self::close_event_direct(sender, old.join_ref.as_deref(), &frame.topic); + } + let Some(factory) = self.router.route(&frame.topic) else { + Self::reply_direct( + sender, + frame.join_ref.as_deref(), + frame.reference.as_deref(), + &frame.topic, + Status::Error, + ); + return; + }; + let (tx, rx) = smarm::channel::>(); + let conn_pid = smarm::self_pid(); + let topic = frame.topic.clone(); + let join_ref = frame.join_ref.clone(); + let reference = frame.reference; + let ws = sender.clone(); + let bus = self.bus.clone(); + smarm::spawn(move || { + run_channel(conn_pid, factory, topic, join_ref, reference, payload, ws, bus, rx) + }); + self.joined + .insert(frame.topic, Joined { join_ref: frame.join_ref, tx }); + } + EV_LEAVE => match self.joined.remove(&frame.topic) { + Some(j) => { + if j.tx.send(In::Leave { reference: frame.reference.clone() }).is_err() { + // Actor already gone (rejected join, earlier + // death): leaving a dead channel is a success. + Self::reply_direct( + sender, + frame.join_ref.as_deref(), + frame.reference.as_deref(), + &frame.topic, + Status::Ok, + ); + } + } + None => Self::reply_direct( + sender, + frame.join_ref.as_deref(), + frame.reference.as_deref(), + &frame.topic, + Status::Error, + ), + }, + _ => { + let pruned_or_unjoined = match self.joined.get(&frame.topic) { + Some(j) => j + .tx + .send(In::Event { reference: frame.reference.clone(), event, payload }) + .is_err(), + None => true, + }; + if pruned_or_unjoined { + // Lazy prune (the pubsub discipline): a dead + // channel's entry goes on first send failure. + self.joined.remove(&frame.topic); + Self::reply_direct( + sender, + frame.join_ref.as_deref(), + frame.reference.as_deref(), + &frame.topic, + Status::Error, + ); + } + } + } + } + + // on_close: nothing explicit — this handler drops when the + // connection actor finishes (or unwinds), every inbound sender + // drops with it, and each channel actor wakes on its closed arm, + // runs terminate, and exits. The whole chain is in-runtime. +} + +// --------------------------------------------------------------------------- +// The channel actor +// --------------------------------------------------------------------------- + +#[allow(clippy::too_many_arguments)] +fn run_channel( + conn_pid: Pid, + factory: Arc>, + topic: String, + join_ref: Option, + reference: Option, + payload: P, + ws: WsSender, + bus: PubSub>, + inbound: Receiver>, +) { + // Linked, per the ratified design: a channel panic is a socket + // event (conn unwinds, every sibling channel follows through the + // sender-drop chain), and an abnormal conn death (drain-deadline + // force-stop) reaps channels immediately instead of waiting on the + // closed-arm wake. + smarm::link(conn_pid); + + let socket = ChannelSocket { + ws, + bus: bus.clone(), + topic: topic.clone(), + join_ref, + cur_ref: RefCell::new(None), + }; + let mut ch = factory.create(&topic); + + socket.set_ref(reference); + match ch.join(&topic, payload, &socket) { + Ok(reply) => { + // Subscribe BEFORE the ok reply: subscribe is a call, so + // once the client observes its join ack, its membership in + // the broadcast table is a fact, not a race (the v0.5 + // on_open discovery, answered structurally). + let bus_rx = match bus.subscribe(&topic) { + Ok(rx) => rx, + Err(_) => { + socket.send_reply(Status::Error, None); + return; + } + }; + socket.send_reply(Status::Ok, Some(&reply)); + socket.set_ref(None); + channel_loop(&mut *ch, &socket, &inbound, &bus_rx); + } + Err(reply) => { + // Rejected: error reply, actor exits, no terminate (the + // channel never existed). The conn-side entry self-prunes + // on its first failed send. + socket.send_reply(Status::Error, Some(&reply)); + } + } +} + +fn channel_loop( + ch: &mut dyn Channel

, + socket: &ChannelSocket

, + inbound: &Receiver>, + bus_rx: &Receiver>>, +) { + // A closed arm stays ready forever (it would starve the other arm + // under priority order) — drop the bus arm from the set once its + // disconnect has been observed. + let mut bus_alive = true; + loop { + let idx = if bus_alive { + smarm::select(&[inbound, bus_rx]) + } else { + smarm::select(&[inbound]) + }; + if idx == 0 { + match inbound.try_recv() { + Ok(Some(In::Event { reference, event, payload })) => { + socket.set_ref(reference); + ch.handle_in(&event, payload, socket); + socket.set_ref(None); + } + Ok(Some(In::Leave { reference })) => { + socket.set_ref(reference); + socket.send_reply(Status::Ok, None); + socket.set_ref(None); + // Generation over: tell the client, then go. The + // subscription dies with the actor (monitor). + let _ = socket.ws.send(P::encode(FrameRef { + join_ref: socket.join_ref.as_deref(), + reference: None, + topic: &socket.topic, + message: MessageRef::Event { event: EV_CLOSE, payload: None }, + })); + ch.terminate(); + return; + } + Ok(None) => {} + Err(_) => { + // Socket gone (conn actor finished or replaced this + // join generation). + ch.terminate(); + return; + } + } + } else { + match bus_rx.try_recv() { + Ok(Some(b)) => { + let out = P::encode(FrameRef { + join_ref: socket.join_ref.as_deref(), + reference: None, + topic: &socket.topic, + message: MessageRef::Event { event: &b.event, payload: Some(&b.payload) }, + }); + if socket.ws.send(out).is_err() { + ch.terminate(); + return; + } + } + Ok(None) => {} + Err(_) => bus_alive = false, // table down (shutdown tail) + } + } + } +} + +// --------------------------------------------------------------------------- +// Tests — runtime-backed, with a toy pipe-delimited codec (no JSON: the +// core must be exercisable without the phoenix feature). +// --------------------------------------------------------------------------- + +#[cfg(test)] +mod tests { + use super::*; + use crate::ws::duplex::Outbound; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::Mutex; + use std::time::Duration; + + /// Toy payload + codec: `jref|ref|topic|kind|event-or-status|payload` + /// with `-` for absent refs. Enough to drive the machinery. + #[derive(Debug, Clone, PartialEq, Eq)] + struct TP(String); + + fn part(o: Option<&str>) -> &str { + o.unwrap_or("-") + } + + impl Encode for TP { + fn encode(f: FrameRef<'_, Self>) -> Message { + let s = match f.message { + MessageRef::Event { event, payload } => format!( + "{}|{}|{}|event|{}|{}", + part(f.join_ref), + part(f.reference), + f.topic, + event, + payload.map(|p| p.0.as_str()).unwrap_or("") + ), + MessageRef::Reply { status, payload } => format!( + "{}|{}|{}|reply|{}|{}", + part(f.join_ref), + part(f.reference), + f.topic, + match status { + Status::Ok => "ok", + Status::Error => "error", + }, + payload.map(|p| p.0.as_str()).unwrap_or("") + ), + }; + Message::Text(s) + } + } + + impl Decode for TP { + fn decode(msg: &Message) -> Result, DecodeError> { + let Message::Text(s) = msg else { + return Err(DecodeError("binary".into())); + }; + let p: Vec<&str> = s.splitn(6, '|').collect(); + if p.len() != 6 || p[3] != "event" { + return Err(DecodeError(format!("bad frame: {s}"))); + } + let opt = |x: &str| (x != "-").then(|| x.to_string()); + Ok(ChannelFrame { + join_ref: opt(p[0]), + reference: opt(p[1]), + topic: p[2].to_string(), + message: ChannelMessage::Event { event: p[4].to_string(), payload: TP(p[5].into()) }, + }) + } + } + + /// A room channel: rejects topic "room:locked"; echoes events back as + /// replies; broadcasts on "shout"; flags terminate. + struct Room { + terminated: Arc, + } + + impl Channel for Room { + fn join(&mut self, topic: &str, payload: TP, _s: &ChannelSocket) -> Result { + if topic == "room:locked" { + Err(TP("no entry".into())) + } else { + Ok(TP(format!("welcome:{}", payload.0))) + } + } + fn handle_in(&mut self, event: &str, payload: TP, s: &ChannelSocket) { + match event { + "shout" => s.broadcast("shouted", payload), + "shout_from" => s.broadcast_from("shouted", payload), + _ => s.reply(Status::Ok, TP(format!("echo:{}:{}", event, payload.0))), + } + } + fn terminate(&mut self) { + self.terminated.store(true, Ordering::SeqCst); + } + } + + struct RoomFactory { + terminated: Arc, + } + impl ChannelFactory for RoomFactory { + fn create(&self, _topic: &str) -> Box> { + Box::new(Room { terminated: self.terminated.clone() }) + } + } + + /// A fake socket: the conn-side handler plus the raw Outbound channel + /// a real duplex loop would drain. `recv_text` pulls the next encoded + /// frame the "client" would see. + struct Harness { + handler: SocketHandler, + sender: WsSender, + out_rx: smarm::Receiver, + } + + impl Harness { + fn new(hub: &ChannelHub) -> Self { + let (tx, out_rx) = smarm::channel(); + Self { + handler: SocketHandler { + bus: hub.bus.clone(), + router: hub.router.clone(), + joined: HashMap::new(), + }, + sender: WsSender { tx }, + out_rx, + } + } + + /// Feed one wire line in, as the duplex loop would. + fn send(&mut self, line: &str) { + self.handler + .on_message(Message::Text(line.to_string()), &self.sender.clone()); + } + + /// Next encoded frame the client would see (2s budget — channel + /// actors answer asynchronously). + fn recv(&self) -> String { + match self.out_rx.recv_timeout(Duration::from_secs(2)) { + Ok(Outbound::Msg(Message::Text(s))) => s, + other => panic!("expected outbound text frame, got {:?}", other.is_ok()), + } + } + } + + fn hub(terminated: &Arc) -> ChannelHub { + ChannelHub::new( + PrefixRouter::new().channel("room:*", RoomFactory { terminated: terminated.clone() }), + ) + } + + fn eventually(mut f: impl FnMut() -> bool) -> bool { + for _ in 0..200 { + if f() { + return true; + } + smarm::sleep(Duration::from_millis(10)); + } + false + } + + #[test] + fn prefix_router_exact_and_prefix_and_miss() { + let t = Arc::new(AtomicBool::new(false)); + let r = PrefixRouter::new() + .channel("room:*", RoomFactory { terminated: t.clone() }) + .channel("lobby", RoomFactory { terminated: t.clone() }); + assert!(r.route("room:a").is_some()); + assert!(r.route("room:").is_some()); // empty wildcard segment still routes + assert!(r.route("lobby").is_some()); + assert!(r.route("lobbyx").is_none()); + assert!(r.route("hall:a").is_none()); + assert!(r.route("room").is_none()); // no colon, no exact hit + } + + #[test] + fn join_ok_replies_then_events_echo() { + let out = Arc::new(Mutex::new(Vec::::new())); + let (out2, t) = (out.clone(), Arc::new(AtomicBool::new(false))); + smarm::run(move || { + let hub = hub(&t); + let mut h = Harness::new(&hub); + h.send("j1|r1|room:a|event|phx_join|hi"); + out2.lock().unwrap().push(h.recv()); + h.send("j1|r2|room:a|event|ping|x"); + out2.lock().unwrap().push(h.recv()); + }); + let v = out.lock().unwrap().clone(); + assert_eq!(v[0], "j1|r1|room:a|reply|ok|welcome:hi"); + assert_eq!(v[1], "j1|r2|room:a|reply|ok|echo:ping:x"); + } + + #[test] + fn join_rejected_replies_error_no_terminate() { + let out = Arc::new(Mutex::new(String::new())); + let (out2, t) = (out.clone(), Arc::new(AtomicBool::new(false))); + let t2 = t.clone(); + smarm::run(move || { + let hub = hub(&t2); + let mut h = Harness::new(&hub); + h.send("j1|r1|room:locked|event|phx_join|hi"); + *out2.lock().unwrap() = h.recv(); + // The rejected channel never existed: terminate must not run. + smarm::sleep(Duration::from_millis(50)); + }); + assert_eq!(*out.lock().unwrap(), "j1|r1|room:locked|reply|error|no entry"); + assert!(!t.load(Ordering::SeqCst)); + } + + #[test] + fn unmatched_topic_join_replies_error() { + let out = Arc::new(Mutex::new(String::new())); + let (out2, t) = (out.clone(), Arc::new(AtomicBool::new(false))); + smarm::run(move || { + let hub = hub(&t); + let mut h = Harness::new(&hub); + h.send("j1|r1|hall:a|event|phx_join|hi"); + *out2.lock().unwrap() = h.recv(); + }); + assert_eq!(*out.lock().unwrap(), "j1|r1|hall:a|reply|error|"); + } + + #[test] + fn heartbeat_replies_ok_without_join() { + let out = Arc::new(Mutex::new(String::new())); + let (out2, t) = (out.clone(), Arc::new(AtomicBool::new(false))); + smarm::run(move || { + let hub = hub(&t); + let mut h = Harness::new(&hub); + h.send("-|hb1|phoenix|event|heartbeat|"); + *out2.lock().unwrap() = h.recv(); + }); + assert_eq!(*out.lock().unwrap(), "-|hb1|phoenix|reply|ok|"); + } + + #[test] + fn event_on_unjoined_topic_replies_error() { + let out = Arc::new(Mutex::new(String::new())); + let (out2, t) = (out.clone(), Arc::new(AtomicBool::new(false))); + smarm::run(move || { + let hub = hub(&t); + let mut h = Harness::new(&hub); + h.send("j1|r1|room:a|event|ping|x"); + *out2.lock().unwrap() = h.recv(); + }); + assert_eq!(*out.lock().unwrap(), "j1|r1|room:a|reply|error|"); + } + + #[test] + fn broadcast_reaches_both_sockets_broadcast_from_skips_sender() { + let got = Arc::new(Mutex::new(Vec::<(u8, String)>::new())); + let (got2, t) = (got.clone(), Arc::new(AtomicBool::new(false))); + smarm::run(move || { + let hub = hub(&t); + let mut a = Harness::new(&hub); + let mut b = Harness::new(&hub); + a.send("j1|r1|room:a|event|phx_join|A"); + b.send("j1|r1|room:a|event|phx_join|B"); + a.recv(); + b.recv(); + + a.send("j1|r2|room:a|event|shout|hello"); + got2.lock().unwrap().push((0, a.recv())); // broadcast includes sender + got2.lock().unwrap().push((1, b.recv())); + + a.send("j1|r3|room:a|event|shout_from|again"); + got2.lock().unwrap().push((1, b.recv())); // only b sees it + // a's next outbound must be the echo of a fresh ping, not a + // stray "again" — proves broadcast_from skipped a. + a.send("j1|r4|room:a|event|ping|x"); + got2.lock().unwrap().push((0, a.recv())); + }); + let v = got.lock().unwrap().clone(); + assert_eq!(v[0], (0, "j1|-|room:a|event|shouted|hello".into())); + assert_eq!(v[1], (1, "j1|-|room:a|event|shouted|hello".into())); + assert_eq!(v[2], (1, "j1|-|room:a|event|shouted|again".into())); + assert_eq!(v[3], (0, "j1|r4|room:a|reply|ok|echo:ping:x".into())); + } + + #[test] + fn leave_replies_ok_sends_close_and_terminates() { + let out = Arc::new(Mutex::new(Vec::::new())); + let (out2, t) = (out.clone(), Arc::new(AtomicBool::new(false))); + let t2 = t.clone(); + smarm::run(move || { + let hub = hub(&t2); + let mut h = Harness::new(&hub); + h.send("j1|r1|room:a|event|phx_join|hi"); + out2.lock().unwrap().push(h.recv()); + h.send("j1|r9|room:a|event|phx_leave|"); + out2.lock().unwrap().push(h.recv()); + out2.lock().unwrap().push(h.recv()); + assert!(eventually(|| t2.load(Ordering::SeqCst))); + // The conn-side entry is gone: a new event errors. + h.send("j1|r10|room:a|event|ping|x"); + out2.lock().unwrap().push(h.recv()); + }); + let v = out.lock().unwrap().clone(); + assert_eq!(v[1], "j1|r9|room:a|reply|ok|"); + assert_eq!(v[2], "j1|-|room:a|event|phx_close|"); + assert_eq!(v[3], "j1|r10|room:a|reply|error|"); + assert!(t.load(Ordering::SeqCst)); + } + + #[test] + fn socket_drop_terminates_channel_and_subscription() { + let t = Arc::new(AtomicBool::new(false)); + let t2 = t.clone(); + smarm::run(move || { + let hub = hub(&t2); + let bus = hub.bus.clone(); + { + let mut h = Harness::new(&hub); + h.send("j1|r1|room:a|event|phx_join|hi"); + h.recv(); + assert!(eventually(|| bus.subscriber_count("room:a").unwrap() == 1)); + } // handler drops -> inbound sender drops -> channel actor exits + assert!(eventually(|| t2.load(Ordering::SeqCst))); + assert!(eventually(|| bus.subscriber_count("room:a").unwrap() == 0)); + }); + assert!(t.load(Ordering::SeqCst)); + } + + #[test] + fn rejoin_replaces_old_generation() { + let out = Arc::new(Mutex::new(Vec::::new())); + let (out2, t) = (out.clone(), Arc::new(AtomicBool::new(false))); + let t2 = t.clone(); + smarm::run(move || { + let hub = hub(&t2); + let mut h = Harness::new(&hub); + h.send("j1|r1|room:a|event|phx_join|one"); + out2.lock().unwrap().push(h.recv()); + h.send("j2|r2|room:a|event|phx_join|two"); + // Old generation is closed (its actor terminates), then the + // new join acks. + out2.lock().unwrap().push(h.recv()); + out2.lock().unwrap().push(h.recv()); + assert!(eventually(|| t2.load(Ordering::SeqCst))); + h.send("j2|r3|room:a|event|ping|x"); + out2.lock().unwrap().push(h.recv()); + }); + let v = out.lock().unwrap().clone(); + assert_eq!(v[1], "j1|-|room:a|event|phx_close|"); + assert_eq!(v[2], "j2|r2|room:a|reply|ok|welcome:two"); + assert_eq!(v[3], "j2|r3|room:a|reply|ok|echo:ping:x"); + } + + #[test] + fn hub_broadcast_reaches_joined_socket() { + let out = Arc::new(Mutex::new(String::new())); + let (out2, t) = (out.clone(), Arc::new(AtomicBool::new(false))); + smarm::run(move || { + let hub = hub(&t); + let mut h = Harness::new(&hub); + h.send("j1|r1|room:a|event|phx_join|hi"); + h.recv(); + hub.broadcast("room:a", "news", TP("server says".into())); + *out2.lock().unwrap() = h.recv(); + }); + assert_eq!(*out.lock().unwrap(), "j1|-|room:a|event|news|server says"); + } +} diff --git a/src/lib.rs b/src/lib.rs index 8004bac..2806104 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -29,6 +29,8 @@ pub mod serve; pub mod sse; pub mod ws; pub mod pubsub; +#[cfg(feature = "channels")] +pub mod channels; // Re-exports — what most users want at the crate root. pub use conn::{Assigns, Body, Conn, HeaderMap, HttpVersion, Method, Params, RespBody, StreamBody}; @@ -36,6 +38,8 @@ pub use plug::{Next, Pipeline, Plug}; pub use router::Router; pub use sse::{EventSender, SseClosed}; pub use pubsub::{PubSub, PubSubDown}; +#[cfg(feature = "channels")] +pub use channels::{Channel, ChannelHub, ChannelSocket, PrefixRouter, Status, TopicRouter}; pub use ws::{Message, WsClosed, WsHandler, WsSender}; pub use serve::{ serve, serve_with, serve_with_shutdown, shutdown_handle, Config, Handle, ShutdownSignal,