wip: Have Bin implement FromMessageParts and Data implement FromMessage

e2e/socketioxide/socketioxide.rs

@@ -27,7 +27,7 @@ fn on_connect(socket: SocketRef, Data(data): Data<PayloadValue>) {
     // keep this handler async to test async message handlers
     socket.on(
         "message-with-ack",
-        |Data::<PayloadValue>(data), ack: AckSender| async move {
+        |ack: AckSender, Data::<PayloadValue>(data)| async move {
             info!("Received event: {:?}", data);
             ack.send(data).ok();
         },

examples/angular-todomvc/src/main.rs

@@ -43,7 +43,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
         s.on(
             "update-store",
-            |s: SocketRef, Data::<Vec<Todo>>(new_todos), State(Todos(todos))| {
+            |s: SocketRef, State(Todos(todos)), Data::<Vec<Todo>>(new_todos)| {
                 info!("Received update-store event: {:?}", new_todos);
 
                 let mut todos = todos.lock().unwrap();

examples/axum-echo-tls/axum_echo-tls.rs

@@ -23,7 +23,7 @@ fn on_connect(socket: SocketRef, Data(data): Data<PayloadValue>) {
 
     socket.on(
         "message-with-ack",
-        |Data::<PayloadValue>(data), ack: AckSender| {
+        |ack: AckSender, Data::<PayloadValue>(data)| {
             info!("Received event: {:?}", data);
             ack.send(data).ok();
         },

examples/axum-echo/axum_echo.rs

@@ -20,7 +20,7 @@ fn on_connect(socket: SocketRef, Data(data): Data<PayloadValue>) {
 
     socket.on(
         "message-with-ack",
-        |Data::<PayloadValue>(data), ack: AckSender| {
+        |ack: AckSender, Data::<PayloadValue>(data)| {
             info!("Received event: {:?}", data);
             ack.send(data).ok();
         },

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

@@ -41,7 +41,7 @@ impl Todos {
     }
 }
 
-pub fn create(s: SocketRef, Data(data): Data<PartialTodo>, ack: AckSender, todos: State<Todos>) {
+pub fn create(s: SocketRef, ack: AckSender, todos: State<Todos>, Data(data): Data<PartialTodo>) {
     let id = Uuid::new_v4();
     let todo = Todo { id, inner: data };
 
@@ -53,12 +53,12 @@ pub fn create(s: SocketRef, Data(data): Data<PartialTodo>, ack: AckSender, todos
     s.broadcast().emit("todo:created", todo).ok();
 }
 
-pub async fn read(Data(id): Data<Uuid>, ack: AckSender, todos: State<Todos>) {
+pub async fn read(ack: AckSender, todos: State<Todos>, Data(id): Data<Uuid>) {
     let todo = todos.get(&id).ok_or(Error::NotFound);
     ack.send(todo).ok();
 }
 
-pub async fn update(s: SocketRef, Data(data): Data<Todo>, ack: AckSender, todos: State<Todos>) {
+pub async fn update(s: SocketRef, ack: AckSender, todos: State<Todos>, Data(data): Data<Todo>) {
     let res = todos
         .get_mut(&data.id)
         .ok_or(Error::NotFound)
@@ -70,7 +70,7 @@ pub async fn update(s: SocketRef, Data(data): Data<Todo>, ack: AckSender, todos:
     ack.send(res).ok();
 }
 
-pub async fn delete(s: SocketRef, Data(id): Data<Uuid>, ack: AckSender, todos: State<Todos>) {
+pub async fn delete(s: SocketRef, ack: AckSender, todos: State<Todos>, Data(id): Data<Uuid>) {
     let res = todos.remove(&id).ok_or(Error::NotFound).map(|_| {
         s.broadcast().emit("todo:deleted", id).ok();
     });

examples/chat/src/main.rs

@@ -70,7 +70,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
         s.on(
             "add user",
-            |s: SocketRef, Data::<String>(username), user_cnt: State<UserCnt>| {
+            |s: SocketRef, user_cnt: State<UserCnt>, Data::<String>(username)| {
                 if s.extensions.get::<Username>().is_some() {
                     return;
                 }

examples/hyper-echo/hyper_echo.rs

@@ -24,7 +24,7 @@ fn on_connect(socket: SocketRef, Data(data): Data<PayloadValue>) {
 
     socket.on(
         "message-with-ack",
-        |Data::<PayloadValue>(data), ack: AckSender| {
+        |ack: AckSender, Data::<PayloadValue>(data)| {
             info!("Received event: {:?}", data);
             ack.send(data).ok();
         },

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

@@ -18,7 +18,7 @@ pub async fn on_connect(socket: SocketRef) {
 
     socket.on(
         "join",
-        |socket: SocketRef, Data::<String>(room), store: State<state::MessageStore>| async move {
+        |socket: SocketRef, store: State<state::MessageStore>, Data::<String>(room)| async move {
             tracing::info!("Received join: {:?}", room);
             let _ = socket.leave_all();
             let _ = socket.join(room.clone());
@@ -29,7 +29,7 @@ pub async fn on_connect(socket: SocketRef) {
 
     socket.on(
         "message",
-        |socket: SocketRef, Data::<MessageIn>(data), store: State<state::MessageStore>| async move {
+        |socket: SocketRef, store: State<state::MessageStore>, Data::<MessageIn>(data)| async move {
             tracing::info!("Received message: {:?}", data);
 
             let response = state::Message {

examples/private-messaging/src/handlers.rs

@@ -53,7 +53,7 @@ pub fn on_connection(
 
     s.on(
         "private message",
-        |s: SocketRef, Data(PrivateMessageReq { to, content }), State(Messages(msg))| {
+        |s: SocketRef, State(Messages(msg)), Data(PrivateMessageReq { to, content })| {
             let user_id = s.extensions.get::<Session>().unwrap().user_id;
             let message = Message {
                 from: user_id,

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

@@ -27,7 +27,7 @@ async fn on_connect(socket: SocketRef) {
 
     socket.on(
         "join",
-        |socket: SocketRef, Data::<String>(room), store: State<state::MessageStore>| async move {
+        |socket: SocketRef, store: State<state::MessageStore>, Data::<String>(room)| async move {
             info!("Received join: {:?}", room);
             let _ = socket.leave_all();
             let _ = socket.join(room.clone());
@@ -38,7 +38,7 @@ async fn on_connect(socket: SocketRef) {
 
     socket.on(
         "message",
-        |socket: SocketRef, Data::<MessageIn>(data), store: State<state::MessageStore>| async move {
+        |socket: SocketRef, store: State<state::MessageStore>, Data::<MessageIn>(data)| async move {
             info!("Received message: {:?}", data);
 
             let response = state::Message {

examples/salvo-echo/salvo_echo.rs

@@ -23,7 +23,7 @@ fn on_connect(socket: SocketRef, Data(data): Data<PayloadValue>) {
 
     socket.on(
         "message-with-ack",
-        |Data::<PayloadValue>(data), ack: AckSender| {
+        |ack: AckSender, Data::<PayloadValue>(data)| {
             info!("Received event: {:?}", data);
             ack.send(data).ok();
         },

examples/viz-echo/viz_echo.rs

@@ -20,7 +20,7 @@ fn on_connect(socket: SocketRef, Data(data): Data<PayloadValue>) {
 
     socket.on(
         "message-with-ack",
-        |Data::<PayloadValue>(data), ack: AckSender| {
+        |ack: AckSender, Data::<PayloadValue>(data)| {
             info!("Received event: {:?}", data);
             ack.send(data).ok();
         },

socketioxide/src/handler/extract.rs

@@ -2,14 +2,16 @@
 //! and [`DisconnectHandler`](super::DisconnectHandler).
 //!
 //! They can be used to extract data from the context of the handler and get specific params. Here are some examples of extractors:
-//! * [`Data`]: extracts and deserialize to json any data, if a deserialization error occurs the handler won't be called:
+//! * [`Data`]: extracts and deserialize to json any data. Because it consumes the event it should be the last argument. If a
+//! deserialization error occurs the handler won't be called:
 //!     - for [`ConnectHandler`](super::ConnectHandler): extracts and deserialize to json the auth data
 //!     - for [`MessageHandler`](super::MessageHandler): extracts and deserialize to json the message data
-//! * [`TryData`]: extracts and deserialize to json any data but with a `Result` type in case of error:
+//! * [`TryData`]: extracts and deserialize to json any data. Because it consumes the event it should be the last argument. In case of
+//! error, a `Result` type is returned;
 //!     - for [`ConnectHandler`](super::ConnectHandler): extracts and deserialize to json the auth data
 //!     - for [`MessageHandler`](super::MessageHandler): extracts and deserialize to json the message data
 //! * [`SocketRef`]: extracts a reference to the [`Socket`]
-//! * [`Bin`]: extract a binary payload for a given message. Because it consumes the event it should be the last argument
+//! * [`Bin`]: extract a binary payload for a given message.
 //! * [`AckSender`]: Can be used to send an ack response to the current message event
 //! * [`ProtocolVersion`](crate::ProtocolVersion): extracts the protocol version
 //! * [`TransportType`](crate::TransportType): extracts the transport type
@@ -129,19 +131,19 @@ where
             .map(Data)
     }
 }
-impl<T, A> FromMessageParts<A> for Data<T>
+impl<T, A> FromMessage<A> for Data<T>
 where
     T: DeserializeOwned,
     A: Adapter,
 {
     type Error = serde_json::Error;
-    fn from_message_parts(
-        _: &Arc<Socket<A>>,
-        v: &mut PayloadValue,
-        _: &Option<i64>,
+    fn from_message(
+        _: Arc<Socket<A>>,
+        mut v: PayloadValue,
+        _: Option<i64>,
     ) -> Result<Self, Self::Error> {
-        upwrap_array(v);
-        v.clone().into_data::<T>().map(Data)
+        upwrap_array(&mut v);
+        v.into_data::<T>().map(Data)
     }
 }
 
@@ -237,14 +239,18 @@ impl<A: Adapter> SocketRef<A> {
 /// An Extractor that returns the binary data of the message.
 /// If there is no binary data, it will contain an empty vec.
 pub struct Bin(pub Vec<Vec<u8>>);
-impl<A: Adapter> FromMessage<A> for Bin {
+impl<A: Adapter> FromMessageParts<A> for Bin {
     type Error = Infallible;
-    fn from_message(
-        _: Arc<Socket<A>>,
-        mut v: PayloadValue,
-        _: Option<i64>,
+    fn from_message_parts(
+        _: &Arc<Socket<A>>,
+        v: &mut PayloadValue,
+        _: &Option<i64>,
     ) -> Result<Self, Infallible> {
-        Ok(Bin(v.extract_binary_payloads()))
+        Ok(Bin(v
+            .binary_payloads_ref()
+            .into_iter()
+            .map(Clone::clone)
+            .collect()))
     }
 }
 

socketioxide/src/handler/message.rs

@@ -29,8 +29,8 @@
 //!       ack.send("ack data").ok();
 //!     });
 //!
-//!     // `Bin` extractor must be the last argument because it consumes the rest of the packet
-//!     s.on("binary_event", |s: SocketRef, TryData::<String>(data), Bin(bin)| {
+//!     // `Data` extractor must be the last argument because it consumes the rest of the packet
+//!     s.on("binary_event", |s: SocketRef, Bin(bin), TryData::<String>(data)| {
 //!       println!("Socket received event with data: {:?} and binary data: {:?}", data, bin);
 //!     })
 //! });
@@ -47,8 +47,8 @@
 //!        tokio::time::sleep(std::time::Duration::from_secs(1)).await;
 //!        println!("Socket received event with data: {}", data);
 //!     });
-//!     // `Bin` extractor must be the last argument because it consumes the rest of the packet
-//!     s.on("/binary_event", move |s: SocketRef, TryData::<String>(data), Bin(bin)| async move {
+//!     // `Data` extractor must be the last argument because it consumes the rest of the packet
+//!     s.on("/binary_event", move |s: SocketRef, Bin(bin), TryData::<String>(data)| async move {
 //!       println!("Socket received event with data: {:?} and binary data: {:?}", data, bin);
 //!     })
 //! });
@@ -60,7 +60,7 @@
 //! # use serde_json::Error;
 //! # use socketioxide::extract::*;
 //! // async named event handler
-//! async fn on_event(s: SocketRef, Data(data): Data<PayloadValue>, ack: AckSender) {
+//! async fn on_event(s: SocketRef, ack: AckSender, Data(data): Data<PayloadValue>) {
 //!     tokio::time::sleep(std::time::Duration::from_secs(1)).await;
 //!     ack.send("Here is my acknowledgment!").ok();   
 //! }

socketioxide/src/io.rs

@@ -295,7 +295,7 @@ impl<A: Adapter> SocketIo<A> {
     ///     // Register an async handler for the "test" event and extract the data as a `MyData` struct
     ///     // Extract the binary payload as a `Vec<Vec<u8>>` with the Bin extractor.
     ///     // It should be the last extractor because it consumes the request
-    ///     socket.on("test", |socket: SocketRef, Data::<MyData>(data), ack: AckSender| async move {
+    ///     socket.on("test", |socket: SocketRef, ack: AckSender, Data::<MyData>(data)| async move {
     ///         println!("Received a test message {:?}", data);
     ///         tokio::time::sleep(std::time::Duration::from_secs(1)).await;
     ///         ack.send(data).ok(); // The data received is sent back to the client through the ack

socketioxide/src/operators.rs

@@ -304,7 +304,7 @@ impl<A: Adapter> ConfOperators<'_, A> {
     /// # use socketioxide::{PayloadValue, SocketIo, extract::*};
     /// let (_, io) = SocketIo::new_svc();
     /// io.ns("/", |socket: SocketRef| {
-    ///     socket.on("test", |socket: SocketRef, Data::<PayloadValue>(data), Bin(bin)| async move {
+    ///     socket.on("test", |socket: SocketRef, Bin(bin), Data::<PayloadValue>(data)| async move {
     ///          // Emit a test message to the client
     ///         socket.emit("test", data).ok();
     ///

socketioxide/src/socket.rs

@@ -218,7 +218,7 @@ impl<A: Adapter> Socket<A> {
     ///     // Register an async handler for the "test" event and extract the data as a `MyData` struct
     ///     // Extract the binary payload as a `Vec<Vec<u8>>` with the Bin extractor.
     ///     // It should be the last extractor because it consumes the request
-    ///     socket.on("test", |socket: SocketRef, Data::<MyData>(data), ack: AckSender| async move {
+    ///     socket.on("test", |socket: SocketRef, ack: AckSender, Data::<MyData>(data)| async move {
     ///         println!("Received a test message {:?}", data);
     ///         tokio::time::sleep(std::time::Duration::from_secs(1)).await;
     ///         ack.send(data).ok(); // The data received is sent back to the client through the ack