feat: per client state

examples/angular-todomvc/src/main.rs

@@ -5,13 +5,14 @@ use socketioxide::{
     extract::{Data, SocketRef, State},
     SocketIo,
 };
+use std::sync::Arc;
 use tower::ServiceBuilder;
 use tower_http::{cors::CorsLayer, services::ServeDir};
 use tracing::info;
 use tracing_subscriber::FmtSubscriber;
 
-#[derive(Default)]
-struct Todos(pub Mutex<Vec<Todo>>);
+#[derive(Default, Clone)]
+struct Todos(Arc<Mutex<Vec<Todo>>>);
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 struct Todo {

examples/basic-crud-application/src/handlers/todo.rs

@@ -1,4 +1,4 @@
-use std::{collections::HashMap, sync::RwLock};
+use std::{collections::HashMap, sync::{RwLock, Arc}};
 
 use serde::{Deserialize, Serialize};
 use socketioxide::extract::{AckSender, Data, SocketRef, State};
@@ -21,8 +21,8 @@ pub struct PartialTodo {
     title: String,
 }
 
-#[derive(Default)]
-pub struct Todos(RwLock<HashMap<Uuid, Todo>>);
+#[derive(Clone, Default)]
+pub struct Todos(Arc<RwLock<HashMap<Uuid, Todo>>>);
 impl Todos {
     fn insert(&self, id: Uuid, todo: Todo) {
         self.0.write().unwrap().insert(id, todo);

examples/chat/src/main.rs

@@ -9,6 +9,7 @@ use tower::ServiceBuilder;
 use tower_http::{cors::CorsLayer, services::ServeDir};
 use tracing::info;
 use tracing_subscriber::FmtSubscriber;
+use std::sync::Arc;
 
 #[derive(Deserialize, Serialize, Debug, Clone)]
 #[serde(transparent)]
@@ -34,11 +35,11 @@ enum Res {
         username: Username,
     },
 }
-
-struct UserCnt(AtomicUsize);
+#[derive(Clone)]
+struct UserCnt(Arc<AtomicUsize>);
 impl UserCnt {
     fn new() -> Self {
-        Self(AtomicUsize::new(0))
+        Self(Arc::new(AtomicUsize::new(0)))
     }
     fn add_user(&self) -> usize {
         self.0.fetch_add(1, std::sync::atomic::Ordering::SeqCst) + 1

examples/loco-rooms-chat/Cargo.toml

@@ -6,7 +6,7 @@ rust-version = "1.70"    # required by loco
 
 [dependencies]
 
-loco-rs = { version = "0.3.1", default-features = false, features = [
+loco-rs = { version = "0.5.0", default-features = false, features = [
     "cli",
     "channels",
 ] }

examples/loco-rooms-chat/src/channels/state.rs

@@ -1,5 +1,6 @@
 use std::collections::{HashMap, VecDeque};
 use tokio::sync::RwLock;
+use std::sync::Arc;
 
 #[derive(serde::Serialize, Clone, Debug)]
 pub struct Message {
@@ -10,9 +11,9 @@ pub struct Message {
 
 pub type RoomStore = HashMap<String, VecDeque<Message>>;
 
-#[derive(Default)]
+#[derive(Default, Clone)]
 pub struct MessageStore {
-    pub messages: RwLock<RoomStore>,
+    pub messages: Arc<RwLock<RoomStore>>,
 }
 
 impl MessageStore {

examples/private-messaging/src/store.rs

@@ -47,8 +47,8 @@ pub struct Message {
     pub content: String,
 }
 
-#[derive(Default)]
-pub struct Sessions(RwLock<HashMap<Uuid, Arc<Session>>>);
+#[derive(Clone, Default)]
+pub struct Sessions(Arc<RwLock<HashMap<Uuid, Arc<Session>>>>);
 
 impl Sessions {
     pub fn get_all_other_sessions(&self, user_id: Uuid) -> Vec<Arc<Session>> {
@@ -69,8 +69,8 @@ impl Sessions {
         self.0.write().unwrap().insert(session.session_id, session);
     }
 }
-#[derive(Default)]
-pub struct Messages(RwLock<Vec<Message>>);
+#[derive(Clone, Default)]
+pub struct Messages(Arc<RwLock<Vec<Message>>>);
 
 impl Messages {
     pub fn add(&self, message: Message) {

examples/react-rooms-chat/Cargo.toml

@@ -6,13 +6,13 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-socketioxide = { version = "0.9.0", features = ["state"] }
+socketioxide = { path = "../../socketioxide", features = ["state"] }
 tokio = { version = "1", features = ["full"] }
 tracing = "0.1"
 tracing-subscriber = "0.3"
 axum = "0.7.2"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
-tower-http = {version = "0.5.0", features = ["cors"]}
+tower-http = { version = "0.5.0", features = ["cors"] }
 tower = "0.4"
 chrono = { version = "0.4", features = ["serde"] }

examples/react-rooms-chat/src/state.rs

@@ -1,4 +1,4 @@
-use std::collections::{HashMap, VecDeque};
+use std::{collections::{HashMap, VecDeque}, sync::Arc};
 use tokio::sync::RwLock;
 
 #[derive(serde::Serialize, Clone, Debug)]
@@ -10,9 +10,9 @@ pub struct Message {
 
 pub type RoomStore = HashMap<String, VecDeque<Message>>;
 
-#[derive(Default)]
+#[derive(Clone, Default)]
 pub struct MessageStore {
-    pub messages: RwLock<RoomStore>,
+    pub messages: Arc<RwLock<RoomStore>>,
 }
 
 impl MessageStore {

socketioxide/src/adapter.rs

@@ -1,4 +1,4 @@
-//! Adapters are responsible for managing the state of the server.
+//! Adapters are responsible for managing the internal state of the server (rooms, sockets, etc...).
 //! When a socket joins or leaves a room, the adapter is responsible for updating the state.
 //! The default adapter is the [`LocalAdapter`], which stores the state in memory.
 //! Other adapters can be made to share the state between multiple servers.

socketioxide/src/client.rs

@@ -22,20 +22,26 @@ use crate::{
 };
 use crate::{ProtocolVersion, SocketIo};
 
-#[derive(Debug)]
 pub struct Client<A: Adapter> {
     pub(crate) config: Arc<SocketIoConfig>,
     ns: RwLock<HashMap<Cow<'static, str>, Arc<Namespace<A>>>>,
+    #[cfg(feature = "state")]
+    pub(crate) state: state::TypeMap![Send + Sync],
 }
 
 impl<A: Adapter> Client<A> {
-    pub fn new(config: Arc<SocketIoConfig>) -> Self {
+    pub fn new(
+        config: Arc<SocketIoConfig>,
+        #[cfg(feature = "state")] mut state: state::TypeMap![Send + Sync],
+    ) -> Self {
         #[cfg(feature = "state")]
-        crate::state::freeze_state();
+        state.freeze();
 
         Self {
             config,
             ns: RwLock::new(HashMap::new()),
+            #[cfg(feature = "state")]
+            state,
         }
     }
 
@@ -346,6 +352,15 @@ impl<A: Adapter> EngineIoHandler for Client<A> {
         }
     }
 }
+impl<A: Adapter> std::fmt::Debug for Client<A> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let mut f = f.debug_struct("Client");
+        f.field("config", &self.config).field("ns", &self.ns);
+        #[cfg(feature = "state")]
+        let f = f.field("state", &self.state);
+        f.finish()
+    }
+}
 
 /// Utility that applies an incoming binary payload to a partial binary packet
 /// waiting to be filled with all the payloads
@@ -382,7 +397,11 @@ mod test {
             connect_timeout: CONNECT_TIMEOUT,
             ..Default::default()
         };
-        let client = Client::<LocalAdapter>::new(std::sync::Arc::new(config));
+        let client = Client::<LocalAdapter>::new(
+            std::sync::Arc::new(config),
+            #[cfg(feature = "state")]
+            Default::default(),
+        );
         client.add_ns("/".into(), || {});
         Arc::new(client)
     }

socketioxide/src/extract/mod.rs

@@ -16,7 +16,7 @@
 //! * [`ProtocolVersion`](crate::ProtocolVersion): extracts the protocol version
 //! * [`TransportType`](crate::TransportType): extracts the transport type
 //! * [`DisconnectReason`](crate::socket::DisconnectReason): extracts the reason of the disconnection
-//! * [`State`]: extracts a reference to a state previously set with [`SocketIoBuilder::with_state`](crate::io::SocketIoBuilder).
+//! * [`State`]: extracts a [`Clone`] of a state previously set with [`SocketIoBuilder::with_state`](crate::io::SocketIoBuilder).
 //! * [`Extension`]: extracts an extension of the given type stored on the called socket by cloning it.
 //! * [`MaybeExtension`]: extracts an extension of the given type if it exists or [`None`] otherwise
 //! * [`HttpExtension`]: extracts an http extension of the given type coming from the request.

socketioxide/src/extract/state.rs

@@ -1,14 +1,12 @@
 use bytes::Bytes;
 
-use crate::state::get_state;
-use std::ops::Deref;
 use std::sync::Arc;
 
 use crate::adapter::Adapter;
 use crate::handler::{FromConnectParts, FromDisconnectParts, FromMessageParts};
 use crate::socket::{DisconnectReason, Socket};
 
-/// An Extractor that contains a reference to a state previously set with [`SocketIoBuilder::with_state`](crate::io::SocketIoBuilder).
+/// An Extractor that contains a [`Clone`] of a state previously set with [`SocketIoBuilder::with_state`](crate::io::SocketIoBuilder).
 /// It implements [`std::ops::Deref`] to access the inner type so you can use it as a normal reference.
 ///
 /// The specified state type must be the same as the one set with [`SocketIoBuilder::with_state`](crate::io::SocketIoBuilder).
@@ -20,11 +18,10 @@ use crate::socket::{DisconnectReason, Socket};
 /// ```
 /// # use socketioxide::{SocketIo, extract::{SocketRef, State}};
 /// # use serde::{Serialize, Deserialize};
-/// # use std::sync::atomic::AtomicUsize;
-/// # use std::sync::atomic::Ordering;
-/// #[derive(Default)]
+/// # use std::sync::{Arc, atomic::{Ordering, AtomicUsize}};
+/// #[derive(Default, Clone)]
 /// struct MyAppData {
-///     user_cnt: AtomicUsize,
+///     user_cnt: Arc<AtomicUsize>,
 /// }
 /// impl MyAppData {
 ///     fn add_user(&self) {
@@ -39,7 +36,7 @@ use crate::socket::{DisconnectReason, Socket};
 ///     state.add_user();
 ///     println!("User count: {}", state.user_cnt.load(Ordering::SeqCst));
 /// });
-pub struct State<T: 'static>(pub &'static T);
+pub struct State<T>(pub T);
 
 /// It was impossible to find the given state and therefore the handler won't be called.
 pub struct StateNotFound<T>(std::marker::PhantomData<T>);
@@ -48,7 +45,7 @@ impl<T> std::fmt::Display for StateNotFound<T> {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         write!(
             f,
-            "State of type {} not found, maybe you forgot to insert it in the extensions map?",
+            "State of type {} not found, maybe you forgot to insert it in the state map?",
             std::any::type_name::<T>()
         )
     }
@@ -60,46 +57,43 @@ impl<T> std::fmt::Debug for StateNotFound<T> {
 }
 impl<T> std::error::Error for StateNotFound<T> {}
 
-impl<A: Adapter, T: Send + Sync + 'static> FromConnectParts<A> for State<T> {
+impl<A: Adapter, T: Clone + Send + Sync + 'static> FromConnectParts<A> for State<T> {
     type Error = StateNotFound<T>;
     fn from_connect_parts(
-        _: &Arc<Socket<A>>,
+        s: &Arc<Socket<A>>,
         _: &Option<String>,
     ) -> Result<Self, StateNotFound<T>> {
-        get_state::<T>()
+        s.get_io()
+            .get_state::<T>()
             .map(State)
             .ok_or(StateNotFound(std::marker::PhantomData))
     }
 }
-impl<A: Adapter, T: Send + Sync + 'static> FromDisconnectParts<A> for State<T> {
+impl<A: Adapter, T: Clone + Send + Sync + 'static> FromDisconnectParts<A> for State<T> {
     type Error = StateNotFound<T>;
     fn from_disconnect_parts(
-        _: &Arc<Socket<A>>,
+        s: &Arc<Socket<A>>,
         _: DisconnectReason,
     ) -> Result<Self, StateNotFound<T>> {
-        get_state::<T>()
+        s.get_io()
+            .get_state::<T>()
             .map(State)
             .ok_or(StateNotFound(std::marker::PhantomData))
     }
 }
-impl<A: Adapter, T: Send + Sync + 'static> FromMessageParts<A> for State<T> {
+impl<A: Adapter, T: Clone + Send + Sync + 'static> FromMessageParts<A> for State<T> {
     type Error = StateNotFound<T>;
     fn from_message_parts(
-        _: &Arc<Socket<A>>,
+        s: &Arc<Socket<A>>,
         _: &mut serde_json::Value,
         _: &mut Vec<Bytes>,
         _: &Option<i64>,
     ) -> Result<Self, StateNotFound<T>> {
-        get_state::<T>()
+        s.get_io()
+            .get_state::<T>()
             .map(State)
             .ok_or(StateNotFound(std::marker::PhantomData))
     }
 }
 
-impl<T> Deref for State<T> {
-    type Target = &'static T;
-    #[inline(always)]
-    fn deref(&self) -> &Self::Target {
-        &self.0
-    }
-}
+super::__impl_deref!(State);