diff --git a/crates/engineioxide/Cargo.toml b/crates/engineioxide/Cargo.toml index 030f98be..dda825d7 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/engine.rs b/crates/engineioxide/src/engine.rs index c0fdabc1..92720df5 100644 --- a/crates/engineioxide/src/engine.rs +++ b/crates/engineioxide/src/engine.rs @@ -99,7 +99,7 @@ impl EngineIo { // channel so that `Socket::closed` resolves and no more packets can be buffered. // When no transport holds the lock (e.g. polling between requests) this resolves // immediately. - socket.internal_rx.lock().await.rx.close(); + socket.internal_rx.lock().await.buffered_rx.close(); #[cfg(feature = "tracing")] tracing::debug!(sid = %socket.id, "session closed, notifying handler"); diff --git a/crates/engineioxide/src/socket.rs b/crates/engineioxide/src/socket.rs index 99c42edb..d338d928 100644 --- a/crates/engineioxide/src/socket.rs +++ b/crates/engineioxide/src/socket.rs @@ -57,7 +57,6 @@ use std::{ collections::VecDeque, fmt::Debug, - ops::{Deref, DerefMut}, sync::{ Arc, atomic::{AtomicBool, AtomicU8, Ordering}, @@ -74,6 +73,7 @@ use smallvec::{SmallVec, smallvec}; use tokio::sync::{ Mutex, mpsc::{self, Receiver, error::TrySendError}, + watch, }; pub use engineioxide_core::Sid; @@ -156,24 +156,21 @@ impl Permit<'_> { #[derive(Debug)] pub(crate) struct InternalRx { - pub rx: Receiver, - pub peeked: Option, + pub buffered_rx: Receiver, + pub peeked_packet: Option, + pub volatile_rx: watch::Receiver>, } -impl InternalRx { - fn new(rx: Receiver) -> Self { - Self { rx, peeked: None } - } -} -impl Deref for InternalRx { - type Target = Receiver; - fn deref(&self) -> &Self::Target { - &self.rx - } -} -impl DerefMut for InternalRx { - fn deref_mut(&mut self) -> &mut Self::Target { - &mut self.rx +impl InternalRx { + fn new( + buffered_rx: Receiver, + volatile_rx: watch::Receiver>, + ) -> Self { + Self { + buffered_rx, + peeked_packet: None, + volatile_rx, + } } } @@ -219,6 +216,11 @@ where /// Channel to send [PacketBuf] to the internal connection pub(crate) 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>, + /// 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>, @@ -255,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(), @@ -262,9 +265,11 @@ where transport: AtomicU8::new(transport as u8), upgrading: AtomicBool::new(false), - internal_rx: Mutex::new(InternalRx::new(internal_rx)), + internal_rx: Mutex::new(InternalRx::new(internal_rx, volatile_rx)), internal_tx, + volatile_tx, + heartbeat_rx: Mutex::new(heartbeat_rx), heartbeat_tx, cancellation_token: CancellationToken::new(), @@ -457,6 +462,7 @@ where /// Immediately closes the socket and the underlying connection. /// The socket will be removed from the `Engine` and the [`Handler`](crate::handler::EngineIoHandler) will be notified. + #[cfg_attr(feature = "tracing", tracing::instrument(skip(self)))] pub fn close(&self, reason: DisconnectReason) { // Try to send a close packet is the connection is still operational. self.send(Packet::Close).ok(); @@ -493,6 +499,55 @@ 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. + /// + /// 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())]) + } + + /// 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 { + 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. + /// + /// 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()); + 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 { @@ -553,6 +608,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, @@ -560,9 +616,11 @@ where transport: AtomicU8::new(TransportType::Websocket as u8), upgrading: AtomicBool::new(false), - internal_rx: Mutex::new(InternalRx::new(internal_rx)), + internal_rx: Mutex::new(InternalRx::new(internal_rx, volatile_rx)), internal_tx, + volatile_tx, + heartbeat_rx: Mutex::new(heartbeat_rx), heartbeat_tx, cancellation_token: CancellationToken::new(), @@ -579,7 +637,7 @@ where let sock_clone = sock.clone(); tokio::spawn(async move { let mut internal_rx = sock_clone.internal_rx.try_lock().unwrap(); - while let Some(packets) = internal_rx.recv().await { + while let Some(packets) = internal_rx.buffered_rx.recv().await { for packet in packets { tx.send(packet).await.unwrap(); } diff --git a/crates/engineioxide/src/transport/polling.rs b/crates/engineioxide/src/transport/polling.rs index 1d618e79..23bdc4bc 100644 --- a/crates/engineioxide/src/transport/polling.rs +++ b/crates/engineioxide/src/transport/polling.rs @@ -13,7 +13,7 @@ use engineioxide_core::{Packet, ProtocolVersion, Sid, TransportType}; use crate::{ DisconnectReason, body::ResponseBody, engine::EngineIo, errors::Error, - handler::EngineIoHandler, transport::make_open_packet, + handler::EngineIoHandler, socket::InternalRx, transport::make_open_packet, }; /// Create a response for http request @@ -116,16 +116,24 @@ where tracing::debug!(%sid, %protocol, supports_binary = socket.supports_binary, "polling request"); let max_payload = engine.config.max_payload; + let payload = { + let InternalRx { + buffered_rx, + volatile_rx, + peeked_packet, + } = &mut *rx; - // Prepend the packet peeked during the previous - // polling request so it is encoded first, ahead of the newly received packets. - let rx_stream = stream::iter(rx.peeked.take()).chain(rx_stream::ReceiverStream::new(&mut rx)); + // Prepend the packet peeked during the previous + // polling request so it is encoded first, ahead of the newly received packets. + let rx_stream = stream::iter(peeked_packet.take()) + .chain(rx_stream::EncoderStream::new(buffered_rx, volatile_rx)); - // Stop waiting for the next packet as soon as the session is closed. The combinator is biased - // towards the encoder completion, so any buffered close packet is still flushed to the client. - let payload = payload::encoder(rx_stream, protocol, socket.supports_binary, max_payload) - .with_cancellation_token(&socket.cancellation_token) - .await; + // Stop waiting for the next packet as soon as the session is closed. The combinator is biased + // towards the encoder completion, so any buffered close packet is still flushed to the client. + payload::encoder(rx_stream, protocol, socket.supports_binary, max_payload) + .with_cancellation_token(&socket.cancellation_token) + .await + }; let Some(Payload { data, @@ -140,7 +148,7 @@ where // set back the peeked packet so it can be read again // on the next polling request - rx.peeked = peeked; + rx.peeked_packet = peeked; #[cfg(feature = "tracing")] tracing::trace!(%sid, %protocol, supports_binary = socket.supports_binary, "sending data: {:?}", data); @@ -214,20 +222,22 @@ where mod rx_stream { use std::{ pin::Pin, - task::{Context, Poll}, + task::{Context, Poll, ready}, }; use futures_core::Stream; - use tokio::sync::mpsc::Receiver; + use pin_project_lite::pin_project; + use tokio::sync::{mpsc, watch}; + use tokio_util::sync::ReusableBoxFuture; - /// [`ReceiverStream`] is a stream that wraps a tokio::sync::mpsc::Receiver by reference. + /// [`ReceiverStream`] is a stream that wraps a [`mpsc::Receiver`] by reference. /// Allowing to use it as a stream even if it is behind a mutex. pub struct ReceiverStream<'a, T> { - inner: &'a mut Receiver, + inner: &'a mut mpsc::Receiver, } impl<'a, T> ReceiverStream<'a, T> { - pub fn new(inner: &'a mut Receiver) -> Self { + pub fn new(inner: &'a mut mpsc::Receiver) -> Self { Self { inner } } } @@ -239,4 +249,91 @@ mod rx_stream { self.inner.poll_recv(cx) } } + + type WatchFutOutput<'a, T> = ( + Result<(), watch::error::RecvError>, + &'a mut watch::Receiver, + ); + + /// Wraps a [`watch::Receiver`] into a [`Stream`] by reference. + /// Allowing to use it as a stream even if it is behind a mutex. + /// + /// Inspired by + pub struct WatchStream<'a, T> { + inner: ReusableBoxFuture<'a, WatchFutOutput<'a, T>>, + } + impl<'a, T: Send + Sync + 'static> WatchStream<'a, T> { + pub fn new(rx: &'a mut watch::Receiver) -> Self { + Self { + inner: ReusableBoxFuture::new(make_future(rx)), + } + } + } + + async fn make_future( + rx: &mut watch::Receiver, + ) -> (Result<(), watch::error::RecvError>, &mut watch::Receiver) { + let result = rx.changed().await; + (result, rx) + } + + impl<'a, T: Clone + Send + Sync + 'static> Stream for WatchStream<'a, T> { + type Item = T; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let (result, rx) = ready!(self.inner.poll(cx)); + match result { + Ok(_) => { + let received = (*rx.borrow_and_update()).clone(); + self.inner.set(make_future(rx)); + Poll::Ready(Some(received)) + } + Err(_) => { + self.inner.set(make_future(rx)); + Poll::Ready(None) + } + } + } + } + + impl Unpin for WatchStream<'_, T> {} + + pin_project! { + /// Combines a [`WatchStream`] and a [`ReceiverStream`] into a single [`Stream`]. + pub struct EncoderStream<'a, T> { + #[pin] + watch: WatchStream<'a, Option>, + #[pin] + rx: ReceiverStream<'a, T>, + } + } + + impl<'a, T: Clone + Send + Sync + 'static> EncoderStream<'a, T> { + pub fn new( + rx: &'a mut mpsc::Receiver, + watch: &'a mut watch::Receiver>, + ) -> Self { + Self { + rx: ReceiverStream::new(rx), + watch: WatchStream::new(watch), + } + } + } + + impl<'a, T: Clone + Send + Sync + 'static> Stream for EncoderStream<'a, T> { + type Item = T; + + fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let this = self.project(); + // low priority: poll the watch stream first, then the rx stream. + // same than with ws transport + match this.rx.poll_next(cx) { + Poll::Ready(v) => Poll::Ready(v), + Poll::Pending => match this.watch.poll_next(cx) { + Poll::Ready(Some(v)) => Poll::Ready(v), + Poll::Ready(None) | Poll::Pending => Poll::Pending, + }, + } + } + } } diff --git a/crates/engineioxide/src/transport/ws.rs b/crates/engineioxide/src/transport/ws.rs index f579fac3..ea7ad418 100644 --- a/crates/engineioxide/src/transport/ws.rs +++ b/crates/engineioxide/src/transport/ws.rs @@ -27,7 +27,7 @@ use engineioxide_core::{Packet, ProtocolVersion, Sid, Str, TransportType}; use crate::{ DisconnectReason, Socket, body::ResponseBody, config::EngineIoConfig, engine::EngineIo, - errors::Error, handler::EngineIoHandler, transport::make_open_packet, + errors::Error, handler::EngineIoHandler, socket::InternalRx, transport::make_open_packet, }; /// Create a response for websocket upgrade @@ -253,6 +253,11 @@ async fn forward_to_socket( S: AsyncRead + AsyncWrite + Unpin + Send + 'static, { let mut internal_rx = socket.internal_rx.try_lock().unwrap(); + let InternalRx { + buffered_rx, + volatile_rx, + .. + } = &mut *internal_rx; // map a packet to a websocket message // It is declared as a macro rather than a closure to avoid ownership issues @@ -288,30 +293,45 @@ async fn forward_to_socket( } // Forward buffered packets until the channel is closed or the session is cancelled. - while let Some(Some(items)) = internal_rx - .recv() - .with_cancellation_token(&socket.cancellation_token) - .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! { + biased; + items = buffered_rx.recv() => { + match items { + Some(packets) => { + for item in packets { + map_fn!(item); + } + while let Ok(packets) = buffered_rx.try_recv() { + for item in packets { + map_fn!(item); + } + } + } + None => break + } } + _ = socket.cancellation_token.cancelled() => break, + Ok(()) = volatile_rx.changed() => { + 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); + } + } + }, } + #[cfg(feature = "tracing")] + tracing::trace!(sid = %socket.id, "ws flush"); tx.flush().await.ok(); } - // Flush any remaining buffered packets before closing the websocket. - while let Ok(items) = internal_rx.try_recv() { - for item in items { - map_fn!(item); - } - } + #[cfg(feature = "tracing")] + tracing::trace!(sid = %socket.id, "ws closing flush"); + tx.flush().await.ok(); tx.close().await.ok(); } diff --git a/crates/engineioxide/tests/volatile.rs b/crates/engineioxide/tests/volatile.rs new file mode 100644 index 00000000..a7dee0c0 --- /dev/null +++ b/crates/engineioxide/tests/volatile.rs @@ -0,0 +1,167 @@ +use std::sync::Arc; + +use bytes::Bytes; +use engineioxide::{ + Str, + handler::EngineIoHandler, + socket::{DisconnectReason, Socket}, +}; +use tokio::sync::mpsc; +use tokio_tungstenite::tungstenite::Message; + +mod fixture; +use fixture::{create_polling_connection, create_server, create_ws_connection, 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, "normal_msg\x1e4volatile_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, "normal\x1e4kept"); +} + +#[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 7e055ed9..9b9b5e87 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. @@ -202,6 +207,10 @@ 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); /// 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, @@ -430,8 +439,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_volatile(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. @@ -777,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/docs/operators/volatile.md b/crates/socketioxide/docs/operators/volatile.md new file mode 100644 index 00000000..1a18ad56 --- /dev/null +++ b/crates/socketioxide/docs/operators/volatile.md @@ -0,0 +1,37 @@ +# Emit a message with volatile behavior + +When set, the emitted 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). + +
+The volatile operator wont have any effect if you use it with emit_with_ack(). +
+ +# 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(); + + + // Same than without volatile. + socket.volatile().to("game_room").emit_with_ack::<_, ()>("update", &42).await.ok(); +}); +``` diff --git a/crates/socketioxide/src/io.rs b/crates/socketioxide/src/io.rs index 15618278..02c970b6 100644 --- a/crates/socketioxide/src/io.rs +++ b/crates/socketioxide/src/io.rs @@ -591,6 +591,12 @@ impl SocketIo { self.get_default_op() } + #[doc = include_str!("../docs/operators/volatile.md")] + #[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/ns.rs b/crates/socketioxide/src/ns.rs index 78236875..5ea4fc9c 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,12 @@ 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(); + sids.filter_map(|sid| sockets.get(&sid)) + .for_each(|socket| socket.send_raw_volatile(data.clone())); + } + fn send_many_with_ack( &self, sids: BroadcastIter<'_>, @@ -354,6 +363,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/operators.rs b/crates/socketioxide/src/operators.rs index efdde4a8..57c10a48 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, } } @@ -91,6 +96,12 @@ impl<'a, A: Adapter> ConfOperators<'a, A> { self.timeout = Some(timeout); self } + + #[doc = include_str!("../docs/operators/volatile.md")] + pub fn volatile(mut self) -> Self { + self.volatile = true; + self + } } // ==== impl ConfOperators consume fns ==== @@ -106,6 +117,14 @@ impl ConfOperators<'_, A> { if !self.socket.connected() { return Err(SendError::Socket(SocketError::Closed)); } + + if self.volatile { + let packet = self.get_packet(event, data)?; + self.socket + .send_raw_volatile(self.socket.parser.encode(packet)); + return Ok(()); + } + let permit = match self.socket.reserve() { Ok(permit) => permit, Err(e) => { @@ -228,6 +247,12 @@ impl BroadcastOperators { self.timeout = Some(timeout); self } + + #[doc = include_str!("../docs/operators/volatile.md")] + pub fn volatile(mut self) -> Self { + self.opts.add_flag(BroadcastFlags::Volatile); + self + } } // ==== impl BroadcastOperators consume fns ==== @@ -336,6 +361,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 4d4d7b68..57a3ce88 100644 --- a/crates/socketioxide/src/socket.rs +++ b/crates/socketioxide/src/socket.rs @@ -639,6 +639,11 @@ impl Socket { BroadcastOperators::from_sock(self.ns.clone(), self.id, self.parser).broadcast() } + #[doc = include_str!("../docs/operators/volatile.md")] + pub fn volatile(&self) -> ConfOperators<'_, A> { + ConfOperators::new(self).volatile() + } + /// # Get the [`SocketIo`] context related to this socket /// /// # Panics @@ -738,6 +743,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, diff --git a/crates/socketioxide/tests/volatile.rs b/crates/socketioxide/tests/volatile.rs new file mode 100644 index 00000000..976f597f --- /dev/null +++ b/crates/socketioxide/tests/volatile.rs @@ -0,0 +1,109 @@ +//! Integration tests for volatile events on socketioxide. +//! 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, mut rx) = mpsc::channel::>(1); + io.ns("/", async move |socket: SocketRef| { + let result = socket.volatile().emit("test", &json!({"key": "val"})); + tx.send(result).await.unwrap(); + }); + + io.new_dummy_sock("/", ()).await; + assert!(rx.recv().await.unwrap().is_ok()); +} + +#[tokio::test] +async fn volatile_emit_second_overwrites() { + use serde_json::json; + let (_svc, io) = SocketIo::new_svc(); + + 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; + 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] +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; + + 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; + 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_req( + &svc, + format!("transport=polling&sid={sender_sid}"), + Method::POST, + Some("42[\"drawing\",\"hello\"]".into()), + ) + .await; + + let response = send_req( + &svc, + format!("transport=polling&sid={receiver_sid}"), + Method::GET, + None, + ) + .await; + assert!( + response.contains("drawing"), + "Expected volatile broadcast with 'drawing', got: {response}" + ); +} diff --git a/e2e/adapter/client.ts b/e2e/adapter/client.ts index 14e08739..ef02079e 100644 --- a/e2e/adapter/client.ts +++ b/e2e/adapter/client.ts @@ -42,6 +42,28 @@ describe("adapter tests", { timeout: 60000 }, () => { } }); + it("should broadcast a volatile packet sent from a socket to all other sockets", async (ctx) => { + const sockets: Socket[] = (ctx as any).sockets; + for (const socket of sockets) { + let msgs: string[] = []; + const prom = new Promise((resolve) => { + for (const socket of sockets) { + socket.once("volatile_broadcast", (data: string) => { + msgs.push(data); + if (msgs.length === sockets.length) resolve(null); + }); + } + }); + + socket.emit("volatile_broadcast"); + await prom; + assert.equal(Object.values(msgs).length, sockets.length); + for (const msg of msgs) { + assert.deepStrictEqual(msg, `hello from ${socket.id}`); + } + } + }); + it("should broadcast a packet sent from a socket to all other sockets and get an ack from each socket", async (ctx) => { const sockets: Socket[] = (ctx as any).sockets; const expected = sockets.map((s) => `ack from ${s.id}`).sort(); diff --git a/e2e/adapter/src/lib.rs b/e2e/adapter/src/lib.rs index a1564858..e779a9e6 100644 --- a/e2e/adapter/src/lib.rs +++ b/e2e/adapter/src/lib.rs @@ -22,6 +22,7 @@ pub async fn handler(s: SocketRef) { // "Broadcast" tests s.on("broadcast", broadcast); + s.on("volatile_broadcast", volatile_broadcast); s.on("fetch_sockets", fetch_sockets); s.on("broadcast_with_ack", broadcast_with_ack); s.on("disconnect_socket", disconnect_socket); @@ -46,6 +47,12 @@ async fn broadcast(io: SocketIo, s: SocketRef) { .await .unwrap(); } +async fn volatile_broadcast(io: SocketIo, s: SocketRef) { + io.volatile() + .emit("volatile_broadcast", &format!("hello from {}", s.id)) + .await + .unwrap(); +} async fn broadcast_with_ack(io: SocketIo, ack: AckSender) { let data: Vec = io .emit_with_ack("broadcast_with_ack", &()) diff --git a/examples/Cargo.lock b/examples/Cargo.lock index 63a364e6..1346a150 100644 --- a/examples/Cargo.lock +++ b/examples/Cargo.lock @@ -1120,6 +1120,7 @@ dependencies = [ "serde", "serde_json", "smallvec", + "tracing", ] [[package]] diff --git a/examples/whiteboard/src/main.rs b/examples/whiteboard/src/main.rs index 52ba121d..15b5a19d 100644 --- a/examples/whiteboard/src/main.rs +++ b/examples/whiteboard/src/main.rs @@ -23,7 +23,13 @@ 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"); }); });