Custom parser support

examples/axum-echo/axum_echo.rs

@@ -1,11 +1,9 @@
 use axum::routing::get;
 use serde_json::Value;
-use socketioxide::{
-    extract::{AckSender, Bin, Data, SocketRef},
-    SocketIo,
-};
+use socketioxide::{extract::{AckSender, Bin, Data, SocketRef}, SocketIo, SocketIoBuilder};
 use tracing::info;
 use tracing_subscriber::FmtSubscriber;
+use socketioxide::parser::MsgpackParser;
 
 fn on_connect(socket: SocketRef, Data(data): Data<Value>) {
     info!("Socket.IO connected: {:?} {:?}", socket.ns(), socket.id);
@@ -32,7 +30,9 @@
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     tracing::subscriber::set_global_default(FmtSubscriber::default())?;
 
-    let (layer, io) = SocketIo::new_layer();
+    let (layer, io) = SocketIoBuilder::new()
+        .with_parser(MsgpackParser::default())
+        .build_layer();
 
     io.ns("/", on_connect);
     io.ns("/custom", on_connect);

examples/axum-echo/index.js

@@ -0,0 +1,13 @@
+import io from "socket.io-client";
+import parser from "./parser.js";
+
+const enc = new TextEncoder();
+const socket = io(`ws://127.0.0.1:3000/`, {
+    // parser: parser
+});
+
+socket.on("connect", () => {
+    socket.emit("message-with-ack", { data: "nice" }, enc.encode("Some nice broadcasting yk =)"), (...response) => {
+        console.log("\x1b[33msqw:broadcast response:", JSON.stringify(response), "\x1b[0m");
+    });
+})

socketioxide/Cargo.toml

@@ -26,6 +26,7 @@ thiserror.workspace = true
 itoa.workspace = true
 hyper.workspace = true
 pin-project-lite.workspace = true
+dyn-clone = "1.0.16"
 
 # Extensions
 dashmap = { version = "5.4.0", optional = true }

socketioxide/src/client.rs

@@ -197,7 +197,7 @@ impl<A: Adapter> EngineIoHandler for Client<A> {
     fn on_message(&self, msg: String, socket: Arc<EIoSocket<SocketData>>) {
         #[cfg(feature = "tracing")]
         tracing::debug!("Received message: {:?}", msg);
-        let packet = match Packet::try_from(msg) {
+        let packet = match self.config.parser.decode_msg(msg, socket.clone()) {
             Ok(packet) => packet,
             Err(_e) => {
                 #[cfg(feature = "tracing")]
@@ -242,42 +242,19 @@ impl<A: Adapter> EngineIoHandler for Client<A> {
     ///
     /// If the packet is complete, it is propagated to the namespace
     fn on_binary(&self, data: Vec<u8>, socket: Arc<EIoSocket<SocketData>>) {
-        if apply_payload_on_packet(data, &socket) {
-            if let Some(packet) = socket.data.partial_bin_packet.lock().unwrap().take() {
-                if let Err(ref err) = self.sock_propagate_packet(packet, socket.id) {
-                    #[cfg(feature = "tracing")]
-                    tracing::debug!(
+        if let Some(packet) = self.config.parser.decode_bin(data, socket.clone()) {
+            if let Err(ref err) = self.sock_propagate_packet(packet, socket.id) {
+                #[cfg(feature = "tracing")]
+                tracing::debug!(
                         "error while propagating packet to socket {}: {}",
                         socket.id,
                         err
                     );
-                    if let Some(reason) = err.into() {
-                        socket.close(reason);
-                    }
+                if let Some(reason) = err.into() {
+                    socket.close(reason);
                 }
             }
         }
     }
 }
 
-/// Utility that applies an incoming binary payload to a partial binary packet
-/// waiting to be filled with all the payloads
-///
-/// Returns true if the packet is complete and should be processed
-fn apply_payload_on_packet(data: Vec<u8>, socket: &EIoSocket<SocketData>) -> bool {
-    #[cfg(feature = "tracing")]
-    tracing::debug!("[sid={}] applying payload on packet", socket.id);
-    if let Some(ref mut packet) = *socket.data.partial_bin_packet.lock().unwrap() {
-        match packet.inner {
-            PacketData::BinaryEvent(_, ref mut bin, _) | PacketData::BinaryAck(ref mut bin, _) => {
-                bin.add_payload(data);
-                bin.is_complete()
-            }
-            _ => unreachable!("partial_bin_packet should only be set for binary packets"),
-        }
-    } else {
-        #[cfg(feature = "tracing")]
-        tracing::debug!("[sid={}] socket received unexpected bin data", socket.id);
-        false
-    }
-}

socketioxide/src/io.rs

@@ -18,6 +18,7 @@
     service::SocketIoService,
     BroadcastError, DisconnectError,
 };
+use crate::parser::{DefaultParser, Parser};
 
 /// Configuration for Socket.IO & Engine.IO
 #[derive(Debug, Clone)]
@@ -34,6 +35,9 @@
     ///
     /// Defaults to 45 seconds.
     pub connect_timeout: Duration,
+
+    /// A custom parser that encodes and decodes packets
+    pub parser: Box<dyn Parser>
 }
 
 impl Default for SocketIoConfig {
@@ -45,6 +49,7 @@
             },
             ack_timeout: Duration::from_secs(5),
             connect_timeout: Duration::from_secs(45),
+            parser: Box::new(DefaultParser::default())
         }
     }
 }
@@ -152,6 +157,13 @@
         self
     }
 
+    /// Sets a custom [`Parser`] for encoding and decoding packets for this [`SocketIoBuilder`].
+    /// Can be used to implement a custom protocol.
+    pub fn with_parser<P : Parser + Send + Sync + 'static>(mut self, parser: P) -> Self {
+        self.config.parser = Box::new(parser);
+        self
+    }
+
     /// Sets a custom [`Adapter`] for this [`SocketIoBuilder`]
     pub fn with_adapter<B: Adapter>(self) -> SocketIoBuilder<B> {
         SocketIoBuilder {

socketioxide/src/lib.rs

@@ -264,6 +264,7 @@
 pub mod packet;
 pub mod service;
 pub mod socket;
+pub mod parser;
 
 pub use engineioxide::TransportType;
 pub use errors::{AckError, AdapterError, BroadcastError, DisconnectError, SendError, SocketError};

socketioxide/src/operators.rs

@@ -463,7 +463,7 @@ impl<A: Adapter> Operators<A> {
     }
 }
 
-#[cfg(feature = "test-utils")]
+// #[cfg(feature = "test-utils")]
 impl<A: Adapter> Operators<A> {
     #[allow(dead_code)]
     pub(crate) fn is_broadcast(&self) -> bool {