feat(socketio): Async adapter

.github/workflows/github-ci.yml

@@ -5,8 +5,6 @@ on:
     branches:
       - main
   pull_request:
-    branches:
-      - main
 
 jobs:
   format:

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/engine.rs

@@ -115,12 +115,12 @@ mod tests {
             println!("socket disconnect {} {:?}", socket.id, reason);
         }
 
-        fn on_message(&self, msg: Str, socket: Arc<Socket<Self::Data>>) {
+        fn on_message(self: &Arc<Self>, msg: Str, socket: Arc<Socket<Self::Data>>) {
             println!("Ping pong message {:?}", msg);
             socket.emit(msg).ok();
         }
 
-        fn on_binary(&self, data: Bytes, socket: Arc<Socket<Self::Data>>) {
+        fn on_binary(self: &Arc<Self>, data: Bytes, socket: Arc<Socket<Self::Data>>) {
             println!("Ping pong binary message {:?}", data);
             socket.emit_binary(data).ok();
         }

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
@@ -60,8 +60,8 @@ pub trait EngineIoHandler: std::fmt::Debug + Send + Sync + 'static {
     fn on_disconnect(&self, socket: Arc<Socket<Self::Data>>, reason: DisconnectReason);
 
     /// Called when a message is received from the client.
-    fn on_message(&self, msg: Str, socket: Arc<Socket<Self::Data>>);
+    fn on_message(self: &Arc<Self>, msg: Str, socket: Arc<Socket<Self::Data>>);
 
     /// Called when a binary message is received from the client.
-    fn on_binary(&self, data: Bytes, socket: Arc<Socket<Self::Data>>);
+    fn on_binary(self: &Arc<Self>, data: Bytes, socket: Arc<Socket<Self::Data>>);
 }

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/engineioxide/src/str.rs

@@ -100,6 +100,11 @@ impl From<Str> for String {
         unsafe { String::from_utf8_unchecked(vec) }
     }
 }
+impl From<Str> for Vec<u8> {
+    fn from(value: Str) -> Self {
+        Vec::from(value.0)
+    }
+}
 impl Serialize for Str {
     fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
     where

crates/engineioxide/tests/disconnect_reason.rs

@@ -39,12 +39,12 @@ impl EngineIoHandler for MyHandler {
         self.disconnect_tx.try_send(reason).unwrap();
     }
 
-    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).ok();
     }
 
-    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).ok();
     }

crates/parser-common/src/de.rs

@@ -7,9 +7,7 @@ use socketioxide_core::{
     Str, Value,
 };
 
-pub fn deserialize_packet(
-    data: Str,
-) -> Result<(Packet, Option<usize>), ParseError<serde_json::Error>> {
+pub fn deserialize_packet(data: Str) -> Result<(Packet, Option<usize>), ParseError> {
     if data.is_empty() {
         return Err(ParseError::InvalidPacketType);
     }

crates/parser-common/src/lib.rs

@@ -45,7 +45,7 @@ use bytes::Bytes;
 use serde::{Deserialize, Serialize};
 use socketioxide_core::{
     packet::{Packet, PacketData},
-    parser::{Parse, ParseError, ParserState},
+    parser::{Parse, ParseError, ParserError, ParserState},
     Str, Value,
 };
 
@@ -59,17 +59,11 @@ mod value;
 pub struct CommonParser;
 
 impl Parse for CommonParser {
-    type EncodeError = serde_json::Error;
-    type DecodeError = serde_json::Error;
     fn encode(self, packet: Packet) -> Value {
         ser::serialize_packet(packet)
     }
 
-    fn decode_str(
-        self,
-        state: &ParserState,
-        value: Str,
-    ) -> Result<Packet, ParseError<Self::DecodeError>> {
+    fn decode_str(self, state: &ParserState, value: Str) -> Result<Packet, ParseError> {
         let (packet, incoming_binary_cnt) = de::deserialize_packet(value)?;
         if packet.inner.is_binary() {
             let incoming_binary_cnt = incoming_binary_cnt.ok_or(ParseError::InvalidAttachments)?;
@@ -87,11 +81,7 @@ impl Parse for CommonParser {
         }
     }
 
-    fn decode_bin(
-        self,
-        state: &ParserState,
-        data: Bytes,
-    ) -> Result<Packet, ParseError<Self::DecodeError>> {
+    fn decode_bin(self, state: &ParserState, data: Bytes) -> Result<Packet, ParseError> {
         let packet = &mut *state.partial_bin_packet.lock().unwrap();
         match packet {
             Some(Packet {
@@ -117,41 +107,41 @@ impl Parse for CommonParser {
         self,
         data: &T,
         event: Option<&str>,
-    ) -> Result<Value, Self::EncodeError> {
-        value::to_value(data, event)
+    ) -> Result<Value, ParserError> {
+        value::to_value(data, event).map_err(ParserError::new)
     }
 
     #[inline]
     fn decode_value<'de, T: Deserialize<'de>>(
         self,
         value: &'de mut Value,
         with_event: bool,
-    ) -> Result<T, Self::DecodeError> {
-        value::from_value(value, with_event)
+    ) -> Result<T, ParserError> {
+        value::from_value(value, with_event).map_err(ParserError::new)
     }
 
     fn decode_default<'de, T: Deserialize<'de>>(
         self,
         value: Option<&'de Value>,
-    ) -> Result<T, Self::DecodeError> {
+    ) -> Result<T, ParserError> {
         if let Some(value) = value {
             let data = value
                 .as_str()
                 .expect("CommonParser only supports string values");
-            serde_json::from_str(data)
+            serde_json::from_str(data).map_err(ParserError::new)
         } else {
-            serde_json::from_str("{}")
+            serde_json::from_str("{}").map_err(ParserError::new)
         }
     }
 
-    fn encode_default<T: ?Sized + Serialize>(self, data: &T) -> Result<Value, Self::EncodeError> {
-        let value = serde_json::to_string(data)?;
+    fn encode_default<T: ?Sized + Serialize>(self, data: &T) -> Result<Value, ParserError> {
+        let value = serde_json::to_string(data).map_err(ParserError::new)?;
         Ok(Value::Str(Str::from(value), None))
     }
 
     #[inline]
-    fn read_event(self, value: &Value) -> Result<&str, Self::DecodeError> {
-        value::read_event(value)
+    fn read_event(self, value: &Value) -> Result<&str, ParserError> {
+        value::read_event(value).map_err(ParserError::new)
     }
 }
 
@@ -590,16 +580,16 @@ mod test {
     fn decode_default_none() {
         // Common parser should deserialize by default to an empty map to match the behavior of the
         // socket.io client when deserializing incoming connect message without an auth payload.
-        let data: serde_json::Result<HashMap<String, ()>> = CommonParser.decode_default(None);
+        let data = CommonParser.decode_default::<HashMap<String, ()>>(None);
         assert!(matches!(data, Ok(d) if d.is_empty()));
     }
 
     #[test]
     fn decode_default_some() {
         // Common parser should deserialize by default to an empty map to match the behavior of the
         // socket.io client when deserializing incoming connect message without an auth payload.
-        let data: serde_json::Result<String> =
-            CommonParser.decode_default(Some(&Value::Str("\"test\"".into(), None)));
+        let data =
+            CommonParser.decode_default::<String>(Some(&Value::Str("\"test\"".into(), None)));
         assert!(matches!(data, Ok(d) if d == "test"));
     }
 

crates/parser-msgpack/src/de.rs

@@ -9,11 +9,11 @@ use rmp::{
 use rmp_serde::decode::Error as DecodeError;
 use socketioxide_core::{
     packet::{Packet, PacketData},
-    parser::ParseError,
+    parser::{ParseError, ParserError},
     Str, Value,
 };
 
-pub fn deserialize_packet(buff: Bytes) -> Result<Packet, ParseError<DecodeError>> {
+pub fn deserialize_packet(buff: Bytes) -> Result<Packet, ParseError> {
     let mut reader = Cursor::new(buff);
     let maplen = read_map_len(&mut reader).map_err(|e| {
         use DecodeError::*;
@@ -22,17 +22,17 @@ pub fn deserialize_packet(buff: Bytes) -> Result<Packet, ParseError<DecodeError>
             ValueReadError::InvalidDataRead(e) => InvalidDataRead(e),
             ValueReadError::TypeMismatch(e) => TypeMismatch(e),
         };
-        ParseError::ParserError(e)
+        ParseError::ParserError(ParserError::new(e))
     })?;
 
     // Bound check to prevent DoS attacks.
     // other implementations might add some other keys that we don't support
     // Therefore, we limit the number of keys to 20
     if maplen == 0 || maplen > 20 {
-        Err(DecodeError::Uncategorized(format!(
+        Err(ParserError::new(DecodeError::Uncategorized(format!(
             "packet length too big or empty: {}",
             maplen
-        )))?;
+        ))))?;
     }
 
     let mut index = 0xff;
@@ -41,7 +41,8 @@ pub fn deserialize_packet(buff: Bytes) -> Result<Packet, ParseError<DecodeError>
     let mut id = None;
 
     for _ in 0..maplen {
-        parse_key_value(&mut reader, &mut index, &mut nsp, &mut data_pos, &mut id)?;
+        parse_key_value(&mut reader, &mut index, &mut nsp, &mut data_pos, &mut id)
+            .map_err(ParserError::new)?;
     }
     let buff = reader.into_inner();
     let mut data = buff.slice(data_pos.clone());
@@ -64,7 +65,8 @@ pub fn deserialize_packet(buff: Bytes) -> Result<Packet, ParseError<DecodeError>
             struct ErrorMessage {
                 message: String,
             }
-            let ErrorMessage { message } = rmp_serde::decode::from_slice(&buff[data_pos])?;
+            let ErrorMessage { message } =
+                rmp_serde::decode::from_slice(&buff[data_pos]).map_err(ParserError::new)?;
             PacketData::ConnectError(message)
         }
         5 => PacketData::BinaryEvent(data, id),