diff --git a/src/channels/phoenix.rs b/src/channels/phoenix.rs new file mode 100644 index 0000000..88eb7c8 --- /dev/null +++ b/src/channels/phoenix.rs @@ -0,0 +1,223 @@ +//! Phoenix V2 wire codec — feature `"phoenix"` (implies `"channels"`, +//! spends dependency slot #4 on `serde`/`serde_json`, ratified +//! 2026-06-12). +//! +//! The wire format phoenix.js speaks: one JSON array per frame, +//! +//! ```text +//! [join_ref, ref, topic, event, payload] +//! ``` +//! +//! refs are strings or `null`; replies are `event: "phx_reply"` with +//! `payload: {"status": "ok"|"error", "response": }`. +//! +//! # `Json`, not a blanket impl on `P` +//! +//! The ratified sketch said "JsonCodec blanket impl via serde". A +//! literal `impl Encode for P` is +//! coherence poison: features are additive across a dependency graph, +//! so the moment ANY crate enables `"phoenix"`, every hand-written +//! `Encode` impl in every other crate stops compiling (the compiler +//! cannot prove a foreign type will never gain `Serialize`). That +//! breaks the design's own promise that `"channels"`-only users supply +//! their own codecs. The blanket therefore lives one newtype away: +//! [`Json`] is V2-JSON for ANY serde-capable `T` — including +//! `Json` for schemaless payloads — while the +//! `Encode`/`Decode` namespace stays open. +//! +//! # Strictness +//! +//! Decoding is exactly as strict as `T`'s `Deserialize`: a phoenix.js +//! `{}` join payload fails against a `T` with required fields (the +//! socket closes `1002`). Use `#[serde(default)]`, `Option` fields, or +//! `Json` where the client is loose. Encoding a `T` +//! that fails `Serialize` (non-string map keys and friends) panics — +//! that is a bug in the payload type, and the panic rides the +//! documented channel-panic contract (linked teardown). + +use super::{ + ChannelFrame, ChannelMessage, Decode, DecodeError, Encode, FrameRef, MessageRef, Status, + EV_REPLY, +}; +use crate::ws::Message; + +use serde::de::DeserializeOwned; +use serde::Serialize; +use serde_json::{json, Map, Value}; + +/// V2-JSON payload wrapper: `Json` speaks the phoenix wire for any +/// `T: Serialize + DeserializeOwned`. Derefs to `T`. +#[derive(Debug, Clone, PartialEq, Eq, Default)] +pub struct Json(pub T); + +impl std::ops::Deref for Json { + type Target = T; + fn deref(&self) -> &T { + &self.0 + } +} +impl std::ops::DerefMut for Json { + fn deref_mut(&mut self) -> &mut T { + &mut self.0 + } +} +impl From for Json { + fn from(t: T) -> Self { + Json(t) + } +} + +fn empty_object() -> Value { + Value::Object(Map::new()) +} + +fn to_value(t: &T) -> Value { + serde_json::to_value(t).expect("phoenix codec: payload failed to serialize") +} + +impl Encode for Json { + fn encode(f: FrameRef<'_, Self>) -> Message { + let (event, payload) = match f.message { + MessageRef::Event { event, payload } => { + (event, payload.map(|p| to_value(&p.0)).unwrap_or_else(empty_object)) + } + MessageRef::Reply { status, payload } => ( + EV_REPLY, + json!({ + "status": match status { + Status::Ok => "ok", + Status::Error => "error", + }, + "response": payload.map(|p| to_value(&p.0)).unwrap_or_else(empty_object), + }), + ), + }; + let arr = json!([f.join_ref, f.reference, f.topic, event, payload]); + Message::Text(arr.to_string()) + } +} + +impl Decode for Json { + fn decode(msg: &Message) -> Result, DecodeError> { + let Message::Text(s) = msg else { + return Err(DecodeError("phoenix V2 frames are text".into())); + }; + let (join_ref, reference, topic, event, payload): ( + Option, + Option, + String, + String, + Value, + ) = serde_json::from_str(s).map_err(|e| DecodeError(e.to_string()))?; + + let message = if event == EV_REPLY { + // Clients don't reply in practice; decoded for symmetry + // (and ignored upstream). + let status = match payload.get("status").and_then(Value::as_str) { + Some("ok") => Status::Ok, + _ => Status::Error, + }; + let response = payload.get("response").cloned().unwrap_or_else(empty_object); + let payload = serde_json::from_value::(response) + .map_err(|e| DecodeError(format!("reply response: {e}")))?; + ChannelMessage::Reply { status, payload: Some(Json(payload)) } + } else { + let payload = serde_json::from_value::(payload) + .map_err(|e| DecodeError(format!("payload for '{event}': {e}")))?; + ChannelMessage::Event { event, payload: Json(payload) } + }; + Ok(ChannelFrame { join_ref, reference, topic, message }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + type V = Json; + + fn parse(m: &Message) -> Value { + let Message::Text(s) = m else { panic!("not text") }; + serde_json::from_str(s).unwrap() + } + + #[test] + fn decodes_phoenix_js_join_frame() { + let m = Message::Text(r#"["3","3","room:lobby","phx_join",{"token":"t"}]"#.into()); + let f = V::decode(&m).unwrap(); + assert_eq!(f.join_ref.as_deref(), Some("3")); + assert_eq!(f.reference.as_deref(), Some("3")); + assert_eq!(f.topic, "room:lobby"); + let ChannelMessage::Event { event, payload } = f.message else { panic!() }; + assert_eq!(event, "phx_join"); + assert_eq!(payload.0, json!({"token": "t"})); + } + + #[test] + fn decodes_heartbeat_null_refs_allowed() { + let m = Message::Text(r#"[null,"7","phoenix","heartbeat",{}]"#.into()); + let f = V::decode(&m).unwrap(); + assert_eq!(f.join_ref, None); + assert_eq!(f.topic, "phoenix"); + } + + #[test] + fn encodes_ok_reply_with_status_response_wrapper() { + let p = Json(json!({"user_count": 2})); + let m = V::encode(FrameRef { + join_ref: Some("3"), + reference: Some("4"), + topic: "room:lobby", + message: MessageRef::Reply { status: Status::Ok, payload: Some(&p) }, + }); + assert_eq!( + parse(&m), + json!(["3", "4", "room:lobby", "phx_reply", + {"status": "ok", "response": {"user_count": 2}}]) + ); + } + + #[test] + fn encodes_payloadless_reply_and_event_as_empty_object() { + let m = V::encode(FrameRef { + join_ref: None, + reference: Some("9"), + topic: "phoenix", + message: MessageRef::Reply { status: Status::Ok, payload: None }, + }); + assert_eq!( + parse(&m), + json!([null, "9", "phoenix", "phx_reply", {"status": "ok", "response": {}}]) + ); + let m = V::encode(FrameRef { + join_ref: Some("3"), + reference: None, + topic: "room:lobby", + message: MessageRef::Event { event: "phx_close", payload: None }, + }); + assert_eq!(parse(&m), json!(["3", null, "room:lobby", "phx_close", {}])); + } + + #[test] + fn typed_payload_round_trip_and_strictness() { + #[derive(serde::Serialize, serde::Deserialize, Debug, PartialEq)] + struct Chat { + body: String, + } + let m = Message::Text(r#"["1","2","room:a","new_msg",{"body":"hi"}]"#.into()); + let f = Json::::decode(&m).unwrap(); + let ChannelMessage::Event { payload, .. } = f.message else { panic!() }; + assert_eq!(payload.0, Chat { body: "hi".into() }); + + // Missing required field: the codec is as strict as T. + let m = Message::Text(r#"["1","2","room:a","new_msg",{}]"#.into()); + assert!(Json::::decode(&m).is_err()); + } + + #[test] + fn garbage_and_binary_are_decode_errors() { + assert!(V::decode(&Message::Text("not json".into())).is_err()); + assert!(V::decode(&Message::Text(r#"{"a":1}"#.into())).is_err()); + assert!(V::decode(&Message::Binary(vec![1, 2, 3])).is_err()); + } +} diff --git a/tests/integration.rs b/tests/integration.rs index fa401a0..d2636e9 100644 --- a/tests/integration.rs +++ b/tests/integration.rs @@ -1408,3 +1408,199 @@ fn shutdown_with_open_chat_terminates() { let _ = a.read_to_end(&mut rest); let _ = b.read_to_end(&mut rest); } + +// =========================================================================== +// Channels (v0.6) — phoenix V2 wire over a real socket +// =========================================================================== + +#[cfg(feature = "phoenix")] +mod channels_wire { + use super::*; + use serde_json::{json, Value}; + use urus::channels::phoenix::Json; + use urus::channels::{Channel, ChannelHub, ChannelSocket, PrefixRouter, Status}; + + type P = Json; + + /// join rejects payloads carrying "deny"; "shout" broadcasts to the + /// room (sender included); anything else echoes as an ok reply. + struct Lobby; + + impl Channel

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

) -> Result { + if payload.0.get("deny").is_some() { + Err(Json(json!({"reason": "denied"}))) + } else { + Ok(Json(json!({"joined": topic}))) + } + } + 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}))), + } + } + } + + fn channels_pipeline() -> Pipeline { + let hub: std::sync::Arc>> = + std::sync::Arc::new(std::sync::OnceLock::new()); + Pipeline::new().plug(Router::new().get("/socket", move |c: Conn, _n: Next| { + // Hub construction is in-runtime only (it spawns the pubsub + // table); the NON-static OnceLock is what lets shutdown + // drain the table (the v0.5 pattern). + let hub = hub.clone(); + let hub = hub.get_or_init(|| { + ChannelHub::new(PrefixRouter::new().channel("room:*", |_: &str| { + Box::new(Lobby) as Box> + })) + }); + hub.upgrade(c) + })) + } + + const CHAN_HANDSHAKE: &[u8] = + b"GET /socket HTTP/1.1\r\nHost: x\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n\ + Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\nSec-WebSocket-Version: 13\r\n\r\n"; + + fn chan_connect(port: u16) -> TcpStream { + let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap(); + s.set_read_timeout(Some(Duration::from_secs(5))).unwrap(); + s.write_all(CHAN_HANDSHAKE).unwrap(); + let head = String::from_utf8(read_response_head(&mut s)).unwrap(); + assert!(head.starts_with("HTTP/1.1 101"), "handshake: {head}"); + s + } + + fn chan_send(s: &mut TcpStream, frame: Value) { + ws_send(s, &Frame::new(Opcode::Text, frame.to_string())); + } + + fn chan_read(s: &mut TcpStream, buf: &mut Vec) -> Value { + let f = ws_read_frame(s, buf); + assert_eq!(f.opcode, Opcode::Text, "expected V2 text frame"); + serde_json::from_slice(&f.payload).expect("server sent non-JSON frame") + } + + #[test] + fn channels_join_heartbeat_event_broadcast_leave() { + let port = spawn_server(channels_pipeline()); + + let mut a = chan_connect(port); + let mut a_buf = Vec::new(); + + // Join — the reply IS the subscription ack. + chan_send(&mut a, json!(["1", "1", "room:lobby", "phx_join", {}])); + assert_eq!( + chan_read(&mut a, &mut a_buf), + json!(["1", "1", "room:lobby", "phx_reply", + {"status": "ok", "response": {"joined": "room:lobby"}}]) + ); + + // Transport heartbeat. + chan_send(&mut a, json!([null, "2", "phoenix", "heartbeat", {}])); + assert_eq!( + chan_read(&mut a, &mut a_buf), + json!([null, "2", "phoenix", "phx_reply", {"status": "ok", "response": {}}]) + ); + + // Unrouted topic: error reply. + chan_send(&mut a, json!(["9", "3", "hall:x", "phx_join", {}])); + assert_eq!( + chan_read(&mut a, &mut a_buf), + json!(["9", "3", "hall:x", "phx_reply", {"status": "error", "response": {}}]) + ); + + // Rejected join: the channel's error payload comes back. + chan_send(&mut a, json!(["9", "4", "room:vip", "phx_join", {"deny": true}])); + assert_eq!( + chan_read(&mut a, &mut a_buf), + json!(["9", "4", "room:vip", "phx_reply", + {"status": "error", "response": {"reason": "denied"}}]) + ); + + // Second member; a's shout reaches both (sender included). + let mut b = chan_connect(port); + let mut b_buf = Vec::new(); + chan_send(&mut b, json!(["1", "1", "room:lobby", "phx_join", {}])); + chan_read(&mut b, &mut b_buf); + + chan_send(&mut a, json!(["1", "5", "room:lobby", "shout", {"body": "hi"}])); + let push = json!(["1", null, "room:lobby", "shouted", {"body": "hi"}]); + assert_eq!(chan_read(&mut a, &mut a_buf), push); + assert_eq!(chan_read(&mut b, &mut b_buf), push); + + // Plain event echoes as a correlated reply. + chan_send(&mut a, json!(["1", "6", "room:lobby", "wave", {}])); + assert_eq!( + chan_read(&mut a, &mut a_buf), + json!(["1", "6", "room:lobby", "phx_reply", + {"status": "ok", "response": {"echo": "wave"}}]) + ); + + // Leave: ok reply, then phx_close; the topic is then unjoined. + chan_send(&mut a, json!(["1", "7", "room:lobby", "phx_leave", {}])); + assert_eq!( + chan_read(&mut a, &mut a_buf), + json!(["1", "7", "room:lobby", "phx_reply", {"status": "ok", "response": {}}]) + ); + assert_eq!( + chan_read(&mut a, &mut a_buf), + json!(["1", null, "room:lobby", "phx_close", {}]) + ); + chan_send(&mut a, json!(["1", "8", "room:lobby", "wave", {}])); + assert_eq!( + chan_read(&mut a, &mut a_buf), + json!(["1", "8", "room:lobby", "phx_reply", {"status": "error", "response": {}}]) + ); + + // a is gone from the room: b's shout comes back to b only, and + // a's connection stays a working transport (heartbeat). + chan_send(&mut b, json!(["1", "9", "room:lobby", "shout", {"n": 2}])); + assert_eq!( + chan_read(&mut b, &mut b_buf), + json!(["1", null, "room:lobby", "shouted", {"n": 2}]) + ); + chan_send(&mut a, json!([null, "10", "phoenix", "heartbeat", {}])); + assert_eq!( + chan_read(&mut a, &mut a_buf), + json!([null, "10", "phoenix", "phx_reply", {"status": "ok", "response": {}}]) + ); + } + + #[test] + fn channels_codec_garbage_closes_1002() { + let port = spawn_server(channels_pipeline()); + let mut s = chan_connect(port); + let mut buf = Vec::new(); + ws_send(&mut s, &Frame::new(Opcode::Text, "not a v2 frame")); + let f = ws_read_frame(&mut s, &mut buf); + assert_eq!(f.opcode, Opcode::Close); + assert_eq!(u16::from_be_bytes([f.payload[0], f.payload[1]]), 1002); + } + + #[test] + fn shutdown_with_open_channel_terminates() { + // The whole exit chain under one roof: drain force-stops the + // conn actor parked in its select -> linked channel actor dies + // (and/or wakes on the dropped inbound arm) -> its subscription + // is monitor-pruned -> the non-static hub's last handle drops + // with the drained pipeline -> pubsub table exits -> AllDone. + let (port, handle, done_rx) = + spawn_server_with_handle(channels_pipeline(), Duration::from_millis(300)); + + let mut a = chan_connect(port); + let mut a_buf = Vec::new(); + chan_send(&mut a, json!(["1", "1", "room:lobby", "phx_join", {}])); + let reply = chan_read(&mut a, &mut a_buf); + assert_eq!(reply[3], "phx_reply"); // joined: channel actor live + + handle.shutdown(); + done_rx + .recv_timeout(Duration::from_secs(5)) + .expect("serve did not return: channel actor, hub, or table outlived the drain"); + + let mut rest = Vec::new(); + let _ = a.read_to_end(&mut rest); + } +}