feat: typestate pattern for SocketIoBuilder + tests fixes

Author: Totodore

crates/engineioxide/Readme.md

@@ -53,10 +53,10 @@ impl EngineIoHandler for MyHandler {
         let cnt = self.user_cnt.fetch_sub(1, Ordering::Relaxed) - 1;
         socket.emit(cnt.to_string()).ok();
     }
-    fn on_message(&self, msg: Str, socket: Arc<Socket<SocketState>>) {
+    fn on_message(self: &Arc<Self>, msg: Str, socket: Arc<Socket<SocketState>>) {
         *socket.data.id.lock().unwrap() = msg.into(); // bind a provided user id to a socket
     }
-    fn on_binary(&self, data: Bytes, socket: Arc<Socket<SocketState>>) { }
+    fn on_binary(self: &Arc<Self>, data: Bytes, socket: Arc<Socket<SocketState>>) { }
 }
 
 // Create a new engineio layer

crates/engineioxide/src/config.rs

@@ -15,8 +15,8 @@
 //!     type Data = ();
 //!     fn on_connect(self: Arc<Self>, socket: Arc<Socket<()>>) { }
 //!     fn on_disconnect(&self, socket: Arc<Socket<()>>, reason: DisconnectReason) { }
-//!     fn on_message(&self, msg: Str, socket: Arc<Socket<()>>) { }
-//!     fn on_binary(&self, data: Bytes, socket: Arc<Socket<()>>) { }
+//!     fn on_message(self: &Arc<Self>, msg: Str, socket: Arc<Socket<()>>) { }
+//!     fn on_binary(self: &Arc<Self>, data: Bytes, socket: Arc<Socket<()>>) { }
 //! }
 //!
 //! let config = EngineIoConfig::builder()
@@ -150,12 +150,12 @@ impl EngineIoConfigBuilder {
     ///         println!("socket disconnect {}", socket.id);
     ///     }
     ///
-    ///     fn on_message(&self, msg: Str, socket: Arc<Socket<()>>) {
+    ///     fn on_message(self: &Arc<Self>, msg: Str, socket: Arc<Socket<()>>) {
     ///         println!("Ping pong message {:?}", msg);
     ///         socket.emit(msg).unwrap();
     ///     }
     ///
-    ///     fn on_binary(&self, data: Bytes, socket: Arc<Socket<()>>) {
+    ///     fn on_binary(self: &Arc<Self>, data: Bytes, socket: Arc<Socket<()>>) {
     ///         println!("Ping pong binary message {:?}", data);
     ///         socket.emit_binary(data).unwrap();
     ///     }

crates/engineioxide/src/handler.rs

@@ -30,10 +30,10 @@
 //!         let cnt = self.user_cnt.fetch_sub(1, Ordering::Relaxed) - 1;
 //!         socket.emit(cnt.to_string()).ok();
 //!     }
-//!     fn on_message(&self, msg: Str, socket: Arc<Socket<SocketState>>) {
+//!     fn on_message(self: &Arc<Self>, msg: Str, socket: Arc<Socket<SocketState>>) {
 //!         *socket.data.id.lock().unwrap() = msg.into(); // bind a provided user id to a socket
 //!     }
-//!     fn on_binary(&self, data: Bytes, socket: Arc<Socket<SocketState>>) { }
+//!     fn on_binary(self: &Arc<Self>, data: Bytes, socket: Arc<Socket<SocketState>>) { }
 //! }
 //!
 //! // Create an engine io service with the given handler

crates/engineioxide/src/layer.rs

@@ -15,8 +15,8 @@
 //!     type Data = ();
 //!     fn on_connect(self: Arc<Self>, socket: Arc<Socket<()>>) { }
 //!     fn on_disconnect(&self, socket: Arc<Socket<()>>, reason: DisconnectReason) { }
-//!     fn on_message(&self, msg: Str, socket: Arc<Socket<()>>) { }
-//!     fn on_binary(&self, data: Bytes, socket: Arc<Socket<()>>) { }
+//!     fn on_message(self: &Arc<Self>, msg: Str, socket: Arc<Socket<()>>) { }
+//!     fn on_binary(self: &Arc<Self>, data: Bytes, socket: Arc<Socket<()>>) { }
 //! }
 //! // Create a new engineio layer
 //! let layer = EngineIoLayer::new(Arc::new(MyHandler));

crates/engineioxide/src/service/mod.rs

@@ -17,8 +17,8 @@
 //!     type Data = ();
 //!     fn on_connect(self: Arc<Self>, socket: Arc<Socket<()>>) { }
 //!     fn on_disconnect(&self, socket: Arc<Socket<()>>, reason: DisconnectReason) { }
-//!     fn on_message(&self, msg: Str, socket: Arc<Socket<()>>) { }
-//!     fn on_binary(&self, data: Bytes, socket: Arc<Socket<()>>) { }
+//!     fn on_message(self: &Arc<Self>, msg: Str, socket: Arc<Socket<()>>) { }
+//!     fn on_binary(self: &Arc<Self>, data: Bytes, socket: Arc<Socket<()>>) { }
 //! }
 //!
 //! // Create a new engine.io service that will return a 404 not found response for other requests

crates/engineioxide/src/socket.rs

@@ -46,10 +46,10 @@
 //!     fn on_disconnect(&self, socket: Arc<Socket<SocketState>>, reason: DisconnectReason) {
 //!         let cnt = self.user_cnt.fetch_sub(1, Ordering::Relaxed) - 1;
 //!     }
-//!     fn on_message(&self, msg: Str, socket: Arc<Socket<SocketState>>) {
+//!     fn on_message(self: &Arc<Self>, msg: Str, socket: Arc<Socket<SocketState>>) {
 //!         *socket.data.id.lock().unwrap() = msg.into(); // bind a provided user id to a socket
 //!     }
-//!     fn on_binary(&self, data: Bytes, socket: Arc<Socket<SocketState>>) { }
+//!     fn on_binary(self: &Arc<Self>, data: Bytes, socket: Arc<Socket<SocketState>>) { }
 //! }
 //!
 //! let svc = EngineIoService::new(Arc::new(MyHandler::default()));

crates/socketioxide-core/src/adapter.rs

@@ -180,17 +180,22 @@ pub trait SocketEmitter: Send + Sync + 'static {
 
 /// An adapter is responsible for managing the state of the namespace.
 /// This adapter can be implemented to share the state between multiple servers.
-/// The default adapter is the [`LocalAdapter`], which stores the state in memory.
+///
+/// A [`CoreLocalAdapter`] instance will be given when constructing this type, it will allow
+/// you to manipulate local sockets (emitting, fetching data, broadcasting).
 pub trait CoreAdapter<E: SocketEmitter>: Sized + Send + Sync + 'static {
-    /// An error that can occur when using the adapter. The default [`LocalAdapter`] has an [`Infallible`] error.
+    /// An error that can occur when using the adapter.
     type Error: StdError + Into<AdapterError> + Send + 'static;
     /// A shared state between all the namespace [`CoreAdapter`].
     /// This can be used to share a connection for example.
     type State: Send + Sync + 'static;
     /// A stream that emits the acknowledgments of multiple sockets.
     type AckStream: Stream<Item = AckStreamItem<E::AckError>> + FusedStream + Send + 'static;
 
-    /// Creates a new adapter with the given state and socket server.
+    /// Creates a new adapter with the given state and local adapter.
+    ///
+    /// The state is used to share a common state between all your adapters. E.G. a connection to a remote system.
+    /// The local adapter is used to manipulate the local sockets.
     fn new(state: &Self::State, local: CoreLocalAdapter<E>) -> Self;
 
     /// Initializes the adapter.
@@ -245,7 +250,7 @@ pub trait CoreAdapter<E: SocketEmitter>: Sized + Send + Sync + 'static {
     /// and return a stream of ack responses.
     ///
     /// This method does not have default implementation because GAT cannot have default impls.
-    /// https://github.com/rust-lang/rust/issues/29661
+    /// <https://github.com/rust-lang/rust/issues/29661>
     fn broadcast_with_ack(
         &self,
         packet: Packet,

crates/socketioxide-core/src/errors.rs

@@ -8,7 +8,7 @@ pub enum SocketError {
     /// The socket channel is full.
     /// You might need to increase the channel size with the [`SocketIoBuilder::max_buffer_size`] method.
     ///
-    /// [`SocketIoBuilder::max_buffer_size`]: crate::SocketIoBuilder#method.max_buffer_size
+    /// [`SocketIoBuilder::max_buffer_size`]: https://docs.rs/socketioxide/latest/socketioxide/struct.SocketIoBuilder.html#method.max_buffer_size
     #[error("internal channel full error")]
     InternalChannelFull,
 
@@ -17,7 +17,7 @@ pub enum SocketError {
     Closed,
 }
 
-/// Error type for the [`Adapter`](crate::adapter::Adapter) trait.
+/// Error type for the [`CoreAdapter`](crate::adapter::CoreAdapter) trait.
 #[derive(Debug, thiserror::Error)]
 pub struct AdapterError(#[from] pub Box<dyn std::error::Error + Send>);
 impl fmt::Display for AdapterError {
@@ -37,7 +37,7 @@ pub enum DisconnectError {
     /// The socket channel is full.
     /// You might need to increase the channel size with the [`SocketIoBuilder::max_buffer_size`] method.
     ///
-    /// [`SocketIoBuilder::max_buffer_size`]: crate::SocketIoBuilder#method.max_buffer_size
+    /// [`SocketIoBuilder::max_buffer_size`]: https://docs.rs/socketioxide/latest/socketioxide/struct.SocketIoBuilder.html#method.max_buffer_size
     #[error("internal channel full error")]
     InternalChannelFull,
 

crates/socketioxide/docs/operators/emit_with_ack.md

@@ -10,7 +10,7 @@ To receive acknowledgments, an [`AckStream`] is returned. It can be used in two
 * As a [`Future`]: This will yield the first acknowledgment response received from the client, useful when expecting only one acknowledgment.
 
 # Errors
-If packet encoding fails, an [`EncodeError`] is **immediately** returned.
+If packet encoding fails, an [`ParserError`] is **immediately** returned.
 
 If the socket is full or if it is closed before receiving the acknowledgment,
 a [`SendError::Socket`] will be **immediately** returned, and the value to send will be given back.
@@ -28,7 +28,7 @@ an [`AckError::Decode`] will be yielded.
 [`AckError::Socket`]: crate::AckError::Socket
 [`AckError::Socket(SocketError::Closed)`]: crate::SocketError::Closed
 [`SendError::Socket`]: crate::SendError::Socket
-[`EncodeError`]: crate::EncodeError
+[`ParserError`]: crate::ParserError
 [`io::get_socket()`]: crate::SocketIo#method.get_socket
 
 # Single-socket example

crates/socketioxide/src/client.rs

@@ -37,7 +37,7 @@ pub struct Client<A: Adapter> {
     pub(crate) state: state::TypeMap![Send + Sync],
 }
 
-/// ==== impl Client ====
+// ==== impl Client ====
 
 impl<A: Adapter> Client<A> {
     pub fn new(
@@ -148,7 +148,7 @@ impl<A: Adapter> Client<A> {
     }
 
     /// Adds a new namespace handler
-    pub async fn add_ns<C, T>(&self, path: Cow<'static, str>, callback: C) -> Result<(), A::Error>
+    pub fn add_ns<C, T>(&self, path: Cow<'static, str>, callback: C)
     where
         C: ConnectHandler<A, T>,
         T: Send + Sync + 'static,
@@ -162,9 +162,7 @@ impl<A: Adapter> Client<A> {
             &self.adapter_state,
             self.config.parser,
         );
-        ns.adapter.clone().init().await?;
         self.nsps.write().unwrap().insert(path, ns);
-        Ok(())
     }
 
     pub fn add_dyn_ns<C, T>(&self, path: String, callback: C) -> Result<(), matchit::InsertError>
@@ -179,6 +177,16 @@ impl<A: Adapter> Client<A> {
         self.router.write().unwrap().insert(path, ns)
     }
 
+    /// Initializes all the namespace handlers
+    ///
+    /// If an any error occurs while initializing a namespace, it is immediately returned
+    pub async fn init_nsps(&self) -> Result<(), A::Error> {
+        let nsps: Vec<_> = self.nsps.read().unwrap().values().cloned().collect();
+        let futures = nsps.into_iter().map(|ns| ns.adapter.clone().init());
+        futures_util::future::try_join_all(futures).await?;
+        Ok(())
+    }
+
     /// Deletes a namespace handler and closes all the connections to it
     pub fn delete_ns(&self, path: &str) {
         #[cfg(feature = "v4")]