docs(channels): example, README section, ROADMAP v0.6 mark-done — chunk 4

examples/channels_chat.rs (required-features phoenix): ephemeral room:* and persistent session:* side by side over phoenix.js V2 JSON, stdin-driven graceful shutdown (the ws_chat pattern). Smoke-tested live: rejoin after transport drop lands on the same instance (state counter persists across three transports), broadcasts carry the new join generation's ref. README: channels section (feature split, core trait walkthrough, session persistence semantics incl. the re-entrant-join contract). ROADMAP: v0.6 marked done with the full as-landed decision list, chunked as committed.
2026-06-12 15:26:21 +00:00
parent 0de2baa72d
commit 86c8d31b93
4 changed files with 225 additions and 4 deletions
@@ -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"]
@@ -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<T>` newtype.
+
+```rust
+use urus::{Channel, ChannelHub, ChannelSocket, PrefixRouter, Status};
+use urus::channels::phoenix::Json;
+use serde_json::{json, Value};
+
+type P = Json<Value>;
+
+#[derive(Default)]
+struct Room;
+
+impl Channel<P> for Room {
+    fn join(&mut self, topic: &str, payload: P, _s: &ChannelSocket<P>) -> Result<P, P> {
+        Ok(Json(json!({ "joined": topic })))   // Err(..) rejects
+    }
+    fn handle_in(&mut self, event: &str, payload: P, s: &ChannelSocket<P>) {
+        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>("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<P> 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::<ByToken>("session:*", |_: &str| {
+    Box::new(Room::default()) as Box<dyn Channel<P>>
+})
+```
+
+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.

@@ -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<Arc<dyn
+  ChannelFactory>>`.
+- chunk 2: the serde blanket lives behind the `Json<T>` 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
@@ -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<OnceLock<...>>` 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<T>` is the phoenix codec's opt-in newtype.** The
+//!   `"channels"` core is wire-neutral (`Encode`/`Decode`, zero deps);
+//!   `Json<Value>` here buys phoenix.js V2 interop. Typed payloads work
+//!   the same way: `Json<MyPayload>` 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<Value>;
+
+/// One chat room. State lives across callbacks (and, for `session:*`
+/// topics, across transports).
+#[derive(Default)]
+struct Room {
+    name: String,
+    msgs_seen: u64,
+}
+
+impl Channel<P> for Room {
+    fn join(&mut self, topic: &str, payload: P, _s: &ChannelSocket<P>) -> Result<P, P> {
+        // 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<P>) {
+        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<P> 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<OnceLock<ChannelHub<P>>> = 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>("room:*")
+                    .channel_session::<BySessionName>("session:*", |_: &str| {
+                        Box::new(Room::default()) as Box<dyn Channel<P>>
+                    }),
+            )
+        });
+        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");
+}