From 0ed449cc255fdac6717cd4a124cf49d16f802897 Mon Sep 17 00:00:00 2001 From: Liam Date: Sat, 20 Jun 2026 21:04:50 +0200 Subject: [PATCH 01/11] feat: implement support for volatile events with new VolatileSocket wrapper and operator flags --- crates/socketioxide-core/src/adapter/mod.rs | 13 ++- crates/socketioxide/src/io.rs | 10 ++ crates/socketioxide/src/operators.rs | 19 ++++ crates/socketioxide/src/socket.rs | 111 ++++++++++++++++++++ 4 files changed, 152 insertions(+), 1 deletion(-) diff --git a/crates/socketioxide-core/src/adapter/mod.rs b/crates/socketioxide-core/src/adapter/mod.rs index 7e055ed9..3524c803 100644 --- a/crates/socketioxide-core/src/adapter/mod.rs +++ b/crates/socketioxide-core/src/adapter/mod.rs @@ -38,6 +38,11 @@ pub enum BroadcastFlags { Local = 0x01, /// Broadcast to all clients except the sender Broadcast = 0x02, + /// The event may be dropped if the client is not ready to receive it + /// (e.g. the connection is buffering or not connected). + /// This is useful for events that are not critical, like position updates in a game. + /// See [socket.io volatile events](https://socket.io/docs/v4/emitting-events/#volatile-events). + Volatile = 0x04, } /// Options that can be used to modify the behavior of the broadcast methods. @@ -430,8 +435,14 @@ impl CoreLocalAdapter { return Ok(()); } + let is_volatile = opts.has_flag(BroadcastFlags::Volatile); let data = self.emitter.parser().encode(packet); - self.emitter.send_many(sids, data) + if is_volatile { + _ = self.emitter.send_many(sids, data); + Ok(()) + } else { + self.emitter.send_many(sids, data) + } } /// Broadcasts the packet to the sockets that match the [`BroadcastOptions`] and return a stream of ack responses. diff --git a/crates/socketioxide/src/io.rs b/crates/socketioxide/src/io.rs index 15618278..045822e3 100644 --- a/crates/socketioxide/src/io.rs +++ b/crates/socketioxide/src/io.rs @@ -591,6 +591,16 @@ impl SocketIo { self.get_default_op() } + /// _Alias for `io.of("/").unwrap().volatile()`_. If the **default namespace "/" is not found** this fn will panic! + /// + /// Sets the volatile flag on the broadcast operators. Volatile events may be dropped + /// if the client is not ready to receive them. + /// See [socket.io volatile events](https://socket.io/docs/v4/emitting-events/#volatile-events). + #[inline] + pub fn volatile(&self) -> BroadcastOperators { + self.get_default_op().volatile() + } + #[cfg(feature = "state")] pub(crate) fn get_state(&self) -> Option { self.0.state.try_get::().cloned() diff --git a/crates/socketioxide/src/operators.rs b/crates/socketioxide/src/operators.rs index efdde4a8..594b8977 100644 --- a/crates/socketioxide/src/operators.rs +++ b/crates/socketioxide/src/operators.rs @@ -91,6 +91,15 @@ impl<'a, A: Adapter> ConfOperators<'a, A> { self.timeout = Some(timeout); self } + + /// Sets the volatile flag for the emit. When set, the event may be dropped + /// if the client is not ready to receive it (e.g. the connection is buffering or not connected). + /// This is useful for events that are not critical, such as position updates in a game. + /// + /// See [socket.io volatile events](https://socket.io/docs/v4/emitting-events/#volatile-events). + pub fn volatile(self) -> BroadcastOperators { + BroadcastOperators::from(self).volatile() + } } // ==== impl ConfOperators consume fns ==== @@ -228,6 +237,16 @@ impl BroadcastOperators { self.timeout = Some(timeout); self } + + /// Sets the volatile flag for the emit. When set, the event may be dropped + /// if the client is not ready to receive it (e.g. the connection is buffering or not connected). + /// This is useful for events that are not critical, such as position updates in a game. + /// + /// See [socket.io volatile events](https://socket.io/docs/v4/emitting-events/#volatile-events). + pub fn volatile(mut self) -> Self { + self.opts.add_flag(BroadcastFlags::Volatile); + self + } } // ==== impl BroadcastOperators consume fns ==== diff --git a/crates/socketioxide/src/socket.rs b/crates/socketioxide/src/socket.rs index 4d4d7b68..7439340b 100644 --- a/crates/socketioxide/src/socket.rs +++ b/crates/socketioxide/src/socket.rs @@ -278,6 +278,94 @@ impl From for RemoteActionError { } } +/// A wrapper around a [`Socket`] that marks emitted events as volatile. +/// +/// Volatile events may be dropped if the client is not ready to receive them +/// (e.g. the underlying connection is buffering or the socket is not connected). +/// This is useful for events that are not critical, such as position updates in a game. +/// +/// See [socket.io volatile events](https://socket.io/docs/v4/emitting-events/#volatile-events). +/// +/// # Example +/// ``` +/// # use socketioxide::{SocketIo, extract::*}; +/// # use serde::Serialize; +/// #[derive(Serialize)] +/// struct GameState { x: f64, y: f64 } +/// +/// let (_, io) = SocketIo::new_svc(); +/// io.ns("/", async |socket: SocketRef| { +/// // Position updates may be dropped if the connection is slow +/// socket.volatile().emit("position", &GameState { x: 1.0, y: 2.0 }); +/// }); +/// ``` +pub struct VolatileSocket<'a, A: Adapter = LocalAdapter> { + socket: &'a Socket, +} + +impl VolatileSocket<'_, A> { + /// Emit a volatile event to the client. If the socket is not connected or + /// the internal buffer is full, the event is silently dropped. + pub fn emit(&self, event: impl AsRef, data: &T) { + if !self.socket.connected() { + #[cfg(feature = "tracing")] + tracing::debug!(?self.socket.id, "dropping volatile event: socket not connected"); + return; + } + + let permit = match self.socket.reserve() { + Ok(permit) => permit, + Err(_e) => { + #[cfg(feature = "tracing")] + tracing::debug!(?_e, ?self.socket.id, "dropping volatile event: cannot reserve"); + return; + } + }; + + let ns = self.socket.ns.path.clone(); + let data = match self.socket.parser.encode_value(data, Some(event.as_ref())) { + Ok(data) => data, + Err(_e) => { + #[cfg(feature = "tracing")] + tracing::debug!(?_e, "dropping volatile event: encode error"); + return; + } + }; + + permit.send(Packet::event(ns, data), self.socket.parser); + } + + #[doc = include_str!("../docs/operators/to.md")] + pub fn to(self, rooms: impl RoomParam) -> BroadcastOperators { + self.socket.to(rooms).volatile() + } + + #[doc = include_str!("../docs/operators/within.md")] + pub fn within(self, rooms: impl RoomParam) -> BroadcastOperators { + self.socket.within(rooms).volatile() + } + + #[doc = include_str!("../docs/operators/except.md")] + pub fn except(self, rooms: impl RoomParam) -> BroadcastOperators { + self.socket.except(rooms).volatile() + } + + #[doc = include_str!("../docs/operators/local.md")] + pub fn local(self) -> BroadcastOperators { + self.socket.local().volatile() + } + + #[doc = include_str!("../docs/operators/broadcast.md")] + pub fn broadcast(self) -> BroadcastOperators { + self.socket.broadcast().volatile() + } + + #[doc = include_str!("../docs/operators/timeout.md")] + pub fn timeout(self, timeout: Duration) -> BroadcastOperators { + BroadcastOperators::from(ConfOperators::new(self.socket).timeout(timeout)).volatile() + } +} + /// A Socket represents a client connected to a namespace. /// It is used to send and receive messages from the client, join and leave rooms, etc. /// The socket struct itself should not be used directly, but through a [`SocketRef`](crate::extract::SocketRef). @@ -639,6 +727,29 @@ impl Socket { BroadcastOperators::from_sock(self.ns.clone(), self.id, self.parser).broadcast() } + /// Returns a [`VolatileSocket`] wrapper that emits volatile events to this socket. + /// Volatile events may be dropped if the client is not ready to receive them + /// (e.g. the underlying connection is buffering or the socket is not connected). + /// + /// See [`VolatileSocket`] for more details. + /// + /// # Example + /// ``` + /// # use socketioxide::{SocketIo, extract::SocketRef}; + /// # use serde::Serialize; + /// #[derive(Serialize)] + /// struct GameState { x: f64, y: f64 } + /// + /// let (_, io) = SocketIo::new_svc(); + /// io.ns("/", async |socket: SocketRef| { + /// socket.volatile().emit("position", &GameState { x: 1.0, y: 2.0 }); + /// socket.volatile().to("room1").emit("update", &42); + /// }); + /// ``` + pub fn volatile(&self) -> VolatileSocket<'_, A> { + VolatileSocket { socket: self } + } + /// # Get the [`SocketIo`] context related to this socket /// /// # Panics From 35d867ba89b1c388b27c0b68d0de9f00954752b9 Mon Sep 17 00:00:00 2001 From: Liam Date: Sun, 21 Jun 2026 11:56:28 +0200 Subject: [PATCH 02/11] feat(socketio,engineio): add volatile flag for emitting events --- crates/engineioxide/Cargo.toml | 2 +- crates/engineioxide/src/socket.rs | 53 +++++++++++++++ .../engineioxide/src/transport/polling/mod.rs | 23 ++++++- .../src/transport/polling/payload/encoder.rs | 65 +++++++++++++++---- .../src/transport/polling/payload/mod.rs | 11 ++-- crates/engineioxide/src/transport/ws.rs | 43 +++++++++--- crates/socketioxide-core/src/adapter/mod.rs | 8 ++- crates/socketioxide/src/ns.rs | 18 +++++ crates/socketioxide/src/socket.rs | 41 +++++++----- 9 files changed, 214 insertions(+), 50 deletions(-) diff --git a/crates/engineioxide/Cargo.toml b/crates/engineioxide/Cargo.toml index b604481d..45fcad51 100644 --- a/crates/engineioxide/Cargo.toml +++ b/crates/engineioxide/Cargo.toml @@ -28,7 +28,7 @@ http-body.workspace = true serde.workspace = true serde_json.workspace = true thiserror.workspace = true -tokio = { workspace = true, features = ["rt", "time"] } +tokio = { workspace = true, features = ["rt", "time", "macros"] } tokio-util.workspace = true tower-service.workspace = true tower-layer.workspace = true diff --git a/crates/engineioxide/src/socket.rs b/crates/engineioxide/src/socket.rs index 222e3d8f..ecc23b13 100644 --- a/crates/engineioxide/src/socket.rs +++ b/crates/engineioxide/src/socket.rs @@ -78,6 +78,7 @@ use smallvec::{SmallVec, smallvec}; use tokio::sync::{ Mutex, mpsc::{self, Receiver, error::TrySendError}, + watch, }; pub use engineioxide_core::Sid; @@ -212,6 +213,13 @@ where /// Channel to send [PacketBuf] to the internal connection internal_tx: mpsc::Sender, + /// Channel to send volatile [PacketBuf]s that bypass the internal buffer. + /// Uses a [`watch`](tokio::sync::watch) channel so only the latest volatile + /// message is retained; subsequent volatile sends overwrite the previous one. + volatile_tx: watch::Sender>, + /// Receiver for the volatile channel, read by the transport with priority. + pub(crate) volatile_rx: watch::Receiver>, + /// Internal channel to receive Pong [`Packets`](Packet) (v4 protocol) or Ping (v3 protocol) in the heartbeat job /// which is running in a separate task heartbeat_rx: Mutex>, @@ -249,6 +257,7 @@ where ) -> Self { let (internal_tx, internal_rx) = mpsc::channel(config.max_buffer_size); let (heartbeat_tx, heartbeat_rx) = mpsc::channel(1); + let (volatile_tx, volatile_rx) = watch::channel(None); Self { id: Sid::new(), @@ -259,6 +268,9 @@ where internal_rx: Mutex::new(PeekableReceiver::new(internal_rx)), internal_tx, + volatile_rx, + volatile_tx, + heartbeat_rx: Mutex::new(heartbeat_rx), heartbeat_tx, cancellation_token: CancellationToken::new(), @@ -488,6 +500,43 @@ where TrySendError::Closed(p) => TrySendError::Closed(p.into_binary()), }) } + + /// Try to send a volatile message bypassing the internal buffer channel. + /// Volatile messages may be dropped if the transport is not ready to + /// receive them. + #[inline] + pub fn emit_volatile(&self, msg: impl Into) -> bool { + self.send_volatile(smallvec![Packet::Message(msg.into())]) + } + + /// Try to send a volatile binary message bypassing the internal buffer channel. + /// Volatile messages may be dropped if the transport is not ready to + /// receive them. + #[inline] + pub fn emit_binary_volatile>(&self, data: B) -> bool { + if self.protocol == ProtocolVersion::V3 { + self.send_volatile(smallvec![Packet::BinaryV3(data.into())]) + } else { + self.send_volatile(smallvec![Packet::Binary(data.into())]) + } + } + + /// Try to send a volatile message with multiple adjacent binary payloads. + /// The message and all binary payloads are sent atomically as a single + /// volatile write. + #[inline] + pub fn emit_many_volatile(&self, msg: Str, data: VecDeque) -> bool { + let mut packets = SmallVec::with_capacity(1 + data.len()); + packets.push(Packet::Message(msg)); + for bin in data { + packets.push(Packet::Binary(bin)); + } + self.send_volatile(packets) + } + + pub(crate) fn send_volatile(&self, packets: PacketBuf) -> bool { + self.volatile_tx.send(Some(packets)).is_ok() + } } impl std::fmt::Debug for Socket { @@ -548,6 +597,7 @@ where ) -> (Arc>, tokio::sync::mpsc::Receiver) { let (internal_tx, internal_rx) = mpsc::channel(buffer_size); let (heartbeat_tx, heartbeat_rx) = mpsc::channel(1); + let (volatile_tx, volatile_rx) = watch::channel(None); let sock = Self { id: sid, @@ -558,6 +608,9 @@ where internal_rx: Mutex::new(PeekableReceiver::new(internal_rx)), internal_tx, + volatile_rx, + volatile_tx, + heartbeat_rx: Mutex::new(heartbeat_rx), heartbeat_tx, cancellation_token: CancellationToken::new(), diff --git a/crates/engineioxide/src/transport/polling/mod.rs b/crates/engineioxide/src/transport/polling/mod.rs index e5207d35..bb3b8979 100644 --- a/crates/engineioxide/src/transport/polling/mod.rs +++ b/crates/engineioxide/src/transport/polling/mod.rs @@ -130,11 +130,28 @@ where let max_payload = engine.config.max_payload; + // Read the latest volatile packet from the watch channel, if any. + // Because the watch channel only retains the most recent value, + // any volatile messages that were overwritten since the last poll + // are automatically discarded. + let volatile_packets: Vec<_> = { + let mut rx = socket.volatile_rx.clone(); + let value = rx.borrow_and_update(); + value.as_ref().map(|p| vec![p.clone()]).unwrap_or_default() + }; + #[cfg(feature = "v3")] - let Payload { data, has_binary } = - payload::encoder(rx, protocol, socket.supports_binary, max_payload).await?; + let Payload { data, has_binary } = payload::encoder( + rx, + protocol, + socket.supports_binary, + max_payload, + volatile_packets, + ) + .await?; #[cfg(not(feature = "v3"))] - let Payload { data, has_binary } = payload::encoder(rx, protocol, max_payload).await?; + let Payload { data, has_binary } = + payload::encoder(rx, protocol, max_payload, volatile_packets).await?; #[cfg(feature = "tracing")] tracing::debug!("[sid={sid}] sending data: {:?}", data); diff --git a/crates/engineioxide/src/transport/polling/payload/encoder.rs b/crates/engineioxide/src/transport/polling/payload/encoder.rs index a56a9c9a..406669ac 100644 --- a/crates/engineioxide/src/transport/polling/payload/encoder.rs +++ b/crates/engineioxide/src/transport/polling/payload/encoder.rs @@ -75,6 +75,7 @@ async fn recv_packet( pub async fn v4_encoder( mut rx: MutexGuard<'_, PeekableReceiver>, max_payload: u64, + volatile_packets: Vec, ) -> Result { use crate::transport::polling::payload::PACKET_SEPARATOR_V4; @@ -82,6 +83,18 @@ pub async fn v4_encoder( tracing::debug!("encoding payload with v4 encoder"); let mut data: String = String::new(); + // Encode volatile packets first so they bypass the main channel backlog + for packets in volatile_packets { + for packet in packets { + let packet: String = packet.into(); + + if !data.is_empty() { + data.push(std::char::from_u32(PACKET_SEPARATOR_V4 as u32).unwrap()); + } + data.push_str(&packet); + } + } + // Send all packets in the buffer const PUNCTUATION_LEN: usize = 1; while let Some(packets) = @@ -175,6 +188,7 @@ pub fn v3_string_packet_encoder(packet: Packet, data: &mut bytes::BytesMut) { pub async fn v3_binary_encoder( mut rx: MutexGuard<'_, PeekableReceiver>, max_payload: u64, + volatile_packets: Vec, ) -> Result { let mut data = bytes::BytesMut::new(); let mut packet_buffer: Vec = Vec::new(); @@ -189,6 +203,18 @@ pub async fn v3_binary_encoder( // buffer all packets to find if there is binary packets let mut has_binary = false; + // Encode volatile packets first so they bypass the main channel backlog + for packets in volatile_packets { + for packet in packets { + if packet.is_binary() { + has_binary = true; + } + const PUNCTUATION_LEN: usize = 2; + estimated_size += packet.get_size_hint(false) + max_packet_size_len + PUNCTUATION_LEN; + packet_buffer.push(packet); + } + } + while let Some(packets) = try_recv_packet(&mut rx, estimated_size, max_payload, false) { for packet in packets { if packet.is_binary() { @@ -236,6 +262,7 @@ pub async fn v3_binary_encoder( pub async fn v3_string_encoder( mut rx: MutexGuard<'_, PeekableReceiver>, max_payload: u64, + volatile_packets: Vec, ) -> Result { let mut data = bytes::BytesMut::new(); @@ -245,6 +272,14 @@ pub async fn v3_string_encoder( const PUNCTUATION_LEN: usize = 2; // number of digits of the max packet size, used to approximate the payload size let max_packet_size_len = max_payload.checked_ilog10().unwrap_or(0) as usize + 1; + + // Encode volatile packets first so they bypass the main channel backlog + for packets in volatile_packets { + for packet in packets { + v3_string_packet_encoder(packet, &mut data); + } + } + while let Some(packets) = try_recv_packet( &mut rx, data.len() + PUNCTUATION_LEN + max_packet_size_len, @@ -291,7 +326,7 @@ mod tests { .unwrap(); tx.try_send(smallvec::smallvec![Packet::Message("hello€".into())]) .unwrap(); - let Payload { data, .. } = v4_encoder(rx, MAX_PAYLOAD).await.unwrap(); + let Payload { data, .. } = v4_encoder(rx, MAX_PAYLOAD, vec![]).await.unwrap(); assert_eq!(data, PAYLOAD.as_bytes()); } @@ -309,7 +344,7 @@ mod tests { ]) .unwrap(); }); - let Payload { data, .. } = v4_encoder(rx, MAX_PAYLOAD).await.unwrap(); + let Payload { data, .. } = v4_encoder(rx, MAX_PAYLOAD, vec![]).await.unwrap(); assert_eq!(data, PAYLOAD); } @@ -330,17 +365,17 @@ mod tests { .unwrap(); { let rx = mutex.lock().await; - let Payload { data, .. } = v4_encoder(rx, MAX_PAYLOAD).await.unwrap(); + let Payload { data, .. } = v4_encoder(rx, MAX_PAYLOAD, vec![]).await.unwrap(); assert_eq!(data, "4hello€".as_bytes()); } { let rx = mutex.lock().await; - let Payload { data, .. } = v4_encoder(rx, MAX_PAYLOAD + 10).await.unwrap(); + let Payload { data, .. } = v4_encoder(rx, MAX_PAYLOAD + 10, vec![]).await.unwrap(); assert_eq!(data, "bAQIDBA==\x1e4hello€".as_bytes()); } { let rx = mutex.lock().await; - let Payload { data, .. } = v4_encoder(rx, MAX_PAYLOAD + 10).await.unwrap(); + let Payload { data, .. } = v4_encoder(rx, MAX_PAYLOAD + 10, vec![]).await.unwrap(); assert_eq!(data, "4hello€".as_bytes()); } } @@ -380,7 +415,7 @@ mod tests { .unwrap(); let Payload { data, has_binary, .. - } = v3_string_encoder(rx, MAX_PAYLOAD).await.unwrap(); + } = v3_string_encoder(rx, MAX_PAYLOAD, vec![]).await.unwrap(); assert_eq!(data, PAYLOAD.as_bytes()); assert!(!has_binary); } @@ -404,18 +439,22 @@ mod tests { .unwrap(); { let rx = mutex.lock().await; - let Payload { data, .. } = v3_string_encoder(rx, MAX_PAYLOAD).await.unwrap(); + let Payload { data, .. } = v3_string_encoder(rx, MAX_PAYLOAD, vec![]).await.unwrap(); assert_eq!(data, "7:4hello€".as_bytes()); } { let rx = mutex.lock().await; - let Payload { data, .. } = v3_string_encoder(rx, MAX_PAYLOAD + 10).await.unwrap(); + let Payload { data, .. } = v3_string_encoder(rx, MAX_PAYLOAD + 10, vec![]) + .await + .unwrap(); assert_eq!(data, "10:b4AQIDBA==".as_bytes()); } { // Next call drains one of the remaining Message packets. let rx = mutex.lock().await; - let Payload { data, .. } = v3_string_encoder(rx, MAX_PAYLOAD + 10).await.unwrap(); + let Payload { data, .. } = v3_string_encoder(rx, MAX_PAYLOAD + 10, vec![]) + .await + .unwrap(); assert_eq!(data, "7:4hello€".as_bytes()); } } @@ -438,7 +477,7 @@ mod tests { .unwrap(); let Payload { data, has_binary, .. - } = v3_binary_encoder(rx, MAX_PAYLOAD).await.unwrap(); + } = v3_binary_encoder(rx, MAX_PAYLOAD, vec![]).await.unwrap(); assert_eq!(*data, PAYLOAD); assert!(has_binary); } @@ -467,7 +506,7 @@ mod tests { }); let Payload { data, has_binary, .. - } = v3_binary_encoder(rx, MAX_PAYLOAD).await.unwrap(); + } = v3_binary_encoder(rx, MAX_PAYLOAD, vec![]).await.unwrap(); assert_eq!(*data, PAYLOAD); assert!(has_binary); } @@ -495,12 +534,12 @@ mod tests { .unwrap(); { let rx = mutex.lock().await; - let Payload { data, .. } = v3_binary_encoder(rx, MAX_PAYLOAD).await.unwrap(); + let Payload { data, .. } = v3_binary_encoder(rx, MAX_PAYLOAD, vec![]).await.unwrap(); assert_eq!(*data, PAYLOAD); } { let rx = mutex.lock().await; - let Payload { data, .. } = v3_binary_encoder(rx, MAX_PAYLOAD).await.unwrap(); + let Payload { data, .. } = v3_binary_encoder(rx, MAX_PAYLOAD, vec![]).await.unwrap(); assert_eq!(data, "7:4hello€7:4hello€".as_bytes()); } } diff --git a/crates/engineioxide/src/transport/polling/payload/mod.rs b/crates/engineioxide/src/transport/polling/payload/mod.rs index db62cc96..9b95efab 100644 --- a/crates/engineioxide/src/transport/polling/payload/mod.rs +++ b/crates/engineioxide/src/transport/polling/payload/mod.rs @@ -70,21 +70,24 @@ pub async fn encoder( #[allow(unused_variables)] protocol: ProtocolVersion, #[cfg(feature = "v3")] supports_binary: bool, max_payload: u64, + volatile_packets: Vec, ) -> Result { #[cfg(feature = "v3")] { match protocol { - ProtocolVersion::V4 => encoder::v4_encoder(rx, max_payload).await, + ProtocolVersion::V4 => encoder::v4_encoder(rx, max_payload, volatile_packets).await, ProtocolVersion::V3 if supports_binary => { - encoder::v3_binary_encoder(rx, max_payload).await + encoder::v3_binary_encoder(rx, max_payload, volatile_packets).await + } + ProtocolVersion::V3 => { + encoder::v3_string_encoder(rx, max_payload, volatile_packets).await } - ProtocolVersion::V3 => encoder::v3_string_encoder(rx, max_payload).await, } } #[cfg(not(feature = "v3"))] { - encoder::v4_encoder(rx, max_payload).await + encoder::v4_encoder(rx, max_payload, volatile_packets).await } } diff --git a/crates/engineioxide/src/transport/ws.rs b/crates/engineioxide/src/transport/ws.rs index 565a9940..ed59e76b 100644 --- a/crates/engineioxide/src/transport/ws.rs +++ b/crates/engineioxide/src/transport/ws.rs @@ -249,6 +249,7 @@ async fn forward_to_socket( S: AsyncRead + AsyncWrite + Unpin + Send + 'static, { let mut internal_rx = socket.internal_rx.try_lock().unwrap(); + let mut volatile_rx = socket.volatile_rx.clone(); // map a packet to a websocket message // It is declared as a macro rather than a closure to avoid ownership issues @@ -288,18 +289,40 @@ async fn forward_to_socket( }; } - while let Some(items) = internal_rx.recv().await { - for item in items { - map_fn!(item); - } - // For every available packet we continue to send until the channel is drained - while let Ok(items) = internal_rx.try_recv() { - for item in items { - map_fn!(item); + loop { + tokio::select! { + // Priority: main channel is checked first so regular events + // are never starved by a flood of volatile ones. + items = internal_rx.recv() => { + match items { + Some(items) => { + for item in items { + map_fn!(item); + } + // For every available packet we continue to send until the channel is drained + while let Ok(items) = internal_rx.try_recv() { + for item in items { + map_fn!(item); + } + } + tx.flush().await.ok(); + } + None => break, + } + } + // Lower priority: volatile watch channel. + // Only the latest volatile message is retained; earlier ones + // are overwritten if they haven't been consumed yet. + Ok(()) = volatile_rx.changed() => { + let value = volatile_rx.borrow_and_update().clone(); + if let Some(packets) = value { + for item in packets { + map_fn!(item); + } + tx.flush().await.ok(); + } } } - - tx.flush().await.ok(); } } /// Send a Engine.IO [`OpenPacket`] to initiate a websocket connection diff --git a/crates/socketioxide-core/src/adapter/mod.rs b/crates/socketioxide-core/src/adapter/mod.rs index 3524c803..e5f5447d 100644 --- a/crates/socketioxide-core/src/adapter/mod.rs +++ b/crates/socketioxide-core/src/adapter/mod.rs @@ -207,6 +207,12 @@ pub trait SocketEmitter: Send + Sync + 'static { fn get_remote_sockets(&self, sids: BroadcastIter<'_>) -> Vec; /// Send data to the list of socket ids. fn send_many(&self, sids: BroadcastIter<'_>, data: Value) -> Result<(), Vec>; + /// Send data to the list of socket ids with volatile semantics. + /// Errors are silently discarded; packets may be dropped if the + /// transport is not ready. + fn send_many_volatile(&self, sids: BroadcastIter<'_>, data: Value) { + _ = self.send_many(sids, data); + } /// Send data to the list of socket ids and get a stream of acks and the number of expected acks. fn send_many_with_ack( &self, @@ -438,7 +444,7 @@ impl CoreLocalAdapter { let is_volatile = opts.has_flag(BroadcastFlags::Volatile); let data = self.emitter.parser().encode(packet); if is_volatile { - _ = self.emitter.send_many(sids, data); + self.emitter.send_many_volatile(sids, data); Ok(()) } else { self.emitter.send_many(sids, data) diff --git a/crates/socketioxide/src/ns.rs b/crates/socketioxide/src/ns.rs index 78236875..c4b94a0f 100644 --- a/crates/socketioxide/src/ns.rs +++ b/crates/socketioxide/src/ns.rs @@ -227,6 +227,9 @@ trait InnerEmitter: Send + Sync + 'static { fn get_all_sids(&self, filter: &dyn Fn(&Sid) -> bool) -> Vec; /// Send data to the list of socket ids. fn send_many(&self, sids: BroadcastIter<'_>, data: Value) -> Result<(), Vec>; + /// Send data to the list of socket ids with volatile semantics. + /// Errors are silently discarded. + fn send_many_volatile(&self, sids: BroadcastIter<'_>, data: Value); /// Send data to the list of socket ids and get a stream of acks. fn send_many_with_ack( &self, @@ -268,6 +271,15 @@ impl InnerEmitter for Namespace { if errs.is_empty() { Ok(()) } else { Err(errs) } } + fn send_many_volatile(&self, sids: BroadcastIter<'_>, data: Value) { + let sockets = self.sockets.read().unwrap(); + for sid in sids { + if let Some(socket) = sockets.get(&sid) { + socket.send_raw_volatile(data.clone()); + } + } + } + fn send_many_with_ack( &self, sids: BroadcastIter<'_>, @@ -354,6 +366,12 @@ impl SocketEmitter for Emitter { } } + fn send_many_volatile(&self, sids: BroadcastIter<'_>, data: Value) { + if let Some(ns) = self.ns.upgrade() { + ns.send_many_volatile(sids, data); + } + } + fn send_many_with_ack( &self, sids: BroadcastIter<'_>, diff --git a/crates/socketioxide/src/socket.rs b/crates/socketioxide/src/socket.rs index 7439340b..13bdf2af 100644 --- a/crates/socketioxide/src/socket.rs +++ b/crates/socketioxide/src/socket.rs @@ -309,30 +309,21 @@ impl VolatileSocket<'_, A> { pub fn emit(&self, event: impl AsRef, data: &T) { if !self.socket.connected() { #[cfg(feature = "tracing")] - tracing::debug!(?self.socket.id, "dropping volatile event: socket not connected"); + tracing::debug!( + ?self.socket.id, + "dropping volatile event: socket not connected" + ); return; } - let permit = match self.socket.reserve() { - Ok(permit) => permit, - Err(_e) => { - #[cfg(feature = "tracing")] - tracing::debug!(?_e, ?self.socket.id, "dropping volatile event: cannot reserve"); - return; - } - }; - let ns = self.socket.ns.path.clone(); - let data = match self.socket.parser.encode_value(data, Some(event.as_ref())) { - Ok(data) => data, - Err(_e) => { - #[cfg(feature = "tracing")] - tracing::debug!(?_e, "dropping volatile event: encode error"); - return; - } + let Ok(data) = self.socket.parser.encode_value(data, Some(event.as_ref())) else { + return; }; - permit.send(Packet::event(ns, data), self.socket.parser); + let packet = Packet::event(ns, data); + self.socket + .send_raw_volatile(self.socket.parser.encode(packet)); } #[doc = include_str!("../docs/operators/to.md")] @@ -849,6 +840,20 @@ impl Socket { Ok(()) } + pub(crate) fn send_raw_volatile(&self, value: Value) { + match value { + Value::Str(msg, None) => { + self.esocket.emit_volatile(msg); + } + Value::Str(msg, Some(bin_payloads)) => { + self.esocket.emit_many_volatile(msg, bin_payloads); + } + Value::Bytes(bin) => { + self.esocket.emit_binary_volatile(bin); + } + } + } + pub(crate) fn send_with_ack_permit( &self, mut packet: Packet, From 78e4f2897136144980426badd67811ebe6c36372 Mon Sep 17 00:00:00 2001 From: Liam Date: Sun, 21 Jun 2026 14:17:59 +0200 Subject: [PATCH 03/11] docs: add warnings about potential message ordering issues with volatile events --- crates/engineioxide/src/socket.rs | 3 +++ crates/socketioxide/src/socket.rs | 5 +++++ 2 files changed, 8 insertions(+) diff --git a/crates/engineioxide/src/socket.rs b/crates/engineioxide/src/socket.rs index ecc23b13..8f938523 100644 --- a/crates/engineioxide/src/socket.rs +++ b/crates/engineioxide/src/socket.rs @@ -504,6 +504,9 @@ where /// Try to send a volatile message bypassing the internal buffer channel. /// Volatile messages may be dropped if the transport is not ready to /// receive them. + /// + /// Because volatile messages bypass the main mpsc buffer queue, they may + /// arrive out of order relative to regular messages. #[inline] pub fn emit_volatile(&self, msg: impl Into) -> bool { self.send_volatile(smallvec![Packet::Message(msg.into())]) diff --git a/crates/socketioxide/src/socket.rs b/crates/socketioxide/src/socket.rs index 13bdf2af..7be8e0a1 100644 --- a/crates/socketioxide/src/socket.rs +++ b/crates/socketioxide/src/socket.rs @@ -284,6 +284,11 @@ impl From for RemoteActionError { /// (e.g. the underlying connection is buffering or the socket is not connected). /// This is useful for events that are not critical, such as position updates in a game. /// +/// Because volatile events use a separate channel that bypasses the main +/// mpsc buffer, they may arrive **out of order** relative to regular events +/// emitted around the same time. Only use volatile when ordering relative to +/// regular events is not important. +/// /// See [socket.io volatile events](https://socket.io/docs/v4/emitting-events/#volatile-events). /// /// # Example From 8861111124d54753d33c88e221472d6a2ccba2a5 Mon Sep 17 00:00:00 2001 From: Liam Date: Fri, 26 Jun 2026 09:20:50 +0200 Subject: [PATCH 04/11] refactor: replace VolatileSocket with operator flag and optimize volatile message loop in transport --- crates/engineioxide/Cargo.toml | 2 +- crates/engineioxide/src/socket.rs | 9 ++ crates/engineioxide/src/transport/ws.rs | 53 +++++----- .../socketioxide/docs/operators/volatile.md | 28 +++++ crates/socketioxide/src/operators.rs | 37 ++++--- crates/socketioxide/src/socket.rs | 100 ++---------------- 6 files changed, 93 insertions(+), 136 deletions(-) create mode 100644 crates/socketioxide/docs/operators/volatile.md diff --git a/crates/engineioxide/Cargo.toml b/crates/engineioxide/Cargo.toml index 45fcad51..b604481d 100644 --- a/crates/engineioxide/Cargo.toml +++ b/crates/engineioxide/Cargo.toml @@ -28,7 +28,7 @@ http-body.workspace = true serde.workspace = true serde_json.workspace = true thiserror.workspace = true -tokio = { workspace = true, features = ["rt", "time", "macros"] } +tokio = { workspace = true, features = ["rt", "time"] } tokio-util.workspace = true tower-service.workspace = true tower-layer.workspace = true diff --git a/crates/engineioxide/src/socket.rs b/crates/engineioxide/src/socket.rs index 8f938523..b046bacb 100644 --- a/crates/engineioxide/src/socket.rs +++ b/crates/engineioxide/src/socket.rs @@ -507,6 +507,9 @@ where /// /// Because volatile messages bypass the main mpsc buffer queue, they may /// arrive out of order relative to regular messages. + /// + /// Returns `true` if the message was queued for sending, `false` if it + /// was dropped (channel full or transport shutting down). #[inline] pub fn emit_volatile(&self, msg: impl Into) -> bool { self.send_volatile(smallvec![Packet::Message(msg.into())]) @@ -515,6 +518,9 @@ where /// Try to send a volatile binary message bypassing the internal buffer channel. /// Volatile messages may be dropped if the transport is not ready to /// receive them. + /// + /// Returns `true` if the message was queued for sending, `false` if it + /// was dropped. #[inline] pub fn emit_binary_volatile>(&self, data: B) -> bool { if self.protocol == ProtocolVersion::V3 { @@ -527,6 +533,9 @@ where /// Try to send a volatile message with multiple adjacent binary payloads. /// The message and all binary payloads are sent atomically as a single /// volatile write. + /// + /// Returns `true` if the message was queued for sending, `false` if it + /// was dropped. #[inline] pub fn emit_many_volatile(&self, msg: Str, data: VecDeque) -> bool { let mut packets = SmallVec::with_capacity(1 + data.len()); diff --git a/crates/engineioxide/src/transport/ws.rs b/crates/engineioxide/src/transport/ws.rs index ed59e76b..0e2ffbe0 100644 --- a/crates/engineioxide/src/transport/ws.rs +++ b/crates/engineioxide/src/transport/ws.rs @@ -290,37 +290,32 @@ async fn forward_to_socket( } loop { - tokio::select! { - // Priority: main channel is checked first so regular events - // are never starved by a flood of volatile ones. - items = internal_rx.recv() => { - match items { - Some(items) => { - for item in items { - map_fn!(item); - } - // For every available packet we continue to send until the channel is drained - while let Ok(items) = internal_rx.try_recv() { - for item in items { - map_fn!(item); - } - } - tx.flush().await.ok(); - } - None => break, - } + // Priority: wait for and drain the main channel. + // Volatile events are checked after each main-channel cycle. + let items = match internal_rx.recv().await { + Some(items) => items, + None => break, + }; + for item in items { + map_fn!(item); + } + while let Ok(items) = internal_rx.try_recv() { + for item in items { + map_fn!(item); } - // Lower priority: volatile watch channel. - // Only the latest volatile message is retained; earlier ones - // are overwritten if they haven't been consumed yet. - Ok(()) = volatile_rx.changed() => { - let value = volatile_rx.borrow_and_update().clone(); - if let Some(packets) = value { - for item in packets { - map_fn!(item); - } - tx.flush().await.ok(); + } + tx.flush().await.ok(); + + // Check volatile channel after main is drained. + // `has_changed` is non-blocking; if no volatile data is + // pending we simply go back to waiting for the main channel. + if volatile_rx.has_changed().unwrap_or(false) { + let val = volatile_rx.borrow_and_update().clone(); + if let Some(packets) = val { + for item in packets { + map_fn!(item); } + tx.flush().await.ok(); } } } diff --git a/crates/socketioxide/docs/operators/volatile.md b/crates/socketioxide/docs/operators/volatile.md new file mode 100644 index 00000000..42023754 --- /dev/null +++ b/crates/socketioxide/docs/operators/volatile.md @@ -0,0 +1,28 @@ +# Set the volatile flag for the emit. When set, the event may be dropped +if the client is not ready to receive it (e.g. the connection is buffering +or not connected). This is useful for events that are not critical, such +as position updates in a game. + +Because volatile events use a separate channel that bypasses the main +mpsc buffer, they may arrive **out of order** relative to regular events +emitted around the same time. Only use volatile when ordering relative to +regular events is not important. + +See [socket.io volatile events](https://socket.io/docs/v4/emitting-events/#volatile-events). + +# Example +```rust +# use socketioxide::{SocketIo, extract::*}; +# use serde::Serialize; +#[derive(Serialize)] +struct GameState { x: f64, y: f64 } + +let (_, io) = SocketIo::new_svc(); +io.ns("/", async |socket: SocketRef| { + // Direct volatile emit — may be dropped if the socket is not ready + socket.volatile().emit("position", &GameState { x: 1.0, y: 2.0 }).ok(); + + // Volatile broadcast to a room + socket.volatile().to("game_room").emit("update", &42).await.ok(); +}); +``` diff --git a/crates/socketioxide/src/operators.rs b/crates/socketioxide/src/operators.rs index 594b8977..c92bf534 100644 --- a/crates/socketioxide/src/operators.rs +++ b/crates/socketioxide/src/operators.rs @@ -30,6 +30,7 @@ use socketioxide_core::{ /// Chainable operators to configure the message to be sent. pub struct ConfOperators<'a, A: Adapter = LocalAdapter> { timeout: Option, + volatile: bool, socket: &'a Socket, } /// Chainable operators to select sockets to send a message to and to configure the message to be sent. @@ -42,7 +43,10 @@ pub struct BroadcastOperators { impl From> for BroadcastOperators { fn from(conf: ConfOperators<'_, A>) -> Self { - let opts = BroadcastOptions::new(conf.socket.id); + let mut opts = BroadcastOptions::new(conf.socket.id); + if conf.volatile { + opts.add_flag(BroadcastFlags::Volatile); + } Self { timeout: conf.timeout, ns: conf.socket.ns.clone(), @@ -57,6 +61,7 @@ impl<'a, A: Adapter> ConfOperators<'a, A> { pub(crate) fn new(sender: &'a Socket) -> Self { Self { timeout: None, + volatile: false, socket: sender, } } @@ -92,13 +97,10 @@ impl<'a, A: Adapter> ConfOperators<'a, A> { self } - /// Sets the volatile flag for the emit. When set, the event may be dropped - /// if the client is not ready to receive it (e.g. the connection is buffering or not connected). - /// This is useful for events that are not critical, such as position updates in a game. - /// - /// See [socket.io volatile events](https://socket.io/docs/v4/emitting-events/#volatile-events). - pub fn volatile(self) -> BroadcastOperators { - BroadcastOperators::from(self).volatile() + #[doc = include_str!("../docs/operators/volatile.md")] + pub fn volatile(mut self) -> Self { + self.volatile = true; + self } } @@ -110,6 +112,18 @@ impl ConfOperators<'_, A> { event: impl AsRef, data: &T, ) -> Result<(), SendError> { + if self.volatile { + if !self.socket.connected() { + return Ok(()); + } + let Ok(packet) = self.get_packet(event, data) else { + return Ok(()); + }; + self.socket + .send_raw_volatile(self.socket.parser.encode(packet)); + return Ok(()); + } + use crate::SocketError; use crate::socket::PermitExt; if !self.socket.connected() { @@ -238,11 +252,7 @@ impl BroadcastOperators { self } - /// Sets the volatile flag for the emit. When set, the event may be dropped - /// if the client is not ready to receive it (e.g. the connection is buffering or not connected). - /// This is useful for events that are not critical, such as position updates in a game. - /// - /// See [socket.io volatile events](https://socket.io/docs/v4/emitting-events/#volatile-events). + #[doc = include_str!("../docs/operators/volatile.md")] pub fn volatile(mut self) -> Self { self.opts.add_flag(BroadcastFlags::Volatile); self @@ -355,6 +365,7 @@ impl<'a, A: Adapter> Clone for ConfOperators<'a, A> { fn clone(&self) -> Self { Self { timeout: self.timeout, + volatile: self.volatile, socket: self.socket, } } diff --git a/crates/socketioxide/src/socket.rs b/crates/socketioxide/src/socket.rs index 7be8e0a1..1073cef5 100644 --- a/crates/socketioxide/src/socket.rs +++ b/crates/socketioxide/src/socket.rs @@ -278,90 +278,6 @@ impl From for RemoteActionError { } } -/// A wrapper around a [`Socket`] that marks emitted events as volatile. -/// -/// Volatile events may be dropped if the client is not ready to receive them -/// (e.g. the underlying connection is buffering or the socket is not connected). -/// This is useful for events that are not critical, such as position updates in a game. -/// -/// Because volatile events use a separate channel that bypasses the main -/// mpsc buffer, they may arrive **out of order** relative to regular events -/// emitted around the same time. Only use volatile when ordering relative to -/// regular events is not important. -/// -/// See [socket.io volatile events](https://socket.io/docs/v4/emitting-events/#volatile-events). -/// -/// # Example -/// ``` -/// # use socketioxide::{SocketIo, extract::*}; -/// # use serde::Serialize; -/// #[derive(Serialize)] -/// struct GameState { x: f64, y: f64 } -/// -/// let (_, io) = SocketIo::new_svc(); -/// io.ns("/", async |socket: SocketRef| { -/// // Position updates may be dropped if the connection is slow -/// socket.volatile().emit("position", &GameState { x: 1.0, y: 2.0 }); -/// }); -/// ``` -pub struct VolatileSocket<'a, A: Adapter = LocalAdapter> { - socket: &'a Socket, -} - -impl VolatileSocket<'_, A> { - /// Emit a volatile event to the client. If the socket is not connected or - /// the internal buffer is full, the event is silently dropped. - pub fn emit(&self, event: impl AsRef, data: &T) { - if !self.socket.connected() { - #[cfg(feature = "tracing")] - tracing::debug!( - ?self.socket.id, - "dropping volatile event: socket not connected" - ); - return; - } - - let ns = self.socket.ns.path.clone(); - let Ok(data) = self.socket.parser.encode_value(data, Some(event.as_ref())) else { - return; - }; - - let packet = Packet::event(ns, data); - self.socket - .send_raw_volatile(self.socket.parser.encode(packet)); - } - - #[doc = include_str!("../docs/operators/to.md")] - pub fn to(self, rooms: impl RoomParam) -> BroadcastOperators { - self.socket.to(rooms).volatile() - } - - #[doc = include_str!("../docs/operators/within.md")] - pub fn within(self, rooms: impl RoomParam) -> BroadcastOperators { - self.socket.within(rooms).volatile() - } - - #[doc = include_str!("../docs/operators/except.md")] - pub fn except(self, rooms: impl RoomParam) -> BroadcastOperators { - self.socket.except(rooms).volatile() - } - - #[doc = include_str!("../docs/operators/local.md")] - pub fn local(self) -> BroadcastOperators { - self.socket.local().volatile() - } - - #[doc = include_str!("../docs/operators/broadcast.md")] - pub fn broadcast(self) -> BroadcastOperators { - self.socket.broadcast().volatile() - } - - #[doc = include_str!("../docs/operators/timeout.md")] - pub fn timeout(self, timeout: Duration) -> BroadcastOperators { - BroadcastOperators::from(ConfOperators::new(self.socket).timeout(timeout)).volatile() - } -} - /// A Socket represents a client connected to a namespace. /// It is used to send and receive messages from the client, join and leave rooms, etc. /// The socket struct itself should not be used directly, but through a [`SocketRef`](crate::extract::SocketRef). @@ -723,11 +639,9 @@ impl Socket { BroadcastOperators::from_sock(self.ns.clone(), self.id, self.parser).broadcast() } - /// Returns a [`VolatileSocket`] wrapper that emits volatile events to this socket. - /// Volatile events may be dropped if the client is not ready to receive them - /// (e.g. the underlying connection is buffering or the socket is not connected). - /// - /// See [`VolatileSocket`] for more details. + /// Returns a [`ConfOperators`] with the volatile flag set, so that any + /// subsequent `emit()` will drop the event instead of buffering it if the + /// client is not ready to receive it. /// /// # Example /// ``` @@ -738,12 +652,12 @@ impl Socket { /// /// let (_, io) = SocketIo::new_svc(); /// io.ns("/", async |socket: SocketRef| { - /// socket.volatile().emit("position", &GameState { x: 1.0, y: 2.0 }); - /// socket.volatile().to("room1").emit("update", &42); + /// socket.volatile().emit("position", &GameState { x: 1.0, y: 2.0 }).ok(); + /// socket.volatile().to("room1").emit("update", &42).await.ok(); /// }); /// ``` - pub fn volatile(&self) -> VolatileSocket<'_, A> { - VolatileSocket { socket: self } + pub fn volatile(&self) -> ConfOperators<'_, A> { + ConfOperators::new(self).volatile() } /// # Get the [`SocketIo`] context related to this socket From 91a04de2037ee684ecc399290dfe1382b38a3c99 Mon Sep 17 00:00:00 2001 From: Liam Date: Fri, 26 Jun 2026 11:53:12 +0200 Subject: [PATCH 05/11] test: update v3_string_encoder call to include missing argument in polling payload test --- crates/engineioxide/src/transport/polling/payload/encoder.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/engineioxide/src/transport/polling/payload/encoder.rs b/crates/engineioxide/src/transport/polling/payload/encoder.rs index 406669ac..bb3e46b7 100644 --- a/crates/engineioxide/src/transport/polling/payload/encoder.rs +++ b/crates/engineioxide/src/transport/polling/payload/encoder.rs @@ -393,7 +393,7 @@ mod tests { let rx = mutex.lock().await; tx.try_send(smallvec::smallvec![Packet::Message("𝕊".into())]) .unwrap(); - let Payload { data, .. } = v3_string_encoder(rx, MAX_PAYLOAD).await.unwrap(); + let Payload { data, .. } = v3_string_encoder(rx, MAX_PAYLOAD, vec![]).await.unwrap(); assert_eq!(data, PAYLOAD.as_bytes()); } From fabc57b2fa0755caa7e4e5a30184e0d8413c94c1 Mon Sep 17 00:00:00 2001 From: Liam Date: Fri, 26 Jun 2026 19:05:51 +0200 Subject: [PATCH 06/11] refactor: replace volatile packet collection with lazy watch receiver polling in encoders to capture packets arriving during encoding --- .../engineioxide/src/transport/polling/mod.rs | 14 +- .../src/transport/polling/payload/encoder.rs | 208 +++++++++++++----- .../src/transport/polling/payload/mod.rs | 12 +- 3 files changed, 160 insertions(+), 74 deletions(-) diff --git a/crates/engineioxide/src/transport/polling/mod.rs b/crates/engineioxide/src/transport/polling/mod.rs index bb3b8979..bb8826c9 100644 --- a/crates/engineioxide/src/transport/polling/mod.rs +++ b/crates/engineioxide/src/transport/polling/mod.rs @@ -130,15 +130,7 @@ where let max_payload = engine.config.max_payload; - // Read the latest volatile packet from the watch channel, if any. - // Because the watch channel only retains the most recent value, - // any volatile messages that were overwritten since the last poll - // are automatically discarded. - let volatile_packets: Vec<_> = { - let mut rx = socket.volatile_rx.clone(); - let value = rx.borrow_and_update(); - value.as_ref().map(|p| vec![p.clone()]).unwrap_or_default() - }; + let mut volatile_rx = socket.volatile_rx.clone(); #[cfg(feature = "v3")] let Payload { data, has_binary } = payload::encoder( @@ -146,12 +138,12 @@ where protocol, socket.supports_binary, max_payload, - volatile_packets, + &mut volatile_rx, ) .await?; #[cfg(not(feature = "v3"))] let Payload { data, has_binary } = - payload::encoder(rx, protocol, max_payload, volatile_packets).await?; + payload::encoder(rx, protocol, max_payload, &mut volatile_rx).await?; #[cfg(feature = "tracing")] tracing::debug!("[sid={sid}] sending data: {:?}", data); diff --git a/crates/engineioxide/src/transport/polling/payload/encoder.rs b/crates/engineioxide/src/transport/polling/payload/encoder.rs index bb3e46b7..3b8e9096 100644 --- a/crates/engineioxide/src/transport/polling/payload/encoder.rs +++ b/crates/engineioxide/src/transport/polling/payload/encoder.rs @@ -75,7 +75,7 @@ async fn recv_packet( pub async fn v4_encoder( mut rx: MutexGuard<'_, PeekableReceiver>, max_payload: u64, - volatile_packets: Vec, + volatile_rx: &mut tokio::sync::watch::Receiver>, ) -> Result { use crate::transport::polling::payload::PACKET_SEPARATOR_V4; @@ -83,23 +83,24 @@ pub async fn v4_encoder( tracing::debug!("encoding payload with v4 encoder"); let mut data: String = String::new(); - // Encode volatile packets first so they bypass the main channel backlog - for packets in volatile_packets { - for packet in packets { - let packet: String = packet.into(); - - if !data.is_empty() { - data.push(std::char::from_u32(PACKET_SEPARATOR_V4 as u32).unwrap()); - } - data.push_str(&packet); - } - } + // Encode any pending volatile packets first so they are included in + // the current response even if the main channel has a backlog. + // The watch receiver is checked again at each iteration of the main + // channel drain loop, so volatile packets arriving during encoding + // (between .await points) are also captured. + encode_volatile_packets_v4(&mut data, volatile_rx); // Send all packets in the buffer const PUNCTUATION_LEN: usize = 1; - while let Some(packets) = - try_recv_packet(&mut rx, data.len() + PUNCTUATION_LEN, max_payload, true) - { + loop { + // Check for volatile data before each main channel read + encode_volatile_packets_v4(&mut data, volatile_rx); + + let Some(packets) = + try_recv_packet(&mut rx, data.len() + PUNCTUATION_LEN, max_payload, true) + else { + break; + }; for packet in packets { let packet: String = packet.into(); @@ -126,6 +127,30 @@ pub async fn v4_encoder( Ok(Payload::new(data.into(), false)) } +/// Encode any pending volatile packets from the watch receiver into the +/// v4 payload string. The watch is checked lazily (only when new data is +/// available) so volatile packets arriving during encoding are captured. +fn encode_volatile_packets_v4( + data: &mut String, + volatile_rx: &mut tokio::sync::watch::Receiver>, +) { + use crate::transport::polling::payload::PACKET_SEPARATOR_V4; + + if !volatile_rx.has_changed().unwrap_or(false) { + return; + } + let value = volatile_rx.borrow_and_update().clone(); + if let Some(packets) = value { + for packet in packets { + let packet: String = packet.into(); + if !data.is_empty() { + data.push(std::char::from_u32(PACKET_SEPARATOR_V4 as u32).unwrap()); + } + data.push_str(&packet); + } + } +} + /// Encode one packet into a *binary* payload according to the /// [engine.io v3 protocol](https://github.com/socketio/engine.io-protocol/tree/v3#payload) #[cfg(feature = "v3")] @@ -188,7 +213,7 @@ pub fn v3_string_packet_encoder(packet: Packet, data: &mut bytes::BytesMut) { pub async fn v3_binary_encoder( mut rx: MutexGuard<'_, PeekableReceiver>, max_payload: u64, - volatile_packets: Vec, + volatile_rx: &mut tokio::sync::watch::Receiver>, ) -> Result { let mut data = bytes::BytesMut::new(); let mut packet_buffer: Vec = Vec::new(); @@ -203,19 +228,28 @@ pub async fn v3_binary_encoder( // buffer all packets to find if there is binary packets let mut has_binary = false; - // Encode volatile packets first so they bypass the main channel backlog - for packets in volatile_packets { - for packet in packets { - if packet.is_binary() { - has_binary = true; - } - const PUNCTUATION_LEN: usize = 2; - estimated_size += packet.get_size_hint(false) + max_packet_size_len + PUNCTUATION_LEN; - packet_buffer.push(packet); - } - } + // Encode any pending volatile packets first. Checked again inside the + // main drain loop so volatile packets arriving during encoding are captured. + buffer_volatile_packets( + &mut packet_buffer, + &mut estimated_size, + &mut has_binary, + max_packet_size_len, + volatile_rx, + ); - while let Some(packets) = try_recv_packet(&mut rx, estimated_size, max_payload, false) { + loop { + buffer_volatile_packets( + &mut packet_buffer, + &mut estimated_size, + &mut has_binary, + max_packet_size_len, + volatile_rx, + ); + + let Some(packets) = try_recv_packet(&mut rx, estimated_size, max_payload, false) else { + break; + }; for packet in packets { if packet.is_binary() { has_binary = true; @@ -256,13 +290,40 @@ pub async fn v3_binary_encoder( Ok(Payload::new(data.freeze(), has_binary)) } +/// Buffer any pending volatile packets from the watch receiver into the +/// packet buffer. Called at each iteration so volatile packets arriving +/// during encoding (between .await points) are captured. +#[cfg(feature = "v3")] +fn buffer_volatile_packets( + packet_buffer: &mut Vec, + estimated_size: &mut usize, + has_binary: &mut bool, + max_packet_size_len: usize, + volatile_rx: &mut tokio::sync::watch::Receiver>, +) { + if !volatile_rx.has_changed().unwrap_or(false) { + return; + } + let value = volatile_rx.borrow_and_update().clone(); + if let Some(packets) = value { + for packet in packets { + if packet.is_binary() { + *has_binary = true; + } + const PUNCTUATION_LEN: usize = 2; + *estimated_size += packet.get_size_hint(false) + max_packet_size_len + PUNCTUATION_LEN; + packet_buffer.push(packet); + } + } +} + /// Encode multiple packet packet into a *string* payload according to the /// [engine.io v3 protocol](https://github.com/socketio/engine.io-protocol/tree/v3#payload) #[cfg(feature = "v3")] pub async fn v3_string_encoder( mut rx: MutexGuard<'_, PeekableReceiver>, max_payload: u64, - volatile_packets: Vec, + volatile_rx: &mut tokio::sync::watch::Receiver>, ) -> Result { let mut data = bytes::BytesMut::new(); @@ -273,19 +334,21 @@ pub async fn v3_string_encoder( // number of digits of the max packet size, used to approximate the payload size let max_packet_size_len = max_payload.checked_ilog10().unwrap_or(0) as usize + 1; - // Encode volatile packets first so they bypass the main channel backlog - for packets in volatile_packets { - for packet in packets { - v3_string_packet_encoder(packet, &mut data); - } - } - - while let Some(packets) = try_recv_packet( - &mut rx, - data.len() + PUNCTUATION_LEN + max_packet_size_len, - max_payload, - true, - ) { + // Encode any pending volatile packets first; checked again in the main + // drain loop so volatile packets arriving during encoding are captured. + encode_volatile_v3_string(&mut data, volatile_rx); + + loop { + encode_volatile_v3_string(&mut data, volatile_rx); + + let Some(packets) = try_recv_packet( + &mut rx, + data.len() + PUNCTUATION_LEN + max_packet_size_len, + max_payload, + true, + ) else { + break; + }; for packet in packets { v3_string_packet_encoder(packet, &mut data); } @@ -302,6 +365,25 @@ pub async fn v3_string_encoder( Ok(Payload::new(data.freeze(), false)) } +/// Encode any pending volatile packets from the watch receiver into the +/// v3 string payload. Called at each iteration so volatile packets +/// arriving during encoding are captured. +#[cfg(feature = "v3")] +fn encode_volatile_v3_string( + data: &mut bytes::BytesMut, + volatile_rx: &mut tokio::sync::watch::Receiver>, +) { + if !volatile_rx.has_changed().unwrap_or(false) { + return; + } + let value = volatile_rx.borrow_and_update().clone(); + if let Some(packets) = value { + for packet in packets { + v3_string_packet_encoder(packet, data); + } + } +} + #[cfg(test)] mod tests { use bytes::Bytes; @@ -312,8 +394,14 @@ mod tests { use super::*; const MAX_PAYLOAD: u64 = 100_000; + fn dummy_volatile_rx() -> tokio::sync::watch::Receiver> { + let (_, rx) = tokio::sync::watch::channel(None); + rx + } + #[tokio::test] async fn encode_v4_payload() { + let mut vr = dummy_volatile_rx(); const PAYLOAD: &str = "4hello€\x1ebAQIDBA==\x1e4hello€"; let (tx, rx) = tokio::sync::mpsc::channel::(10); let rx = Mutex::new(PeekableReceiver::new(rx)); @@ -326,12 +414,13 @@ mod tests { .unwrap(); tx.try_send(smallvec::smallvec![Packet::Message("hello€".into())]) .unwrap(); - let Payload { data, .. } = v4_encoder(rx, MAX_PAYLOAD, vec![]).await.unwrap(); + let Payload { data, .. } = v4_encoder(rx, MAX_PAYLOAD, &mut vr).await.unwrap(); assert_eq!(data, PAYLOAD.as_bytes()); } #[tokio::test] async fn encode_v4_payload_parked_poll_multi_packet_batch() { + let mut vr = dummy_volatile_rx(); const PAYLOAD: &str = "4hello€\x1ebAQIDBA=="; let (tx, rx) = tokio::sync::mpsc::channel::(10); let rx = Mutex::new(PeekableReceiver::new(rx)); @@ -344,12 +433,13 @@ mod tests { ]) .unwrap(); }); - let Payload { data, .. } = v4_encoder(rx, MAX_PAYLOAD, vec![]).await.unwrap(); + let Payload { data, .. } = v4_encoder(rx, MAX_PAYLOAD, &mut vr).await.unwrap(); assert_eq!(data, PAYLOAD); } #[tokio::test] async fn max_payload_v4() { + let mut vr = dummy_volatile_rx(); const MAX_PAYLOAD: u64 = 10; let (tx, rx) = tokio::sync::mpsc::channel::(10); let mutex = Mutex::new(PeekableReceiver::new(rx)); @@ -365,17 +455,17 @@ mod tests { .unwrap(); { let rx = mutex.lock().await; - let Payload { data, .. } = v4_encoder(rx, MAX_PAYLOAD, vec![]).await.unwrap(); + let Payload { data, .. } = v4_encoder(rx, MAX_PAYLOAD, &mut vr).await.unwrap(); assert_eq!(data, "4hello€".as_bytes()); } { let rx = mutex.lock().await; - let Payload { data, .. } = v4_encoder(rx, MAX_PAYLOAD + 10, vec![]).await.unwrap(); + let Payload { data, .. } = v4_encoder(rx, MAX_PAYLOAD + 10, &mut vr).await.unwrap(); assert_eq!(data, "bAQIDBA==\x1e4hello€".as_bytes()); } { let rx = mutex.lock().await; - let Payload { data, .. } = v4_encoder(rx, MAX_PAYLOAD + 10, vec![]).await.unwrap(); + let Payload { data, .. } = v4_encoder(rx, MAX_PAYLOAD + 10, &mut vr).await.unwrap(); assert_eq!(data, "4hello€".as_bytes()); } } @@ -383,6 +473,7 @@ mod tests { #[cfg(feature = "v3")] #[tokio::test] async fn encode_v3_string_payload_utf16_length() { + let mut vr = dummy_volatile_rx(); // Length must be the number of UTF-16 code units to match the // engine.io v3 JS reference implementation. The message "4𝕊" // (packet type '4' + non-BMP codepoint U+1D54A) has 2 codepoints @@ -393,13 +484,14 @@ mod tests { let rx = mutex.lock().await; tx.try_send(smallvec::smallvec![Packet::Message("𝕊".into())]) .unwrap(); - let Payload { data, .. } = v3_string_encoder(rx, MAX_PAYLOAD, vec![]).await.unwrap(); + let Payload { data, .. } = v3_string_encoder(rx, MAX_PAYLOAD, &mut vr).await.unwrap(); assert_eq!(data, PAYLOAD.as_bytes()); } #[cfg(feature = "v3")] #[tokio::test] async fn encode_v3b64_payload() { + let mut vr = dummy_volatile_rx(); const PAYLOAD: &str = "7:4hello€10:b4AQIDBA==7:4hello€"; let (tx, rx) = tokio::sync::mpsc::channel::(10); let mutex = Mutex::new(PeekableReceiver::new(rx)); @@ -415,7 +507,7 @@ mod tests { .unwrap(); let Payload { data, has_binary, .. - } = v3_string_encoder(rx, MAX_PAYLOAD, vec![]).await.unwrap(); + } = v3_string_encoder(rx, MAX_PAYLOAD, &mut vr).await.unwrap(); assert_eq!(data, PAYLOAD.as_bytes()); assert!(!has_binary); } @@ -423,6 +515,7 @@ mod tests { #[cfg(feature = "v3")] #[tokio::test] async fn max_payload_v3_b64() { + let mut vr = dummy_volatile_rx(); const MAX_PAYLOAD: u64 = 10; let (tx, rx) = tokio::sync::mpsc::channel::(10); @@ -439,12 +532,12 @@ mod tests { .unwrap(); { let rx = mutex.lock().await; - let Payload { data, .. } = v3_string_encoder(rx, MAX_PAYLOAD, vec![]).await.unwrap(); + let Payload { data, .. } = v3_string_encoder(rx, MAX_PAYLOAD, &mut vr).await.unwrap(); assert_eq!(data, "7:4hello€".as_bytes()); } { let rx = mutex.lock().await; - let Payload { data, .. } = v3_string_encoder(rx, MAX_PAYLOAD + 10, vec![]) + let Payload { data, .. } = v3_string_encoder(rx, MAX_PAYLOAD + 10, &mut vr) .await .unwrap(); assert_eq!(data, "10:b4AQIDBA==".as_bytes()); @@ -452,7 +545,7 @@ mod tests { { // Next call drains one of the remaining Message packets. let rx = mutex.lock().await; - let Payload { data, .. } = v3_string_encoder(rx, MAX_PAYLOAD + 10, vec![]) + let Payload { data, .. } = v3_string_encoder(rx, MAX_PAYLOAD + 10, &mut vr) .await .unwrap(); assert_eq!(data, "7:4hello€".as_bytes()); @@ -462,6 +555,7 @@ mod tests { #[cfg(feature = "v3")] #[tokio::test] async fn encode_v3binary_payload() { + let mut vr = dummy_volatile_rx(); const PAYLOAD: [u8; 20] = [ 0, 9, 255, 52, 104, 101, 108, 108, 111, 226, 130, 172, 1, 5, 255, 4, 1, 2, 3, 4, ]; @@ -477,7 +571,7 @@ mod tests { .unwrap(); let Payload { data, has_binary, .. - } = v3_binary_encoder(rx, MAX_PAYLOAD, vec![]).await.unwrap(); + } = v3_binary_encoder(rx, MAX_PAYLOAD, &mut vr).await.unwrap(); assert_eq!(*data, PAYLOAD); assert!(has_binary); } @@ -485,6 +579,7 @@ mod tests { #[cfg(feature = "v3")] #[tokio::test] async fn encode_v3binary_payload_parked_poll_multi_packet_batch() { + let mut vr = dummy_volatile_rx(); // When the v3 binary encoder is parked on an empty buffer and a // multi-packet batch arrives, every packet must be encoded with the // same framing (binary framing if any packet in the batch is binary). @@ -506,7 +601,7 @@ mod tests { }); let Payload { data, has_binary, .. - } = v3_binary_encoder(rx, MAX_PAYLOAD, vec![]).await.unwrap(); + } = v3_binary_encoder(rx, MAX_PAYLOAD, &mut vr).await.unwrap(); assert_eq!(*data, PAYLOAD); assert!(has_binary); } @@ -514,6 +609,7 @@ mod tests { #[cfg(feature = "v3")] #[tokio::test] async fn max_payload_v3_binary() { + let mut vr = dummy_volatile_rx(); const MAX_PAYLOAD: u64 = 25; const PAYLOAD: [u8; 23] = [ @@ -534,12 +630,12 @@ mod tests { .unwrap(); { let rx = mutex.lock().await; - let Payload { data, .. } = v3_binary_encoder(rx, MAX_PAYLOAD, vec![]).await.unwrap(); + let Payload { data, .. } = v3_binary_encoder(rx, MAX_PAYLOAD, &mut vr).await.unwrap(); assert_eq!(*data, PAYLOAD); } { let rx = mutex.lock().await; - let Payload { data, .. } = v3_binary_encoder(rx, MAX_PAYLOAD, vec![]).await.unwrap(); + let Payload { data, .. } = v3_binary_encoder(rx, MAX_PAYLOAD, &mut vr).await.unwrap(); assert_eq!(data, "7:4hello€7:4hello€".as_bytes()); } } diff --git a/crates/engineioxide/src/transport/polling/payload/mod.rs b/crates/engineioxide/src/transport/polling/payload/mod.rs index 9b95efab..beafbdb4 100644 --- a/crates/engineioxide/src/transport/polling/payload/mod.rs +++ b/crates/engineioxide/src/transport/polling/payload/mod.rs @@ -70,24 +70,22 @@ pub async fn encoder( #[allow(unused_variables)] protocol: ProtocolVersion, #[cfg(feature = "v3")] supports_binary: bool, max_payload: u64, - volatile_packets: Vec, + volatile_rx: &mut tokio::sync::watch::Receiver>, ) -> Result { #[cfg(feature = "v3")] { match protocol { - ProtocolVersion::V4 => encoder::v4_encoder(rx, max_payload, volatile_packets).await, + ProtocolVersion::V4 => encoder::v4_encoder(rx, max_payload, volatile_rx).await, ProtocolVersion::V3 if supports_binary => { - encoder::v3_binary_encoder(rx, max_payload, volatile_packets).await - } - ProtocolVersion::V3 => { - encoder::v3_string_encoder(rx, max_payload, volatile_packets).await + encoder::v3_binary_encoder(rx, max_payload, volatile_rx).await } + ProtocolVersion::V3 => encoder::v3_string_encoder(rx, max_payload, volatile_rx).await, } } #[cfg(not(feature = "v3"))] { - encoder::v4_encoder(rx, max_payload, volatile_packets).await + encoder::v4_encoder(rx, max_payload, volatile_rx).await } } From 06f547fee77ba9ea1d95fb917dea8e191feb8d1c Mon Sep 17 00:00:00 2001 From: Liam Date: Sat, 27 Jun 2026 13:44:57 +0200 Subject: [PATCH 07/11] trigger ci From 0a12ccaadb7d94ffb514c6b4ecaac1b65e41acd5 Mon Sep 17 00:00:00 2001 From: Liam Date: Mon, 29 Jun 2026 21:59:53 +0200 Subject: [PATCH 08/11] feat: implement comprehensive test coverage --- .../src/transport/polling/payload/encoder.rs | 211 ++++++++++++++++++ crates/engineioxide/tests/volatile.rs | 112 ++++++++++ crates/socketioxide/tests/volatile.rs | 82 +++++++ 3 files changed, 405 insertions(+) create mode 100644 crates/engineioxide/tests/volatile.rs create mode 100644 crates/socketioxide/tests/volatile.rs diff --git a/crates/engineioxide/src/transport/polling/payload/encoder.rs b/crates/engineioxide/src/transport/polling/payload/encoder.rs index 3b8e9096..95bcc392 100644 --- a/crates/engineioxide/src/transport/polling/payload/encoder.rs +++ b/crates/engineioxide/src/transport/polling/payload/encoder.rs @@ -639,4 +639,215 @@ mod tests { assert_eq!(data, "7:4hello€7:4hello€".as_bytes()); } } + + fn make_volatile_chan( + packets: PacketBuf, + ) -> ( + tokio::sync::watch::Sender>, + tokio::sync::watch::Receiver>, + ) { + let (tx, rx) = tokio::sync::watch::channel(None); + tx.send(Some(packets)).unwrap(); + (tx, rx) + } + + #[tokio::test] + async fn v4_volatile_before_normal() { + let (tx, rx) = tokio::sync::mpsc::channel::(10); + let mutex = Mutex::new(PeekableReceiver::new(rx)); + tx.try_send(smallvec::smallvec![Packet::Message("foo".into())]) + .unwrap(); + tx.try_send(smallvec::smallvec![Packet::Message("bar".into())]) + .unwrap(); + let (_volatile_tx, mut vr) = + make_volatile_chan(smallvec::smallvec![Packet::Message("v1".into())]); + + let rx = mutex.lock().await; + let Payload { data, .. } = v4_encoder(rx, MAX_PAYLOAD, &mut vr).await.unwrap(); + assert_eq!(data, "4v1\x1e4foo\x1e4bar".as_bytes()); + } + + #[tokio::test] + async fn v4_normal_before_volatile() { + let (tx, rx) = tokio::sync::mpsc::channel::(10); + let mutex = Mutex::new(PeekableReceiver::new(rx)); + tx.try_send(smallvec::smallvec![Packet::Message("foo".into())]) + .unwrap(); + tx.try_send(smallvec::smallvec![Packet::Message("bar".into())]) + .unwrap(); + let (_volatile_tx, mut vr) = + make_volatile_chan(smallvec::smallvec![Packet::Message("v1".into())]); + + let rx = mutex.lock().await; + let Payload { data, .. } = v4_encoder(rx, MAX_PAYLOAD, &mut vr).await.unwrap(); + assert_eq!(data, "4v1\x1e4foo\x1e4bar".as_bytes()); + } + + #[tokio::test] + async fn v4_volatile_overwrite() { + let (tx, rx) = tokio::sync::mpsc::channel::(10); + let mutex = Mutex::new(PeekableReceiver::new(rx)); + tx.try_send(smallvec::smallvec![Packet::Message("normal".into())]) + .unwrap(); + let (volatile_tx, mut vr) = tokio::sync::watch::channel(None); + volatile_tx + .send(Some(smallvec::smallvec![Packet::Message("dropped".into())])) + .unwrap(); + volatile_tx + .send(Some(smallvec::smallvec![Packet::Message("kept".into())])) + .unwrap(); + + let rx = mutex.lock().await; + let Payload { data, .. } = v4_encoder(rx, MAX_PAYLOAD, &mut vr).await.unwrap(); + assert_eq!(data, "4kept\x1e4normal".as_bytes()); + } + + #[tokio::test] + async fn v4_volatile_mid_encoding() { + let (main_tx, rx) = tokio::sync::mpsc::channel::(10); + let mutex = Mutex::new(PeekableReceiver::new(rx)); + let (volatile_tx, mut vr) = tokio::sync::watch::channel(None); + main_tx + .try_send(smallvec::smallvec![Packet::Message("first".into())]) + .unwrap(); + main_tx + .try_send(smallvec::smallvec![Packet::Message("second".into())]) + .unwrap(); + + let rx = mutex.lock().await; + volatile_tx + .send(Some(smallvec::smallvec![Packet::Message("mid".into())])) + .unwrap(); + + let Payload { data, .. } = v4_encoder(rx, MAX_PAYLOAD, &mut vr).await.unwrap(); + assert_eq!(data, "4mid\x1e4first\x1e4second".as_bytes()); + } + + #[tokio::test] + async fn v4_volatile_only_no_drain() { + let (_main_tx, rx) = tokio::sync::mpsc::channel::(10); + let mutex = Mutex::new(PeekableReceiver::new(rx)); + let (_volatile_tx, mut vr) = + make_volatile_chan(smallvec::smallvec![Packet::Message("v_only".into())]); + drop(_main_tx); + + let rx = mutex.lock().await; + let Payload { data, .. } = v4_encoder(rx, MAX_PAYLOAD, &mut vr).await.unwrap(); + assert_eq!(data, "4v_only".as_bytes()); + } + + #[cfg(feature = "v3")] + #[tokio::test] + async fn v3_string_volatile_mixed() { + let (tx, rx) = tokio::sync::mpsc::channel::(10); + let mutex = Mutex::new(PeekableReceiver::new(rx)); + tx.try_send(smallvec::smallvec![Packet::Message("foo".into())]) + .unwrap(); + tx.try_send(smallvec::smallvec![Packet::Message("bar".into())]) + .unwrap(); + let (_volatile_tx, mut vr) = + make_volatile_chan(smallvec::smallvec![Packet::Message("v1".into())]); + + let rx = mutex.lock().await; + let Payload { + data, has_binary, .. + } = v3_string_encoder(rx, MAX_PAYLOAD, &mut vr).await.unwrap(); + assert_eq!(data, "3:4v14:4foo4:4bar".as_bytes()); + assert!(!has_binary); + } + + #[cfg(feature = "v3")] + #[tokio::test] + async fn v3_string_volatile_overwrite() { + let (tx, rx) = tokio::sync::mpsc::channel::(10); + let mutex = Mutex::new(PeekableReceiver::new(rx)); + tx.try_send(smallvec::smallvec![Packet::Message("normal".into())]) + .unwrap(); + let (volatile_tx, mut vr) = tokio::sync::watch::channel(None); + volatile_tx + .send(Some(smallvec::smallvec![Packet::Message("drop1".into())])) + .unwrap(); + volatile_tx + .send(Some(smallvec::smallvec![Packet::Message("keep1".into())])) + .unwrap(); + + let rx = mutex.lock().await; + let Payload { + data, has_binary, .. + } = v3_string_encoder(rx, MAX_PAYLOAD, &mut vr).await.unwrap(); + assert_eq!(data, "6:4keep17:4normal".as_bytes()); + assert!(!has_binary); + } + + #[cfg(feature = "v3")] + #[tokio::test] + async fn v3_binary_volatile_mixed() { + let (tx, rx) = tokio::sync::mpsc::channel::(10); + let mutex = Mutex::new(PeekableReceiver::new(rx)); + tx.try_send(smallvec::smallvec![Packet::Message("foo".into())]) + .unwrap(); + let (_volatile_tx, mut vr) = make_volatile_chan(smallvec::smallvec![Packet::BinaryV3( + Bytes::from_static(&[1, 2, 3, 4]) + )]); + + let rx = mutex.lock().await; + let Payload { + data, has_binary, .. + } = v3_binary_encoder(rx, MAX_PAYLOAD, &mut vr).await.unwrap(); + assert!(has_binary); + assert_eq!( + &data[..8], + &[0x01, 0x05, 0xff, 0x04, 0x01, 0x02, 0x03, 0x04][..] + ); + assert_eq!(&data[8..], &[0x00, 0x04, 0xff, 0x34, 0x66, 0x6f, 0x6f][..]); + } + + #[cfg(feature = "v3")] + #[tokio::test] + async fn v3_binary_volatile_overwrite() { + let (tx, rx) = tokio::sync::mpsc::channel::(10); + let mutex = Mutex::new(PeekableReceiver::new(rx)); + tx.try_send(smallvec::smallvec![Packet::Message("normal".into())]) + .unwrap(); + let (volatile_tx, mut vr) = tokio::sync::watch::channel(None); + volatile_tx + .send(Some(smallvec::smallvec![Packet::Message("drop_v".into())])) + .unwrap(); + volatile_tx + .send(Some(smallvec::smallvec![Packet::BinaryV3( + Bytes::from_static(&[9, 9]) + )])) + .unwrap(); + + let rx = mutex.lock().await; + let Payload { + data, has_binary, .. + } = v3_binary_encoder(rx, MAX_PAYLOAD, &mut vr).await.unwrap(); + assert!(has_binary); + assert_eq!(&data[..6], &[0x01, 0x03, 0xff, 0x04, 0x09, 0x09][..]); + assert_eq!( + &data[6..], + &[0x00, 0x07, 0xff, 0x34, 0x6e, 0x6f, 0x72, 0x6d, 0x61, 0x6c][..] + ); + } + + #[cfg(feature = "v3")] + #[tokio::test] + async fn v3_binary_volatile_determines_has_binary() { + let (tx, rx) = tokio::sync::mpsc::channel::(10); + let mutex = Mutex::new(PeekableReceiver::new(rx)); + tx.try_send(smallvec::smallvec![Packet::Message("foo".into())]) + .unwrap(); + let (_volatile_tx, mut vr) = make_volatile_chan(smallvec::smallvec![Packet::BinaryV3( + Bytes::from_static(&[1, 2, 3]) + )]); + + let rx = mutex.lock().await; + let Payload { + data: _, + has_binary, + .. + } = v3_binary_encoder(rx, MAX_PAYLOAD, &mut vr).await.unwrap(); + assert!(has_binary); + } } diff --git a/crates/engineioxide/tests/volatile.rs b/crates/engineioxide/tests/volatile.rs new file mode 100644 index 00000000..743b2807 --- /dev/null +++ b/crates/engineioxide/tests/volatile.rs @@ -0,0 +1,112 @@ +use std::sync::Arc; + +use bytes::Bytes; +use engineioxide::{ + Str, + handler::EngineIoHandler, + socket::{DisconnectReason, Socket}, +}; +use tokio::sync::mpsc; + +mod fixture; +use fixture::{create_polling_connection, create_server, send_req}; + +#[derive(Debug, Clone)] +struct VolatileHandler { + socket_tx: mpsc::Sender>>, +} +impl EngineIoHandler for VolatileHandler { + type Data = (); + fn on_connect(self: Arc, socket: Arc>) { + self.socket_tx.try_send(socket).ok(); + } + fn on_disconnect(&self, _socket: Arc>, _reason: DisconnectReason) {} + fn on_message(self: &Arc, msg: Str, socket: Arc>) { + if msg == "trigger_volatile" { + socket.emit_volatile("volatile_response"); + } else if msg == "echo" { + socket.emit(msg).ok(); + } + } + fn on_binary(self: &Arc, data: Bytes, socket: Arc>) { + socket.emit_binary(data).ok(); + } +} + +#[tokio::test] +async fn volatile_message_arrives_via_polling() { + let (socket_tx, _socket_rx) = mpsc::channel(10); + let mut svc = create_server(VolatileHandler { socket_tx }).await; + let sid = create_polling_connection(&mut svc).await; + + send_req( + &mut svc, + format!("transport=polling&sid={sid}"), + http::Method::POST, + Some("4trigger_volatile".into()), + ) + .await; + + let response = send_req( + &mut svc, + format!("transport=polling&sid={sid}"), + http::Method::GET, + None, + ) + .await; + // send_req skips the first character (packet type '4') + assert_eq!(response, "volatile_response"); +} + +#[tokio::test] +async fn mixed_volatile_and_normal_via_polling() { + let (socket_tx, mut socket_rx) = mpsc::channel(10); + let mut svc = create_server(VolatileHandler { socket_tx }).await; + let sid = create_polling_connection(&mut svc).await; + + let socket = socket_rx + .recv() + .await + .expect("socket not received from on_connect"); + + // Send a normal message AND a volatile + socket.emit("normal_msg").ok(); + assert!(socket.emit_volatile("volatile_msg")); + + let response = send_req( + &mut svc, + format!("transport=polling&sid={sid}"), + http::Method::GET, + None, + ) + .await; + // Volatile should have priority and appear first, before normal. + // send_req skips only the first character (the leading '4' of the volatile). + assert_eq!(response, "volatile_msg\x1e4normal_msg"); +} + +#[tokio::test] +async fn volatile_overwrite_only_latest_survives() { + let (socket_tx, mut socket_rx) = mpsc::channel(10); + let mut svc = create_server(VolatileHandler { socket_tx }).await; + let sid = create_polling_connection(&mut svc).await; + + let socket = socket_rx + .recv() + .await + .expect("socket not received from on_connect"); + + assert!(socket.emit_volatile("dropped")); + assert!(socket.emit_volatile("kept")); + assert!(socket.emit("normal").is_ok()); + + let response = send_req( + &mut svc, + format!("transport=polling&sid={sid}"), + http::Method::GET, + None, + ) + .await; + // After send_req's skip(1): "kept" then \x1e separator then "4normal" + assert_eq!(response, "kept\x1e4normal"); +} diff --git a/crates/socketioxide/tests/volatile.rs b/crates/socketioxide/tests/volatile.rs new file mode 100644 index 00000000..4c176c6a --- /dev/null +++ b/crates/socketioxide/tests/volatile.rs @@ -0,0 +1,82 @@ +//! Integration tests for volatile events on socketioxide. +//! Verifies the volatile operator API and that volatile emits +//! silently drop errors rather than propagating them. +mod utils; + +use socketioxide::{SocketIo, extract::SocketRef}; + +#[tokio::test] +async fn volatile_emit_returns_ok() { + use serde_json::json; + let (_svc, io) = SocketIo::new_svc(); + + let (tx, rx) = std::sync::mpsc::channel(); + io.ns("/", async move |socket: SocketRef| { + let result = socket.volatile().emit("test", &json!({"key": "val"})); + tx.send(result).unwrap(); + }); + + io.new_dummy_sock("/", ()).await; + assert!(rx.recv().unwrap().is_ok()); +} + +#[tokio::test] +async fn volatile_emit_broadcast_does_not_panic() { + let (_svc, io) = SocketIo::new_svc(); + + let (tx, rx) = std::sync::mpsc::channel(); + io.ns("/", async move |socket: SocketRef| { + // Broadcast with volatile flag should complete without panicking + socket + .within("room") + .volatile() + .emit("event", &"data") + .await + .ok(); + tx.send(()).unwrap(); + }); + + io.new_dummy_sock("/", ()).await; + rx.recv().unwrap(); +} + +#[tokio::test] +async fn io_volatile_composes() { + let (_svc, io) = SocketIo::new_svc(); + io.ns("/", |_: SocketRef| async {}); + + io.new_dummy_sock("/", ()).await; + + // Volatile on the io handle delegates to the default namespace + let _ = io.volatile(); + let _ = io.of("/").unwrap().volatile(); +} + +#[tokio::test] +async fn volatile_emit_on_disconnected_socket_returns_ok() { + use serde_json::json; + let (_svc, io) = SocketIo::new_svc(); + + let (tx, rx) = std::sync::mpsc::channel(); + io.ns("/", { + let tx = tx.clone(); + async move |_: SocketRef| { + tx.send(()).unwrap(); + } + }); + + io.new_dummy_sock("/", ()).await; + rx.recv().unwrap(); + + // At this point the dummy socket's connect handler has run. + // The socket is technically connected — test that volatile + // emit returns Ok(()) without error. + let (tx2, rx2) = std::sync::mpsc::channel(); + io.ns("/test", async move |socket: SocketRef| { + let result = socket.volatile().emit("event", &json!({"data": 42})); + tx2.send(result).unwrap(); + }); + + io.new_dummy_sock("/test", ()).await; + assert!(rx2.recv().unwrap().is_ok()); +} From b150bb431dcb237c893e9d017ffb19925c33ca8f Mon Sep 17 00:00:00 2001 From: Liam Date: Mon, 29 Jun 2026 23:28:29 +0200 Subject: [PATCH 09/11] fix: move websocket flush after volatile check and add payload encoder tests for volatile packets --- .../src/transport/polling/payload/encoder.rs | 171 ++++++++++++++++++ crates/engineioxide/src/transport/ws.rs | 3 +- 2 files changed, 172 insertions(+), 2 deletions(-) diff --git a/crates/engineioxide/src/transport/polling/payload/encoder.rs b/crates/engineioxide/src/transport/polling/payload/encoder.rs index 95bcc392..a1fe1588 100644 --- a/crates/engineioxide/src/transport/polling/payload/encoder.rs +++ b/crates/engineioxide/src/transport/polling/payload/encoder.rs @@ -850,4 +850,175 @@ mod tests { } = v3_binary_encoder(rx, MAX_PAYLOAD, &mut vr).await.unwrap(); assert!(has_binary); } + + #[tokio::test] + async fn v4_volatile_arrives_during_parked_poll() { + let (tx, rx) = tokio::sync::mpsc::channel::(10); + let mutex = Mutex::new(PeekableReceiver::new(rx)); + let (volatile_tx, mut vr) = tokio::sync::watch::channel(None); + let volatile_tx = std::sync::Arc::new(std::sync::Mutex::new(volatile_tx)); + + let tx_clone = tx.clone(); + let vt = volatile_tx.clone(); + tokio::spawn(async move { + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + vt.lock() + .unwrap() + .send(Some(smallvec::smallvec![Packet::Message( + "volatile".into() + )])) + .unwrap(); + tx_clone + .try_send(smallvec::smallvec![Packet::Message("normal".into())]) + .unwrap(); + }); + + let rx = mutex.lock().await; + let Payload { data, .. } = v4_encoder(rx, MAX_PAYLOAD, &mut vr).await.unwrap(); + assert_eq!(data, "4normal".as_bytes()); + + drop(tx); + let rx = mutex.lock().await; + let Payload { data, .. } = v4_encoder(rx, MAX_PAYLOAD, &mut vr).await.unwrap(); + assert_eq!(data, "4volatile".as_bytes()); + } + + #[tokio::test] + async fn v4_volatile_pushes_past_max_payload() { + const SMALL_LIMIT: u64 = 12; + let (tx, rx) = tokio::sync::mpsc::channel::(10); + let mutex = Mutex::new(PeekableReceiver::new(rx)); + let (_volatile_tx, mut vr) = + make_volatile_chan(smallvec::smallvec![Packet::Message("big_vol".into())]); + tx.try_send(smallvec::smallvec![Packet::Message("normal".into())]) + .unwrap(); + tx.try_send(smallvec::smallvec![Packet::Message("extra".into())]) + .unwrap(); + + let rx = mutex.lock().await; + let Payload { data, .. } = v4_encoder(rx, SMALL_LIMIT, &mut vr).await.unwrap(); + assert_eq!(data, "4big_vol".as_bytes()); + + let rx = mutex.lock().await; + let Payload { data, .. } = v4_encoder(rx, MAX_PAYLOAD, &mut vr).await.unwrap(); + assert_eq!(data, "4normal\x1e4extra".as_bytes()); + } + + #[cfg(feature = "v3")] + #[tokio::test] + async fn v3_string_volatile_during_parked_poll() { + let (tx, rx) = tokio::sync::mpsc::channel::(10); + let mutex = Mutex::new(PeekableReceiver::new(rx)); + let (volatile_tx, mut vr) = tokio::sync::watch::channel(None); + let volatile_tx = std::sync::Arc::new(std::sync::Mutex::new(volatile_tx)); + + let tx_clone = tx.clone(); + let vt = volatile_tx.clone(); + tokio::spawn(async move { + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + vt.lock() + .unwrap() + .send(Some(smallvec::smallvec![Packet::Message("v".into())])) + .unwrap(); + tx_clone + .try_send(smallvec::smallvec![Packet::Message("n".into())]) + .unwrap(); + }); + + let rx = mutex.lock().await; + let Payload { data, .. } = v3_string_encoder(rx, MAX_PAYLOAD, &mut vr).await.unwrap(); + assert_eq!(data, "2:4n".as_bytes()); + + drop(tx); + let rx = mutex.lock().await; + let Payload { data, .. } = v3_string_encoder(rx, MAX_PAYLOAD, &mut vr).await.unwrap(); + assert_eq!(data, "2:4v".as_bytes()); + } + + #[cfg(feature = "v3")] + #[tokio::test] + async fn v3_string_volatile_max_payload() { + const SMALL_LIMIT: u64 = 12; + let (tx, rx) = tokio::sync::mpsc::channel::(10); + let mutex = Mutex::new(PeekableReceiver::new(rx)); + let (_volatile_tx, mut vr) = + make_volatile_chan(smallvec::smallvec![Packet::Message("big".into())]); + tx.try_send(smallvec::smallvec![Packet::Message("normal".into())]) + .unwrap(); + tx.try_send(smallvec::smallvec![Packet::Message("extra".into())]) + .unwrap(); + + let rx = mutex.lock().await; + let Payload { + data, has_binary, .. + } = v3_string_encoder(rx, SMALL_LIMIT, &mut vr).await.unwrap(); + assert_eq!(data, "4:4big".as_bytes()); + assert!(!has_binary); + + let rx = mutex.lock().await; + let Payload { data, .. } = v3_string_encoder(rx, MAX_PAYLOAD, &mut vr).await.unwrap(); + assert_eq!(data, "7:4normal6:4extra".as_bytes()); + } + + #[cfg(feature = "v3")] + #[tokio::test] + async fn v3_binary_volatile_during_parked_poll() { + let (tx, rx) = tokio::sync::mpsc::channel::(10); + let mutex = Mutex::new(PeekableReceiver::new(rx)); + let (volatile_tx, mut vr) = tokio::sync::watch::channel(None); + let volatile_tx = std::sync::Arc::new(std::sync::Mutex::new(volatile_tx)); + + let tx_clone = tx.clone(); + let vt = volatile_tx.clone(); + tokio::spawn(async move { + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + vt.lock() + .unwrap() + .send(Some(smallvec::smallvec![Packet::BinaryV3( + Bytes::from_static(&[1, 2]) + )])) + .unwrap(); + tx_clone + .try_send(smallvec::smallvec![Packet::Message("n".into())]) + .unwrap(); + }); + + let rx = mutex.lock().await; + let Payload { + data, has_binary, .. + } = v3_binary_encoder(rx, MAX_PAYLOAD, &mut vr).await.unwrap(); + assert!(!has_binary); + assert_eq!(&data[..], "2:4n".as_bytes()); + + drop(tx); + let rx = mutex.lock().await; + let Payload { + data, has_binary, .. + } = v3_binary_encoder(rx, MAX_PAYLOAD, &mut vr).await.unwrap(); + assert!(has_binary); + assert_eq!(&data[..6], &[0x01, 0x03, 0xff, 0x04, 0x01, 0x02][..]); + } + + #[cfg(feature = "v3")] + #[tokio::test] + async fn v3_binary_volatile_max_payload() { + const SMALL_LIMIT: u64 = 15; + let (tx, rx) = tokio::sync::mpsc::channel::(10); + let mutex = Mutex::new(PeekableReceiver::new(rx)); + let (_volatile_tx, mut vr) = + make_volatile_chan(smallvec::smallvec![Packet::Message("big_volatile".into())]); + tx.try_send(smallvec::smallvec![Packet::Message("after".into())]) + .unwrap(); + + let rx = mutex.lock().await; + let Payload { + data, has_binary, .. + } = v3_binary_encoder(rx, SMALL_LIMIT, &mut vr).await.unwrap(); + assert!(!has_binary); + assert_eq!(data, "13:4big_volatile".as_bytes()); + + let rx = mutex.lock().await; + let Payload { data, .. } = v3_binary_encoder(rx, MAX_PAYLOAD, &mut vr).await.unwrap(); + assert_eq!(data, "6:4after".as_bytes()); + } } diff --git a/crates/engineioxide/src/transport/ws.rs b/crates/engineioxide/src/transport/ws.rs index 0e2ffbe0..c0ffbaf6 100644 --- a/crates/engineioxide/src/transport/ws.rs +++ b/crates/engineioxide/src/transport/ws.rs @@ -304,7 +304,6 @@ async fn forward_to_socket( map_fn!(item); } } - tx.flush().await.ok(); // Check volatile channel after main is drained. // `has_changed` is non-blocking; if no volatile data is @@ -315,9 +314,9 @@ async fn forward_to_socket( for item in packets { map_fn!(item); } - tx.flush().await.ok(); } } + tx.flush().await.ok(); } } /// Send a Engine.IO [`OpenPacket`] to initiate a websocket connection From 804212cb48aaa16c039e3b86e7bc7ce3e8cb0e18 Mon Sep 17 00:00:00 2001 From: Liam Date: Tue, 30 Jun 2026 01:17:35 +0200 Subject: [PATCH 10/11] fix: ensure volatile packets are correctly captured and flushed during transport polling stalls --- .../src/transport/polling/payload/encoder.rs | 70 ++++++++++++++---- crates/engineioxide/src/transport/ws.rs | 2 + crates/socketioxide/tests/volatile.rs | 74 +++++++++++++++++++ examples/Cargo.lock | 6 +- examples/whiteboard/src/main.rs | 4 +- 5 files changed, 136 insertions(+), 20 deletions(-) diff --git a/crates/engineioxide/src/transport/polling/payload/encoder.rs b/crates/engineioxide/src/transport/polling/payload/encoder.rs index a1fe1588..c509cee3 100644 --- a/crates/engineioxide/src/transport/polling/payload/encoder.rs +++ b/crates/engineioxide/src/transport/polling/payload/encoder.rs @@ -122,6 +122,17 @@ pub async fn v4_encoder( let packet: String = packet.into(); data.push_str(&packet); } + + // Check for volatile packets that arrived during the parked recv + let mut volatile_data = String::new(); + encode_volatile_packets_v4(&mut volatile_data, volatile_rx); + if !volatile_data.is_empty() { + if !data.is_empty() { + volatile_data.push(std::char::from_u32(PACKET_SEPARATOR_V4 as u32).unwrap()); + } + volatile_data.push_str(&data); + data = volatile_data; + } } Ok(Payload::new(data.into(), false)) @@ -263,11 +274,11 @@ pub async fn v3_binary_encoder( } if has_binary { - for packet in packet_buffer { + for packet in packet_buffer.drain(..) { v3_bin_packet_encoder(packet, &mut data); } } else { - for packet in packet_buffer { + for packet in packet_buffer.drain(..) { v3_string_packet_encoder(packet, &mut data); } } @@ -275,7 +286,24 @@ pub async fn v3_binary_encoder( // If there is no packet in the buffer, wait for the next packet if data.is_empty() { let packets = recv_packet(&mut rx).await?; - has_binary = packets.iter().any(|p| p.is_binary()); + has_binary = packets.iter().any(|p| p.is_binary()) || has_binary; + + // Check for volatile that arrived during the park + buffer_volatile_packets( + &mut packet_buffer, + &mut estimated_size, + &mut has_binary, + max_packet_size_len, + volatile_rx, + ); + + for packet in packet_buffer.drain(..) { + if has_binary { + v3_bin_packet_encoder(packet, &mut data); + } else { + v3_string_packet_encoder(packet, &mut data); + } + } for packet in packets { if has_binary { v3_bin_packet_encoder(packet, &mut data); @@ -360,6 +388,13 @@ pub async fn v3_string_encoder( for packet in packets { v3_string_packet_encoder(packet, &mut data); } + + let mut volatile_data = bytes::BytesMut::new(); + encode_volatile_v3_string(&mut volatile_data, volatile_rx); + if !volatile_data.is_empty() { + volatile_data.unsplit(data); + data = volatile_data; + } } Ok(Payload::new(data.freeze(), false)) @@ -875,12 +910,13 @@ mod tests { let rx = mutex.lock().await; let Payload { data, .. } = v4_encoder(rx, MAX_PAYLOAD, &mut vr).await.unwrap(); - assert_eq!(data, "4normal".as_bytes()); + // After the fix: volatile arrived during park, now captured same payload + assert_eq!(data, "4volatile\x1e4normal".as_bytes()); drop(tx); let rx = mutex.lock().await; - let Payload { data, .. } = v4_encoder(rx, MAX_PAYLOAD, &mut vr).await.unwrap(); - assert_eq!(data, "4volatile".as_bytes()); + let result = v4_encoder(rx, MAX_PAYLOAD, &mut vr).await; + assert!(result.is_err()); // no more data } #[tokio::test] @@ -927,12 +963,12 @@ mod tests { let rx = mutex.lock().await; let Payload { data, .. } = v3_string_encoder(rx, MAX_PAYLOAD, &mut vr).await.unwrap(); - assert_eq!(data, "2:4n".as_bytes()); + assert_eq!(data, "2:4v2:4n".as_bytes()); drop(tx); let rx = mutex.lock().await; - let Payload { data, .. } = v3_string_encoder(rx, MAX_PAYLOAD, &mut vr).await.unwrap(); - assert_eq!(data, "2:4v".as_bytes()); + let result = v3_string_encoder(rx, MAX_PAYLOAD, &mut vr).await; + assert!(result.is_err()); } #[cfg(feature = "v3")] @@ -987,16 +1023,18 @@ mod tests { let Payload { data, has_binary, .. } = v3_binary_encoder(rx, MAX_PAYLOAD, &mut vr).await.unwrap(); - assert!(!has_binary); - assert_eq!(&data[..], "2:4n".as_bytes()); + // After fix: volatile binary + normal message both captured, has_binary=true + assert!(has_binary); + // Volatile binary: 0x01 0x03 0xFF 0x04 0x01 0x02 (6 bytes) + // Normal message in binary frame: 0x00 0x02 0xFF 0x34 0x6E (5 bytes) + assert_eq!(data.len(), 11); + assert_eq!(&data[..6], &[0x01, 0x03, 0xff, 0x04, 0x01, 0x02][..]); + assert_eq!(&data[6..], &[0x00, 0x02, 0xff, 0x34, 0x6e][..]); drop(tx); let rx = mutex.lock().await; - let Payload { - data, has_binary, .. - } = v3_binary_encoder(rx, MAX_PAYLOAD, &mut vr).await.unwrap(); - assert!(has_binary); - assert_eq!(&data[..6], &[0x01, 0x03, 0xff, 0x04, 0x01, 0x02][..]); + let result = v3_binary_encoder(rx, MAX_PAYLOAD, &mut vr).await; + assert!(result.is_err()); } #[cfg(feature = "v3")] diff --git a/crates/engineioxide/src/transport/ws.rs b/crates/engineioxide/src/transport/ws.rs index c0ffbaf6..ec7451ef 100644 --- a/crates/engineioxide/src/transport/ws.rs +++ b/crates/engineioxide/src/transport/ws.rs @@ -311,6 +311,8 @@ async fn forward_to_socket( if volatile_rx.has_changed().unwrap_or(false) { let val = volatile_rx.borrow_and_update().clone(); if let Some(packets) = val { + #[cfg(feature = "tracing")] + tracing::info!(sid = ?socket.id, "ws volatile check: flushing {:?}", &packets); for item in packets { map_fn!(item); } diff --git a/crates/socketioxide/tests/volatile.rs b/crates/socketioxide/tests/volatile.rs index 4c176c6a..cfc273af 100644 --- a/crates/socketioxide/tests/volatile.rs +++ b/crates/socketioxide/tests/volatile.rs @@ -1,8 +1,11 @@ //! Integration tests for volatile events on socketioxide. //! Verifies the volatile operator API and that volatile emits //! silently drop errors rather than propagating them. +mod fixture; mod utils; +use fixture::{create_polling_connection, create_server, send_req}; +use http::Method; use socketioxide::{SocketIo, extract::SocketRef}; #[tokio::test] @@ -80,3 +83,74 @@ async fn volatile_emit_on_disconnected_socket_returns_ok() { io.new_dummy_sock("/test", ()).await; assert!(rx2.recv().unwrap().is_ok()); } + +#[tokio::test] +async fn volatile_broadcast_arrives_via_polling_transport() { + let (svc, io) = create_server().await; + + io.ns("/", |s: SocketRef| async move { + s.on( + "drawing", + |s: SocketRef, socketioxide::extract::Data::(data)| async move { + s.broadcast().volatile().emit("drawing", &data).await.ok(); + }, + ); + }); + + let sender_sid = create_polling_connection(&svc).await; + let receiver_sid = create_polling_connection(&svc).await; + + // Drain any queued handshake data from the connections + send_req( + &svc, + format!("transport=polling&sid={sender_sid}"), + Method::GET, + None, + ) + .await; + send_req( + &svc, + format!("transport=polling&sid={receiver_sid}"), + Method::GET, + None, + ) + .await; + // Respond to pings to keep sessions alive + send_req( + &svc, + format!("transport=polling&sid={sender_sid}"), + Method::POST, + Some("3".into()), + ) + .await; + send_req( + &svc, + format!("transport=polling&sid={receiver_sid}"), + Method::POST, + Some("3".into()), + ) + .await; + + // Send drawing event from sender + send_req( + &svc, + format!("transport=polling&sid={sender_sid}"), + Method::POST, + Some("42[\"drawing\",\"hello\"]".into()), + ) + .await; + + // Poll receiver — should receive the volatile broadcast + let response = send_req( + &svc, + format!("transport=polling&sid={receiver_sid}"), + Method::GET, + None, + ) + .await; + // send_req skips first char (engine.io message type '4') + assert!( + response.contains("drawing"), + "Expected volatile broadcast with 'drawing', got: {response}" + ); +} diff --git a/examples/Cargo.lock b/examples/Cargo.lock index 24d5140e..a941f4b9 100644 --- a/examples/Cargo.lock +++ b/examples/Cargo.lock @@ -1081,7 +1081,7 @@ dependencies = [ [[package]] name = "engineioxide" -version = "0.17.3" +version = "0.17.5" dependencies = [ "base64", "bytes", @@ -3789,7 +3789,7 @@ dependencies = [ [[package]] name = "socketioxide" -version = "0.18.3" +version = "0.18.4" dependencies = [ "bytes", "engineioxide", @@ -3814,7 +3814,7 @@ dependencies = [ [[package]] name = "socketioxide-core" -version = "0.18.0" +version = "0.18.1" dependencies = [ "arbitrary", "bytes", diff --git a/examples/whiteboard/src/main.rs b/examples/whiteboard/src/main.rs index 52ba121d..c78454ae 100644 --- a/examples/whiteboard/src/main.rs +++ b/examples/whiteboard/src/main.rs @@ -23,7 +23,9 @@ async fn main() -> Result<(), Box> { io.ns("/", async |s: SocketRef| { s.on("drawing", async |s: SocketRef, Data::(data)| { - s.broadcast().emit("drawing", &data).await.unwrap(); + info!("Drawing event received, broadcasting with volatile"); + s.broadcast().volatile().emit("drawing", &data).await.unwrap(); + info!("Volatile broadcast completed"); }); }); From e40d1ad218bbff7d62c6f2cf3646a4dfecd0fd39 Mon Sep 17 00:00:00 2001 From: Liam Date: Mon, 6 Jul 2026 18:18:43 +0200 Subject: [PATCH 11/11] refactor: improve volatile event handling in engineioxide transport using tokio::select and unify documentation via include_str --- crates/engineioxide/Cargo.toml | 2 +- crates/engineioxide/src/transport/ws.rs | 62 ++++++++++------- crates/engineioxide/tests/volatile.rs | 57 ++++++++++++++- crates/socketioxide-core/src/adapter/mod.rs | 6 +- crates/socketioxide/src/io.rs | 6 +- crates/socketioxide/src/ns.rs | 7 +- crates/socketioxide/src/socket.rs | 18 +---- crates/socketioxide/tests/volatile.rs | 77 ++++----------------- 8 files changed, 117 insertions(+), 118 deletions(-) diff --git a/crates/engineioxide/Cargo.toml b/crates/engineioxide/Cargo.toml index b604481d..45fcad51 100644 --- a/crates/engineioxide/Cargo.toml +++ b/crates/engineioxide/Cargo.toml @@ -28,7 +28,7 @@ http-body.workspace = true serde.workspace = true serde_json.workspace = true thiserror.workspace = true -tokio = { workspace = true, features = ["rt", "time"] } +tokio = { workspace = true, features = ["rt", "time", "macros"] } tokio-util.workspace = true tower-service.workspace = true tower-layer.workspace = true diff --git a/crates/engineioxide/src/transport/ws.rs b/crates/engineioxide/src/transport/ws.rs index ec7451ef..68ddb948 100644 --- a/crates/engineioxide/src/transport/ws.rs +++ b/crates/engineioxide/src/transport/ws.rs @@ -290,31 +290,45 @@ async fn forward_to_socket( } loop { - // Priority: wait for and drain the main channel. - // Volatile events are checked after each main-channel cycle. - let items = match internal_rx.recv().await { - Some(items) => items, - None => break, - }; - for item in items { - map_fn!(item); - } - while let Ok(items) = internal_rx.try_recv() { - for item in items { - map_fn!(item); + tokio::select! { + items = internal_rx.recv() => { + match items { + Some(packets) => { + for item in packets { + map_fn!(item); + } + while let Ok(packets) = internal_rx.try_recv() { + for item in packets { + map_fn!(item); + } + } + } + None => { + // Main channel closed, flush and drain remaining volatile + tx.flush().await.ok(); + if volatile_rx.has_changed().unwrap_or(false) { + let val = volatile_rx.borrow_and_update().clone(); + if let Some(packets) = val { + for item in packets { + map_fn!(item); + } + tx.flush().await.ok(); + } + } + break; + } + } } - } - - // Check volatile channel after main is drained. - // `has_changed` is non-blocking; if no volatile data is - // pending we simply go back to waiting for the main channel. - if volatile_rx.has_changed().unwrap_or(false) { - let val = volatile_rx.borrow_and_update().clone(); - if let Some(packets) = val { - #[cfg(feature = "tracing")] - tracing::info!(sid = ?socket.id, "ws volatile check: flushing {:?}", &packets); - for item in packets { - map_fn!(item); + result = volatile_rx.changed() => { + if let Ok(()) = result { + let val = volatile_rx.borrow_and_update().clone(); + if let Some(packets) = val { + #[cfg(feature = "tracing")] + tracing::debug!(sid = ?socket.id, "ws volatile flush: {} packets", packets.len()); + for item in packets { + map_fn!(item); + } + } } } } diff --git a/crates/engineioxide/tests/volatile.rs b/crates/engineioxide/tests/volatile.rs index 743b2807..3867ab2a 100644 --- a/crates/engineioxide/tests/volatile.rs +++ b/crates/engineioxide/tests/volatile.rs @@ -7,9 +7,10 @@ use engineioxide::{ socket::{DisconnectReason, Socket}, }; use tokio::sync::mpsc; +use tokio_tungstenite::tungstenite::Message; mod fixture; -use fixture::{create_polling_connection, create_server, send_req}; +use fixture::{create_polling_connection, create_server, create_ws_connection, send_req}; #[derive(Debug, Clone)] struct VolatileHandler { @@ -110,3 +111,57 @@ async fn volatile_overwrite_only_latest_survives() { // After send_req's skip(1): "kept" then \x1e separator then "4normal" assert_eq!(response, "kept\x1e4normal"); } + +#[tokio::test] +async fn volatile_only_arrives_via_polling() { + let (socket_tx, mut socket_rx) = mpsc::channel(10); + let mut svc = create_server(VolatileHandler { socket_tx }).await; + let sid = create_polling_connection(&mut svc).await; + + let socket = socket_rx + .recv() + .await + .expect("socket not received from on_connect"); + + assert!(socket.emit_volatile("volatile_only")); + + let response = send_req( + &mut svc, + format!("transport=polling&sid={sid}"), + http::Method::GET, + None, + ) + .await; + assert_eq!(response, "volatile_only"); +} + +#[tokio::test] +async fn volatile_only_arrives_via_ws() { + use std::time::Duration; + + use futures_util::StreamExt; + + let (socket_tx, mut socket_rx) = mpsc::channel(10); + let mut svc = create_server(VolatileHandler { socket_tx }).await; + let mut stream = create_ws_connection(&mut svc).await; + + let socket = socket_rx + .recv() + .await + .expect("socket not received from on_connect"); + + let _open_packet = stream.next().await.unwrap().unwrap(); + + let ping = stream.next().await.unwrap().unwrap(); + assert_eq!(ping, Message::Text("2".into())); + + assert!(socket.emit_volatile("volatile_only")); + + let volatile = tokio::time::timeout(Duration::from_millis(100), stream.next()).await; + + let msg = volatile + .expect("timeout: volatile-only packet was never flushed over websocket") + .unwrap() + .unwrap(); + assert_eq!(msg, Message::Text("4volatile_only".into())); +} diff --git a/crates/socketioxide-core/src/adapter/mod.rs b/crates/socketioxide-core/src/adapter/mod.rs index e5f5447d..9b9b5e87 100644 --- a/crates/socketioxide-core/src/adapter/mod.rs +++ b/crates/socketioxide-core/src/adapter/mod.rs @@ -210,9 +210,7 @@ pub trait SocketEmitter: Send + Sync + 'static { /// Send data to the list of socket ids with volatile semantics. /// Errors are silently discarded; packets may be dropped if the /// transport is not ready. - fn send_many_volatile(&self, sids: BroadcastIter<'_>, data: Value) { - _ = self.send_many(sids, data); - } + fn send_many_volatile(&self, sids: BroadcastIter<'_>, data: Value); /// Send data to the list of socket ids and get a stream of acks and the number of expected acks. fn send_many_with_ack( &self, @@ -794,6 +792,8 @@ mod test { Ok(()) } + fn send_many_volatile(&self, _: BroadcastIter<'_>, _: Value) {} + fn send_many_with_ack( &self, _: BroadcastIter<'_>, diff --git a/crates/socketioxide/src/io.rs b/crates/socketioxide/src/io.rs index 045822e3..02c970b6 100644 --- a/crates/socketioxide/src/io.rs +++ b/crates/socketioxide/src/io.rs @@ -591,11 +591,7 @@ impl SocketIo { self.get_default_op() } - /// _Alias for `io.of("/").unwrap().volatile()`_. If the **default namespace "/" is not found** this fn will panic! - /// - /// Sets the volatile flag on the broadcast operators. Volatile events may be dropped - /// if the client is not ready to receive them. - /// See [socket.io volatile events](https://socket.io/docs/v4/emitting-events/#volatile-events). + #[doc = include_str!("../docs/operators/volatile.md")] #[inline] pub fn volatile(&self) -> BroadcastOperators { self.get_default_op().volatile() diff --git a/crates/socketioxide/src/ns.rs b/crates/socketioxide/src/ns.rs index c4b94a0f..5ea4fc9c 100644 --- a/crates/socketioxide/src/ns.rs +++ b/crates/socketioxide/src/ns.rs @@ -273,11 +273,8 @@ impl InnerEmitter for Namespace { fn send_many_volatile(&self, sids: BroadcastIter<'_>, data: Value) { let sockets = self.sockets.read().unwrap(); - for sid in sids { - if let Some(socket) = sockets.get(&sid) { - socket.send_raw_volatile(data.clone()); - } - } + sids.filter_map(|sid| sockets.get(&sid)) + .for_each(|socket| socket.send_raw_volatile(data.clone())); } fn send_many_with_ack( diff --git a/crates/socketioxide/src/socket.rs b/crates/socketioxide/src/socket.rs index 1073cef5..57a3ce88 100644 --- a/crates/socketioxide/src/socket.rs +++ b/crates/socketioxide/src/socket.rs @@ -639,23 +639,7 @@ impl Socket { BroadcastOperators::from_sock(self.ns.clone(), self.id, self.parser).broadcast() } - /// Returns a [`ConfOperators`] with the volatile flag set, so that any - /// subsequent `emit()` will drop the event instead of buffering it if the - /// client is not ready to receive it. - /// - /// # Example - /// ``` - /// # use socketioxide::{SocketIo, extract::SocketRef}; - /// # use serde::Serialize; - /// #[derive(Serialize)] - /// struct GameState { x: f64, y: f64 } - /// - /// let (_, io) = SocketIo::new_svc(); - /// io.ns("/", async |socket: SocketRef| { - /// socket.volatile().emit("position", &GameState { x: 1.0, y: 2.0 }).ok(); - /// socket.volatile().to("room1").emit("update", &42).await.ok(); - /// }); - /// ``` + #[doc = include_str!("../docs/operators/volatile.md")] pub fn volatile(&self) -> ConfOperators<'_, A> { ConfOperators::new(self).volatile() } diff --git a/crates/socketioxide/tests/volatile.rs b/crates/socketioxide/tests/volatile.rs index cfc273af..976f597f 100644 --- a/crates/socketioxide/tests/volatile.rs +++ b/crates/socketioxide/tests/volatile.rs @@ -1,87 +1,45 @@ //! Integration tests for volatile events on socketioxide. -//! Verifies the volatile operator API and that volatile emits -//! silently drop errors rather than propagating them. +//! Verifies volatile emits flow through the transport correctly. mod fixture; mod utils; use fixture::{create_polling_connection, create_server, send_req}; use http::Method; use socketioxide::{SocketIo, extract::SocketRef}; +use tokio::sync::mpsc; #[tokio::test] async fn volatile_emit_returns_ok() { use serde_json::json; let (_svc, io) = SocketIo::new_svc(); - let (tx, rx) = std::sync::mpsc::channel(); + let (tx, mut rx) = mpsc::channel::>(1); io.ns("/", async move |socket: SocketRef| { let result = socket.volatile().emit("test", &json!({"key": "val"})); - tx.send(result).unwrap(); + tx.send(result).await.unwrap(); }); io.new_dummy_sock("/", ()).await; - assert!(rx.recv().unwrap().is_ok()); + assert!(rx.recv().await.unwrap().is_ok()); } #[tokio::test] -async fn volatile_emit_broadcast_does_not_panic() { - let (_svc, io) = SocketIo::new_svc(); - - let (tx, rx) = std::sync::mpsc::channel(); - io.ns("/", async move |socket: SocketRef| { - // Broadcast with volatile flag should complete without panicking - socket - .within("room") - .volatile() - .emit("event", &"data") - .await - .ok(); - tx.send(()).unwrap(); - }); - - io.new_dummy_sock("/", ()).await; - rx.recv().unwrap(); -} - -#[tokio::test] -async fn io_volatile_composes() { - let (_svc, io) = SocketIo::new_svc(); - io.ns("/", |_: SocketRef| async {}); - - io.new_dummy_sock("/", ()).await; - - // Volatile on the io handle delegates to the default namespace - let _ = io.volatile(); - let _ = io.of("/").unwrap().volatile(); -} - -#[tokio::test] -async fn volatile_emit_on_disconnected_socket_returns_ok() { +async fn volatile_emit_second_overwrites() { use serde_json::json; let (_svc, io) = SocketIo::new_svc(); - let (tx, rx) = std::sync::mpsc::channel(); - io.ns("/", { - let tx = tx.clone(); - async move |_: SocketRef| { - tx.send(()).unwrap(); - } + let (tx, mut rx) = mpsc::channel::<(Result<(), _>, Result<(), _>)>(1); + io.ns("/", async move |socket: SocketRef| { + let first = socket.volatile().emit("test", &json!({"key": "first"})); + let second = socket.volatile().emit("test", &json!({"key": "second"})); + tx.send((first, second)).await.unwrap(); }); io.new_dummy_sock("/", ()).await; - rx.recv().unwrap(); - - // At this point the dummy socket's connect handler has run. - // The socket is technically connected — test that volatile - // emit returns Ok(()) without error. - let (tx2, rx2) = std::sync::mpsc::channel(); - io.ns("/test", async move |socket: SocketRef| { - let result = socket.volatile().emit("event", &json!({"data": 42})); - tx2.send(result).unwrap(); - }); - - io.new_dummy_sock("/test", ()).await; - assert!(rx2.recv().unwrap().is_ok()); + let (first, _second) = rx.recv().await.unwrap(); + // Both should return Ok(()); the second emit overwrites the first + // in the watch channel (it's not "false" since the send succeeds). + assert!(first.is_ok()); } #[tokio::test] @@ -100,7 +58,6 @@ async fn volatile_broadcast_arrives_via_polling_transport() { let sender_sid = create_polling_connection(&svc).await; let receiver_sid = create_polling_connection(&svc).await; - // Drain any queued handshake data from the connections send_req( &svc, format!("transport=polling&sid={sender_sid}"), @@ -115,7 +72,6 @@ async fn volatile_broadcast_arrives_via_polling_transport() { None, ) .await; - // Respond to pings to keep sessions alive send_req( &svc, format!("transport=polling&sid={sender_sid}"), @@ -131,7 +87,6 @@ async fn volatile_broadcast_arrives_via_polling_transport() { ) .await; - // Send drawing event from sender send_req( &svc, format!("transport=polling&sid={sender_sid}"), @@ -140,7 +95,6 @@ async fn volatile_broadcast_arrives_via_polling_transport() { ) .await; - // Poll receiver — should receive the volatile broadcast let response = send_req( &svc, format!("transport=polling&sid={receiver_sid}"), @@ -148,7 +102,6 @@ async fn volatile_broadcast_arrives_via_polling_transport() { None, ) .await; - // send_req skips first char (engine.io message type '4') assert!( response.contains("drawing"), "Expected volatile broadcast with 'drawing', got: {response}"