Merge branch 'main' into ci-heaptrack-job

Author: Totodore

Cargo.toml

@@ -39,5 +39,5 @@ pin-project-lite = "0.2.13"
 tracing-subscriber = { version = "0.3.17", features = ["env-filter"] }
 criterion = { version = "0.5.1", features = ["html_reports"] }
 axum = "0.7.2"
-salvo = { version = "0.65.0", features = ["tower-compat"] }
+salvo = { version = "0.66.0", features = ["tower-compat"] }
 rust_socketio = { version = "0.4.2", features = ["async"] }

engineioxide/Cargo.toml

@@ -33,7 +33,7 @@ http-body-util.workspace = true
 pin-project-lite.workspace = true
 hyper-util = { workspace = true, features = ["tokio"] }
 
-base64 = "0.21.0"
+base64 = "0.22.0"
 bytes = "1.4.0"
 rand = "0.8.5"
 

engineioxide/src/transport/ws.rs

@@ -182,7 +182,11 @@ where
                 }
                 p => return Err(Error::BadPacket(p)),
             },
-            Message::Binary(data) => {
+            Message::Binary(mut data) => {
+                if socket.protocol == ProtocolVersion::V3 && !data.is_empty() {
+                    // The first byte is the message type, which we don't need.
+                    let _ = data.remove(0);
+                }
                 engine.handler.on_binary(data, socket.clone());
                 Ok(())
             }
@@ -212,7 +216,11 @@ where
         macro_rules! map_fn {
             ($item:ident) => {
                 let res = match $item {
-                    Packet::Binary(bin) | Packet::BinaryV3(bin) => {
+                    Packet::Binary(mut bin) | Packet::BinaryV3(mut bin) => {
+                        if socket.protocol == ProtocolVersion::V3 {
+                            // v3 protocol requires packet type as the first byte
+                            bin.insert(0, 0x04);
+                        }
                         tx.feed(Message::Binary(bin)).await
                     }
                     Packet::Close => {