Feat: io client shared in socketdata

e2e/engineioxide/engineioxide.rs

@@ -22,7 +22,7 @@ struct MyHandler;
 impl EngineIoHandler for MyHandler {
     type Data = ();
 
-    fn on_connect(&self, socket: Arc<Socket<Self::Data>>) {
+    fn on_connect(self: Arc<Self>, socket: Arc<Socket<Self::Data>>) {
         println!("socket connect {}", socket.id);
     }
     fn on_disconnect(&self, socket: Arc<Socket<Self::Data>>, reason: DisconnectReason) {
@@ -54,7 +54,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         .max_payload(1e6 as u64)
         .build();
 
-    let svc = EngineIoService::with_config(MyHandler, config);
+    let svc = EngineIoService::with_config(Arc::new(MyHandler), config);
 
     let listener = TcpListener::bind("127.0.0.1:3000").await?;
 

engineioxide/Readme.md

@@ -42,7 +42,7 @@ struct SocketState {
 impl EngineIoHandler for MyHandler {
     type Data = SocketState;
 
-    fn on_connect(&self, socket: Arc<Socket<SocketState>>) { 
+    fn on_connect(self: Arc<Self>, socket: Arc<Socket<SocketState>>) { 
         let cnt = self.user_cnt.fetch_add(1, Ordering::Relaxed) + 1;
         socket.emit(cnt.to_string()).ok();
     }
@@ -57,7 +57,7 @@ impl EngineIoHandler for MyHandler {
 }
 
 // Create a new engineio layer
-let layer = EngineIoLayer::new(MyHandler::default());
+let layer = EngineIoLayer::new(Arc::new(MyHandler::default()));
 
 let app = axum::Router::<()>::new()
     .route("/", get(|| async { "Hello, World!" }))

engineioxide/src/config.rs

@@ -13,7 +13,7 @@
 //!
 //! impl EngineIoHandler for MyHandler {
 //!     type Data = ();
-//!     fn on_connect(&self, socket: Arc<Socket<()>>) { }
+//!     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<()>>) { }
@@ -27,7 +27,7 @@
 //!     .build();
 //!
 //! // Create an engine io service with a custom config
-//! let svc = EngineIoService::with_config(MyHandler, config);
+//! let svc = EngineIoService::with_config(Arc::new(MyHandler), config);
 //! ```
 
 use std::{borrow::Cow, time::Duration};
@@ -143,7 +143,7 @@ impl EngineIoConfigBuilder {
     /// impl EngineIoHandler for MyHandler {
     ///
     ///     type Data = ();
-    ///     fn on_connect(&self, socket: Arc<Socket<()>>) {
+    ///     fn on_connect(self: Arc<Self>, socket: Arc<Socket<()>>) {
     ///         println!("socket connect {}", socket.id);
     ///     }
     ///     fn on_disconnect(&self, socket: Arc<Socket<()>>, reason: DisconnectReason) {

engineioxide/src/engine.rs

@@ -21,15 +21,15 @@ pub struct EngineIo<H: EngineIoHandler> {
     sockets: SocketMap<Socket<H::Data>>,
 
     /// The handler for the engine.io server that will be called when events are received
-    pub handler: H,
+    pub handler: Arc<H>,
 
     /// The config for the engine.io server
     pub config: EngineIoConfig,
 }
 
 impl<H: EngineIoHandler> EngineIo<H> {
     /// Create a new Engine.IO server with a [`EngineIoHandler`] and a [`EngineIoConfig`]
-    pub fn new(handler: H, config: EngineIoConfig) -> Self {
+    pub fn new(handler: Arc<H>, config: EngineIoConfig) -> Self {
         Self {
             sockets: RwLock::new(HashMap::new()),
             config,
@@ -64,7 +64,7 @@ impl<H: EngineIoHandler> EngineIo<H> {
             .write()
             .unwrap()
             .insert(socket.id, socket.clone());
-        self.handler.on_connect(socket.clone());
+        self.handler.clone().on_connect(socket.clone());
         socket
     }
 
@@ -107,7 +107,7 @@ mod tests {
     impl EngineIoHandler for MockHandler {
         type Data = ();
 
-        fn on_connect(&self, socket: Arc<Socket<Self::Data>>) {
+        fn on_connect(self: Arc<Self>, socket: Arc<Socket<Self::Data>>) {
             println!("socket connect {}", socket.id);
         }
 
@@ -126,10 +126,14 @@ mod tests {
         }
     }
 
+    fn create_engine() -> Arc<EngineIo<MockHandler>> {
+        let config = EngineIoConfig::default();
+        Arc::new(EngineIo::new(Arc::new(MockHandler), config))
+    }
+
     #[tokio::test]
     async fn create_session() {
-        let config = EngineIoConfig::default();
-        let engine = Arc::new(EngineIo::new(MockHandler, config));
+        let engine = create_engine();
         let socket = engine.create_session(
             ProtocolVersion::V4,
             TransportType::Polling,
@@ -144,8 +148,7 @@ mod tests {
 
     #[tokio::test]
     async fn close_session() {
-        let config = EngineIoConfig::default();
-        let engine = Arc::new(EngineIo::new(MockHandler, config));
+        let engine = create_engine();
         let socket = engine.create_session(
             ProtocolVersion::V4,
             TransportType::Polling,
@@ -160,8 +163,7 @@ mod tests {
 
     #[tokio::test]
     async fn get_socket() {
-        let config = EngineIoConfig::default();
-        let engine = Arc::new(EngineIo::new(MockHandler, config));
+        let engine = create_engine();
         let socket = engine.create_session(
             ProtocolVersion::V4,
             TransportType::Polling,

engineioxide/src/handler.rs

@@ -22,7 +22,7 @@
 //! impl EngineIoHandler for MyHandler {
 //!     type Data = SocketState;
 //!
-//!     fn on_connect(&self, socket: Arc<Socket<SocketState>>) {
+//!     fn on_connect(self: Arc<Self>, socket: Arc<Socket<SocketState>>) {
 //!         let cnt = self.user_cnt.fetch_add(1, Ordering::Relaxed) + 1;
 //!         socket.emit(cnt.to_string()).ok();
 //!     }
@@ -37,7 +37,7 @@
 //! }
 //!
 //! // Create an engine io service with the given handler
-//! let svc = EngineIoService::new(MyHandler::default());
+//! let svc = EngineIoService::new(Arc::new(MyHandler::default()));
 //! ```
 use std::sync::Arc;
 
@@ -54,7 +54,7 @@ pub trait EngineIoHandler: std::fmt::Debug + Send + Sync + 'static {
     type Data: Default + Send + Sync + 'static;
 
     /// Called when a new socket is connected.
-    fn on_connect(&self, socket: Arc<Socket<Self::Data>>);
+    fn on_connect(self: Arc<Self>, socket: Arc<Socket<Self::Data>>);
 
     /// Called when a socket is disconnected with a [`DisconnectReason`]
     fn on_disconnect(&self, socket: Arc<Socket<Self::Data>>, reason: DisconnectReason);
@@ -65,23 +65,3 @@ pub trait EngineIoHandler: std::fmt::Debug + Send + Sync + 'static {
     /// Called when a binary message is received from the client.
     fn on_binary(&self, data: Bytes, socket: Arc<Socket<Self::Data>>);
 }
-
-impl<T: EngineIoHandler> EngineIoHandler for Arc<T> {
-    type Data = T::Data;
-
-    fn on_connect(&self, socket: Arc<Socket<Self::Data>>) {
-        (**self).on_connect(socket)
-    }
-
-    fn on_disconnect(&self, socket: Arc<Socket<Self::Data>>, reason: DisconnectReason) {
-        (**self).on_disconnect(socket, reason)
-    }
-
-    fn on_message(&self, msg: Str, socket: Arc<Socket<Self::Data>>) {
-        (**self).on_message(msg, socket)
-    }
-
-    fn on_binary(&self, data: Bytes, socket: Arc<Socket<Self::Data>>) {
-        (**self).on_binary(data, socket)
-    }
-}

engineioxide/src/layer.rs

@@ -13,19 +13,20 @@
 //!
 //! impl EngineIoHandler for MyHandler {
 //!     type Data = ();
-//!     fn on_connect(&self, socket: Arc<Socket<()>>) { }
+//!     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<()>>) { }
 //! }
 //! // Create a new engineio layer
-//! let layer = EngineIoLayer::new(MyHandler);
+//! let layer = EngineIoLayer::new(Arc::new(MyHandler));
 //!
 //! let app = axum::Router::<()>::new()
 //!     .route("/", get(|| async { "Hello, World!" }))
 //!     .layer(layer);
 //! // Spawn the axum server
 //! ```
+use std::sync::Arc;
 use tower::Layer;
 
 use crate::{config::EngineIoConfig, handler::EngineIoHandler, service::EngineIoService};
@@ -34,13 +35,13 @@ use crate::{config::EngineIoConfig, handler::EngineIoHandler, service::EngineIoS
 #[derive(Debug, Clone)]
 pub struct EngineIoLayer<H: EngineIoHandler> {
     config: EngineIoConfig,
-    handler: H,
+    handler: Arc<H>,
 }
 
 impl<H: EngineIoHandler> EngineIoLayer<H> {
     /// Create a new [`EngineIoLayer`] with a given [`Handler`](crate::handler::EngineIoHandler)
     /// and a default [`EngineIoConfig`]
-    pub fn new(handler: H) -> Self {
+    pub fn new(handler: Arc<H>) -> Self {
         Self {
             config: EngineIoConfig::default(),
             handler,
@@ -49,7 +50,7 @@ impl<H: EngineIoHandler> EngineIoLayer<H> {
 
     /// Create a new [`EngineIoLayer`] with a given [`Handler`](crate::handler::EngineIoHandler)
     /// and a custom [`EngineIoConfig`]
-    pub fn from_config(handler: H, config: EngineIoConfig) -> Self {
+    pub fn from_config(handler: Arc<H>, config: EngineIoConfig) -> Self {
         Self { config, handler }
     }
 }

engineioxide/src/service/mod.rs

@@ -15,14 +15,14 @@
 //!
 //! impl EngineIoHandler for MyHandler {
 //!     type Data = ();
-//!     fn on_connect(&self, socket: Arc<Socket<()>>) { }
+//!     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<()>>) { }
 //! }
 //!
 //! // Create a new engine.io service that will return a 404 not found response for other requests
-//! let service = EngineIoService::new(MyHandler)
+//! let service = EngineIoService::new(Arc::new(MyHandler))
 //!     .into_make_service(); // Create a MakeService from the EngineIoService to give it to hyper
 //! ```
 
@@ -63,23 +63,23 @@ pub struct EngineIoService<H: EngineIoHandler, S = NotFoundService> {
 impl<H: EngineIoHandler> EngineIoService<H, NotFoundService> {
     /// Create a new [`EngineIoService`] with a [`NotFoundService`] as the inner service.
     /// If the request is not an `EngineIo` request, it will always return a 404 response.
-    pub fn new(handler: H) -> Self {
+    pub fn new(handler: Arc<H>) -> Self {
         EngineIoService::with_config(handler, EngineIoConfig::default())
     }
     /// Create a new [`EngineIoService`] with a custom config
-    pub fn with_config(handler: H, config: EngineIoConfig) -> Self {
+    pub fn with_config(handler: Arc<H>, config: EngineIoConfig) -> Self {
         EngineIoService::with_config_inner(NotFoundService, handler, config)
     }
 }
 
 impl<S: Clone, H: EngineIoHandler> EngineIoService<H, S> {
     /// Create a new [`EngineIoService`] with a custom inner service.
-    pub fn with_inner(inner: S, handler: H) -> Self {
+    pub fn with_inner(inner: S, handler: Arc<H>) -> Self {
         EngineIoService::with_config_inner(inner, handler, EngineIoConfig::default())
     }
 
     /// Create a new [`EngineIoService`] with a custom inner service and a custom config.
-    pub fn with_config_inner(inner: S, handler: H, config: EngineIoConfig) -> Self {
+    pub fn with_config_inner(inner: S, handler: Arc<H>, config: EngineIoConfig) -> Self {
         EngineIoService {
             inner,
             engine: Arc::new(EngineIo::new(handler, config)),

engineioxide/src/socket.rs

@@ -28,7 +28,7 @@
 //! impl EngineIoHandler for MyHandler {
 //!     type Data = SocketState;
 //!
-//!     fn on_connect(&self, socket: Arc<Socket<SocketState>>) {
+//!     fn on_connect(self: Arc<Self>, socket: Arc<Socket<SocketState>>) {
 //!         // Get the request made to initialize the connection
 //!         // and check that the authorization header is correct
 //!         let connected = socket.req_parts.headers.get("Authorization")
@@ -52,7 +52,7 @@
 //!     fn on_binary(&self, data: Bytes, socket: Arc<Socket<SocketState>>) { }
 //! }
 //!
-//! let svc = EngineIoService::new(MyHandler::default());
+//! let svc = EngineIoService::new(Arc::new(MyHandler::default()));
 //! ```
 use std::{
     sync::{

engineioxide/tests/disconnect_reason.rs

@@ -31,7 +31,7 @@ struct MyHandler {
 impl EngineIoHandler for MyHandler {
     type Data = ();
 
-    fn on_connect(&self, socket: Arc<Socket<()>>) {
+    fn on_connect(self: Arc<Self>, socket: Arc<Socket<()>>) {
         println!("socket connect {}", socket.id);
     }
     fn on_disconnect(&self, socket: Arc<Socket<()>>, reason: DisconnectReason) {

engineioxide/tests/fixture.rs

@@ -1,6 +1,7 @@
 use std::{
     collections::VecDeque,
     net::{IpAddr, Ipv4Addr, SocketAddr},
+    sync::Arc,
     time::Duration,
 };
 
@@ -83,7 +84,7 @@ pub async fn create_server<H: EngineIoHandler>(handler: H, port: u16) {
 
     let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), port);
 
-    let svc = EngineIoService::with_config(handler, config);
+    let svc = EngineIoService::with_config(Arc::new(handler), config);
 
     let listener = TcpListener::bind(&addr).await.unwrap();
     tokio::spawn(async move {