From 2a6cab3fd40a84a02648e9db86d3567237e67ad2 Mon Sep 17 00:00:00 2001 From: Liam Date: Sat, 11 Jul 2026 22:05:51 +0200 Subject: [PATCH 1/9] refactor(core): move AckStream, DropStream, ResponseHandlers and ChanStream into core behind remote-adapter feature Moves the duplicated stream constructs from all three remote adapters (redis, postgres, mongodb) into socketioxide-core behind the existing remote-adapter feature flag. Removes 737 lines of duplicated code. - DropStream: generic cleanup-on-drop stream wrapper - AckStream: generic ack stream merging local + remote - ResponseHandlers: generic type alias for HashMap> - ChanStream: generic mpsc::Receiver stream wrapper Each adapter now provides a decode function passed as a fn pointer to AckStream, avoiding orphan rule issues with trait impls on foreign types like Vec. Closes #756 --- Cargo.lock | 4 + crates/socketioxide-core/Cargo.toml | 12 +- crates/socketioxide-core/src/adapter/mod.rs | 2 + .../socketioxide-core/src/adapter/stream.rs | 272 ++++++++++++++++++ crates/socketioxide-mongodb/src/lib.rs | 13 +- crates/socketioxide-mongodb/src/stream.rs | 258 ++--------------- crates/socketioxide-postgres/src/lib.rs | 14 +- crates/socketioxide-postgres/src/stream.rs | 271 ++--------------- crates/socketioxide-redis/src/lib.rs | 18 +- crates/socketioxide-redis/src/stream.rs | 238 +-------------- 10 files changed, 365 insertions(+), 737 deletions(-) create mode 100644 crates/socketioxide-core/src/adapter/stream.rs diff --git a/Cargo.lock b/Cargo.lock index e3e1ec2e..14008782 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2617,11 +2617,15 @@ dependencies = [ "bytes", "engineioxide-core", "futures-core", + "futures-util", + "pin-project-lite", "rmp-serde", "serde", "serde_json", "smallvec", "thiserror", + "tokio", + "tracing", ] [[package]] diff --git a/crates/socketioxide-core/Cargo.toml b/crates/socketioxide-core/Cargo.toml index 09bf24cf..e85dd37f 100644 --- a/crates/socketioxide-core/Cargo.toml +++ b/crates/socketioxide-core/Cargo.toml @@ -13,7 +13,12 @@ license.workspace = true readme.workspace = true [features] -remote-adapter = [] +remote-adapter = [ + "dep:pin-project-lite", + "dep:futures-util", + "dep:tokio", + "dep:tracing", +] [dependencies] bytes.workspace = true @@ -21,7 +26,11 @@ engineioxide-core = { version = "0.2", path = "../engineioxide-core" } serde.workspace = true thiserror.workspace = true futures-core.workspace = true +futures-util = { workspace = true, optional = true } +pin-project-lite = { workspace = true, optional = true } smallvec = { workspace = true, features = ["serde"] } +tokio = { workspace = true, features = ["sync", "time"], optional = true } +tracing = { workspace = true, optional = true } [target."cfg(fuzzing)".dependencies] arbitrary = { version = "1.4.2", features = ["derive"] } @@ -29,6 +38,7 @@ arbitrary = { version = "1.4.2", features = ["derive"] } [dev-dependencies] serde_json.workspace = true rmp-serde.workspace = true +tokio = { workspace = true, features = ["macros", "rt"] } [lints] workspace = true diff --git a/crates/socketioxide-core/src/adapter/mod.rs b/crates/socketioxide-core/src/adapter/mod.rs index 9b9b5e87..55b3014e 100644 --- a/crates/socketioxide-core/src/adapter/mod.rs +++ b/crates/socketioxide-core/src/adapter/mod.rs @@ -27,6 +27,8 @@ use errors::{AdapterError, BroadcastError, SocketError}; pub mod errors; #[cfg(feature = "remote-adapter")] pub mod remote_packet; +#[cfg(feature = "remote-adapter")] +pub mod stream; /// A room identifier pub type Room = Cow<'static, str>; diff --git a/crates/socketioxide-core/src/adapter/stream.rs b/crates/socketioxide-core/src/adapter/stream.rs new file mode 100644 index 00000000..ef7e6bd7 --- /dev/null +++ b/crates/socketioxide-core/src/adapter/stream.rs @@ -0,0 +1,272 @@ +use std::{ + collections::HashMap, + fmt, + pin::Pin, + sync::{Arc, Mutex}, + task::{self, Poll}, + time::Duration, +}; + +use futures_core::{FusedStream, Stream}; +use futures_util::{StreamExt, stream::TakeUntil}; +use pin_project_lite::pin_project; +use serde::de::DeserializeOwned; +use tokio::{sync::mpsc, time}; + +use crate::Sid; +use crate::adapter::AckStreamItem; +use crate::adapter::remote_packet::{Response, ResponseType}; + +/// A map of request IDs to response senders, used to route responses to the correct awaiting stream. +pub type ResponseHandlers = HashMap>; + +pin_project! { + /// A [`Stream`] that wraps an [`mpsc::Receiver`]. + pub struct ChanStream { + #[pin] + rx: mpsc::Receiver, + } +} +impl ChanStream { + /// Create a new `ChanStream` from an [`mpsc::Receiver`]. + pub fn new(rx: mpsc::Receiver) -> Self { + Self { rx } + } +} +impl Stream for ChanStream { + type Item = T; + + fn poll_next(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll> { + self.project().rx.poll_recv(cx) + } +} + +pin_project! { + /// A stream that unsubscribes from its source channel when dropped. + pub struct DropStream { + #[pin] + stream: S, + req_id: Sid, + handlers: Arc>>, + } + impl PinnedDrop for DropStream { + fn drop(this: Pin<&mut Self>) { + let stream = this.project(); + let chan = stream.req_id; + tracing::debug!(?chan, "dropping stream"); + stream.handlers.lock().unwrap().remove(chan); + } + } +} +impl DropStream { + /// Create a new `DropStream` that will remove the handler entry on drop. + pub fn new(stream: S, handlers: Arc>>, req_id: Sid) -> Self { + Self { + stream, + handlers, + req_id, + } + } +} +impl Stream for DropStream { + type Item = S::Item; + + fn poll_next(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll> { + self.project().stream.poll_next(cx) + } +} +impl FusedStream for DropStream { + fn is_terminated(&self) -> bool { + self.stream.is_terminated() + } +} + +pin_project! { + /// A stream of acknowledgement messages received from the local and remote servers. + /// It merges the local ack stream with the remote ack stream from all the servers. + // The server_cnt is the number of servers that are expected to send a AckCount message. + // It is decremented each time a AckCount message is received. + // + // The ack_cnt is the number of acks that are expected to be received. It is the sum of all the the ack counts. + // And it is decremented each time an ack is received. + // + // Therefore an exhausted stream correspond to `ack_cnt == 0` and `server_cnt == 0`. + pub struct AckStream { + #[pin] + local: S, + #[pin] + remote: DropStream, T>, + decode: fn(R::Item) -> Result, Box>, + ack_cnt: u32, + total_ack_cnt: usize, + serv_cnt: u16, + } +} + +impl AckStream { + /// Create a new `AckStream` wrapping the given local and remote streams. + pub fn new( + local: S, + remote: R, + decode: fn(R::Item) -> Result, Box>, + timeout: Duration, + serv_cnt: u16, + req_id: Sid, + handlers: Arc>>, + ) -> Self { + let remote = remote.take_until(time::sleep(timeout)); + let remote = DropStream::new(remote, handlers, req_id); + Self { + local, + remote, + decode, + ack_cnt: 0, + total_ack_cnt: 0, + serv_cnt, + } + } + + /// Create a new `AckStream` for local-only use, with an empty remote stream. + pub fn new_empty_remote(local: S, empty_remote: R, decode: fn(R::Item) -> Result, Box>) -> Self { + let handlers = Arc::new(Mutex::new(ResponseHandlers::::new())); + let remote = empty_remote.take_until(time::sleep(Duration::ZERO)); + let remote = DropStream::new(remote, handlers, Sid::ZERO); + Self { + local, + remote, + decode, + ack_cnt: 0, + total_ack_cnt: 0, + serv_cnt: 0, + } + } +} + +impl AckStream +where + Err: DeserializeOwned + fmt::Debug, + S: Stream> + FusedStream, +{ + /// Poll the remote stream. First the count of acks is received, then the acks are received. + /// We expect `serv_cnt` of `BroadcastAckCount` messages to be received, then we expect + /// `ack_cnt` of `BroadcastAck` messages. + fn poll_remote( + self: Pin<&mut Self>, + cx: &mut task::Context<'_>, + ) -> Poll>> { + if FusedStream::is_terminated(&self) { + return Poll::Ready(None); + } + let mut projection = self.project(); + loop { + match projection.remote.as_mut().poll_next(cx) { + Poll::Pending => return Poll::Pending, + Poll::Ready(None) => return Poll::Ready(None), + Poll::Ready(Some(item)) => match (projection.decode)(item) { + Ok(Response { + node_id: uid, + r#type: ResponseType::BroadcastAckCount(count), + }) if *projection.serv_cnt > 0 => { + tracing::trace!(?uid, "receiving broadcast ack count {count}"); + *projection.ack_cnt += count; + *projection.total_ack_cnt += count as usize; + *projection.serv_cnt -= 1; + } + Ok(Response { + node_id: uid, + r#type: ResponseType::BroadcastAck((sid, res)), + }) if *projection.ack_cnt > 0 => { + tracing::trace!(?uid, "receiving broadcast ack {sid} {:?}", res); + *projection.ack_cnt -= 1; + return Poll::Ready(Some((sid, res))); + } + Ok(Response { node_id: uid, .. }) => { + tracing::warn!(?uid, "unexpected response type"); + } + Err(e) => { + tracing::warn!("error decoding ack response: {e}"); + } + }, + } + } + } +} + +impl Stream for AckStream +where + Err: DeserializeOwned + fmt::Debug, + S: Stream> + FusedStream, +{ + type Item = AckStreamItem; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll> { + match self.as_mut().project().local.poll_next(cx) { + Poll::Pending => match self.poll_remote(cx) { + Poll::Pending => Poll::Pending, + Poll::Ready(Some(item)) => Poll::Ready(Some(item)), + Poll::Ready(None) => Poll::Pending, + }, + Poll::Ready(Some(item)) => Poll::Ready(Some(item)), + Poll::Ready(None) => self.poll_remote(cx), + } + } + + fn size_hint(&self) -> (usize, Option) { + let (lower, upper) = self.local.size_hint(); + (lower, upper.map(|upper| upper + self.total_ack_cnt)) + } +} + +impl FusedStream for AckStream +where + Err: DeserializeOwned + fmt::Debug, + S: Stream> + FusedStream, +{ + /// The stream is terminated if: + /// * The local stream is terminated. + /// * All the servers have sent the expected ack count. + /// * We have received all the expected acks. + fn is_terminated(&self) -> bool { + let remote_term = (self.ack_cnt == 0 && self.serv_cnt == 0) || self.remote.is_terminated(); + self.local.is_terminated() && remote_term + } +} + +impl fmt::Debug for AckStream { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("AckStream") + .field("ack_cnt", &self.ack_cnt) + .field("total_ack_cnt", &self.total_ack_cnt) + .field("serv_cnt", &self.serv_cnt) + .finish() + } +} + +#[cfg(test)] +mod tests { + use futures_core::FusedStream; + use futures_util::StreamExt; + use crate::{Sid, Value}; + + use super::AckStream; + + #[tokio::test] + async fn local_ack_stream_should_have_a_closed_remote() { + let sid = Sid::new(); + let local = futures_util::stream::once(async move { + (sid, Ok::<_, ()>(Value::Str("local".into(), None))) + }); + let empty_remote = futures_util::stream::empty::<()>(); + let stream: AckStream<_, _, (), ()> = AckStream::new_empty_remote(local, empty_remote, |_| unreachable!()); + futures_util::pin_mut!(stream); + assert_eq!(stream.ack_cnt, 0); + assert_eq!(stream.total_ack_cnt, 0); + assert_eq!(stream.serv_cnt, 0); + let data = stream.next().await; + assert!( + matches!(data, Some((id, Ok(Value::Str(msg, None)))) if id == sid && msg == "local") + ); + assert_eq!(stream.next().await, None); + assert!(stream.is_terminated()); + } +} diff --git a/crates/socketioxide-mongodb/src/lib.rs b/crates/socketioxide-mongodb/src/lib.rs index d68a38f1..60c392a8 100644 --- a/crates/socketioxide-mongodb/src/lib.rs +++ b/crates/socketioxide-mongodb/src/lib.rs @@ -110,10 +110,10 @@ use socketioxide_core::{ adapter::{ BroadcastOptions, CoreAdapter, CoreLocalAdapter, DefinedAdapter, RemoteSocketData, Room, RoomParam, SocketEmitter, Spawnable, + stream::{AckStream, ChanStream, DropStream, ResponseHandlers}, }, packet::Packet, }; -use stream::{AckStream, ChanStream, DropStream}; use tokio::sync::mpsc; use drivers::{Driver, Item, ItemHeader}; @@ -303,8 +303,6 @@ impl From> for AdapterError { } } -pub(crate) type ResponseHandlers = HashMap>; - /// The mongodb adapter with the [mongodb](drivers::mongodb::mongodb_client) driver. #[cfg(feature = "mongodb")] pub type MongoDbAdapter = CustomMongoDbAdapter; @@ -326,14 +324,14 @@ pub struct CustomMongoDbAdapter { /// A map of nodes liveness, with the last time remote nodes were seen alive. nodes_liveness: Mutex>, /// A map of response handlers used to await for responses from the remote servers. - responses: Arc>, + responses: Arc>>, } impl DefinedAdapter for CustomMongoDbAdapter {} impl CoreAdapter for CustomMongoDbAdapter { type Error = Error; type State = MongoDbAdapterCtr; - type AckStream = AckStream; + type AckStream = AckStream, Item, E::AckError>; type InitRes = InitRes; fn new(state: &Self::State, local: CoreLocalAdapter) -> Self { @@ -429,7 +427,7 @@ impl CoreAdapter for CustomMongoDbAdapter if opts.is_local(self.uid) { tracing::debug!(?opts, "broadcast with ack is local"); let (local, _) = self.local.broadcast_with_ack(packet, opts, timeout); - let stream = AckStream::new_local(local); + let stream = AckStream::new_empty_remote(local, ChanStream::::new(mpsc::channel::(1).1), stream::decode_mongodb_ack); return Ok(stream); } let req = RequestOut::new(self.uid, RequestTypeOut::BroadcastWithAck(&packet), &opts); @@ -451,7 +449,8 @@ impl CoreAdapter for CustomMongoDbAdapter Ok(AckStream::new( local, - rx, + ChanStream::new(rx), + stream::decode_mongodb_ack, timeout, remote_serv_cnt, req_id, diff --git a/crates/socketioxide-mongodb/src/stream.rs b/crates/socketioxide-mongodb/src/stream.rs index eeb64467..02c5af79 100644 --- a/crates/socketioxide-mongodb/src/stream.rs +++ b/crates/socketioxide-mongodb/src/stream.rs @@ -1,241 +1,16 @@ -use std::{ - fmt, - pin::Pin, - sync::{Arc, Mutex}, - task::{self, Poll}, - time::Duration, -}; +use std::fmt; -use futures_core::{FusedStream, Stream}; -use futures_util::{StreamExt, stream::TakeUntil}; -use pin_project_lite::pin_project; use serde::de::DeserializeOwned; -use socketioxide_core::{ - Sid, - adapter::AckStreamItem, - adapter::remote_packet::{Response, ResponseType}, -}; -use tokio::{sync::mpsc, time}; +use socketioxide_core::adapter::remote_packet::Response; -use crate::{ResponseHandlers, drivers::Item}; +use crate::drivers::Item; -pin_project! { - /// A stream of acknowledgement messages received from the local and remote servers. - /// It merges the local ack stream with the remote ack stream from all the servers. - // The server_cnt is the number of servers that are expected to send a AckCount message. - // It is decremented each time a AckCount message is received. - // - // The ack_cnt is the number of acks that are expected to be received. It is the sum of all the the ack counts. - // And it is decremented each time an ack is received. - // - // Therefore an exhausted stream correspond to `ack_cnt == 0` and `server_cnt == 0`. - pub struct AckStream { - #[pin] - local: S, - #[pin] - remote: DropStream>, - ack_cnt: u32, - total_ack_cnt: usize, - serv_cnt: u16, - } -} - -impl AckStream { - pub fn new( - local: S, - rx: mpsc::Receiver, - timeout: Duration, - serv_cnt: u16, - req_id: Sid, - handlers: Arc>, - ) -> Self { - let remote = ChanStream::new(rx).take_until(time::sleep(timeout)); - let remote = DropStream::new(remote, handlers, req_id); - Self { - local, - remote, - ack_cnt: 0, - total_ack_cnt: 0, - serv_cnt, - } - } - pub fn new_local(local: S) -> Self { - let handlers = Arc::new(Mutex::new(ResponseHandlers::new())); - let rx = mpsc::channel(1).1; - let remote = ChanStream::new(rx).take_until(time::sleep(Duration::ZERO)); - let remote = DropStream::new(remote, handlers, Sid::ZERO); - Self { - local, - remote, - ack_cnt: 0, - total_ack_cnt: 0, - serv_cnt: 0, - } - } -} -impl AckStream -where - Err: DeserializeOwned + fmt::Debug, - S: Stream> + FusedStream, -{ - /// Poll the remote stream. First the count of acks is received, then the acks are received. - /// We expect `serv_cnt` of `BroadcastAckCount` messages to be received, then we expect - /// `ack_cnt` of `BroadcastAck` messages. - fn poll_remote( - self: Pin<&mut Self>, - cx: &mut task::Context<'_>, - ) -> Poll>> { - // remote stream is not fused, so we need to check if it is terminated - if FusedStream::is_terminated(&self) { - return Poll::Ready(None); - } - let mut projection = self.project(); - loop { - match projection.remote.as_mut().poll_next(cx) { - Poll::Pending => return Poll::Pending, - Poll::Ready(None) => return Poll::Ready(None), - Poll::Ready(Some(Item { header, data, .. })) => { - let res = rmp_serde::from_slice::>(&data); - match res { - Ok(Response { - node_id: uid, - r#type: ResponseType::BroadcastAckCount(count), - }) if *projection.serv_cnt > 0 => { - tracing::trace!(?uid, ?header, "receiving broadcast ack count {count}"); - *projection.ack_cnt += count; - *projection.total_ack_cnt += count as usize; - *projection.serv_cnt -= 1; - } - Ok(Response { - node_id: uid, - r#type: ResponseType::BroadcastAck((sid, res)), - }) if *projection.ack_cnt > 0 => { - tracing::trace!( - ?uid, - ?header, - "receiving broadcast ack {sid} {:?}", - res - ); - *projection.ack_cnt -= 1; - return Poll::Ready(Some((sid, res))); - } - Ok(Response { node_id: uid, .. }) => { - tracing::warn!(?uid, ?header, "unexpected response type"); - } - Err(e) => { - tracing::warn!("error decoding ack response: {e}"); - } - } - } - } - } - } -} -impl Stream for AckStream -where - E: DeserializeOwned + fmt::Debug, - S: Stream> + FusedStream, -{ - type Item = AckStreamItem; - fn poll_next(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll> { - match self.as_mut().project().local.poll_next(cx) { - Poll::Pending => match self.poll_remote(cx) { - Poll::Pending => Poll::Pending, - Poll::Ready(Some(item)) => Poll::Ready(Some(item)), - Poll::Ready(None) => Poll::Pending, - }, - Poll::Ready(Some(item)) => Poll::Ready(Some(item)), - Poll::Ready(None) => self.poll_remote(cx), - } - } - - fn size_hint(&self) -> (usize, Option) { - let (lower, upper) = self.local.size_hint(); - (lower, upper.map(|upper| upper + self.total_ack_cnt)) - } -} - -impl FusedStream for AckStream -where - Err: DeserializeOwned + fmt::Debug, - S: Stream> + FusedStream, -{ - /// The stream is terminated if: - /// * The local stream is terminated. - /// * All the servers have sent the expected ack count. - /// * We have received all the expected acks. - fn is_terminated(&self) -> bool { - // remote stream is terminated if the timeout is reached - let remote_term = (self.ack_cnt == 0 && self.serv_cnt == 0) || self.remote.is_terminated(); - self.local.is_terminated() && remote_term - } -} -impl fmt::Debug for AckStream { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("AckStream") - .field("ack_cnt", &self.ack_cnt) - .field("total_ack_cnt", &self.total_ack_cnt) - .field("serv_cnt", &self.serv_cnt) - .finish() - } -} - -pin_project! { - /// A stream of messages received from a channel. - pub struct ChanStream { - #[pin] - rx: mpsc::Receiver - } -} -impl ChanStream { - pub fn new(rx: mpsc::Receiver) -> Self { - Self { rx } - } -} -impl Stream for ChanStream { - type Item = Item; - - fn poll_next(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll> { - self.project().rx.poll_recv(cx) - } -} -pin_project! { - /// A stream that unsubscribes from its source channel when dropped. - pub struct DropStream { - #[pin] - stream: S, - req_id: Sid, - handlers: Arc> - } - impl PinnedDrop for DropStream { - fn drop(this: Pin<&mut Self>) { - let stream = this.project(); - let chan = stream.req_id; - tracing::debug!(?chan, "dropping stream"); - stream.handlers.lock().unwrap().remove(chan); - } - } -} -impl DropStream { - pub fn new(stream: S, handlers: Arc>, req_id: Sid) -> Self { - Self { - stream, - handlers, - req_id, - } - } -} -impl Stream for DropStream { - type Item = S::Item; - - fn poll_next(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll> { - self.project().stream.poll_next(cx) - } -} -impl FusedStream for DropStream { - fn is_terminated(&self) -> bool { - self.stream.is_terminated() - } +/// Decode a MongoDB `Item`'s msgpack data into a `Response`. +pub fn decode_mongodb_ack( + item: Item, +) -> Result, Box> { + rmp_serde::from_slice::>(&item.data) + .map_err(|e| Box::new(e) as Box) } #[cfg(test)] @@ -243,8 +18,10 @@ mod tests { use futures_core::FusedStream; use futures_util::StreamExt; use socketioxide_core::{Sid, Value}; + use tokio::sync::mpsc; - use super::AckStream; + use crate::stream::decode_mongodb_ack; + use socketioxide_core::adapter::stream::{AckStream, ChanStream}; #[tokio::test] async fn local_ack_stream_should_have_a_closed_remote() { @@ -252,18 +29,15 @@ mod tests { let local = futures_util::stream::once(async move { (sid, Ok::<_, ()>(Value::Str("local".into(), None))) }); - let stream = AckStream::new_local(local); + let (_, rx) = mpsc::channel::(1); + let stream: AckStream<_, _, crate::drivers::Item, ()> = AckStream::new_empty_remote(local, ChanStream::new(rx), decode_mongodb_ack::<()>); futures_util::pin_mut!(stream); - assert_eq!(stream.ack_cnt, 0); - assert_eq!(stream.total_ack_cnt, 0); - assert_eq!(stream.serv_cnt, 0); - assert!(!stream.local.is_terminated()); - assert!(!stream.is_terminated()); + assert!(!FusedStream::is_terminated(&stream)); let data = stream.next().await; assert!( matches!(data, Some((id, Ok(Value::Str(msg, None)))) if id == sid && msg == "local") ); assert_eq!(stream.next().await, None); - assert!(stream.is_terminated()); + assert!(FusedStream::is_terminated(&stream)); } } diff --git a/crates/socketioxide-postgres/src/lib.rs b/crates/socketioxide-postgres/src/lib.rs index b10eb7cd..909e2d81 100644 --- a/crates/socketioxide-postgres/src/lib.rs +++ b/crates/socketioxide-postgres/src/lib.rs @@ -67,6 +67,7 @@ use socketioxide_core::{ RequestIn, RequestOut, RequestTypeIn, RequestTypeOut, Response, ResponseType, ResponseTypeId, }, + stream::{AckStream, ChanStream, ResponseHandlers}, }, packet::Packet, }; @@ -83,7 +84,7 @@ use tokio::{sync::mpsc, task::AbortHandle}; use crate::{ drivers::Notification, - stream::{AckPayload, AckStream, ChanStream}, + stream::AckPayload, }; pub mod drivers; @@ -333,8 +334,6 @@ pub type SqlxAdapter = CustomPostgresAdapter; pub type TokioPostgresAdapter = CustomPostgresAdapter; -type ResponseHandlers = HashMap>; - /// The postgres adapter implementation. /// It is generic over the [`Driver`] used to communicate with the postgres server. /// And over the [`SocketEmitter`] used to communicate with the local server. This allows to @@ -350,7 +349,7 @@ pub struct CustomPostgresAdapter { /// A map of nodes liveness, with the last time remote nodes were seen alive. nodes_liveness: Mutex>, /// A map of response handlers used to await for responses from the remote servers. - responses: Arc>, + responses: Arc>>, ev_stream_task: OnceLock, hb_task: OnceLock, @@ -361,7 +360,7 @@ impl DefinedAdapter for CustomPostgresAdapter {} impl CoreAdapter for CustomPostgresAdapter { type Error = Error; type State = PostgresAdapterCtr; - type AckStream = AckStream; + type AckStream = AckStream, ResponsePayload, E::AckError>; type InitRes = InitRes; fn new(state: &Self::State, local: CoreLocalAdapter) -> Self { @@ -494,7 +493,7 @@ impl CoreAdapter for CustomPostgresAdapter if opts.is_local(self.local.server_id()) { tracing::debug!(?opts, "broadcast with ack is local"); let (local, _) = self.local.broadcast_with_ack(packet, opts, timeout); - let stream = AckStream::new_local(local); + let stream = AckStream::new_empty_remote(local, futures_util::stream::empty::().boxed(), stream::decode_postgres_ack); return Ok(stream); } let req = RequestOut::new( @@ -535,6 +534,7 @@ impl CoreAdapter for CustomPostgresAdapter Ok(AckStream::new( local, remote, + stream::decode_postgres_ack, timeout, remote_serv_cnt, req_id, @@ -1231,7 +1231,7 @@ impl ResponsePacket { } } #[derive(Debug, Deserialize, Serialize)] -pub(crate) enum ResponsePayload { +pub enum ResponsePayload { Data(Box), Attachment { id: i64, is_binary: bool }, } diff --git a/crates/socketioxide-postgres/src/stream.rs b/crates/socketioxide-postgres/src/stream.rs index 0a805a2b..4618585b 100644 --- a/crates/socketioxide-postgres/src/stream.rs +++ b/crates/socketioxide-postgres/src/stream.rs @@ -1,260 +1,34 @@ -use std::{ - fmt, - pin::Pin, - sync::{Arc, Mutex}, - task::{self, Poll}, - time::Duration, -}; +use std::fmt; -use futures_core::{FusedStream, Stream, stream::BoxStream}; -use futures_util::{StreamExt, stream::TakeUntil}; -use pin_project_lite::pin_project; -use serde::{Deserialize, de::DeserializeOwned}; +use serde::Deserialize; use serde_json::value::RawValue; -use socketioxide_core::{ - Sid, - adapter::{ - AckStreamItem, - remote_packet::{Response, ResponseType}, - }, -}; -use tokio::{sync::mpsc, time}; - -use crate::{ResponseHandlers, drivers::Notification}; +use socketioxide_core::adapter::remote_packet::Response; /// A resolved ack-response payload carrying either a JSON `RawValue` (inline data or a /// JSON-encoded attachment) or raw msgpack bytes (a binary attachment). The format /// has to travel with the payload because the eventual `Response` is decoded -/// downstream in [`AckStream::poll_remote`], where the right serde codec is selected. +/// downstream, where the right serde codec is selected. pub enum AckPayload { Json(Box), MsgPack(Vec), } impl AckPayload { - fn deserialize<'de, T: Deserialize<'de>>(&'de self) -> Result> { - let v = match &self { - AckPayload::Json(rv) => serde_json::from_str::(rv.get())?, - AckPayload::MsgPack(bytes) => rmp_serde::from_slice::(bytes)?, - }; - - Ok(v) - } -} - -pin_project! { - /// A stream of acknowledgement messages received from the local and remote servers. - /// It merges the local ack stream with the remote ack stream from all the servers. - // The server_cnt is the number of servers that are expected to send a AckCount message. - // It is decremented each time a AckCount message is received. - // - // The ack_cnt is the number of acks that are expected to be received. It is the sum of all the the ack counts. - // And it is decremented each time an ack is received. - // - // Therefore an exhausted stream correspond to `ack_cnt == 0` and `server_cnt == 0`. - pub struct AckStream { - #[pin] - local: S, - #[pin] - remote: DropStream, time::Sleep>>, - ack_cnt: u32, - total_ack_cnt: usize, - serv_cnt: u16, - } -} - -impl AckStream { - /// Build an ack stream backed by a remote `ResponsePayload` channel. Attachment payloads are - /// resolved lazily, driven by polling — not by a background task — so dropping the stream - /// cancels any still-pending fetch. - pub(crate) fn new( - local: S, - remote: BoxStream<'static, AckPayload>, - timeout: Duration, - serv_cnt: u16, - req_sid: Sid, - handlers: Arc>, - ) -> Self { - let remote = remote.take_until(time::sleep(timeout)); - let remote = DropStream::new(remote, handlers, req_sid); - Self { - local, - ack_cnt: 0, - total_ack_cnt: 0, - serv_cnt, - remote, - } - } - - pub(crate) fn new_local(local: S) -> Self { - let handlers = Arc::new(Mutex::new(ResponseHandlers::new())); - let empty: BoxStream<'static, AckPayload> = futures_util::stream::empty().boxed(); - let remote = empty.take_until(time::sleep(Duration::ZERO)); - let remote = DropStream::new(remote, handlers, Sid::ZERO); - Self { - local, - remote, - ack_cnt: 0, - total_ack_cnt: 0, - serv_cnt: 0, + fn deserialize<'de, T: Deserialize<'de>>(&'de self) -> Result> { + match &self { + AckPayload::Json(rv) => serde_json::from_str::(rv.get()) + .map_err(|e| Box::new(e) as Box), + AckPayload::MsgPack(bytes) => rmp_serde::from_slice::(bytes) + .map_err(|e| Box::new(e) as Box), } } } -impl AckStream -where - Err: DeserializeOwned + fmt::Debug, - S: Stream> + FusedStream, -{ - /// Poll the remote stream. First the count of acks is receivedhen the acks are received. - /// We expect `serv_cnt` of `BroadcastAckCount` messages to be receivedhen we expect - /// `ack_cnt` of `BroadcastAck` messages. - fn poll_remote( - self: Pin<&mut Self>, - cx: &mut task::Context<'_>, - ) -> Poll>> { - // remote stream is not fused, so we need to check if it is terminated - if FusedStream::is_terminated(&self) { - return Poll::Ready(None); - } - let mut projection = self.project(); - loop { - match projection.remote.as_mut().poll_next(cx) { - Poll::Pending => return Poll::Pending, - Poll::Ready(None) => return Poll::Ready(None), - Poll::Ready(Some(notif)) => match notif.deserialize() { - Ok(Response { - node_id: uid, - r#type: ResponseType::BroadcastAckCount(count), - }) if *projection.serv_cnt > 0 => { - tracing::trace!(?uid, "receiving broadcast ack count {count}"); - *projection.ack_cnt += count; - *projection.total_ack_cnt += count as usize; - *projection.serv_cnt -= 1; - } - Ok(Response { - node_id: uid, - r#type: ResponseType::BroadcastAck((sid, res)), - }) if *projection.ack_cnt > 0 => { - tracing::trace!(?uid, "receiving broadcast ack {sid} {:?}", res); - *projection.ack_cnt -= 1; - return Poll::Ready(Some((sid, res))); - } - Ok(Response { node_id: uid, .. }) => { - tracing::warn!(?uid, "unexpected response type"); - } - Err(e) => { - tracing::warn!("error decoding ack response: {e}"); - } - }, - } - } - } -} -impl Stream for AckStream -where - E: DeserializeOwned + fmt::Debug, - S: Stream> + FusedStream, -{ - type Item = AckStreamItem; - fn poll_next(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll> { - match self.as_mut().project().local.poll_next(cx) { - Poll::Pending => match self.poll_remote(cx) { - Poll::Pending => Poll::Pending, - Poll::Ready(Some(item)) => Poll::Ready(Some(item)), - Poll::Ready(None) => Poll::Pending, - }, - Poll::Ready(Some(item)) => Poll::Ready(Some(item)), - Poll::Ready(None) => self.poll_remote(cx), - } - } - fn size_hint(&self) -> (usize, Option) { - let (lower, upper) = self.local.size_hint(); - (lower, upper.map(|upper| upper + self.total_ack_cnt)) - } -} - -impl FusedStream for AckStream -where - Err: DeserializeOwned + fmt::Debug, - S: Stream> + FusedStream, -{ - /// The stream is terminated if: - /// * The local stream is terminated. - /// * All the servers have sent the expected ack count. - /// * We have received all the expected acks. - fn is_terminated(&self) -> bool { - // remote stream is terminated if the timeout is reached - let remote_term = (self.ack_cnt == 0 && self.serv_cnt == 0) || self.remote.is_terminated(); - self.local.is_terminated() && remote_term - } -} -impl fmt::Debug for AckStream { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("AckStream") - .field("ack_cnt", &self.ack_cnt) - .field("total_ack_cnt", &self.total_ack_cnt) - .field("serv_cnt", &self.serv_cnt) - .finish() - } -} -pin_project! { - /// A stream of messages received from a channel. - pub struct ChanStream { - #[pin] - rx: mpsc::Receiver - } -} -impl ChanStream { - pub fn new(rx: mpsc::Receiver) -> Self { - Self { rx } - } -} -impl Stream for ChanStream { - type Item = T; - - fn poll_next(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll> { - self.project().rx.poll_recv(cx) - } -} - -pin_project! { - /// A stream that unsubscribes from its source channel when dropped. - pub struct DropStream { - #[pin] - stream: S, - req_id: Sid, - handlers: Arc> - } - impl PinnedDrop for DropStream { - fn drop(this: Pin<&mut Self>) { - let stream = this.project(); - let chan = stream.req_id; - tracing::debug!(?chan, "dropping stream"); - stream.handlers.lock().unwrap().remove(chan); - } - } -} -impl DropStream { - pub fn new(stream: S, handlers: Arc>, req_id: Sid) -> Self { - Self { - stream, - handlers, - req_id, - } - } -} -impl Stream for DropStream { - type Item = S::Item; - - fn poll_next(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll> { - self.project().stream.poll_next(cx) - } -} -impl FusedStream for DropStream { - fn is_terminated(&self) -> bool { - self.stream.is_terminated() - } +/// Decode an `AckPayload` into a `Response` using either JSON or msgpack. +pub fn decode_postgres_ack( + item: AckPayload, +) -> Result, Box> { + item.deserialize() } #[cfg(test)] @@ -263,7 +37,8 @@ mod tests { use futures_util::StreamExt; use socketioxide_core::{Sid, Value}; - use super::AckStream; + use crate::stream::{AckPayload, decode_postgres_ack}; + use socketioxide_core::adapter::stream::AckStream; #[tokio::test] async fn local_ack_stream_should_have_a_closed_remote() { @@ -271,18 +46,16 @@ mod tests { let local = futures_util::stream::once(async move { (sid, Ok::<_, ()>(Value::Str("local".into(), None))) }); - let stream = AckStream::<_>::new_local(local); + let empty: futures_core::stream::BoxStream<'static, AckPayload> = + futures_util::stream::empty().boxed(); + let stream: AckStream<_, _, crate::ResponsePayload, ()> = AckStream::new_empty_remote(local, empty, decode_postgres_ack::<()>); futures_util::pin_mut!(stream); - assert_eq!(stream.ack_cnt, 0); - assert_eq!(stream.total_ack_cnt, 0); - assert_eq!(stream.serv_cnt, 0); - assert!(!stream.local.is_terminated()); - assert!(!stream.is_terminated()); + assert!(!FusedStream::is_terminated(&stream)); let data = stream.next().await; assert!( matches!(data, Some((id, Ok(Value::Str(msg, None)))) if id == sid && msg == "local") ); assert_eq!(stream.next().await, None); - assert!(stream.is_terminated()); + assert!(FusedStream::is_terminated(&stream)); } } diff --git a/crates/socketioxide-redis/src/lib.rs b/crates/socketioxide-redis/src/lib.rs index a1747672..fa86bff3 100644 --- a/crates/socketioxide-redis/src/lib.rs +++ b/crates/socketioxide-redis/src/lib.rs @@ -165,7 +165,7 @@ use socketioxide_core::{ }, packet::Packet, }; -use stream::{AckStream, DropStream}; +use socketioxide_core::adapter::stream::{AckStream, DropStream, ResponseHandlers}; use tokio::{sync::mpsc, time}; /// Drivers are an abstraction over the pub/sub backend used by the adapter. @@ -346,8 +346,6 @@ impl RedisAdapterCtr { } } -pub(crate) type ResponseHandlers = HashMap>>; - /// The redis adapter with the fred driver. #[cfg(feature = "fred")] pub type FredAdapter = CustomRedisAdapter; @@ -378,14 +376,14 @@ pub struct CustomRedisAdapter { /// format: `{prefix}-request#{path}#`. req_chan: String, /// A map of response handlers used to await for responses from the remote servers. - responses: Arc>, + responses: Arc>>>, } impl DefinedAdapter for CustomRedisAdapter {} impl CoreAdapter for CustomRedisAdapter { type Error = Error; type State = RedisAdapterCtr; - type AckStream = AckStream; + type AckStream = AckStream>, Vec, E::AckError>; type InitRes = InitRes; fn new(state: &Self::State, local: CoreLocalAdapter) -> Self { @@ -504,7 +502,7 @@ impl CoreAdapter for CustomRedisAdapter { if opts.is_local(self.uid) { tracing::debug!(?opts, "broadcast with ack is local"); let (local, _) = self.local.broadcast_with_ack(packet, opts, timeout); - let stream = AckStream::new_local(local); + let stream = AckStream::new_empty_remote(local, MessageStream::>::new_empty(), stream::decode_redis_ack); return Ok(stream); } let req = RequestOut::new(self.uid, RequestTypeOut::BroadcastWithAck(&packet), &opts); @@ -528,6 +526,7 @@ impl CoreAdapter for CustomRedisAdapter { Ok(AckStream::new( local, remote, + stream::decode_redis_ack, timeout, remote_serv_cnt, req_id, @@ -959,6 +958,7 @@ pub fn read_req_id(data: &[u8]) -> Option { mod tests { use super::*; use futures_util::stream::{self, FusedStream, StreamExt}; + use crate::stream::decode_redis_ack; use socketioxide_core::{Str, Value, adapter::AckStreamItem}; use std::convert::Infallible; @@ -990,10 +990,11 @@ mod tests { fn new_stub_ack_stream( remote: MessageStream>, timeout: Duration, - ) -> AckStream>> { + ) -> AckStream>, MessageStream>, Vec, ()> { AckStream::new( stream::empty::>(), remote, + decode_redis_ack, timeout, 2, Sid::new(), @@ -1063,9 +1064,10 @@ mod tests { let handlers = Arc::new(Mutex::new(HashMap::new())); let id = Sid::new(); handlers.lock().unwrap().insert(id, tx); - let stream = AckStream::new( + let stream: AckStream<_, _, Vec, ()> = AckStream::new( stream::empty::>(), remote, + decode_redis_ack, Duration::from_secs(10), 2, id, diff --git a/crates/socketioxide-redis/src/stream.rs b/crates/socketioxide-redis/src/stream.rs index b7d24584..e296f70c 100644 --- a/crates/socketioxide-redis/src/stream.rs +++ b/crates/socketioxide-redis/src/stream.rs @@ -1,224 +1,18 @@ -use std::{ - fmt, - pin::Pin, - sync::{Arc, Mutex}, - task::{self, Poll}, - time::Duration, -}; +use std::fmt; -use futures_core::{FusedStream, Stream}; -use futures_util::{StreamExt, stream::TakeUntil}; -use pin_project_lite::pin_project; use serde::de::DeserializeOwned; use socketioxide_core::{ Sid, - adapter::AckStreamItem, - adapter::remote_packet::{Response, ResponseType}, + adapter::remote_packet::Response, }; -use tokio::time; - -use crate::{ResponseHandlers, drivers::MessageStream}; - -pin_project! { - /// A stream of acknowledgement messages received from the local and remote servers. - /// It merges the local ack stream with the remote ack stream from all the servers. - // The server_cnt is the number of servers that are expected to send a AckCount message. - // It is decremented each time a AckCount message is received. - // - // The ack_cnt is the number of acks that are expected to be received. It is the sum of all the the ack counts. - // And it is decremented each time an ack is received. - // - // Therefore an exhausted stream correspond to `ack_cnt == 0` and `server_cnt == 0`. - pub struct AckStream { - #[pin] - local: S, - #[pin] - remote: DropStream>, time::Sleep>>, - ack_cnt: u32, - total_ack_cnt: usize, - serv_cnt: u16, - } -} - -impl AckStream { - pub fn new( - local: S, - remote: MessageStream>, - timeout: Duration, - serv_cnt: u16, - req_id: Sid, - handlers: Arc>, - ) -> Self { - let remote = remote.take_until(time::sleep(timeout)); - let remote = DropStream::new(remote, handlers, req_id); - Self { - local, - remote, - ack_cnt: 0, - total_ack_cnt: 0, - serv_cnt, - } - } - pub fn new_local(local: S) -> Self { - let handlers = Arc::new(Mutex::new(ResponseHandlers::new())); - let remote = MessageStream::new_empty().take_until(time::sleep(Duration::ZERO)); - let remote = DropStream::new(remote, handlers, Sid::ZERO); - Self { - local, - remote, - ack_cnt: 0, - total_ack_cnt: 0, - serv_cnt: 0, - } - } -} -impl AckStream -where - Err: DeserializeOwned + fmt::Debug, - S: Stream> + FusedStream, -{ - /// Poll the remote stream. First the count of acks is received, then the acks are received. - /// We expect `serv_cnt` of `BroadcastAckCount` messages to be received, then we expect - /// `ack_cnt` of `BroadcastAck` messages. - fn poll_remote( - mut self: Pin<&mut Self>, - cx: &mut task::Context<'_>, - ) -> Poll>> { - // remote stream is not fused, so we need to check if it is terminated - if FusedStream::is_terminated(&self) { - return Poll::Ready(None); - } - - let projection = self.as_mut().project(); - match projection.remote.poll_next(cx) { - Poll::Pending => Poll::Pending, - Poll::Ready(None) => Poll::Ready(None), - Poll::Ready(Some(item)) => { - let res = rmp_serde::from_slice::<(Sid, Response)>(&item); - match res { - Ok(( - req_id, - Response { - node_id: uid, - r#type: ResponseType::BroadcastAckCount(count), - }, - )) if *projection.serv_cnt > 0 => { - tracing::trace!(?uid, ?req_id, "receiving broadcast ack count {count}"); - *projection.ack_cnt += count; - *projection.total_ack_cnt += count as usize; - *projection.serv_cnt -= 1; - self.poll_remote(cx) - } - Ok(( - req_id, - Response { - node_id: uid, - r#type: ResponseType::BroadcastAck((sid, res)), - }, - )) if *projection.ack_cnt > 0 => { - tracing::trace!(?uid, ?req_id, "receiving broadcast ack {sid} {:?}", res); - *projection.ack_cnt -= 1; - Poll::Ready(Some((sid, res))) - } - Ok((req_id, Response { node_id: uid, .. })) => { - tracing::warn!(?uid, ?req_id, ?self, "unexpected response type"); - self.poll_remote(cx) - } - Err(e) => { - tracing::warn!("error decoding ack response: {e}"); - self.poll_remote(cx) - } - } - } - } - } -} -impl Stream for AckStream -where - E: DeserializeOwned + fmt::Debug, - S: Stream> + FusedStream, -{ - type Item = AckStreamItem; - fn poll_next(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll> { - match self.as_mut().project().local.poll_next(cx) { - Poll::Pending => match self.poll_remote(cx) { - Poll::Pending => Poll::Pending, - Poll::Ready(Some(item)) => Poll::Ready(Some(item)), - Poll::Ready(None) => Poll::Pending, - }, - Poll::Ready(Some(item)) => Poll::Ready(Some(item)), - Poll::Ready(None) => self.poll_remote(cx), - } - } - - fn size_hint(&self) -> (usize, Option) { - let (lower, upper) = self.local.size_hint(); - (lower, upper.map(|upper| upper + self.total_ack_cnt)) - } -} - -impl FusedStream for AckStream -where - Err: DeserializeOwned + fmt::Debug, - S: Stream> + FusedStream, -{ - /// The stream is terminated if: - /// * The local stream is terminated. - /// * All the servers have sent the expected ack count. - /// * We have received all the expected acks. - fn is_terminated(&self) -> bool { - // remote stream is terminated if the timeout is reached - let remote_term = (self.ack_cnt == 0 && self.serv_cnt == 0) || self.remote.is_terminated(); - self.local.is_terminated() && remote_term - } -} -impl fmt::Debug for AckStream { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("AckStream") - .field("ack_cnt", &self.ack_cnt) - .field("total_ack_cnt", &self.total_ack_cnt) - .field("serv_cnt", &self.serv_cnt) - .finish() - } -} -pin_project! { - /// A stream that unsubscribes from its source channel when dropped. - pub struct DropStream { - #[pin] - stream: S, - req_id: Sid, - handlers: Arc> - } - impl PinnedDrop for DropStream { - fn drop(this: Pin<&mut Self>) { - let stream = this.project(); - let chan = stream.req_id; - tracing::debug!(?chan, "dropping stream"); - stream.handlers.lock().unwrap().remove(chan); - } - } -} -impl DropStream { - pub fn new(stream: S, handlers: Arc>, req_id: Sid) -> Self { - Self { - stream, - handlers, - req_id, - } - } -} -impl Stream for DropStream { - type Item = S::Item; - - fn poll_next(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll> { - self.project().stream.poll_next(cx) - } -} -impl FusedStream for DropStream { - fn is_terminated(&self) -> bool { - self.stream.is_terminated() - } +/// Decode a msgpack-encoded `(Sid, Response)` tuple from raw bytes. +pub fn decode_redis_ack( + item: Vec, +) -> Result, Box> { + rmp_serde::from_slice::<(Sid, Response)>(&item) + .map(|(_, response)| response) + .map_err(|e| Box::new(e) as Box) } #[cfg(test)] @@ -227,7 +21,9 @@ mod tests { use futures_util::StreamExt; use socketioxide_core::{Sid, Value}; - use super::AckStream; + use crate::drivers::MessageStream; + use crate::stream::decode_redis_ack; + use socketioxide_core::adapter::stream::AckStream; #[tokio::test] async fn local_ack_stream_should_have_a_closed_remote() { @@ -235,18 +31,14 @@ mod tests { let local = futures_util::stream::once(async move { (sid, Ok::<_, ()>(Value::Str("local".into(), None))) }); - let stream = AckStream::new_local(local); + let stream: AckStream<_, _, Vec, ()> = AckStream::new_empty_remote(local, MessageStream::>::new_empty(), decode_redis_ack::<()>); futures_util::pin_mut!(stream); - assert_eq!(stream.ack_cnt, 0); - assert_eq!(stream.total_ack_cnt, 0); - assert_eq!(stream.serv_cnt, 0); - assert!(!stream.local.is_terminated()); - assert!(!stream.is_terminated()); + assert!(!FusedStream::is_terminated(&stream)); let data = stream.next().await; assert!( matches!(data, Some((id, Ok(Value::Str(msg, None)))) if id == sid && msg == "local") ); assert_eq!(stream.next().await, None); - assert!(stream.is_terminated()); + assert!(FusedStream::is_terminated(&stream)); } } From 520e7ab469d5ba9939d0f9b01ca2552a92187a83 Mon Sep 17 00:00:00 2001 From: Liam Date: Sat, 11 Jul 2026 22:23:00 +0200 Subject: [PATCH 2/9] fix: add module docs and AckDecoder type alias to silence clippy warnings --- crates/socketioxide-core/src/adapter/mod.rs | 1 + crates/socketioxide-core/src/adapter/stream.rs | 9 ++++++--- 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/crates/socketioxide-core/src/adapter/mod.rs b/crates/socketioxide-core/src/adapter/mod.rs index 55b3014e..03b4f47d 100644 --- a/crates/socketioxide-core/src/adapter/mod.rs +++ b/crates/socketioxide-core/src/adapter/mod.rs @@ -28,6 +28,7 @@ pub mod errors; #[cfg(feature = "remote-adapter")] pub mod remote_packet; #[cfg(feature = "remote-adapter")] +#[doc = "Shared stream constructs (`AckStream`, `DropStream`, `ResponseHandlers`) for remote adapters."] pub mod stream; /// A room identifier diff --git a/crates/socketioxide-core/src/adapter/stream.rs b/crates/socketioxide-core/src/adapter/stream.rs index ef7e6bd7..99bc884c 100644 --- a/crates/socketioxide-core/src/adapter/stream.rs +++ b/crates/socketioxide-core/src/adapter/stream.rs @@ -20,6 +20,9 @@ use crate::adapter::remote_packet::{Response, ResponseType}; /// A map of request IDs to response senders, used to route responses to the correct awaiting stream. pub type ResponseHandlers = HashMap>; +/// Decoder function that converts a raw remote payload item into a [`Response`]. +pub type AckDecoder = fn(I) -> Result, Box>; + pin_project! { /// A [`Stream`] that wraps an [`mpsc::Receiver`]. pub struct ChanStream { @@ -96,7 +99,7 @@ pin_project! { local: S, #[pin] remote: DropStream, T>, - decode: fn(R::Item) -> Result, Box>, + decode: AckDecoder, ack_cnt: u32, total_ack_cnt: usize, serv_cnt: u16, @@ -108,7 +111,7 @@ impl AckStream { pub fn new( local: S, remote: R, - decode: fn(R::Item) -> Result, Box>, + decode: AckDecoder, timeout: Duration, serv_cnt: u16, req_id: Sid, @@ -127,7 +130,7 @@ impl AckStream { } /// Create a new `AckStream` for local-only use, with an empty remote stream. - pub fn new_empty_remote(local: S, empty_remote: R, decode: fn(R::Item) -> Result, Box>) -> Self { + pub fn new_empty_remote(local: S, empty_remote: R, decode: AckDecoder) -> Self { let handlers = Arc::new(Mutex::new(ResponseHandlers::::new())); let remote = empty_remote.take_until(time::sleep(Duration::ZERO)); let remote = DropStream::new(remote, handlers, Sid::ZERO); From 26cd198975dc77eb6ffeaa90a2a62e0c445f2f04 Mon Sep 17 00:00:00 2001 From: Liam Date: Sun, 12 Jul 2026 14:45:03 +0200 Subject: [PATCH 3/9] fix: run cargo fmt and fix postgres tokio_postgres ChanStream import --- crates/socketioxide-core/src/adapter/stream.rs | 5 +++-- crates/socketioxide-mongodb/src/lib.rs | 6 +++++- crates/socketioxide-mongodb/src/stream.rs | 3 ++- .../src/drivers/tokio_postgres.rs | 2 +- crates/socketioxide-postgres/src/lib.rs | 14 ++++++++------ crates/socketioxide-postgres/src/stream.rs | 7 +++++-- crates/socketioxide-redis/src/lib.rs | 10 +++++++--- crates/socketioxide-redis/src/stream.rs | 11 ++++++----- 8 files changed, 37 insertions(+), 21 deletions(-) diff --git a/crates/socketioxide-core/src/adapter/stream.rs b/crates/socketioxide-core/src/adapter/stream.rs index 99bc884c..6425b25d 100644 --- a/crates/socketioxide-core/src/adapter/stream.rs +++ b/crates/socketioxide-core/src/adapter/stream.rs @@ -247,9 +247,9 @@ impl fmt::Debug for AckStream { #[cfg(test)] mod tests { + use crate::{Sid, Value}; use futures_core::FusedStream; use futures_util::StreamExt; - use crate::{Sid, Value}; use super::AckStream; @@ -260,7 +260,8 @@ mod tests { (sid, Ok::<_, ()>(Value::Str("local".into(), None))) }); let empty_remote = futures_util::stream::empty::<()>(); - let stream: AckStream<_, _, (), ()> = AckStream::new_empty_remote(local, empty_remote, |_| unreachable!()); + let stream: AckStream<_, _, (), ()> = + AckStream::new_empty_remote(local, empty_remote, |_| unreachable!()); futures_util::pin_mut!(stream); assert_eq!(stream.ack_cnt, 0); assert_eq!(stream.total_ack_cnt, 0); diff --git a/crates/socketioxide-mongodb/src/lib.rs b/crates/socketioxide-mongodb/src/lib.rs index 60c392a8..5c633c94 100644 --- a/crates/socketioxide-mongodb/src/lib.rs +++ b/crates/socketioxide-mongodb/src/lib.rs @@ -427,7 +427,11 @@ impl CoreAdapter for CustomMongoDbAdapter if opts.is_local(self.uid) { tracing::debug!(?opts, "broadcast with ack is local"); let (local, _) = self.local.broadcast_with_ack(packet, opts, timeout); - let stream = AckStream::new_empty_remote(local, ChanStream::::new(mpsc::channel::(1).1), stream::decode_mongodb_ack); + let stream = AckStream::new_empty_remote( + local, + ChanStream::::new(mpsc::channel::(1).1), + stream::decode_mongodb_ack, + ); return Ok(stream); } let req = RequestOut::new(self.uid, RequestTypeOut::BroadcastWithAck(&packet), &opts); diff --git a/crates/socketioxide-mongodb/src/stream.rs b/crates/socketioxide-mongodb/src/stream.rs index 02c5af79..ef5022cf 100644 --- a/crates/socketioxide-mongodb/src/stream.rs +++ b/crates/socketioxide-mongodb/src/stream.rs @@ -30,7 +30,8 @@ mod tests { (sid, Ok::<_, ()>(Value::Str("local".into(), None))) }); let (_, rx) = mpsc::channel::(1); - let stream: AckStream<_, _, crate::drivers::Item, ()> = AckStream::new_empty_remote(local, ChanStream::new(rx), decode_mongodb_ack::<()>); + let stream: AckStream<_, _, crate::drivers::Item, ()> = + AckStream::new_empty_remote(local, ChanStream::new(rx), decode_mongodb_ack::<()>); futures_util::pin_mut!(stream); assert!(!FusedStream::is_terminated(&stream)); let data = stream.next().await; diff --git a/crates/socketioxide-postgres/src/drivers/tokio_postgres.rs b/crates/socketioxide-postgres/src/drivers/tokio_postgres.rs index c6186cd8..d4bcd8c7 100644 --- a/crates/socketioxide-postgres/src/drivers/tokio_postgres.rs +++ b/crates/socketioxide-postgres/src/drivers/tokio_postgres.rs @@ -7,7 +7,7 @@ use futures_util::{StreamExt, sink, stream}; use tokio::sync::mpsc; use tokio_postgres::{AsyncMessage, Client, Config, Socket, tls::MakeTlsConnect}; -use crate::stream::ChanStream; +use socketioxide_core::adapter::stream::ChanStream; use super::{Driver, INIT_LOCK_KEY}; diff --git a/crates/socketioxide-postgres/src/lib.rs b/crates/socketioxide-postgres/src/lib.rs index 909e2d81..0fe26a97 100644 --- a/crates/socketioxide-postgres/src/lib.rs +++ b/crates/socketioxide-postgres/src/lib.rs @@ -82,10 +82,7 @@ use std::{ }; use tokio::{sync::mpsc, task::AbortHandle}; -use crate::{ - drivers::Notification, - stream::AckPayload, -}; +use crate::{drivers::Notification, stream::AckPayload}; pub mod drivers; mod stream; @@ -360,7 +357,8 @@ impl DefinedAdapter for CustomPostgresAdapter {} impl CoreAdapter for CustomPostgresAdapter { type Error = Error; type State = PostgresAdapterCtr; - type AckStream = AckStream, ResponsePayload, E::AckError>; + type AckStream = + AckStream, ResponsePayload, E::AckError>; type InitRes = InitRes; fn new(state: &Self::State, local: CoreLocalAdapter) -> Self { @@ -493,7 +491,11 @@ impl CoreAdapter for CustomPostgresAdapter if opts.is_local(self.local.server_id()) { tracing::debug!(?opts, "broadcast with ack is local"); let (local, _) = self.local.broadcast_with_ack(packet, opts, timeout); - let stream = AckStream::new_empty_remote(local, futures_util::stream::empty::().boxed(), stream::decode_postgres_ack); + let stream = AckStream::new_empty_remote( + local, + futures_util::stream::empty::().boxed(), + stream::decode_postgres_ack, + ); return Ok(stream); } let req = RequestOut::new( diff --git a/crates/socketioxide-postgres/src/stream.rs b/crates/socketioxide-postgres/src/stream.rs index 4618585b..cb051065 100644 --- a/crates/socketioxide-postgres/src/stream.rs +++ b/crates/socketioxide-postgres/src/stream.rs @@ -14,7 +14,9 @@ pub enum AckPayload { } impl AckPayload { - fn deserialize<'de, T: Deserialize<'de>>(&'de self) -> Result> { + fn deserialize<'de, T: Deserialize<'de>>( + &'de self, + ) -> Result> { match &self { AckPayload::Json(rv) => serde_json::from_str::(rv.get()) .map_err(|e| Box::new(e) as Box), @@ -48,7 +50,8 @@ mod tests { }); let empty: futures_core::stream::BoxStream<'static, AckPayload> = futures_util::stream::empty().boxed(); - let stream: AckStream<_, _, crate::ResponsePayload, ()> = AckStream::new_empty_remote(local, empty, decode_postgres_ack::<()>); + let stream: AckStream<_, _, crate::ResponsePayload, ()> = + AckStream::new_empty_remote(local, empty, decode_postgres_ack::<()>); futures_util::pin_mut!(stream); assert!(!FusedStream::is_terminated(&stream)); let data = stream.next().await; diff --git a/crates/socketioxide-redis/src/lib.rs b/crates/socketioxide-redis/src/lib.rs index fa86bff3..3fdd8516 100644 --- a/crates/socketioxide-redis/src/lib.rs +++ b/crates/socketioxide-redis/src/lib.rs @@ -156,6 +156,7 @@ use serde::{Serialize, de::DeserializeOwned}; use socketioxide_core::adapter::remote_packet::{ RequestIn, RequestOut, RequestTypeIn, RequestTypeOut, Response, ResponseType, ResponseTypeId, }; +use socketioxide_core::adapter::stream::{AckStream, DropStream, ResponseHandlers}; use socketioxide_core::{ Sid, Uid, adapter::errors::{AdapterError, BroadcastError}, @@ -165,7 +166,6 @@ use socketioxide_core::{ }, packet::Packet, }; -use socketioxide_core::adapter::stream::{AckStream, DropStream, ResponseHandlers}; use tokio::{sync::mpsc, time}; /// Drivers are an abstraction over the pub/sub backend used by the adapter. @@ -502,7 +502,11 @@ impl CoreAdapter for CustomRedisAdapter { if opts.is_local(self.uid) { tracing::debug!(?opts, "broadcast with ack is local"); let (local, _) = self.local.broadcast_with_ack(packet, opts, timeout); - let stream = AckStream::new_empty_remote(local, MessageStream::>::new_empty(), stream::decode_redis_ack); + let stream = AckStream::new_empty_remote( + local, + MessageStream::>::new_empty(), + stream::decode_redis_ack, + ); return Ok(stream); } let req = RequestOut::new(self.uid, RequestTypeOut::BroadcastWithAck(&packet), &opts); @@ -957,8 +961,8 @@ pub fn read_req_id(data: &[u8]) -> Option { #[cfg(test)] mod tests { use super::*; - use futures_util::stream::{self, FusedStream, StreamExt}; use crate::stream::decode_redis_ack; + use futures_util::stream::{self, FusedStream, StreamExt}; use socketioxide_core::{Str, Value, adapter::AckStreamItem}; use std::convert::Infallible; diff --git a/crates/socketioxide-redis/src/stream.rs b/crates/socketioxide-redis/src/stream.rs index e296f70c..673422b9 100644 --- a/crates/socketioxide-redis/src/stream.rs +++ b/crates/socketioxide-redis/src/stream.rs @@ -1,10 +1,7 @@ use std::fmt; use serde::de::DeserializeOwned; -use socketioxide_core::{ - Sid, - adapter::remote_packet::Response, -}; +use socketioxide_core::{Sid, adapter::remote_packet::Response}; /// Decode a msgpack-encoded `(Sid, Response)` tuple from raw bytes. pub fn decode_redis_ack( @@ -31,7 +28,11 @@ mod tests { let local = futures_util::stream::once(async move { (sid, Ok::<_, ()>(Value::Str("local".into(), None))) }); - let stream: AckStream<_, _, Vec, ()> = AckStream::new_empty_remote(local, MessageStream::>::new_empty(), decode_redis_ack::<()>); + let stream: AckStream<_, _, Vec, ()> = AckStream::new_empty_remote( + local, + MessageStream::>::new_empty(), + decode_redis_ack::<()>, + ); futures_util::pin_mut!(stream); assert!(!FusedStream::is_terminated(&stream)); let data = stream.next().await; From 83adf15eed93fa08336b11e9931e03baddeeaf70 Mon Sep 17 00:00:00 2001 From: Liam Date: Sun, 12 Jul 2026 14:55:09 +0200 Subject: [PATCH 4/9] fix: add docs for ResponsePayload and extract TestAckStream type alias --- crates/socketioxide-postgres/src/lib.rs | 1 + crates/socketioxide-redis/src/lib.rs | 8 ++++---- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/crates/socketioxide-postgres/src/lib.rs b/crates/socketioxide-postgres/src/lib.rs index 0fe26a97..a748158a 100644 --- a/crates/socketioxide-postgres/src/lib.rs +++ b/crates/socketioxide-postgres/src/lib.rs @@ -1232,6 +1232,7 @@ impl ResponsePacket { self.node_id == node_id } } +/// A response payload received from the postgres NOTIFY channel. #[derive(Debug, Deserialize, Serialize)] pub enum ResponsePayload { Data(Box), diff --git a/crates/socketioxide-redis/src/lib.rs b/crates/socketioxide-redis/src/lib.rs index 3fdd8516..8eee219c 100644 --- a/crates/socketioxide-redis/src/lib.rs +++ b/crates/socketioxide-redis/src/lib.rs @@ -966,6 +966,9 @@ mod tests { use socketioxide_core::{Str, Value, adapter::AckStreamItem}; use std::convert::Infallible; + type TestAckStream = + AckStream>, MessageStream>, Vec, ()>; + #[derive(Clone)] struct StubDriver; impl Driver for StubDriver { @@ -991,10 +994,7 @@ mod tests { Ok(0) } } - fn new_stub_ack_stream( - remote: MessageStream>, - timeout: Duration, - ) -> AckStream>, MessageStream>, Vec, ()> { + fn new_stub_ack_stream(remote: MessageStream>, timeout: Duration) -> TestAckStream { AckStream::new( stream::empty::>(), remote, From 0e098be79a768ac945b79dae7cba07af7e84b499 Mon Sep 17 00:00:00 2001 From: Liam Carvajal Date: Sun, 12 Jul 2026 21:59:35 +0200 Subject: [PATCH 5/9] Update crates/socketioxide-core/src/adapter/mod.rs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Théodore Prévot --- crates/socketioxide-core/src/adapter/mod.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/socketioxide-core/src/adapter/mod.rs b/crates/socketioxide-core/src/adapter/mod.rs index 03b4f47d..f530f6e3 100644 --- a/crates/socketioxide-core/src/adapter/mod.rs +++ b/crates/socketioxide-core/src/adapter/mod.rs @@ -28,7 +28,7 @@ pub mod errors; #[cfg(feature = "remote-adapter")] pub mod remote_packet; #[cfg(feature = "remote-adapter")] -#[doc = "Shared stream constructs (`AckStream`, `DropStream`, `ResponseHandlers`) for remote adapters."] +//! Shared stream constructs (`AckStream`, `DropStream`, `ResponseHandlers`) for remote adapters. pub mod stream; /// A room identifier From 48fa11b8d9e57ba3985938cf91b49955934881a6 Mon Sep 17 00:00:00 2001 From: Liam Date: Sun, 12 Jul 2026 22:11:36 +0200 Subject: [PATCH 6/9] fix: add default-features=false, move module docs into stream.rs, add 11 AckStream tests --- crates/socketioxide-core/src/adapter/mod.rs | 1 - .../socketioxide-core/src/adapter/stream.rs | 338 +++++++++++++++++- 2 files changed, 336 insertions(+), 3 deletions(-) diff --git a/crates/socketioxide-core/src/adapter/mod.rs b/crates/socketioxide-core/src/adapter/mod.rs index f530f6e3..55b3014e 100644 --- a/crates/socketioxide-core/src/adapter/mod.rs +++ b/crates/socketioxide-core/src/adapter/mod.rs @@ -28,7 +28,6 @@ pub mod errors; #[cfg(feature = "remote-adapter")] pub mod remote_packet; #[cfg(feature = "remote-adapter")] -//! Shared stream constructs (`AckStream`, `DropStream`, `ResponseHandlers`) for remote adapters. pub mod stream; /// A room identifier diff --git a/crates/socketioxide-core/src/adapter/stream.rs b/crates/socketioxide-core/src/adapter/stream.rs index 6425b25d..070b41a7 100644 --- a/crates/socketioxide-core/src/adapter/stream.rs +++ b/crates/socketioxide-core/src/adapter/stream.rs @@ -1,3 +1,6 @@ +//! Shared stream constructs ([`AckStream`], [`DropStream`], [`ResponseHandlers`]) +//! for remote adapters. + use std::{ collections::HashMap, fmt, @@ -247,12 +250,51 @@ impl fmt::Debug for AckStream { #[cfg(test)] mod tests { - use crate::{Sid, Value}; - use futures_core::FusedStream; + use std::sync::{Arc, Mutex}; + use std::time::Duration; + + use crate::adapter::AckStreamItem; + use crate::adapter::remote_packet::{Response, ResponseType}; + use crate::{Sid, Uid, Value}; + use futures_core::{FusedStream, Stream}; use futures_util::StreamExt; use super::AckStream; + type MockError = Box; + + fn identity(item: Result, MockError>) -> Result, MockError> { + item + } + + fn mock_ack_count(count: u32) -> Result, MockError> { + Ok(Response { + node_id: Uid::ZERO, + r#type: ResponseType::BroadcastAckCount(count), + }) + } + + fn mock_ack(sid: Sid, msg: &'static str) -> Result, MockError> { + Ok(Response { + node_id: Uid::ZERO, + r#type: ResponseType::BroadcastAck((sid, Ok(Value::Str(msg.into(), None)))), + }) + } + + fn mock_all_rooms() -> Result, MockError> { + Ok(Response { + node_id: Uid::ZERO, + r#type: ResponseType::AllRooms(Default::default()), + }) + } + + fn decode_error() -> Result, MockError> { + Err(Box::new(std::io::Error::new( + std::io::ErrorKind::Other, + "decode failure", + ))) + } + #[tokio::test] async fn local_ack_stream_should_have_a_closed_remote() { let sid = Sid::new(); @@ -273,4 +315,296 @@ mod tests { assert_eq!(stream.next().await, None); assert!(stream.is_terminated()); } + + #[tokio::test] + async fn remote_broadcast_ack_count_updates_counters() { + let local = futures_util::stream::empty::>(); + let remote = futures_util::stream::once(async { mock_ack_count(3) }); + let handlers = Arc::new(Mutex::new(super::ResponseHandlers::<()>::new())); + + let stream = AckStream::new( + local, + remote, + identity::<()>, + Duration::from_secs(5), + 1, + Sid::new(), + handlers, + ); + futures_util::pin_mut!(stream); + + assert_eq!(stream.serv_cnt, 1); + let item = stream.next().await; + assert!( + item.is_none(), + "BroadcastAckCount should not yield to stream" + ); + assert_eq!(stream.serv_cnt, 0); + assert_eq!(stream.ack_cnt, 3); + assert_eq!(stream.total_ack_cnt, 3); + } + + #[tokio::test] + async fn remote_broadcast_ack_yields_item() { + let sid = Sid::new(); + let local = futures_util::stream::empty::>(); + + // First item: BroadcastAckCount(1), second: BroadcastAck + let remote = futures_util::stream::iter([mock_ack_count(1), mock_ack(sid, "ack")]); + let handlers = Arc::new(Mutex::new(super::ResponseHandlers::<()>::new())); + + let stream = AckStream::new( + local, + remote, + identity::<()>, + Duration::from_secs(5), + 1, + Sid::new(), + handlers, + ); + futures_util::pin_mut!(stream); + + // First poll: BroadcastAckCount is consumed internally + // Second poll: BroadcastAck is yielded + let data = stream.next().await; + assert!(matches!(data, Some((id, Ok(Value::Str(msg, None)))) if id == sid && msg == "ack")); + } + + #[tokio::test] + async fn mixed_local_and_remote_yields_local_first() { + let local_sid = Sid::new(); + let remote_sid = Sid::new(); + let local = futures_util::stream::once(async move { + (local_sid, Ok::<_, ()>(Value::Str("local".into(), None))) + }); + let remote = + futures_util::stream::iter([mock_ack_count(1), mock_ack(remote_sid, "remote")]); + let handlers = Arc::new(Mutex::new(super::ResponseHandlers::<()>::new())); + + let stream = AckStream::new( + local, + remote, + identity::<()>, + Duration::from_secs(5), + 1, + Sid::new(), + handlers, + ); + futures_util::pin_mut!(stream); + + let first = stream.next().await; + assert!( + matches!(first, Some((id, Ok(Value::Str(msg, None)))) if id == local_sid && msg == "local") + ); + + let second = stream.next().await; + assert!( + matches!(second, Some((id, Ok(Value::Str(msg, None)))) if id == remote_sid && msg == "remote") + ); + } + + #[tokio::test] + async fn unexpected_response_type_is_skipped() { + let local = futures_util::stream::empty::>(); + let remote = futures_util::stream::once(async { mock_all_rooms() }); + let handlers = Arc::new(Mutex::new(super::ResponseHandlers::<()>::new())); + + let stream = AckStream::new( + local, + remote, + identity::<()>, + Duration::from_secs(5), + 1, + Sid::new(), + handlers, + ); + futures_util::pin_mut!(stream); + + let result = stream.next().await; + assert!( + result.is_none(), + "Unexpected response type should not yield" + ); + } + + #[tokio::test] + async fn decode_error_is_skipped() { + let local = futures_util::stream::empty::>(); + let remote = futures_util::stream::once(async { decode_error() }); + let handlers = Arc::new(Mutex::new(super::ResponseHandlers::<()>::new())); + + let stream = AckStream::new( + local, + remote, + identity::<()>, + Duration::from_secs(5), + 1, + Sid::new(), + handlers, + ); + futures_util::pin_mut!(stream); + + let result = stream.next().await; + assert!(result.is_none(), "Decode error should not yield"); + } + + #[tokio::test] + async fn terminated_when_counters_zero_and_local_exhausted() { + let local = futures_util::stream::empty::>(); + let remote = futures_util::stream::pending::, MockError>>(); + let handlers = Arc::new(Mutex::new(super::ResponseHandlers::<()>::new())); + + let stream = AckStream::new( + local, + remote, + identity::<()>, + Duration::from_secs(5), + 0, + Sid::new(), + handlers, + ); + futures_util::pin_mut!(stream); + + assert!( + stream.is_terminated(), + "Should be terminated when local exhausted, serv_cnt==0, ack_cnt==0" + ); + } + + #[tokio::test] + async fn size_hint_incorporates_remote_acks() { + let sid = Sid::new(); + let local = futures_util::stream::once(async move { + (sid, Ok::<_, ()>(Value::Str("local".into(), None))) + }); + let remote = futures_util::stream::empty(); + let handlers = Arc::new(Mutex::new(super::ResponseHandlers::<()>::new())); + + let mut stream = AckStream::new( + local, + remote, + |_: Result, MockError>| unreachable!(), + Duration::from_secs(5), + 0, + Sid::new(), + handlers, + ); + stream.total_ack_cnt = 5; + + let (lower, upper) = stream.size_hint(); + assert_eq!(lower, 1); + assert_eq!(upper, Some(6)); + } + + #[tokio::test] + async fn broadcast_ack_not_yielded_when_ack_cnt_zero() { + let sid = Sid::new(); + let local = futures_util::stream::empty::>(); + let remote = futures_util::stream::once(async { mock_ack(sid, "ack") }); + let handlers = Arc::new(Mutex::new(super::ResponseHandlers::<()>::new())); + + let stream = AckStream::new( + local, + remote, + identity::<()>, + Duration::from_secs(5), + 1, + Sid::new(), + handlers, + ); + futures_util::pin_mut!(stream); + + let result = stream.next().await; + assert!( + result.is_none(), + "BroadcastAck should not yield when ack_cnt is 0" + ); + } + + #[tokio::test] + async fn multiple_broadcast_acks_delivered_in_order() { + let sid1 = Sid::new(); + let sid2 = Sid::new(); + let local = futures_util::stream::empty::>(); + let remote = futures_util::stream::iter([ + mock_ack_count(2), + mock_ack(sid1, "first"), + mock_ack(sid2, "second"), + ]); + let handlers = Arc::new(Mutex::new(super::ResponseHandlers::<()>::new())); + + let stream = AckStream::new( + local, + remote, + identity::<()>, + Duration::from_secs(5), + 1, + Sid::new(), + handlers, + ); + futures_util::pin_mut!(stream); + + let first = stream.next().await; + assert!( + matches!(first, Some((id, Ok(Value::Str(msg, None)))) if id == sid1 && msg == "first") + ); + + let second = stream.next().await; + assert!( + matches!(second, Some((id, Ok(Value::Str(msg, None)))) if id == sid2 && msg == "second") + ); + + assert_eq!(stream.next().await, None); + } + + #[tokio::test] + async fn broadcast_ack_count_guard_ignores_when_serv_cnt_zero() { + let local = futures_util::stream::empty::>(); + let remote = futures_util::stream::once(async { mock_ack_count(5) }); + let handlers = Arc::new(Mutex::new(super::ResponseHandlers::<()>::new())); + + let stream = AckStream::new( + local, + remote, + identity::<()>, + Duration::from_secs(5), + 0, + Sid::new(), + handlers, + ); + futures_util::pin_mut!(stream); + + let result = stream.next().await; + assert!( + result.is_none(), + "BroadcastAckCount should be ignored when serv_cnt is 0" + ); + assert_eq!(stream.ack_cnt, 0); + } + + #[tokio::test] + async fn serv_cnt_decremented_for_each_server() { + let local = futures_util::stream::empty::>(); + let handlers = Arc::new(Mutex::new(super::ResponseHandlers::<()>::new())); + + let remote = + futures_util::stream::iter([mock_ack_count(1), mock_ack_count(1), mock_ack_count(1)]); + + let stream = AckStream::new( + local, + remote, + identity::<()>, + Duration::from_secs(5), + 3, + Sid::new(), + handlers, + ); + futures_util::pin_mut!(stream); + + assert_eq!(stream.serv_cnt, 3); + let _ = stream.next().await; + assert_eq!(stream.serv_cnt, 0); + assert_eq!(stream.ack_cnt, 3); + assert_eq!(stream.total_ack_cnt, 3); + } } From faa54f136b1aaba21f398b89defa3cef2469f9a4 Mon Sep 17 00:00:00 2001 From: Liam Date: Sun, 12 Jul 2026 22:21:48 +0200 Subject: [PATCH 7/9] fix: use std::io::Error::other instead of Error::new --- crates/socketioxide-core/src/adapter/stream.rs | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/crates/socketioxide-core/src/adapter/stream.rs b/crates/socketioxide-core/src/adapter/stream.rs index 070b41a7..8b6e5348 100644 --- a/crates/socketioxide-core/src/adapter/stream.rs +++ b/crates/socketioxide-core/src/adapter/stream.rs @@ -289,10 +289,7 @@ mod tests { } fn decode_error() -> Result, MockError> { - Err(Box::new(std::io::Error::new( - std::io::ErrorKind::Other, - "decode failure", - ))) + Err(Box::new(std::io::Error::other("decode failure"))) } #[tokio::test] From 114506d1760909b59ea428d99e2615370fc54966 Mon Sep 17 00:00:00 2001 From: Liam Date: Sun, 19 Jul 2026 01:40:53 +0200 Subject: [PATCH 8/9] refactor: use SocketEmitter GAT in AckStream, swap decoder params, replace new_empty_remote with Either, move decode fns to lib.rs - AckStream reduces 4 params to 3 - AckDecoder swapped param order (item first, error second) - new_empty_remote removed; adapters use Either> - decode fns moved from stream.rs to lib.rs in redis/mongodb - Duplicate adapter tests removed (covered in core) - Redis/MongoDB stream.rs deleted; Postgres stream.rs trimmed to AckPayload only --- .../socketioxide-core/src/adapter/stream.rs | 284 +++++++++--------- crates/socketioxide-mongodb/src/lib.rs | 26 +- crates/socketioxide-mongodb/src/stream.rs | 44 --- crates/socketioxide-postgres/src/lib.rs | 23 +- crates/socketioxide-postgres/src/stream.rs | 49 +-- crates/socketioxide-redis/src/lib.rs | 129 +------- crates/socketioxide-redis/src/stream.rs | 45 --- 7 files changed, 183 insertions(+), 417 deletions(-) delete mode 100644 crates/socketioxide-mongodb/src/stream.rs delete mode 100644 crates/socketioxide-redis/src/stream.rs diff --git a/crates/socketioxide-core/src/adapter/stream.rs b/crates/socketioxide-core/src/adapter/stream.rs index 8b6e5348..e2b41948 100644 --- a/crates/socketioxide-core/src/adapter/stream.rs +++ b/crates/socketioxide-core/src/adapter/stream.rs @@ -20,11 +20,13 @@ use crate::Sid; use crate::adapter::AckStreamItem; use crate::adapter::remote_packet::{Response, ResponseType}; +use crate::adapter::SocketEmitter; + /// A map of request IDs to response senders, used to route responses to the correct awaiting stream. pub type ResponseHandlers = HashMap>; /// Decoder function that converts a raw remote payload item into a [`Response`]. -pub type AckDecoder = fn(I) -> Result, Box>; +pub type AckDecoder = fn(I) -> Result, Box>; pin_project! { /// A [`Stream`] that wraps an [`mpsc::Receiver`]. @@ -34,7 +36,6 @@ pin_project! { } } impl ChanStream { - /// Create a new `ChanStream` from an [`mpsc::Receiver`]. pub fn new(rx: mpsc::Receiver) -> Self { Self { rx } } @@ -65,7 +66,6 @@ pin_project! { } } impl DropStream { - /// Create a new `DropStream` that will remove the handler entry on drop. pub fn new(stream: S, handlers: Arc>>, req_id: Sid) -> Self { Self { stream, @@ -89,32 +89,23 @@ impl FusedStream for DropStream { pin_project! { /// A stream of acknowledgement messages received from the local and remote servers. - /// It merges the local ack stream with the remote ack stream from all the servers. - // The server_cnt is the number of servers that are expected to send a AckCount message. - // It is decremented each time a AckCount message is received. - // - // The ack_cnt is the number of acks that are expected to be received. It is the sum of all the the ack counts. - // And it is decremented each time an ack is received. - // - // Therefore an exhausted stream correspond to `ack_cnt == 0` and `server_cnt == 0`. - pub struct AckStream { + pub struct AckStream { #[pin] - local: S, + local: E::AckStream, #[pin] remote: DropStream, T>, - decode: AckDecoder, + decode: AckDecoder, ack_cnt: u32, total_ack_cnt: usize, serv_cnt: u16, } } -impl AckStream { - /// Create a new `AckStream` wrapping the given local and remote streams. +impl AckStream { pub fn new( - local: S, + local: E::AckStream, remote: R, - decode: AckDecoder, + decode: AckDecoder, timeout: Duration, serv_cnt: u16, req_id: Sid, @@ -131,35 +122,16 @@ impl AckStream { serv_cnt, } } - - /// Create a new `AckStream` for local-only use, with an empty remote stream. - pub fn new_empty_remote(local: S, empty_remote: R, decode: AckDecoder) -> Self { - let handlers = Arc::new(Mutex::new(ResponseHandlers::::new())); - let remote = empty_remote.take_until(time::sleep(Duration::ZERO)); - let remote = DropStream::new(remote, handlers, Sid::ZERO); - Self { - local, - remote, - decode, - ack_cnt: 0, - total_ack_cnt: 0, - serv_cnt: 0, - } - } } -impl AckStream +impl AckStream where - Err: DeserializeOwned + fmt::Debug, - S: Stream> + FusedStream, + E::AckError: DeserializeOwned + fmt::Debug, { - /// Poll the remote stream. First the count of acks is received, then the acks are received. - /// We expect `serv_cnt` of `BroadcastAckCount` messages to be received, then we expect - /// `ack_cnt` of `BroadcastAck` messages. fn poll_remote( self: Pin<&mut Self>, cx: &mut task::Context<'_>, - ) -> Poll>> { + ) -> Poll>> { if FusedStream::is_terminated(&self) { return Poll::Ready(None); } @@ -198,12 +170,11 @@ where } } -impl Stream for AckStream +impl Stream for AckStream where - Err: DeserializeOwned + fmt::Debug, - S: Stream> + FusedStream, + E::AckError: DeserializeOwned + fmt::Debug, { - type Item = AckStreamItem; + type Item = AckStreamItem; fn poll_next(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll> { match self.as_mut().project().local.poll_next(cx) { @@ -223,22 +194,17 @@ where } } -impl FusedStream for AckStream +impl FusedStream for AckStream where - Err: DeserializeOwned + fmt::Debug, - S: Stream> + FusedStream, + E::AckError: DeserializeOwned + fmt::Debug, { - /// The stream is terminated if: - /// * The local stream is terminated. - /// * All the servers have sent the expected ack count. - /// * We have received all the expected acks. fn is_terminated(&self) -> bool { let remote_term = (self.ack_cnt == 0 && self.serv_cnt == 0) || self.remote.is_terminated(); self.local.is_terminated() && remote_term } } -impl fmt::Debug for AckStream { +impl fmt::Debug for AckStream { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("AckStream") .field("ack_cnt", &self.ack_cnt) @@ -250,79 +216,121 @@ impl fmt::Debug for AckStream { #[cfg(test)] mod tests { + use std::borrow::Cow; use std::sync::{Arc, Mutex}; use std::time::Duration; - use crate::adapter::AckStreamItem; + use crate::adapter::errors::SocketError; use crate::adapter::remote_packet::{Response, ResponseType}; - use crate::{Sid, Uid, Value}; + use crate::adapter::{AckStreamItem, BroadcastIter, RemoteSocketData, SocketEmitter}; + use crate::packet::Packet; + use crate::parser::{Parse, ParserError, test::StubParser}; + use crate::{Sid, Str, Uid, Value}; + use engineioxide_core::Sid as EngineIoSid; use futures_core::{FusedStream, Stream}; use futures_util::StreamExt; + use serde::{Deserialize, Serialize}; use super::AckStream; type MockError = Box; + #[derive(Debug, Serialize, Deserialize, PartialEq)] + struct TestError; + impl std::fmt::Display for TestError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "test error") + } + } + impl std::error::Error for TestError {} + + struct TestEmitter(pub S); + + impl SocketEmitter for TestEmitter + where + S: Stream> + FusedStream + Send + Sync + 'static, + { + type AckError = TestError; + type AckStream = S; + + fn get_all_sids(&self, _filter: impl Fn(&Sid) -> bool) -> Vec { + vec![] + } + fn get_remote_sockets(&self, _sids: BroadcastIter<'_>) -> Vec { + vec![] + } + fn send_many( + &self, + _sids: BroadcastIter<'_>, + _data: Value, + ) -> Result<(), Vec> { + Ok(()) + } + fn send_many_volatile(&self, _sids: BroadcastIter<'_>, _data: Value) {} + fn send_many_with_ack( + &self, + _sids: BroadcastIter<'_>, + _packet: Packet, + _timeout: Option, + ) -> (Self::AckStream, u32) { + unreachable!() + } + fn disconnect_many(&self, _sids: Vec) -> Result<(), Vec> { + Ok(()) + } + fn path(&self) -> &Str { + unreachable!() + } + fn parser(&self) -> StubParser { + StubParser::default() + } + fn server_id(&self) -> Uid { + Uid::ZERO + } + fn ack_timeout(&self) -> Duration { + Duration::from_secs(5) + } + } + fn identity(item: Result, MockError>) -> Result, MockError> { item } - fn mock_ack_count(count: u32) -> Result, MockError> { + fn mock_ack_count(count: u32) -> Result, MockError> { Ok(Response { node_id: Uid::ZERO, r#type: ResponseType::BroadcastAckCount(count), }) } - fn mock_ack(sid: Sid, msg: &'static str) -> Result, MockError> { + fn mock_ack(sid: Sid, msg: &'static str) -> Result, MockError> { Ok(Response { node_id: Uid::ZERO, r#type: ResponseType::BroadcastAck((sid, Ok(Value::Str(msg.into(), None)))), }) } - fn mock_all_rooms() -> Result, MockError> { + fn mock_all_rooms() -> Result, MockError> { Ok(Response { node_id: Uid::ZERO, r#type: ResponseType::AllRooms(Default::default()), }) } - fn decode_error() -> Result, MockError> { + fn decode_error() -> Result, MockError> { Err(Box::new(std::io::Error::other("decode failure"))) } - #[tokio::test] - async fn local_ack_stream_should_have_a_closed_remote() { - let sid = Sid::new(); - let local = futures_util::stream::once(async move { - (sid, Ok::<_, ()>(Value::Str("local".into(), None))) - }); - let empty_remote = futures_util::stream::empty::<()>(); - let stream: AckStream<_, _, (), ()> = - AckStream::new_empty_remote(local, empty_remote, |_| unreachable!()); - futures_util::pin_mut!(stream); - assert_eq!(stream.ack_cnt, 0); - assert_eq!(stream.total_ack_cnt, 0); - assert_eq!(stream.serv_cnt, 0); - let data = stream.next().await; - assert!( - matches!(data, Some((id, Ok(Value::Str(msg, None)))) if id == sid && msg == "local") - ); - assert_eq!(stream.next().await, None); - assert!(stream.is_terminated()); - } - #[tokio::test] async fn remote_broadcast_ack_count_updates_counters() { - let local = futures_util::stream::empty::>(); + let local = futures_util::stream::empty::>(); let remote = futures_util::stream::once(async { mock_ack_count(3) }); - let handlers = Arc::new(Mutex::new(super::ResponseHandlers::<()>::new())); + let handlers = Arc::new(Mutex::new(super::ResponseHandlers::::new())); - let stream = AckStream::new( + let stream: AckStream, _, TestError> = AckStream::new( local, remote, - identity::<()>, + identity::, Duration::from_secs(5), 1, Sid::new(), @@ -332,10 +340,7 @@ mod tests { assert_eq!(stream.serv_cnt, 1); let item = stream.next().await; - assert!( - item.is_none(), - "BroadcastAckCount should not yield to stream" - ); + assert!(item.is_none(), "BroadcastAckCount should not yield"); assert_eq!(stream.serv_cnt, 0); assert_eq!(stream.ack_cnt, 3); assert_eq!(stream.total_ack_cnt, 3); @@ -344,16 +349,14 @@ mod tests { #[tokio::test] async fn remote_broadcast_ack_yields_item() { let sid = Sid::new(); - let local = futures_util::stream::empty::>(); - - // First item: BroadcastAckCount(1), second: BroadcastAck + let local = futures_util::stream::empty::>(); let remote = futures_util::stream::iter([mock_ack_count(1), mock_ack(sid, "ack")]); - let handlers = Arc::new(Mutex::new(super::ResponseHandlers::<()>::new())); + let handlers = Arc::new(Mutex::new(super::ResponseHandlers::::new())); - let stream = AckStream::new( + let stream: AckStream, _, TestError> = AckStream::new( local, remote, - identity::<()>, + identity::, Duration::from_secs(5), 1, Sid::new(), @@ -361,8 +364,6 @@ mod tests { ); futures_util::pin_mut!(stream); - // First poll: BroadcastAckCount is consumed internally - // Second poll: BroadcastAck is yielded let data = stream.next().await; assert!(matches!(data, Some((id, Ok(Value::Str(msg, None)))) if id == sid && msg == "ack")); } @@ -372,16 +373,19 @@ mod tests { let local_sid = Sid::new(); let remote_sid = Sid::new(); let local = futures_util::stream::once(async move { - (local_sid, Ok::<_, ()>(Value::Str("local".into(), None))) + ( + local_sid, + Ok::<_, TestError>(Value::Str("local".into(), None)), + ) }); let remote = futures_util::stream::iter([mock_ack_count(1), mock_ack(remote_sid, "remote")]); - let handlers = Arc::new(Mutex::new(super::ResponseHandlers::<()>::new())); + let handlers = Arc::new(Mutex::new(super::ResponseHandlers::::new())); - let stream = AckStream::new( + let stream: AckStream, _, TestError> = AckStream::new( local, remote, - identity::<()>, + identity::, Duration::from_secs(5), 1, Sid::new(), @@ -393,7 +397,6 @@ mod tests { assert!( matches!(first, Some((id, Ok(Value::Str(msg, None)))) if id == local_sid && msg == "local") ); - let second = stream.next().await; assert!( matches!(second, Some((id, Ok(Value::Str(msg, None)))) if id == remote_sid && msg == "remote") @@ -402,14 +405,14 @@ mod tests { #[tokio::test] async fn unexpected_response_type_is_skipped() { - let local = futures_util::stream::empty::>(); + let local = futures_util::stream::empty::>(); let remote = futures_util::stream::once(async { mock_all_rooms() }); - let handlers = Arc::new(Mutex::new(super::ResponseHandlers::<()>::new())); + let handlers = Arc::new(Mutex::new(super::ResponseHandlers::::new())); - let stream = AckStream::new( + let stream: AckStream, _, TestError> = AckStream::new( local, remote, - identity::<()>, + identity::, Duration::from_secs(5), 1, Sid::new(), @@ -426,14 +429,14 @@ mod tests { #[tokio::test] async fn decode_error_is_skipped() { - let local = futures_util::stream::empty::>(); + let local = futures_util::stream::empty::>(); let remote = futures_util::stream::once(async { decode_error() }); - let handlers = Arc::new(Mutex::new(super::ResponseHandlers::<()>::new())); + let handlers = Arc::new(Mutex::new(super::ResponseHandlers::::new())); - let stream = AckStream::new( + let stream: AckStream, _, TestError> = AckStream::new( local, remote, - identity::<()>, + identity::, Duration::from_secs(5), 1, Sid::new(), @@ -447,14 +450,14 @@ mod tests { #[tokio::test] async fn terminated_when_counters_zero_and_local_exhausted() { - let local = futures_util::stream::empty::>(); - let remote = futures_util::stream::pending::, MockError>>(); - let handlers = Arc::new(Mutex::new(super::ResponseHandlers::<()>::new())); + let local = futures_util::stream::empty::>(); + let remote = futures_util::stream::pending::, MockError>>(); + let handlers = Arc::new(Mutex::new(super::ResponseHandlers::::new())); - let stream = AckStream::new( + let stream: AckStream, _, TestError> = AckStream::new( local, remote, - identity::<()>, + identity::, Duration::from_secs(5), 0, Sid::new(), @@ -462,25 +465,22 @@ mod tests { ); futures_util::pin_mut!(stream); - assert!( - stream.is_terminated(), - "Should be terminated when local exhausted, serv_cnt==0, ack_cnt==0" - ); + assert!(stream.is_terminated()); } #[tokio::test] async fn size_hint_incorporates_remote_acks() { let sid = Sid::new(); let local = futures_util::stream::once(async move { - (sid, Ok::<_, ()>(Value::Str("local".into(), None))) + (sid, Ok::<_, TestError>(Value::Str("local".into(), None))) }); let remote = futures_util::stream::empty(); - let handlers = Arc::new(Mutex::new(super::ResponseHandlers::<()>::new())); + let handlers = Arc::new(Mutex::new(super::ResponseHandlers::::new())); - let mut stream = AckStream::new( + let mut stream: AckStream, _, TestError> = AckStream::new( local, remote, - |_: Result, MockError>| unreachable!(), + |_: Result, MockError>| unreachable!(), Duration::from_secs(5), 0, Sid::new(), @@ -496,14 +496,14 @@ mod tests { #[tokio::test] async fn broadcast_ack_not_yielded_when_ack_cnt_zero() { let sid = Sid::new(); - let local = futures_util::stream::empty::>(); + let local = futures_util::stream::empty::>(); let remote = futures_util::stream::once(async { mock_ack(sid, "ack") }); - let handlers = Arc::new(Mutex::new(super::ResponseHandlers::<()>::new())); + let handlers = Arc::new(Mutex::new(super::ResponseHandlers::::new())); - let stream = AckStream::new( + let stream: AckStream, _, TestError> = AckStream::new( local, remote, - identity::<()>, + identity::, Duration::from_secs(5), 1, Sid::new(), @@ -514,7 +514,7 @@ mod tests { let result = stream.next().await; assert!( result.is_none(), - "BroadcastAck should not yield when ack_cnt is 0" + "BroadcastAck should not yield when ack_cnt==0" ); } @@ -522,18 +522,18 @@ mod tests { async fn multiple_broadcast_acks_delivered_in_order() { let sid1 = Sid::new(); let sid2 = Sid::new(); - let local = futures_util::stream::empty::>(); + let local = futures_util::stream::empty::>(); let remote = futures_util::stream::iter([ mock_ack_count(2), mock_ack(sid1, "first"), mock_ack(sid2, "second"), ]); - let handlers = Arc::new(Mutex::new(super::ResponseHandlers::<()>::new())); + let handlers = Arc::new(Mutex::new(super::ResponseHandlers::::new())); - let stream = AckStream::new( + let stream: AckStream, _, TestError> = AckStream::new( local, remote, - identity::<()>, + identity::, Duration::from_secs(5), 1, Sid::new(), @@ -545,25 +545,23 @@ mod tests { assert!( matches!(first, Some((id, Ok(Value::Str(msg, None)))) if id == sid1 && msg == "first") ); - let second = stream.next().await; assert!( matches!(second, Some((id, Ok(Value::Str(msg, None)))) if id == sid2 && msg == "second") ); - assert_eq!(stream.next().await, None); } #[tokio::test] async fn broadcast_ack_count_guard_ignores_when_serv_cnt_zero() { - let local = futures_util::stream::empty::>(); + let local = futures_util::stream::empty::>(); let remote = futures_util::stream::once(async { mock_ack_count(5) }); - let handlers = Arc::new(Mutex::new(super::ResponseHandlers::<()>::new())); + let handlers = Arc::new(Mutex::new(super::ResponseHandlers::::new())); - let stream = AckStream::new( + let stream: AckStream, _, TestError> = AckStream::new( local, remote, - identity::<()>, + identity::, Duration::from_secs(5), 0, Sid::new(), @@ -572,25 +570,21 @@ mod tests { futures_util::pin_mut!(stream); let result = stream.next().await; - assert!( - result.is_none(), - "BroadcastAckCount should be ignored when serv_cnt is 0" - ); + assert!(result.is_none()); assert_eq!(stream.ack_cnt, 0); } #[tokio::test] async fn serv_cnt_decremented_for_each_server() { - let local = futures_util::stream::empty::>(); - let handlers = Arc::new(Mutex::new(super::ResponseHandlers::<()>::new())); - + let local = futures_util::stream::empty::>(); + let handlers = Arc::new(Mutex::new(super::ResponseHandlers::::new())); let remote = futures_util::stream::iter([mock_ack_count(1), mock_ack_count(1), mock_ack_count(1)]); - let stream = AckStream::new( + let stream: AckStream, _, TestError> = AckStream::new( local, remote, - identity::<()>, + identity::, Duration::from_secs(5), 3, Sid::new(), diff --git a/crates/socketioxide-mongodb/src/lib.rs b/crates/socketioxide-mongodb/src/lib.rs index 5c633c94..6a6a8c6c 100644 --- a/crates/socketioxide-mongodb/src/lib.rs +++ b/crates/socketioxide-mongodb/src/lib.rs @@ -99,7 +99,7 @@ use std::{ }; use futures_core::{Stream, future::Future}; -use futures_util::StreamExt; +use futures_util::{StreamExt, future::Either}; use serde::{Serialize, de::DeserializeOwned}; use socketioxide_core::adapter::remote_packet::{ RequestIn, RequestOut, RequestTypeIn, RequestTypeOut, Response, ResponseType, ResponseTypeId, @@ -122,8 +122,6 @@ use drivers::{Driver, Item, ItemHeader}; /// You can use the provided implementation or implement your own. pub mod drivers; -mod stream; - /// The configuration of the [`MongoDbAdapter`]. #[derive(Debug, Clone)] pub struct MongoDbAdapterConfig { @@ -331,7 +329,7 @@ impl DefinedAdapter for CustomMongoDbAdapter {} impl CoreAdapter for CustomMongoDbAdapter { type Error = Error; type State = MongoDbAdapterCtr; - type AckStream = AckStream, Item, E::AckError>; + type AckStream = Either, Item>>; type InitRes = InitRes; fn new(state: &Self::State, local: CoreLocalAdapter) -> Self { @@ -427,12 +425,7 @@ impl CoreAdapter for CustomMongoDbAdapter if opts.is_local(self.uid) { tracing::debug!(?opts, "broadcast with ack is local"); let (local, _) = self.local.broadcast_with_ack(packet, opts, timeout); - let stream = AckStream::new_empty_remote( - local, - ChanStream::::new(mpsc::channel::(1).1), - stream::decode_mongodb_ack, - ); - return Ok(stream); + return Ok(Either::Left(local)); } let req = RequestOut::new(self.uid, RequestTypeOut::BroadcastWithAck(&packet), &opts); let req_id = req.id; @@ -451,15 +444,15 @@ impl CoreAdapter for CustomMongoDbAdapter .request_timeout .saturating_add(timeout.unwrap_or(self.local.ack_timeout())); - Ok(AckStream::new( + Ok(Either::Right(AckStream::new( local, ChanStream::new(rx), - stream::decode_mongodb_ack, + decode_mongodb_ack, timeout, remote_serv_cnt, req_id, self.responses.clone(), - )) + ))) } async fn disconnect_socket(&self, opts: BroadcastOptions) -> Result<(), BroadcastError> { @@ -889,3 +882,10 @@ impl Spawnable for InitRes { }); } } + +fn decode_mongodb_ack( + item: Item, +) -> Result, Box> { + rmp_serde::from_slice::>(&item.data) + .map_err(|e| Box::new(e) as Box) +} diff --git a/crates/socketioxide-mongodb/src/stream.rs b/crates/socketioxide-mongodb/src/stream.rs deleted file mode 100644 index ef5022cf..00000000 --- a/crates/socketioxide-mongodb/src/stream.rs +++ /dev/null @@ -1,44 +0,0 @@ -use std::fmt; - -use serde::de::DeserializeOwned; -use socketioxide_core::adapter::remote_packet::Response; - -use crate::drivers::Item; - -/// Decode a MongoDB `Item`'s msgpack data into a `Response`. -pub fn decode_mongodb_ack( - item: Item, -) -> Result, Box> { - rmp_serde::from_slice::>(&item.data) - .map_err(|e| Box::new(e) as Box) -} - -#[cfg(test)] -mod tests { - use futures_core::FusedStream; - use futures_util::StreamExt; - use socketioxide_core::{Sid, Value}; - use tokio::sync::mpsc; - - use crate::stream::decode_mongodb_ack; - use socketioxide_core::adapter::stream::{AckStream, ChanStream}; - - #[tokio::test] - async fn local_ack_stream_should_have_a_closed_remote() { - let sid = Sid::new(); - let local = futures_util::stream::once(async move { - (sid, Ok::<_, ()>(Value::Str("local".into(), None))) - }); - let (_, rx) = mpsc::channel::(1); - let stream: AckStream<_, _, crate::drivers::Item, ()> = - AckStream::new_empty_remote(local, ChanStream::new(rx), decode_mongodb_ack::<()>); - futures_util::pin_mut!(stream); - assert!(!FusedStream::is_terminated(&stream)); - let data = stream.next().await; - assert!( - matches!(data, Some((id, Ok(Value::Str(msg, None)))) if id == sid && msg == "local") - ); - assert_eq!(stream.next().await, None); - assert!(FusedStream::is_terminated(&stream)); - } -} diff --git a/crates/socketioxide-postgres/src/lib.rs b/crates/socketioxide-postgres/src/lib.rs index a748158a..024eb010 100644 --- a/crates/socketioxide-postgres/src/lib.rs +++ b/crates/socketioxide-postgres/src/lib.rs @@ -54,7 +54,7 @@ use drivers::Driver; use futures_core::{Stream, stream::BoxStream}; -use futures_util::{StreamExt, pin_mut}; +use futures_util::{StreamExt, future::Either, pin_mut}; use serde::{Deserialize, Serialize, de::DeserializeOwned}; use serde_json::value::RawValue; use socketioxide_core::{ @@ -358,7 +358,7 @@ impl CoreAdapter for CustomPostgresAdapter type Error = Error; type State = PostgresAdapterCtr; type AckStream = - AckStream, ResponsePayload, E::AckError>; + Either, ResponsePayload>>; type InitRes = InitRes; fn new(state: &Self::State, local: CoreLocalAdapter) -> Self { @@ -491,12 +491,7 @@ impl CoreAdapter for CustomPostgresAdapter if opts.is_local(self.local.server_id()) { tracing::debug!(?opts, "broadcast with ack is local"); let (local, _) = self.local.broadcast_with_ack(packet, opts, timeout); - let stream = AckStream::new_empty_remote( - local, - futures_util::stream::empty::().boxed(), - stream::decode_postgres_ack, - ); - return Ok(stream); + return Ok(Either::Left(local)); } let req = RequestOut::new( self.local.server_id(), @@ -533,15 +528,15 @@ impl CoreAdapter for CustomPostgresAdapter .filter_map(future::ready) .boxed(); - Ok(AckStream::new( + Ok(Either::Right(AckStream::new( local, remote, - stream::decode_postgres_ack, + decode_postgres_ack, timeout, remote_serv_cnt, req_id, self.responses.clone(), - )) + ))) } async fn disconnect_socket(&self, opts: BroadcastOptions) -> Result<(), BroadcastError> { @@ -1238,3 +1233,9 @@ pub enum ResponsePayload { Data(Box), Attachment { id: i64, is_binary: bool }, } + +fn decode_postgres_ack( + item: AckPayload, +) -> Result, Box> { + item.deserialize() +} diff --git a/crates/socketioxide-postgres/src/stream.rs b/crates/socketioxide-postgres/src/stream.rs index cb051065..383f1203 100644 --- a/crates/socketioxide-postgres/src/stream.rs +++ b/crates/socketioxide-postgres/src/stream.rs @@ -1,21 +1,17 @@ -use std::fmt; - -use serde::Deserialize; -use serde_json::value::RawValue; -use socketioxide_core::adapter::remote_packet::Response; +use serde::de::DeserializeOwned; /// A resolved ack-response payload carrying either a JSON `RawValue` (inline data or a /// JSON-encoded attachment) or raw msgpack bytes (a binary attachment). The format /// has to travel with the payload because the eventual `Response` is decoded /// downstream, where the right serde codec is selected. pub enum AckPayload { - Json(Box), + Json(Box), MsgPack(Vec), } impl AckPayload { - fn deserialize<'de, T: Deserialize<'de>>( - &'de self, + pub(crate) fn deserialize( + &self, ) -> Result> { match &self { AckPayload::Json(rv) => serde_json::from_str::(rv.get()) @@ -25,40 +21,3 @@ impl AckPayload { } } } - -/// Decode an `AckPayload` into a `Response` using either JSON or msgpack. -pub fn decode_postgres_ack( - item: AckPayload, -) -> Result, Box> { - item.deserialize() -} - -#[cfg(test)] -mod tests { - use futures_core::FusedStream; - use futures_util::StreamExt; - use socketioxide_core::{Sid, Value}; - - use crate::stream::{AckPayload, decode_postgres_ack}; - use socketioxide_core::adapter::stream::AckStream; - - #[tokio::test] - async fn local_ack_stream_should_have_a_closed_remote() { - let sid = Sid::new(); - let local = futures_util::stream::once(async move { - (sid, Ok::<_, ()>(Value::Str("local".into(), None))) - }); - let empty: futures_core::stream::BoxStream<'static, AckPayload> = - futures_util::stream::empty().boxed(); - let stream: AckStream<_, _, crate::ResponsePayload, ()> = - AckStream::new_empty_remote(local, empty, decode_postgres_ack::<()>); - futures_util::pin_mut!(stream); - assert!(!FusedStream::is_terminated(&stream)); - let data = stream.next().await; - assert!( - matches!(data, Some((id, Ok(Value::Str(msg, None)))) if id == sid && msg == "local") - ); - assert_eq!(stream.next().await, None); - assert!(FusedStream::is_terminated(&stream)); - } -} diff --git a/crates/socketioxide-redis/src/lib.rs b/crates/socketioxide-redis/src/lib.rs index 8eee219c..4d19f9bf 100644 --- a/crates/socketioxide-redis/src/lib.rs +++ b/crates/socketioxide-redis/src/lib.rs @@ -151,7 +151,7 @@ use std::{ use drivers::{ChanItem, Driver, MessageStream}; use futures_core::Stream; -use futures_util::StreamExt; +use futures_util::{StreamExt, future::Either}; use serde::{Serialize, de::DeserializeOwned}; use socketioxide_core::adapter::remote_packet::{ RequestIn, RequestOut, RequestTypeIn, RequestTypeOut, Response, ResponseType, ResponseTypeId, @@ -172,8 +172,6 @@ use tokio::{sync::mpsc, time}; /// You can use the provided implementation or implement your own. pub mod drivers; -mod stream; - /// Represent any error that might happen when using this adapter. #[derive(thiserror::Error)] pub enum Error { @@ -383,7 +381,7 @@ impl DefinedAdapter for CustomRedisAdapter {} impl CoreAdapter for CustomRedisAdapter { type Error = Error; type State = RedisAdapterCtr; - type AckStream = AckStream>, Vec, E::AckError>; + type AckStream = Either>, Vec>>; type InitRes = InitRes; fn new(state: &Self::State, local: CoreLocalAdapter) -> Self { @@ -502,11 +500,7 @@ impl CoreAdapter for CustomRedisAdapter { if opts.is_local(self.uid) { tracing::debug!(?opts, "broadcast with ack is local"); let (local, _) = self.local.broadcast_with_ack(packet, opts, timeout); - let stream = AckStream::new_empty_remote( - local, - MessageStream::>::new_empty(), - stream::decode_redis_ack, - ); + let stream = Either::Left(local); return Ok(stream); } let req = RequestOut::new(self.uid, RequestTypeOut::BroadcastWithAck(&packet), &opts); @@ -527,15 +521,15 @@ impl CoreAdapter for CustomRedisAdapter { .request_timeout .saturating_add(timeout.unwrap_or(self.local.ack_timeout())); - Ok(AckStream::new( + Ok(Either::Right(AckStream::new( local, remote, - stream::decode_redis_ack, + decode_redis_ack, timeout, remote_serv_cnt, req_id, self.responses.clone(), - )) + ))) } async fn disconnect_socket(&self, opts: BroadcastOptions) -> Result<(), BroadcastError> { @@ -958,17 +952,22 @@ pub fn read_req_id(data: &[u8]) -> Option { Sid::from_str(str).ok() } +/// Decode a msgpack-encoded `(Sid, Response)` tuple from raw bytes. +fn decode_redis_ack( + item: Vec, +) -> Result, Box> { + rmp_serde::from_slice::<(Sid, Response)>(&item) + .map(|(_, response)| response) + .map_err(|e| Box::new(e) as Box) +} + #[cfg(test)] mod tests { use super::*; - use crate::stream::decode_redis_ack; use futures_util::stream::{self, FusedStream, StreamExt}; use socketioxide_core::{Str, Value, adapter::AckStreamItem}; use std::convert::Infallible; - type TestAckStream = - AckStream>, MessageStream>, Vec, ()>; - #[derive(Clone)] struct StubDriver; impl Driver for StubDriver { @@ -994,102 +993,4 @@ mod tests { Ok(0) } } - fn new_stub_ack_stream(remote: MessageStream>, timeout: Duration) -> TestAckStream { - AckStream::new( - stream::empty::>(), - remote, - decode_redis_ack, - timeout, - 2, - Sid::new(), - Arc::new(Mutex::new(HashMap::new())), - ) - } - - //TODO: test weird behaviours, packets out of orders, etc - #[tokio::test] - async fn ack_stream() { - let (tx, rx) = tokio::sync::mpsc::channel(255); - let remote = MessageStream::new(rx); - let stream = new_stub_ack_stream(remote, Duration::from_secs(10)); - let node_id = Uid::new(); - let req_id = Sid::new(); - - // The two servers will send 2 acks each. - let ack_cnt_res = Response::<()> { - node_id, - r#type: ResponseType::BroadcastAckCount(2), - }; - tx.try_send(rmp_serde::to_vec(&(req_id, &ack_cnt_res)).unwrap()) - .unwrap(); - tx.try_send(rmp_serde::to_vec(&(req_id, &ack_cnt_res)).unwrap()) - .unwrap(); - - let ack_res = Response:: { - node_id, - r#type: ResponseType::BroadcastAck((Sid::new(), Ok(Value::Str(Str::from(""), None)))), - }; - for _ in 0..4 { - tx.try_send(rmp_serde::to_vec(&(req_id, &ack_res)).unwrap()) - .unwrap(); - } - futures_util::pin_mut!(stream); - for _ in 0..4 { - assert!(stream.next().await.is_some()); - } - assert!(stream.is_terminated()); - } - - #[tokio::test] - async fn ack_stream_timeout() { - let (tx, rx) = tokio::sync::mpsc::channel(255); - let remote = MessageStream::new(rx); - let stream = new_stub_ack_stream(remote, Duration::from_millis(50)); - let node_id = Uid::new(); - let req_id = Sid::new(); - // There will be only one ack count and then the stream will timeout. - let ack_cnt_res = Response::<()> { - node_id, - r#type: ResponseType::BroadcastAckCount(2), - }; - tx.try_send(rmp_serde::to_vec(&(req_id, ack_cnt_res)).unwrap()) - .unwrap(); - - futures_util::pin_mut!(stream); - tokio::time::sleep(Duration::from_millis(50)).await; - assert!(stream.next().await.is_none()); - assert!(stream.is_terminated()); - } - - #[tokio::test] - async fn ack_stream_drop() { - let (tx, rx) = tokio::sync::mpsc::channel(255); - let remote = MessageStream::new(rx); - let handlers = Arc::new(Mutex::new(HashMap::new())); - let id = Sid::new(); - handlers.lock().unwrap().insert(id, tx); - let stream: AckStream<_, _, Vec, ()> = AckStream::new( - stream::empty::>(), - remote, - decode_redis_ack, - Duration::from_secs(10), - 2, - id, - handlers.clone(), - ); - drop(stream); - assert!(handlers.lock().unwrap().is_empty(),); - } - - #[test] - fn check_ns_error() { - assert!(matches!( - check_ns::("#"), - Err(InitError::MalformedNamespace) - )); - assert!(matches!( - check_ns::(""), - Err(InitError::MalformedNamespace) - )); - } } diff --git a/crates/socketioxide-redis/src/stream.rs b/crates/socketioxide-redis/src/stream.rs deleted file mode 100644 index 673422b9..00000000 --- a/crates/socketioxide-redis/src/stream.rs +++ /dev/null @@ -1,45 +0,0 @@ -use std::fmt; - -use serde::de::DeserializeOwned; -use socketioxide_core::{Sid, adapter::remote_packet::Response}; - -/// Decode a msgpack-encoded `(Sid, Response)` tuple from raw bytes. -pub fn decode_redis_ack( - item: Vec, -) -> Result, Box> { - rmp_serde::from_slice::<(Sid, Response)>(&item) - .map(|(_, response)| response) - .map_err(|e| Box::new(e) as Box) -} - -#[cfg(test)] -mod tests { - use futures_core::FusedStream; - use futures_util::StreamExt; - use socketioxide_core::{Sid, Value}; - - use crate::drivers::MessageStream; - use crate::stream::decode_redis_ack; - use socketioxide_core::adapter::stream::AckStream; - - #[tokio::test] - async fn local_ack_stream_should_have_a_closed_remote() { - let sid = Sid::new(); - let local = futures_util::stream::once(async move { - (sid, Ok::<_, ()>(Value::Str("local".into(), None))) - }); - let stream: AckStream<_, _, Vec, ()> = AckStream::new_empty_remote( - local, - MessageStream::>::new_empty(), - decode_redis_ack::<()>, - ); - futures_util::pin_mut!(stream); - assert!(!FusedStream::is_terminated(&stream)); - let data = stream.next().await; - assert!( - matches!(data, Some((id, Ok(Value::Str(msg, None)))) if id == sid && msg == "local") - ); - assert_eq!(stream.next().await, None); - assert!(FusedStream::is_terminated(&stream)); - } -} From afee63eaeb3adad072f6e0fb78ebd2391de2e36c Mon Sep 17 00:00:00 2001 From: Liam Date: Sun, 19 Jul 2026 02:34:12 +0200 Subject: [PATCH 9/9] fix: remove unused imports and clippy lints in stream.rs test module --- crates/socketioxide-core/src/adapter/stream.rs | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/crates/socketioxide-core/src/adapter/stream.rs b/crates/socketioxide-core/src/adapter/stream.rs index e2b41948..d236f982 100644 --- a/crates/socketioxide-core/src/adapter/stream.rs +++ b/crates/socketioxide-core/src/adapter/stream.rs @@ -216,7 +216,6 @@ impl fmt::Debug for AckStream { #[cfg(test)] mod tests { - use std::borrow::Cow; use std::sync::{Arc, Mutex}; use std::time::Duration; @@ -224,9 +223,8 @@ mod tests { use crate::adapter::remote_packet::{Response, ResponseType}; use crate::adapter::{AckStreamItem, BroadcastIter, RemoteSocketData, SocketEmitter}; use crate::packet::Packet; - use crate::parser::{Parse, ParserError, test::StubParser}; + use crate::parser::test::StubParser; use crate::{Sid, Str, Uid, Value}; - use engineioxide_core::Sid as EngineIoSid; use futures_core::{FusedStream, Stream}; use futures_util::StreamExt; use serde::{Deserialize, Serialize}; @@ -282,7 +280,7 @@ mod tests { unreachable!() } fn parser(&self) -> StubParser { - StubParser::default() + StubParser } fn server_id(&self) -> Uid { Uid::ZERO