diff --git a/Cargo.toml b/Cargo.toml index 11b62b8..cb48792 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -36,3 +36,8 @@ codegen-units = 1 [[example]] name = "crud" path = "examples/crud.rs" + +[[example]] +name = "channels_chat" +path = "examples/channels_chat.rs" +required-features = ["phoenix"] diff --git a/README.md b/README.md index 8c53e38..6dc71fa 100644 --- a/README.md +++ b/README.md @@ -343,6 +343,87 @@ See [`examples/ws_chat.rs`](examples/ws_chat.rs): rooms as topics, `on_open` subscribes + spawns the relay, `on_message` uses `broadcast_from` so the speaker isn't echoed. +## Channels (v0.6) + +Phoenix-style join/leave/event multiplexing over the WebSocket duplex, +pubsub underneath. Two additive features: + +- `channels` — the wire-neutral core (`Channel`, `ChannelHub`, + `PrefixRouter`, `ChannelSocket`, sessions). Zero new dependencies; + bring your own codec by implementing `Encode`/`Decode` for your + payload type. +- `phoenix` — implies `channels`, adds serde/serde_json and the + phoenix.js V2 wire format via the `Json` newtype. + +```rust +use urus::{Channel, ChannelHub, ChannelSocket, PrefixRouter, Status}; +use urus::channels::phoenix::Json; +use serde_json::{json, Value}; + +type P = Json; + +#[derive(Default)] +struct Room; + +impl Channel

for Room { + fn join(&mut self, topic: &str, payload: P, _s: &ChannelSocket

) -> Result { + Ok(Json(json!({ "joined": topic }))) // Err(..) rejects + } + fn handle_in(&mut self, event: &str, payload: P, s: &ChannelSocket

) { + match event { + "shout" => s.broadcast("shouted", payload), + _ => s.reply(Status::Ok, Json(json!({ "echo": event }))), + } + } +} + +// in-runtime, non-static — the ws_chat OnceLock pattern applies +let hub = ChannelHub::new(PrefixRouter::new().channel_default::("room:*")); +// route handler: hub.upgrade(conn) +// from anywhere with a hub handle: hub.broadcast("room:lobby", "news", payload) +``` + +One channel actor per joined topic per socket, linked to the +connection: callbacks are sequential, a channel panic is a socket +event, and the conn-side handler answers transport heartbeats and +routes joins — `Channel` impls never see frames or refs (`reply` +correlates to the in-flight event invisibly). + +**Session persistence (opt-in).** By default channels cold-start per +join, Phoenix-server style. Implementing `ChannelSession` and +registering with `channel_session` makes the channel actor outlive its +transport: broadcasts that arrive while detached are buffered (the same +`Arc`s the relay path carries — zero re-serialisation) and drained, in +order, under the rejoin's new generation. Buffer cap or TTL exceeded, +explicit leave, or a rejected rejoin tears the session down; the next +join is a cold start. A second transport claiming the same key evicts +the first. + +```rust +struct ByToken; +impl urus::ChannelSession

for ByToken { + type Key = String; + fn session_key(topic: &str, join_payload: &P) -> String { + format!("{topic}/{}", join_payload.0["token"].as_str().unwrap_or("")) + } + // fn buffer_cap() -> usize { 128 } // defaults + // fn ttl() -> Duration { Duration::from_secs(30) } +} + +PrefixRouter::new().channel_session::("session:*", |_: &str| { + Box::new(Room::default()) as Box> +}) +``` + +Session channels must treat `join` as re-entrant: every reattach calls +it again on the same instance (the rejoin ack needs a payload, and the +channel gets to re-auth). + +See [`examples/channels_chat.rs`](examples/channels_chat.rs) +(`cargo run --example channels_chat --features phoenix`): ephemeral +`room:*` and persistent `session:*` side by side over phoenix.js V2 +JSON. + ## Examples ### CRUD with Actor Ownership @@ -414,9 +495,9 @@ This avoids the complexity of async ecosystems while maintaining full concurrenc See [`ROADMAP.md`](ROADMAP.md). Summary: v0.2 (supervised listener pool, graceful shutdown, enforced timeouts, registry names), v0.3 (streaming -bodies, chunked requests, SSE) and v0.4 (WebSocket: handshake, frame -codec, duplex handler API) are done; next up are PubSub (v0.5) and -Phoenix-style channels (v0.6). +bodies, chunked requests, SSE), v0.4 (WebSocket: handshake, frame +codec, duplex handler API), v0.5 (PubSub) and v0.6 (Phoenix-style +channels with opt-in session persistence) are done. Refer to `urus-spec.md` and `urus-v1-build-notes.md` in the artifact persistence for the original design and implementation notes. diff --git a/ROADMAP.md b/ROADMAP.md index a68598f..81c8deb 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -263,7 +263,7 @@ smarm has none). All six design questions ratified by the user pre-code Chunk 2's ws_chat: rooms as topics, `on_open` subscribes (conn-actor pid) + spawns the relay, `broadcast_from(self_pid(), ..)` for no-echo. -## v0.6 — Channels +## v0.6 — Channels — DONE (2026-06-12) Join/leave/event protocol over WebSocket transport, PubSub underneath. Design decisions ratified pre-code (2026-06-12): @@ -329,6 +329,24 @@ reconnect cycles. **Presence:** explicitly out of scope until distribution exists. +**As-landed (2026-06-12), beyond the ratified text — veto by diff:** +- chunk 1: `Channel::join` takes `&mut self` (state from join, and + session rejoins need the same instance); the in-flight ref is + threaded invisibly through the socket; `broadcast` gained an `event` + parameter; `TopicRouter::route` returns `Option>`. +- chunk 2: the serde blanket lives behind the `Json` newtype + (a blanket on bare `P: Serialize` is coherence poison under additive + features); encode failure panics into the linked-teardown contract. +- chunk 3: `ChannelFactory` gained a `#[doc(hidden)] deploy()` seam + (default = the ephemeral linked actor); every attach re-calls + `join()` (re-entrant by contract); rejected rejoin / explicit leave / + cap / TTL all end the session; second transport evicts the first; + `Key: Clone` (registry's pid→key reverse index); registry spawns + eagerly at registration (in-runtime law, same as `ChannelHub::new`); + shutdown composes linklessly via the registry-drop chain — proven by + `shutdown_with_detached_session_terminates`. + --- ## Known bugs diff --git a/examples/channels_chat.rs b/examples/channels_chat.rs new file mode 100644 index 0000000..fa96dd6 --- /dev/null +++ b/examples/channels_chat.rs @@ -0,0 +1,117 @@ +//! Phoenix-style channels (v0.6): join/leave/event protocol over the +//! v0.4 WebSocket duplex, pubsub underneath, phoenix V2 JSON on the +//! wire. +//! +//! cargo run --example channels_chat --features phoenix +//! websocat ws://127.0.0.1:8080/socket (run two of these) +//! ["1","1","room:lobby","phx_join",{"name":"alice"}] +//! ["1","2","room:lobby","shout",{"text":"hi"}] +//! ["1","3","room:lobby","phx_leave",{}] +//! +//! The shapes this demonstrates: +//! +//! - **One `ChannelHub` for the whole app**, built lazily on the first +//! connection via the NON-static `Arc>` pattern (the +//! hub spawns the pubsub table; same in-runtime + shutdown laws as +//! `examples/ws_chat.rs`, see the docs there). +//! +//! - **A channel per joined topic, not per socket.** `Room` never sees +//! frames, refs, or the transport heartbeat — the conn-side handler +//! answers heartbeats, routes `phx_join`, and the per-join channel +//! actor (linked to the connection) runs the callbacks sequentially. +//! +//! - **`Json` is the phoenix codec's opt-in newtype.** The +//! `"channels"` core is wire-neutral (`Encode`/`Decode`, zero deps); +//! `Json` here buys phoenix.js V2 interop. Typed payloads work +//! the same way: `Json` for any serde type. +//! +//! - **`session::*` topics persist across reconnects** (opt-in via +//! `channel_session`): drop a websocat mid-room, reconnect and rejoin +//! within 30s, and the broadcasts you missed drain to you in order — +//! the buffer holds the relay path's own `Arc`s, nothing is copied. +//! The session is keyed by the `"name"` in the join payload. + +use std::sync::{Arc, OnceLock}; + +use serde_json::{json, Value}; +use urus::channels::phoenix::Json; +use urus::{ + serve_with_shutdown, shutdown_handle, Channel, ChannelHub, ChannelSession, ChannelSocket, + Config, Conn, Next, Pipeline, PrefixRouter, Router, Status, +}; + +type P = Json; + +/// One chat room. State lives across callbacks (and, for `session:*` +/// topics, across transports). +#[derive(Default)] +struct Room { + name: String, + msgs_seen: u64, +} + +impl Channel

for Room { + fn join(&mut self, topic: &str, payload: P, _s: &ChannelSocket

) -> Result { + // Re-entrant by contract for session topics: a rejoin lands + // here again on the same instance — re-auth, then ack. + self.name = payload.0["name"].as_str().unwrap_or("anon").to_string(); + Ok(Json(json!({ "joined": topic, "as": self.name, "seen": self.msgs_seen }))) + } + + fn handle_in(&mut self, event: &str, payload: P, s: &ChannelSocket

) { + match event { + "shout" => { + self.msgs_seen += 1; + s.broadcast("shouted", Json(json!({ "from": self.name, "msg": payload.0 }))); + } + _ => s.reply(Status::Ok, Json(json!({ "echo": event }))), + } + } + + fn terminate(&mut self) { + println!("room channel for {:?} is over", self.name); + } +} + +/// Session config for `session:*`: the join payload's `"name"` is the +/// session token. Defaults: 128 buffered broadcasts, 30s TTL. +struct BySessionName; + +impl ChannelSession

for BySessionName { + type Key = String; + fn session_key(topic: &str, join_payload: &P) -> String { + format!("{topic}/{}", join_payload.0["name"].as_str().unwrap_or("anon")) + } +} + +fn main() { + let hub: Arc>> = Arc::new(OnceLock::new()); + + let router = Router::new().get("/socket", move |c: Conn, _n: Next| { + // Hub construction is in-runtime only (it spawns the pubsub + // table and, for session patterns, their registries); the + // non-static cell is what lets shutdown drain them all. + let hub = hub.get_or_init(|| { + ChannelHub::new( + PrefixRouter::new() + .channel_default::("room:*") + .channel_session::("session:*", |_: &str| { + Box::new(Room::default()) as Box> + }), + ) + }); + hub.upgrade(c) + }); + + let (handle, signal) = shutdown_handle(); + std::thread::spawn(move || { + println!("channels_chat on ws://127.0.0.1:8080/socket — press Enter to shut down"); + let mut line = String::new(); + let _ = std::io::stdin().read_line(&mut line); + handle.shutdown(); + }); + + serve_with_shutdown(Config::new("0.0.0.0:8080".parse().unwrap()), Pipeline::new().plug(router), signal) + .unwrap(); + println!("channels_chat: drained, bye"); +}