Repository: 1c3t3a/rust-socketio (language: rust)

Issue:
Add the `BinaryEvent` and the `BinaryAck` message type
Till now the binary messages are ignored. This should be changed.


Discussion / hints:

Fix patch:
diff --git a/.gitignore b/.gitignore
index a3deb7b8..bca8eb6a 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,5 +1,6 @@
 /target
 Cargo.lock
 .vscode
+.idea
 example/node_modules
 example/package-lock.json
diff --git a/Cargo.toml b/Cargo.toml
index 5a7b0235..c141c36c 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -23,3 +23,4 @@ regex = "1.4.2"
 if_chain = "1.0.1"
 byte = "0.2.4"
 websocket = "0.26.2"
+thiserror = "1.0"
diff --git a/README.md b/README.md
index de92399c..54c3d6d4 100644
--- a/README.md
+++ b/README.md
@@ -9,34 +9,45 @@ An implementation of a socket.io client written in the Rust programming language
 ## Example usage
 
 ``` rust
-use rust_socketio::Socket;
+use rust_socketio::{SocketBuilder, Payload};
 use serde_json::json;
-
-fn main() {
-    // connect to a server on localhost with the /admin namespace and
-    // a foo event handler
-    let mut socket = SocketBuilder::new("http://localhost:80")
-          .set_namespace("/admin")
-          .expect("illegal namespace")
-          .on("test", |str| println!("Received: {}", str))
-          .connect()
-          .expect("Connection failed");
-
-    // emit to the "foo" event
-    let payload = json!({"token": 123});
-    socket.emit("foo", &payload.to_string()).expect("Server unreachable");
-
-    // define a callback, that's executed when the ack got acked
-    let ack_callback = |message: String| {
-            println!("Yehaa! My ack got acked?");
-            println!("Ack data: {}", message);
-    };
-    
-    // emit with an ack
-    let ack = socket
-            .emit_with_ack("test", &payload.to_string(), Duration::from_secs(2), ack_callback)
-            .expect("Server unreachable");
- }
+use std::time::Duration;
+
+// define a callback which is called when a payload is received
+let callback = |payload: Payload| {
+       match payload {
+           Payload::String(str) => println!("Received: {}", str),
+           Payload::Binary(bin_data) => println!("Received bytes: {:#?}", bin_data),
+       }
+};
+
+// get a socket that is connected to the admin namespace
+let mut socket = SocketBuilder::new("http://localhost:4200")
+     .set_namespace("/admin")
+     .expect("illegal namespace")
+     .on("test", callback)
+     .on("error", |err| eprintln!("Error: {:#?}", err))
+     .connect()
+     .expect("Connection failed");
+
+// emit to the "foo" event
+let json_payload = json!({"token": 123});
+let payload = Payload::String(json_payload.to_string());
+socket.emit("foo", payload).expect("Server unreachable");
+
+// define a callback, that's executed when the ack got acked
+let ack_callback = |message: Payload| {
+    println!("Yehaa! My ack got acked?");
+    println!("Ack data: {:#?}", message);
+};
+
+let json_payload = json!({"myAckData": 123});
+let payload = Payload::String(json_payload.to_string());
+
+// emit with an ack
+let ack = socket
+    .emit_with_ack("test", payload, Duration::from_secs(2), ack_callback)
+    .expect("Server unreachable");
 ```
 
 The main entry point for using this crate is the `SocketBuilder` which provides a way to easily configure a socket in the needed way. When the `connect` method is called on the builder, it returns a connected client which then could be used to emit messages to certain events. One client can only be connected to one namespace. If you need to listen to the messages in different namespaces you need to allocate multiple sockets.
@@ -47,19 +58,23 @@ Documentation of this crate can be found up on [docs.rs](https://docs.rs/rust_so
 
 ## Current features
 
-This implementation support most of the features of the socket.io protocol. In general the full engine-io protocol is implemented, and concerning the socket.io part only binary events and binary acks are not yet implemented. This implementation generally tries to make use of websockets as often as possible. This means most times only the opening request uses http and as soon as the server mentions that he is able to use websockets, an upgrade is performed. But if this upgrade is not successful or the server does not mention an upgrade possibilty, http-long polling is used (as specified in the protocol specs).
-
+This implementation now supports all of the features of the socket.io protocol mentioned [here](https://github.com/socketio/socket.io-protocol).
+It generally tries to make use of websockets as often as possible. This means most times
+only the opening request uses http and as soon as the server mentions that he is able to use
+websockets, an upgrade  is performed. But if this upgrade is not successful or the server
+does not mention an upgrade possibility, http-long polling is used (as specified in the protocol specs).
 Here's an overview of possible use-cases:
-
-* connecting to a server.
-* register callbacks for the following event types:
+- connecting to a server.
+- register callbacks for the following event types:
     - open
     - close
     - error
     - message
     - custom events like "foo", "on_payment", etc.
-* send json-data to the server (recommended to use serde_json as it provides safe handling of json data).
-* send json-data to the server and receive an ack with a possible message.
+- send JSON data to the server (via `serde_json` which provides safe
+handling).
+- send JSON data to the server and receive an `ack`.
+- send and handle Binary data.
 
 ## Content of this repository
 
diff --git a/example/socket-io.js b/example/socket-io.js
index a2136c3f..fc36be5c 100644
--- a/example/socket-io.js
+++ b/example/socket-io.js
@@ -11,8 +11,24 @@ io.on('connection', client => {
         if (ack) {
             ack('woot');
         }
-      });
+    });
+
+    client.on('binary', data => {
+        var bufView = new Uint8Array(data);
+        console.log("Yehaa binary payload!");
+        for (elem in bufView) {
+            console.log(elem);
+        }
+    });
+
+    client.on('binary', function (arg, ack) {
+        console.log('Ack received, answer with binary')
+        if (ack) {
+            ack(Buffer.from([1, 2, 3]));
+        }
+    });
     client.emit("test", "Hello Wld");
+    client.emit("test", Buffer.from([1, 2, 3]));
 });
 // the socket.io client runs on port 4201
 server.listen(4200);
diff --git a/src/engineio/socket.rs b/src/engineio/socket.rs
index a85de98d..7626c0a4 100644
--- a/src/engineio/socket.rs
+++ b/src/engineio/socket.rs
@@ -1,7 +1,7 @@
 #![allow(unused)]
 use std::thread;
 
-use super::packet::Packet;
+use super::packet::{Packet, PacketId};
 use crate::engineio::transport::TransportClient;
 use crate::error::{Error, Result};
 use std::sync::{
@@ -56,7 +56,17 @@ impl EngineSocket {
         if !self.serving.load(Ordering::Relaxed) {
             return Err(Error::ActionBeforeOpen);
         }
-        self.transport_client.read()?.emit(packet)
+        self.transport_client.read()?.emit(packet, false)
+    }
+
+    /// Sends a socketio binary attachement to the server.
+    pub fn emit_binary_attachement(&mut self, attachement: Vec<u8>) -> Result<()> {
+        if !self.serving.load(Ordering::Relaxed) {
+            return Err(Error::ActionBeforeOpen);
+        }
+        let packet = Packet::new(PacketId::Message, attachement);
+
+        self.transport_client.read()?.emit(packet, true)
     }
 
     /// Registers the `on_open` callback.
diff --git a/src/engineio/transport.rs b/src/engineio/transport.rs
index e750735e..504acd8c 100644
--- a/src/engineio/transport.rs
+++ b/src/engineio/transport.rs
@@ -183,7 +183,7 @@ impl TransportClient {
                         *self.last_ping.lock()? = Instant::now();
 
                         // emit a pong packet to keep trigger the ping cycle on the server
-                        self.emit(Packet::new(PacketId::Pong, Vec::new()))?;
+                        self.emit(Packet::new(PacketId::Pong, Vec::new()), false)?;
 
                         return Ok(());
                     }
@@ -274,8 +274,9 @@ impl TransportClient {
         Err(Error::HandshakeError("Error - invalid url".to_owned()))
     }
 
-    /// Sends a packet to the server
-    pub fn emit(&self, packet: Packet) -> Result<()> {
+    /// Sends a packet to the server. This optionally handles sending of a
+    /// socketio binary attachement via the boolean attribute `is_binary_att`.
+    pub fn emit(&self, packet: Packet, is_binary_att: bool) -> Result<()> {
         if !self.connected.load(Ordering::Relaxed) {
             let error = Error::ActionBeforeOpen;
             self.call_error_callback(format!("{}", error))?;
@@ -289,12 +290,32 @@ impl TransportClient {
         drop(host);
 
         // send a post request with the encoded payload as body
-        let data = encode_payload(vec![packet]);
+        // if this is a binary attachement, then send the raw bytes
+        let data = if is_binary_att {
+            packet.data
+        } else {
+            encode_payload(vec![packet])
+        };
 
         match &self.transport.as_ref() {
             TransportType::Polling(client) => {
+                let data_to_send = if is_binary_att {
+                    // the binary attachement gets `base64` encoded
+                    let mut prefix = vec![b'b'];
+                    prefix.extend(base64::encode(data).as_bytes());
+                    prefix
+                } else {
+                    data
+                };
+
                 let client = client.lock()?;
-                let status = client.post(address).body(data).send()?.status().as_u16();
+                let status = client
+                    .post(address)
+                    .body(data_to_send)
+                    .send()?
+                    .status()
+                    .as_u16();
+
                 drop(client);
 
                 if status != 200 {
@@ -308,7 +329,12 @@ impl TransportClient {
             TransportType::Websocket(_, writer) => {
                 let mut writer = writer.lock()?;
 
-                let dataframe = WsDataFrame::new(true, websocket::dataframe::Opcode::Text, data);
+                let opcode = if is_binary_att {
+                    websocket::dataframe::Opcode::Binary
+                } else {
+                    websocket::dataframe::Opcode::Text
+                };
+                let dataframe = WsDataFrame::new(true, opcode, data);
                 writer.send_dataframe(&dataframe).unwrap();
 
                 Ok(())
@@ -345,7 +371,7 @@ impl TransportClient {
 
         while self.connected.load(Ordering::Relaxed) {
             let data = match &self.transport.as_ref() {
-                // we wont't use the shared client as this blocks the ressource
+                // we wont't use the shared client as this blocks the resource
                 // in the long polling requests
                 TransportType::Polling(_) => {
                     let query_path = self.get_query_path()?;
@@ -361,7 +387,16 @@ impl TransportClient {
                 TransportType::Websocket(receiver, _) => {
                     let mut receiver = receiver.lock()?;
 
-                    receiver.recv_message().unwrap().take_payload()
+                    // if this is a binary payload, we mark it as a message
+                    let received_df = receiver.recv_dataframe()?;
+                    match received_df.opcode {
+                        websocket::dataframe::Opcode::Binary => {
+                            let mut message = vec![b'4'];
+                            message.extend(received_df.take_payload());
+                            message
+                        }
+                        _ => received_df.take_payload(),
+                    }
                 }
             };
 
@@ -379,7 +414,8 @@ impl TransportClient {
                         spawn_scoped!(function(packet.clone()));
                     }
                 }
-                // check for the appropiate action or callback
+
+                // check for the appropriate action or callback
                 match packet.packet_id {
                     PacketId::Message => {
                         let on_data = self.on_data.read()?;
@@ -407,7 +443,7 @@ impl TransportClient {
                     }
                     PacketId::Ping => {
                         last_ping = Instant::now();
-                        self.emit(Packet::new(PacketId::Pong, Vec::new()))?;
+                        self.emit(Packet::new(PacketId::Pong, Vec::new()), false)?;
                     }
                     PacketId::Pong => {
                         // this will never happen as the pong packet is
@@ -548,10 +584,10 @@ mod test {
         assert!(socket.open(SERVER_URL).is_ok());
 
         assert!(socket
-            .emit(Packet::new(
-                PacketId::Message,
-                "HelloWorld".to_owned().into_bytes(),
-            ))
+            .emit(
+                Packet::new(PacketId::Message, "HelloWorld".to_owned().into_bytes(),),
+                false,
+            )
             .is_ok());
 
         socket.on_data = Arc::new(RwLock::new(Some(Box::new(|data| {
@@ -563,10 +599,10 @@ mod test {
 
         // closes the connection
         assert!(socket
-            .emit(Packet::new(
-                PacketId::Message,
-                "CLOSE".to_owned().into_bytes(),
-            ))
+            .emit(
+                Packet::new(PacketId::Message, "CLOSE".to_owned().into_bytes(),),
+                false,
+            )
             .is_ok());
 
         // assert!(socket.poll_cycle().is_ok());
diff --git a/src/error.rs b/src/error.rs
index cd2e9fe1..a1dacc78 100644
--- a/src/error.rs
+++ b/src/error.rs
@@ -1,125 +1,63 @@
 use base64::DecodeError;
-use std::str;
-use std::{
-    fmt::{self, Display, Formatter},
-    num::ParseIntError,
-};
+use std::{num::ParseIntError, str};
+use thiserror::Error;
 use websocket::{client::ParseError, WebSocketError};
 
 /// Enumeration of all possible errors in the `socket.io` context.
-#[derive(Debug)]
+#[derive(Error, Debug)]
+#[non_exhaustive]
 pub enum Error {
+    #[error("Invalid packet id: {0}")]
     InvalidPacketId(u8),
+    #[error("Error while parsing an empty packet")]
     EmptyPacket,
+    #[error("Error while parsing an incomplete packet")]
     IncompletePacket,
+    #[error("Got an invalid packet which did not follow the protocol format")]
     InvalidPacket,
-    Utf8Error(str::Utf8Error),
-    Base64Error(DecodeError),
+    #[error("An error occured while decoding the utf-8 text: {0}")]
+    Utf8Error(#[from] str::Utf8Error),
+    #[error("An error occured while encoding/decoding base64: {0}")]
+    Base64Error(#[from] DecodeError),
+    #[error("Invalid Url: {0}")]
     InvalidUrl(String),
-    ReqwestError(reqwest::Error),
+    #[error("Error during connection via http: {0}")]
+    ReqwestError(#[from] reqwest::Error),
+    #[error("Network request returned with status code: {0}")]
     HttpError(u16),
+    #[error("Got illegal handshake response: {0}")]
     HandshakeError(String),
+    #[error("Called an action before the connection was established")]
     ActionBeforeOpen,
+    #[error("string is not json serializable: {0}")]
     InvalidJson(String),
+    #[error("Did not receive an ack for id: {0}")]
     DidNotReceiveProperAck(i32),
+    #[error("An illegal action (such as setting a callback after being connected) was triggered")]
     IllegalActionAfterOpen,
+    #[error("Specified namespace {0} is not valid")]
     IllegalNamespace(String),
+    #[error("A lock was poisoned")]
     PoisonedLockError,
-    FromWebsocketError(WebSocketError),
-    FromWebsocketParseError(ParseError),
-    FromIoError(std::io::Error),
+    #[error("Got a websocket error: {0}")]
+    FromWebsocketError(#[from] WebSocketError),
+    #[error("Error while parsing the url for the websocket connection: {0}")]
+    FromWebsocketParseError(#[from] ParseError),
+    #[error("Got an IO-Error: {0}")]
+    FromIoError(#[from] std::io::Error),
 }
 
-impl std::error::Error for Error {}
-
 pub(crate) type Result<T> = std::result::Result<T, Error>;
 
-impl From<DecodeError> for Error {
-    fn from(error: DecodeError) -> Self {
-        Self::Base64Error(error)
-    }
-}
-
 impl<T> From<std::sync::PoisonError<T>> for Error {
     fn from(_: std::sync::PoisonError<T>) -> Self {
         Self::PoisonedLockError
     }
 }
 
-impl From<str::Utf8Error> for Error {
-    fn from(error: str::Utf8Error) -> Self {
-        Self::Utf8Error(error)
-    }
-}
-
-impl From<reqwest::Error> for Error {
-    fn from(error: reqwest::Error) -> Self {
-        Self::ReqwestError(error)
-    }
-}
-
 impl From<ParseIntError> for Error {
     fn from(_: ParseIntError) -> Self {
         // this is used for parsing integers from the a packet string
         Self::InvalidPacket
     }
 }
-
-impl From<WebSocketError> for Error {
-    fn from(error: WebSocketError) -> Self {
-        Self::FromWebsocketError(error)
-    }
-}
-
-impl From<ParseError> for Error {
-    fn from(error: ParseError) -> Self {
-        Self::FromWebsocketParseError(error)
-    }
-}
-
-impl From<std::io::Error> for Error {
-    fn from(error: std::io::Error) -> Self {
-        Self::FromIoError(error)
-    }
-}
-
-impl Display for Error {
-    fn fmt(&self, f: &mut Formatter) -> fmt::Result {
-        match &self {
-            Self::InvalidPacketId(id) => write!(f, "Invalid packet id: {}", id),
-            Self::EmptyPacket => write!(f, "Error while parsing an empty packet"),
-            Self::IncompletePacket => write!(f, "Error while parsing an incomplete packet"),
-            Self::Utf8Error(e) => {
-                write!(f, "An error occured while decoding the utf-8 text: {}", e)
-            }
-            Self::Base64Error(e) => {
-                write!(f, "An error occured while encoding/decoding base64: {}", e)
-            }
-            Self
-            ::InvalidUrl(url) => write!(f, "Unable to connect to: {}", url),
-            Self::ReqwestError(error) => {
-                write!(f, "Error during connection via Reqwest: {}", error)
-            }
-            Self::HandshakeError(response) => {
-                write!(f, "Got illegal handshake response: {}", response)
-            }
-            Self::ActionBeforeOpen => {
-                write!(f, "Called an action before the connection was established")
-            }
-            Self::IllegalNamespace(nsp) => write!(f, "Specified namespace {} is not valid", nsp),
-            Self::HttpError(status_code) => write!(
-                f,
-                "Network request returned with status code: {}",
-                status_code
-            ),
-            Self::InvalidJson(string) => write!(f, "string is not json serializable: {}", string),
-            Self::DidNotReceiveProperAck(id) => write!(f, "Did not receive an ack for id: {}", id),
-            Self::IllegalActionAfterOpen => write!(f, "An illegal action (such as setting a callback after being connected) was triggered"),
-            Self::PoisonedLockError => write!(f, "A lock was poisoned"),
-            Self::InvalidPacket => write!(f, "Got an invalid packet which did not follow the protocol format"),
-            Self::FromWebsocketError(error) => write!(f, "Got a websocket error: {}", error),
-            Self::FromWebsocketParseError(error) => write!(f, "Error while parsing the url for the websocket connection: {}", error),
-            Self::FromIoError(error) => write!(f, "Got an IoError: {}", error),
-        }
-    }
-}
diff --git a/src/lib.rs b/src/lib.rs
index 14c0929f..19285979 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -1,37 +1,48 @@
-//! Rust socket.io is a socket.io client written in the Rust Programming Language.
+//! Rust-socket.io is a socket.io client written in the Rust Programming Language.
 //! ## Example usage
 //!
 //! ``` rust
-//! use rust_socketio::SocketBuilder;
+//! use rust_socketio::{SocketBuilder, Payload};
 //! use serde_json::json;
 //! use std::time::Duration;
 //!
+//! // define a callback which is called when a payload is received
+//! let callback = |payload: Payload| {
+//!        match payload {
+//!            Payload::String(str) => println!("Received: {}", str),
+//!            Payload::Binary(bin_data) => println!("Received bytes: {:#?}", bin_data),
+//!        }
+//! };
+//!
 //! // get a socket that is connected to the admin namespace
 //! let mut socket = SocketBuilder::new("http://localhost:4200")
 //!      .set_namespace("/admin")
 //!      .expect("illegal namespace")
-//!      .on("test", |str| println!("Received: {}", str))
-//!      .on("error", |err| eprintln!("Error: {}", err))
+//!      .on("test", callback)
+//!      .on("error", |err| eprintln!("Error: {:#?}", err))
 //!      .connect()
 //!      .expect("Connection failed");
 //!
 //! // emit to the "foo" event
-//! let payload = json!({"token": 123});
-//! socket.emit("foo", &payload.to_string()).expect("Server unreachable");
+//! let json_payload = json!({"token": 123});
+//!
+//! socket.emit("foo", json_payload).expect("Server unreachable");
 //!
 //! // define a callback, that's executed when the ack got acked
-//! let ack_callback = |message: String| {
+//! let ack_callback = |message: Payload| {
 //!     println!("Yehaa! My ack got acked?");
-//!     println!("Ack data: {}", message);
+//!     println!("Ack data: {:#?}", message);
 //! };
 //!
+//! let json_payload = json!({"myAckData": 123});
+//!
 //! // emit with an ack
 //! let ack = socket
-//!     .emit_with_ack("test", &payload.to_string(), Duration::from_secs(2), ack_callback)
+//!     .emit_with_ack("test", json_payload, Duration::from_secs(2), ack_callback)
 //!     .expect("Server unreachable");
 //! ```
 //!
-//! The main entry point for using this crate is the `SocketBuilder` which provides
+//! The main entry point for using this crate is the [`SocketBuilder`] which provides
 //! a way to easily configure a socket in the needed way. When the `connect` method
 //! is called on the builder, it returns a connected client which then could be used
 //! to emit messages to certain events. One client can only be connected to one namespace.
@@ -40,13 +51,12 @@
 //!
 //! ## Current features
 //!
-//! This implementation support most of the features of the socket.io protocol. In general
-//! the full engine-io protocol is implemented, and concerning the socket.io part only binary
-//! events and binary acks are not yet implemented. This implementation generally tries to
-//! make use of websockets as often as possible. This means most times only the opening request
-//! uses http and as soon as the server mentions that he is able to use websockets, an upgrade
-//! is performed. But if this upgrade is not successful or the server does not mention an upgrade
-//! possibilty, http-long polling is used (as specified in the protocol specs).
+//! This implementation now supports all of the features of the socket.io protocol mentioned
+//! [here](https://github.com/socketio/socket.io-protocol).
+//! It generally tries to make use of websockets as often as possible. This means most times
+//! only the opening request uses http and as soon as the server mentions that he is able to use
+//! websockets, an upgrade  is performed. But if this upgrade is not successful or the server
+//! does not mention an upgrade possibility, http-long polling is used (as specified in the protocol specs).
 //!
 //! Here's an overview of possible use-cases:
 //!
@@ -60,6 +70,7 @@
 //! - send JSON data to the server (via `serde_json` which provides safe
 //! handling).
 //! - send JSON data to the server and receive an `ack`.
+//! - send and handle Binary data.
 //!
 #![allow(clippy::rc_buffer)]
 #![warn(clippy::complexity)]
@@ -87,6 +98,7 @@ pub mod socketio;
 pub mod error;
 
 use error::Error;
+pub use socketio::{event::Event, payload::Payload};
 
 use crate::error::Result;
 use std::{sync::Arc, time::Duration};
@@ -117,19 +129,28 @@ impl SocketBuilder {
     /// will be used.
     /// # Example
     /// ```rust
-    /// use rust_socketio::SocketBuilder;
+    /// use rust_socketio::{SocketBuilder, Payload};
     /// use serde_json::json;
     ///
+    ///
+    /// let callback = |payload: Payload| {
+    ///            match payload {
+    ///                Payload::String(str) => println!("Received: {}", str),
+    ///                Payload::Binary(bin_data) => println!("Received bytes: {:#?}", bin_data),
+    ///            }
+    /// };
+    ///
     /// let mut socket = SocketBuilder::new("http://localhost:4200")
     ///     .set_namespace("/admin")
     ///     .expect("illegal namespace")
-    ///     .on("test", |str| println!("Received: {}", str))
+    ///     .on("test", callback)
     ///     .connect()
     ///     .expect("error while connecting");
     ///
     /// // use the socket
-    /// let payload = json!({"token": 123});
-    /// let result = socket.emit("foo", &payload.to_string());
+    /// let json_payload = json!({"token": 123});
+    ///
+    /// let result = socket.emit("foo", json_payload);
     ///
     /// assert!(result.is_ok());
     /// ```
@@ -155,20 +176,26 @@ impl SocketBuilder {
     /// event defined by a string, e.g. `onPayment` or `foo`.
     /// # Example
     /// ```rust
-    /// use rust_socketio::SocketBuilder;
+    /// use rust_socketio::{SocketBuilder, Payload};
+    ///
+    /// let callback = |payload: Payload| {
+    ///            match payload {
+    ///                Payload::String(str) => println!("Received: {}", str),
+    ///                Payload::Binary(bin_data) => println!("Received bytes: {:#?}", bin_data),
+    ///            }
+    /// };
     ///
     /// let socket = SocketBuilder::new("http://localhost:4200")
     ///     .set_namespace("/admin")
     ///     .expect("illegal namespace")
-    ///     .on("test", |str| println!("Received: {}", str))
-    ///     .on("error", |err| eprintln!("Error: {}", err))
+    ///     .on("test", callback)
+    ///     .on("error", |err| eprintln!("Error: {:#?}", err))
     ///     .connect();
     ///
-    ///
     /// ```
     pub fn on<F>(mut self, event: &str, callback: F) -> Self
     where
-        F: FnMut(String) + 'static + Sync + Send,
+        F: FnMut(Payload) + 'static + Sync + Send,
     {
         // unwrapping here is safe as this only returns an error
         // when the client is already connected, which is
@@ -182,19 +209,21 @@ impl SocketBuilder {
     /// value if something goes wrong during connection.
     /// # Example
     /// ```rust
-    /// use rust_socketio::SocketBuilder;
+    /// use rust_socketio::{SocketBuilder, Payload};
     /// use serde_json::json;
     ///
+    ///
     /// let mut socket = SocketBuilder::new("http://localhost:4200")
     ///     .set_namespace("/admin")
     ///     .expect("illegal namespace")
-    ///     .on("test", |str| println!("Received: {}", str))
+    ///     .on("error", |err| println!("Socket error!: {:#?}", err))
     ///     .connect()
     ///     .expect("connection failed");
     ///
     /// // use the socket
-    /// let payload = json!({"token": 123});
-    /// let result = socket.emit("foo", &payload.to_string());
+    /// let json_payload = json!({"token": 123});
+    ///
+    /// let result = socket.emit("foo", json_payload);
     ///
     /// assert!(result.is_ok());
     /// ```
@@ -220,7 +249,7 @@ impl Socket {
     /// after a call to the `connect` method.
     pub(crate) fn on<F>(&mut self, event: &str, callback: F) -> Result<()>
     where
-        F: FnMut(String) + 'static + Sync + Send,
+        F: FnMut(Payload) + 'static + Sync + Send,
     {
         self.transport.on(event.into(), callback)
     }
@@ -240,71 +269,81 @@ impl Socket {
     ///
     /// # Example
     /// ```
-    /// use rust_socketio::SocketBuilder;
+    /// use rust_socketio::{SocketBuilder, Payload};
     /// use serde_json::json;
     ///
     /// let mut socket = SocketBuilder::new("http://localhost:4200")
-    ///     .on("test", |str| println!("Received: {}", str))
+    ///     .on("test", |payload: Payload| println!("Received: {:#?}", payload))
     ///     .connect()
     ///     .expect("connection failed");
     ///
-    /// let payload = json!({"token": 123});
-    /// let result = socket.emit("foo", &payload.to_string());
+    /// let json_payload = json!({"token": 123});
+    ///
+    /// let result = socket.emit("foo", json_payload);
     ///
     /// assert!(result.is_ok());
     /// ```
     #[inline]
-    pub fn emit(&mut self, event: &str, data: &str) -> Result<()> {
-        self.transport.emit(event.into(), data)
+    pub fn emit<E, D>(&mut self, event: E, data: D) -> Result<()>
+    where
+        E: Into<Event>,
+        D: Into<Payload>,
+    {
+        self.transport.emit(event.into(), data.into())
     }
 
     /// Sends a message to the server but `alloc`s an `ack` to check whether the
     /// server responded in a given timespan. This message takes an event, which
     /// could either be one of the common events like "message" or "error" or a
     /// custom event like "foo", as well as a data parameter. But be careful,
-    /// the string needs to be valid JSON. It's even recommended to use a
-    /// library like serde_json to serialize the data properly. It also requires
-    /// a timeout `Duration` in which the client needs to answer. This method
-    /// returns an `Arc<RwLock<Ack>>`. The `Ack` type holds information about
-    /// the `ack` system call, such whether the `ack` got acked fast enough and
-    /// potential data. It is safe to unwrap the data after the `ack` got acked
-    /// from the server. This uses an `RwLock` to reach shared mutability, which
-    /// is needed as the server sets the data on the ack later.
+    /// in case you send a [`Payload::String`], the string needs to be valid JSON.
+    /// It's even recommended to use a library like serde_json to serialize the data properly.
+    /// It also requires a timeout `Duration` in which the client needs to answer.
+    /// If the ack is accked in the correct timespan, the specified callback is
+    /// called. The callback consumes a [`Payload`] which represents the data send
+    /// by the server.
     ///
     /// # Example
     /// ```
-    /// use rust_socketio::SocketBuilder;
+    /// use rust_socketio::{SocketBuilder, Payload};
     /// use serde_json::json;
     /// use std::time::Duration;
     /// use std::thread::sleep;
     ///
     /// let mut socket = SocketBuilder::new("http://localhost:4200")
-    ///     .on("foo", |str| println!("Received: {}", str))
+    ///     .on("foo", |payload: Payload| println!("Received: {:#?}", payload))
     ///     .connect()
     ///     .expect("connection failed");
     ///
     ///
-    /// let payload = json!({"token": 123});
-    /// let ack_callback = |message| { println!("{}", message) };
     ///
-    /// socket.emit_with_ack("foo", &payload.to_string(),
-    /// Duration::from_secs(2), ack_callback).unwrap();
+    /// let ack_callback = |message: Payload| {
+    ///     match message {
+    ///         Payload::String(str) => println!("{}", str),
+    ///         Payload::Binary(bytes) => println!("Received bytes: {:#?}", bytes),
+    ///    }    
+    /// };
+    ///
+    /// let payload = json!({"token": 123});
+    /// socket.emit_with_ack("foo", payload, Duration::from_secs(2), ack_callback).unwrap();
     ///
     /// sleep(Duration::from_secs(2));
     /// ```
     #[inline]
-    pub fn emit_with_ack<F>(
+    pub fn emit_with_ack<F, E, D>(
         &mut self,
-        event: &str,
-        data: &str,
+        event: E,
+        data: D,
         timeout: Duration,
         callback: F,
     ) -> Result<()>
     where
-        F: FnMut(String) + 'static + Send + Sync,
+        F: FnMut(Payload) + 'static + Send + Sync,
+        E: Into<Event>,
+        D: Into<Payload>,
     {
         self.transport
-            .emit_with_ack(event.into(), data, timeout, callback)
+            .emit_with_ack(event.into(), data.into(), timeout, callback)
     }
 
     /// Sets the namespace attribute on a client (used by the builder class)
@@ -326,29 +365,44 @@ mod test {
     fn it_works() {
         let mut socket = Socket::new(SERVER_URL, None);
 
-        let result = socket.on("test", |msg| println!("{}", msg));
+        let result = socket.on("test", |msg| match msg {
+            Payload::String(str) => println!("Received string: {}", str),
+            Payload::Binary(bin) => println!("Received binary data: {:#?}", bin),
+        });
         assert!(result.is_ok());
 
         let result = socket.connect();
         assert!(result.is_ok());
 
         let payload = json!({"token": 123});
-        let result = socket.emit("test", &payload.to_string());
+        let result = socket.emit("test", Payload::String(payload.to_string()));
 
         assert!(result.is_ok());
 
         let mut socket_clone = socket.clone();
-        let ack_callback = move |message: String| {
-            let result = socket_clone.emit("test", &json!({"got ack": true}).to_string());
+        let ack_callback = move |message: Payload| {
+            let result = socket_clone.emit(
+                "test",
+                Payload::String(json!({"got ack": true}).to_string()),
+            );
             assert!(result.is_ok());
 
             println!("Yehaa! My ack got acked?");
-            println!("Ack data: {}", message);
+            match message {
+                Payload::Binary(bin) => {
+                    println!("Received binary Ack");
+                    println!("Data: {:#?}", bin);
+                }
+                Payload::String(str) => {
+                    println!("Received string Ack");
+                    println!("Ack data: {}", str);
+                }
+            }
         };
 
         let ack = socket.emit_with_ack(
             "test",
-            &payload.to_string(),
+            Payload::String(payload.to_string()),
             Duration::from_secs(2),
             ack_callback,
         );
@@ -359,17 +413,30 @@ mod test {
 
     #[test]
     fn test_builder() {
-        let socket_builder = SocketBuilder::new(SERVER_URL).set_namespace("/admin");
-
-        assert!(socket_builder.is_ok());
+        let socket_builder = SocketBuilder::new(SERVER_URL);
 
         let socket = socket_builder
-            .unwrap()
-            .on("error", |err| eprintln!("Error!!: {}", err))
-            .on("test", |str| println!("Received: {}", str))
-            .on("message", |msg| println!("Received: {}", msg))
+            .on("error", |err| eprintln!("Error!!: {:#?}", err))
+            .on("test", |str| println!("Received: {:#?}", str))
+            .on("message", |msg| println!("Received: {:#?}", msg))
             .connect();
 
         assert!(socket.is_ok());
+
+        let mut socket = socket.unwrap();
+        assert!(socket.emit("message", json!("Hello World")).is_ok());
+
+        assert!(socket.emit("binary", vec![46, 88]).is_ok());
+
+        let ack_cb = |payload| {
+            println!("Yehaa the ack got acked");
+            println!("With data: {:#?}", payload);
+        };
+
+        assert!(socket
+            .emit_with_ack("binary", json!("pls ack"), Duration::from_secs(1), ack_cb,)
+            .is_ok());
+
+        sleep(Duration::from_secs(2));
     }
 }
diff --git a/src/socketio/mod.rs b/src/socketio/mod.rs
index d59169c5..674c3594 100644
--- a/src/socketio/mod.rs
+++ b/src/socketio/mod.rs
@@ -1,4 +1,7 @@
 /// Defines the events that could be sent or received.
 pub mod event;
 mod packet;
+/// Defines the types of payload (binary or string), that
+/// could be sent or received.
+pub mod payload;
 pub(crate) mod transport;
diff --git a/src/socketio/packet.rs b/src/socketio/packet.rs
index 5a4ea30d..abbe7b80 100644
--- a/src/socketio/packet.rs
+++ b/src/socketio/packet.rs
@@ -17,9 +17,9 @@ pub enum PacketId {
 
 /// A packet which gets sent or received during in the `socket.io` protocol.
 #[derive(Debug, PartialEq, Eq, Clone)]
-pub struct Packet<'a> {
+pub struct Packet {
     pub packet_type: PacketId,
-    pub nsp: &'a str,
+    pub nsp: String,
     pub data: Option<String>,
     pub binary_data: Option<Vec<u8>>,
     pub id: Option<i32>,
@@ -41,11 +41,11 @@ pub const fn u8_to_packet_id(b: u8) -> Result<PacketId> {
     }
 }
 
-impl<'a> Packet<'a> {
+impl Packet {
     /// Creates an instance.
     pub const fn new(
         packet_type: PacketId,
-        nsp: &'a str,
+        nsp: String,
         data: Option<String>,
         binary_data: Option<Vec<u8>>,
         id: Option<i32>,
@@ -62,6 +62,8 @@ impl<'a> Packet<'a> {
     }
 
     /// Method for encoding from a `Packet` to a `u8` byte stream.
+    /// The binary payload of a packet is not put at the end of the
+    /// stream as it gets handled and send by it's own logic via the socket.
     pub fn encode(&self) -> Bytes {
         // first the packet type
         let mut string = (self.packet_type as u8).to_string();
@@ -88,7 +90,7 @@ impl<'a> Packet<'a> {
         }
 
         let mut buffer = string.into_bytes();
-        if let Some(bin_data) = self.binary_data.as_ref() {
+        if self.binary_data.as_ref().is_some() {
             // check if an event type is present
             let placeholder = if let Some(event_type) = self.data.as_ref() {
                 format!(
@@ -105,7 +107,6 @@ impl<'a> Packet<'a> {
 
             // build the buffers
             buffer.extend(placeholder.into_bytes());
-            buffer.extend(bin_data);
         } else if let Some(data) = self.data.as_ref() {
             buffer.extend(data.to_string().into_bytes());
         }
@@ -114,7 +115,13 @@ impl<'a> Packet<'a> {
     }
 
     /// Decodes a packet given a `Bytes` type.
-    pub fn decode_bytes(payload: &'a Bytes) -> Result<Self> {
+    /// The binary payload of a packet is not put at the end of the
+    /// stream as it gets handled and send by it's own logic via the socket.
+    /// Therefore this method does not return the correct value for the
+    /// binary data, instead the socket is responsible for handling
+    /// this member. This is done because the attachement is usually
+    /// send in another packet.
+    pub fn decode_bytes<'a>(payload: &'a Bytes) -> Result<Self> {
         let mut i = 0;
         let packet_id = u8_to_packet_id(*payload.first().ok_or(Error::EmptyPacket)?)?;
 
@@ -173,7 +180,6 @@ impl<'a> Packet<'a> {
             None
         };
 
-        let mut binary_data = None;
         let data = if payload.get(i + 1).is_some() {
             let start = if id.is_some() { i } else { i + 1 };
 
@@ -209,15 +215,6 @@ impl<'a> Packet<'a> {
 
             match packet_id {
                 PacketId::BinaryAck | PacketId::BinaryEvent => {
-                    binary_data = Some(
-                        payload
-                            .iter()
-                            .skip(end)
-                            .take(payload.len() - end)
-                            .copied()
-                            .collect(),
-                    );
-
                     let re_close = Regex::new(r",]$|]$").unwrap();
                     let mut str = json_data
                         .to_string()
@@ -242,9 +239,9 @@ impl<'a> Packet<'a> {
 
         Ok(Packet::new(
             packet_id,
-            nsp,
+            nsp.to_owned(),
             data,
-            binary_data,
+            None,
             id,
             attachements,
         ))
@@ -266,7 +263,7 @@ mod test {
         assert_eq!(
             Packet::new(
                 PacketId::Connect,
-                "/",
+                "/".to_owned(),
                 Some(String::from("{\"token\":\"123\"}")),
                 None,
                 None,
@@ -282,7 +279,7 @@ mod test {
         assert_eq!(
             Packet::new(
                 PacketId::Connect,
-                "/admin",
+                "/admin".to_owned(),
                 Some(String::from("{\"token\":\"123\"}")),
                 None,
                 None,
@@ -296,7 +293,14 @@ mod test {
         assert!(packet.is_ok());
 
         assert_eq!(
-            Packet::new(PacketId::Disconnect, "/admin", None, None, None, None,),
+            Packet::new(
+                PacketId::Disconnect,
+                "/admin".to_owned(),
+                None,
+                None,
+                None,
+                None,
+            ),
             packet.unwrap()
         );
 
@@ -307,7 +311,7 @@ mod test {
         assert_eq!(
             Packet::new(
                 PacketId::Event,
-                "/",
+                "/".to_owned(),
                 Some(String::from("[\"hello\",1]")),
                 None,
                 None,
@@ -323,7 +327,7 @@ mod test {
         assert_eq!(
             Packet::new(
                 PacketId::Event,
-                "/admin",
+                "/admin".to_owned(),
                 Some(String::from("[\"project:delete\",123]")),
                 None,
                 Some(456),
@@ -339,7 +343,7 @@ mod test {
         assert_eq!(
             Packet::new(
                 PacketId::Ack,
-                "/admin",
+                "/admin".to_owned(),
                 Some(String::from("[]")),
                 None,
                 Some(456),
@@ -355,7 +359,7 @@ mod test {
         assert_eq!(
             Packet::new(
                 PacketId::ConnectError,
-                "/admin",
+                "/admin".to_owned(),
                 Some(String::from("{\"message\":\"Not authorized\"}")),
                 None,
                 None,
@@ -364,17 +368,16 @@ mod test {
             packet.unwrap()
         );
 
-        let payload =
-            Bytes::from_static(b"51-[\"hello\",{\"_placeholder\":true,\"num\":0}]\x01\x02\x03");
+        let payload = Bytes::from_static(b"51-[\"hello\",{\"_placeholder\":true,\"num\":0}]");
         let packet = Packet::decode_bytes(&payload);
         assert!(packet.is_ok());
 
         assert_eq!(
             Packet::new(
                 PacketId::BinaryEvent,
-                "/",
+                "/".to_owned(),
                 Some(String::from("\"hello\"")),
-                Some(vec![1, 2, 3]),
+                None,
                 None,
                 Some(1),
             ),
@@ -382,7 +385,7 @@ mod test {
         );
 
         let payload = Bytes::from_static(
-            b"51-/admin,456[\"project:delete\",{\"_placeholder\":true,\"num\":0}]\x01\x02\x03",
+            b"51-/admin,456[\"project:delete\",{\"_placeholder\":true,\"num\":0}]",
         );
         let packet = Packet::decode_bytes(&payload);
         assert!(packet.is_ok());
@@ -390,26 +393,25 @@ mod test {
         assert_eq!(
             Packet::new(
                 PacketId::BinaryEvent,
-                "/admin",
+                "/admin".to_owned(),
                 Some(String::from("\"project:delete\"")),
-                Some(vec![1, 2, 3]),
+                None,
                 Some(456),
                 Some(1),
             ),
             packet.unwrap()
         );
 
-        let payload =
-            Bytes::from_static(b"61-/admin,456[{\"_placeholder\":true,\"num\":0}]\x03\x02\x01");
+        let payload = Bytes::from_static(b"61-/admin,456[{\"_placeholder\":true,\"num\":0}]");
         let packet = Packet::decode_bytes(&payload);
         assert!(packet.is_ok());
 
         assert_eq!(
             Packet::new(
                 PacketId::BinaryAck,
-                "/admin",
+                "/admin".to_owned(),
+                None,
                 None,
-                Some(vec![3, 2, 1]),
                 Some(456),
                 Some(1),
             ),
@@ -423,7 +425,7 @@ mod test {
     fn test_encode() {
         let packet = Packet::new(
             PacketId::Connect,
-            "/",
+            "/".to_owned(),
             Some(String::from("{\"token\":\"123\"}")),
             None,
             None,
@@ -437,7 +439,7 @@ mod test {
 
         let packet = Packet::new(
             PacketId::Connect,
-            "/admin",
+            "/admin".to_owned(),
             Some(String::from("{\"token\":\"123\"}")),
             None,
             None,
@@ -449,13 +451,20 @@ mod test {
             "0/admin,{\"token\":\"123\"}".to_string().into_bytes()
         );
 
-        let packet = Packet::new(PacketId::Disconnect, "/admin", None, None, None, None);
+        let packet = Packet::new(
+            PacketId::Disconnect,
+            "/admin".to_owned(),
+            None,
+            None,
+            None,
+            None,
+        );
 
         assert_eq!(packet.encode(), "1/admin,".to_string().into_bytes());
 
         let packet = Packet::new(
             PacketId::Event,
-            "/",
+            "/".to_owned(),
             Some(String::from("[\"hello\",1]")),
             None,
             None,
@@ -466,7 +475,7 @@ mod test {
 
         let packet = Packet::new(
             PacketId::Event,
-            "/admin",
+            "/admin".to_owned(),
             Some(String::from("[\"project:delete\",123]")),
             None,
             Some(456),
@@ -482,7 +491,7 @@ mod test {
 
         let packet = Packet::new(
             PacketId::Ack,
-            "/admin",
+            "/admin".to_owned(),
             Some(String::from("[]")),
             None,
             Some(456),
@@ -493,7 +502,7 @@ mod test {
 
         let packet = Packet::new(
             PacketId::ConnectError,
-            "/admin",
+            "/admin".to_owned(),
             Some(String::from("{\"message\":\"Not authorized\"}")),
             None,
             None,
@@ -509,47 +518,50 @@ mod test {
 
         let packet = Packet::new(
             PacketId::BinaryEvent,
-            "/",
+            "/".to_owned(),
             Some(String::from("\"hello\"")),
             Some(vec![1, 2, 3]),
             None,
             Some(1),
         );
 
-        let mut string_part = "51-[\"hello\",{\"_placeholder\":true,\"num\":0}]"
-            .to_string()
-            .into_bytes();
-        string_part.extend(vec![1, 2, 3]);
-        assert_eq!(packet.encode(), string_part);
+        assert_eq!(
+            packet.encode(),
+            "51-[\"hello\",{\"_placeholder\":true,\"num\":0}]"
+                .to_string()
+                .into_bytes()
+        );
 
         let packet = Packet::new(
             PacketId::BinaryEvent,
-            "/admin",
+            "/admin".to_owned(),
             Some(String::from("\"project:delete\"")),
             Some(vec![1, 2, 3]),
             Some(456),
             Some(1),
         );
 
-        let mut string_part = "51-/admin,456[\"project:delete\",{\"_placeholder\":true,\"num\":0}]"
-            .to_string()
-            .into_bytes();
-        string_part.extend(vec![1, 2, 3]);
-        assert_eq!(packet.encode(), string_part);
+        assert_eq!(
+            packet.encode(),
+            "51-/admin,456[\"project:delete\",{\"_placeholder\":true,\"num\":0}]"
+                .to_string()
+                .into_bytes()
+        );
 
         let packet = Packet::new(
             PacketId::BinaryAck,
-            "/admin",
+            "/admin".to_owned(),
             None,
             Some(vec![3, 2, 1]),
             Some(456),
             Some(1),
         );
 
-        let mut string_part = "61-/admin,456[{\"_placeholder\":true,\"num\":0}]"
-            .to_string()
-            .into_bytes();
-        string_part.extend(vec![3, 2, 1]);
-        assert_eq!(packet.encode(), string_part);
+        assert_eq!(
+            packet.encode(),
+            "61-/admin,456[{\"_placeholder\":true,\"num\":0}]"
+                .to_string()
+                .into_bytes()
+        );
     }
 }
diff --git a/src/socketio/payload.rs b/src/socketio/payload.rs
new file mode 100644
index 00000000..694a33f8
--- /dev/null
+++ b/src/socketio/payload.rs
@@ -0,0 +1,40 @@
+/// A type which represents a `payload` in the `socket.io` context.
+/// A payload could either be of the type `Payload::Binary`, which holds
+/// a [`std::vec::Vec<u8>`] that represents the payload or of the type
+/// `Payload::String` which holds a [`std::string::String`]. The enum is
+/// used for both representing data that's send and data that's received.
+#[derive(Debug, PartialEq, Eq, Clone)]
+pub enum Payload {
+    Binary(Vec<u8>),
+    String(String),
+}
+
+impl From<&str> for Payload {
+    fn from(string: &str) -> Self {
+        Self::String(string.to_owned())
+    }
+}
+
+impl From<String> for Payload {
+    fn from(str: String) -> Self {
+        Self::String(str)
+    }
+}
+
+impl From<serde_json::Value> for Payload {
+    fn from(value: serde_json::Value) -> Self {
+        Self::String(value.to_string())
+    }
+}
+
+impl From<Vec<u8>> for Payload {
+    fn from(val: Vec<u8>) -> Self {
+        Self::Binary(val)
+    }
+}
+
+impl From<&[u8]> for Payload {
+    fn from(val: &[u8]) -> Self {
+        Self::Binary(val.to_owned())
+    }
+}
diff --git a/src/socketio/transport.rs b/src/socketio/transport.rs
index 8f69888e..eff94176 100644
--- a/src/socketio/transport.rs
+++ b/src/socketio/transport.rs
@@ -16,7 +16,7 @@ use std::{
     time::{Duration, Instant},
 };
 
-use super::event::Event;
+use super::{event::Event, payload::Payload};
 
 /// The type of a callback function.
 pub(crate) type Callback<I> = RwLock<Box<dyn FnMut(I) + 'static + Sync + Send>>;
@@ -29,7 +29,7 @@ pub struct Ack {
     pub id: i32,
     timeout: Duration,
     time_started: Instant,
-    callback: Callback<String>,
+    callback: Callback<Payload>,
 }
 
 /// Handles communication in the `socket.io` protocol.
@@ -39,8 +39,11 @@ pub struct TransportClient {
     host: Arc<String>,
     connected: Arc<AtomicBool>,
     engineio_connected: Arc<AtomicBool>,
-    on: Arc<Vec<(Event, Callback<String>)>>,
+    on: Arc<Vec<(Event, Callback<Payload>)>>,
     outstanding_acks: Arc<RwLock<Vec<Ack>>>,
+    // used to detect unfinished binary events as, as the attachements
+    // gets send in a seperate packet
+    unfinished_packet: Arc<RwLock<Option<SocketPacket>>>,
     // namespace, for multiplexing messages
     pub(crate) nsp: Arc<Option<String>>,
 }
@@ -55,6 +58,7 @@ impl TransportClient {
             engineio_connected: Arc::new(AtomicBool::default()),
             on: Arc::new(Vec::new()),
             outstanding_acks: Arc::new(RwLock::new(Vec::new())),
+            unfinished_packet: Arc::new(RwLock::new(None)),
             nsp: Arc::new(nsp),
         }
     }
@@ -62,7 +66,7 @@ impl TransportClient {
     /// Registers a new event with some callback function `F`.
     pub fn on<F>(&mut self, event: Event, callback: F) -> Result<()>
     where
-        F: FnMut(String) + 'static + Sync + Send,
+        F: FnMut(Payload) + 'static + Sync + Send,
     {
         Arc::get_mut(&mut self.on)
             .unwrap()
@@ -83,7 +87,7 @@ impl TransportClient {
         // construct the opening packet
         let open_packet = SocketPacket::new(
             SocketPacketId::Connect,
-            &self.nsp.as_ref().as_ref().unwrap_or(&default),
+            self.nsp.as_ref().as_ref().unwrap_or(&default).to_owned(),
             None,
             None,
             None,
@@ -105,26 +109,83 @@ impl TransportClient {
         self.engine_socket.lock()?.emit(engine_packet)
     }
 
+    /// Sends a single binary attachement to the server. This method
+    /// should only be called if a `BinaryEvent` or `BinaryAck` was
+    /// send to the server mentioning this attachement in it's
+    /// `attachements` field.
+    fn send_binary_attachement(&self, attachement: Vec<u8>) -> Result<()> {
+        if !self.engineio_connected.load(Ordering::Relaxed) {
+            return Err(Error::ActionBeforeOpen);
+        }
+
+        self.engine_socket
+            .lock()?
+            .emit_binary_attachement(attachement)
+    }
+
     /// Emits to certain event with given data. The data needs to be JSON,
     /// otherwise this returns an `InvalidJson` error.
-    pub fn emit(&self, event: Event, data: &str) -> Result<()> {
-        if serde_json::from_str::<serde_json::Value>(data).is_err() {
-            return Err(Error::InvalidJson(data.to_owned()));
+    pub fn emit(&self, event: Event, data: Payload) -> Result<()> {
+        let default = String::from("/");
+        let nsp = self.nsp.as_ref().as_ref().unwrap_or(&default);
+
+        let is_string_packet = matches!(&data, &Payload::String(_));
+        let socket_packet = self.build_packet_for_payload(data, event, nsp, None)?;
+
+        if is_string_packet {
+            self.send(&socket_packet)
+        } else {
+            // unwrapping here is safe as this is a binary payload
+            let binary_payload = socket_packet.binary_data.as_ref().unwrap();
+
+            // first send the raw packet announcing the attachement
+            self.send(&socket_packet)?;
+
+            // then send the attachement
+            self.send_binary_attachement(binary_payload.to_owned())
         }
+    }
 
-        let payload = format!("[\"{}\",{}]", String::from(event), data);
+    /// Returns a packet for a payload, could be used for bot binary and non binary
+    /// events and acks.
+    #[inline]
+    fn build_packet_for_payload<'a>(
+        &'a self,
+        payload: Payload,
+        event: Event,
+        nsp: &'a str,
+        id: Option<i32>,
+    ) -> Result<SocketPacket> {
+        match payload {
+            Payload::Binary(bin_data) => Ok(SocketPacket::new(
+                if id.is_some() {
+                    SocketPacketId::BinaryAck
+                } else {
+                    SocketPacketId::BinaryEvent
+                },
+                nsp.to_owned(),
+                Some(serde_json::Value::String(event.into()).to_string()),
+                Some(bin_data),
+                id,
+                Some(1),
+            )),
+            Payload::String(str_data) => {
+                if serde_json::from_str::<serde_json::Value>(&str_data).is_err() {
+                    return Err(Error::InvalidJson(str_data));
+                }
 
-        let default = String::from("/");
-        let socket_packet = SocketPacket::new(
-            SocketPacketId::Event,
-            &self.nsp.as_ref().as_ref().unwrap_or(&default),
-            Some(payload),
-            None,
-            None,
-            None,
-        );
+                let payload = format!("[\"{}\",{}]", String::from(event), str_data);
 
-        self.send(&socket_packet)
+                Ok(SocketPacket::new(
+                    SocketPacketId::Event,
+                    nsp.to_owned(),
+                    Some(payload),
+                    None,
+                    id,
+                    None,
+                ))
+            }
+        }
     }
 
     /// Emits and requests an `ack`. The `ack` returns a `Arc<RwLock<Ack>>` to
@@ -133,29 +194,17 @@ impl TransportClient {
     pub fn emit_with_ack<F>(
         &mut self,
         event: Event,
-        data: &str,
+        data: Payload,
         timeout: Duration,
         callback: F,
     ) -> Result<()>
     where
-        F: FnMut(String) + 'static + Send + Sync,
+        F: FnMut(Payload) + 'static + Send + Sync,
     {
-        if serde_json::from_str::<serde_json::Value>(data).is_err() {
-            return Err(Error::InvalidJson(data.into()));
-        }
-
-        let payload = format!("[\"{}\",{}]", String::from(event), data);
         let id = thread_rng().gen_range(0..999);
-
         let default = String::from("/");
-        let socket_packet = SocketPacket::new(
-            SocketPacketId::Event,
-            &self.nsp.as_ref().as_ref().unwrap_or(&default),
-            Some(payload),
-            None,
-            Some(id),
-            None,
-        );
+        let nsp = self.nsp.as_ref().as_ref().unwrap_or(&default);
+        let socket_packet = self.build_packet_for_payload(data, event, nsp, Some(id))?;
 
         let ack = Ack {
             id,
@@ -176,13 +225,28 @@ impl TransportClient {
     /// engineio client.
     #[inline]
     fn handle_new_message(socket_bytes: &[u8], clone_self: &TransportClient) {
-        // TODO: Refactor the copy as soon as engine.io uses the Bytes type as well
-        if let Ok(socket_packet) = SocketPacket::decode_bytes(&Bytes::copy_from_slice(socket_bytes))
-        {
+        let mut is_finalized_packet = false;
+        // either this is a complete packet or the rest of a binary packet (as attachements are
+        // sent in a seperate packet).
+        let decoded_packet = if clone_self.unfinished_packet.read().unwrap().is_some() {
+            // this must be an attachement, so parse it
+            let mut unfinished_packet = clone_self.unfinished_packet.write().unwrap();
+            let mut finalized_packet = unfinished_packet.take().unwrap();
+            finalized_packet.binary_data = Some(socket_bytes.to_vec());
+
+            is_finalized_packet = true;
+            Ok(finalized_packet)
+        } else {
+            // this is a normal packet, so decode it
+            SocketPacket::decode_bytes(&Bytes::copy_from_slice(socket_bytes))
+        };
+
+        if let Ok(socket_packet) = decoded_packet {
             let default = String::from("/");
-            if socket_packet.nsp != clone_self.nsp.as_ref().as_ref().unwrap_or(&default) {
+            if socket_packet.nsp != *clone_self.nsp.as_ref().as_ref().unwrap_or(&default) {
                 return;
             }
+
             match socket_packet.packet_type {
                 SocketPacketId::Connect => {
                     clone_self.connected.swap(true, Ordering::Relaxed);
@@ -191,12 +255,12 @@ impl TransportClient {
                     clone_self.connected.swap(false, Ordering::Relaxed);
                     if let Some(function) = clone_self.get_event_callback(&Event::Error) {
                         let mut lock = function.1.write().unwrap();
-                        lock(
+                        lock(Payload::String(
                             String::from("Received an ConnectError frame")
                                 + &socket_packet.data.unwrap_or_else(|| {
                                     String::from("\"No error message provided\"")
                                 }),
-                        );
+                        ));
                         drop(lock)
                     }
                 }
@@ -210,10 +274,20 @@ impl TransportClient {
                     Self::handle_ack(socket_packet, clone_self);
                 }
                 SocketPacketId::BinaryEvent => {
-                    // call the callback
+                    // in case of a binary event, check if this is the attachement or not and
+                    // then either handle the event or set the open packet
+                    if is_finalized_packet {
+                        Self::handle_binary_event(socket_packet, clone_self).unwrap();
+                    } else {
+                        *clone_self.unfinished_packet.write().unwrap() = Some(socket_packet);
+                    }
                 }
                 SocketPacketId::BinaryAck => {
-                    // call the callback
+                    if is_finalized_packet {
+                        Self::handle_ack(socket_packet, clone_self);
+                    } else {
+                        *clone_self.unfinished_packet.write().unwrap() = Some(socket_packet);
+                    }
                 }
             }
         }
@@ -239,7 +313,14 @@ impl TransportClient {
                         if let Some(payload) = socket_packet.clone().data {
                             spawn_scoped!({
                                 let mut function = ack.callback.write().unwrap();
-                                function(payload);
+                                function(Payload::String(payload));
+                                drop(function);
+                            });
+                        }
+                        if let Some(payload) = socket_packet.clone().binary_data {
+                            spawn_scoped!({
+                                let mut function = ack.callback.write().unwrap();
+                                function(Payload::Binary(payload));
                                 drop(function);
                             });
                         }
@@ -259,7 +340,7 @@ impl TransportClient {
         let error_callback = move |msg| {
             if let Some(function) = clone_self.get_event_callback(&Event::Error) {
                 let mut lock = function.1.write().unwrap();
-                lock(msg);
+                lock(Payload::String(msg));
                 drop(lock)
             }
         };
@@ -269,7 +350,7 @@ impl TransportClient {
             clone_self.engineio_connected.swap(true, Ordering::Relaxed);
             if let Some(function) = clone_self.get_event_callback(&Event::Connect) {
                 let mut lock = function.1.write().unwrap();
-                lock(String::from("Connection is opened"));
+                lock(Payload::String(String::from("Connection is opened")));
                 drop(lock)
             }
         };
@@ -279,7 +360,7 @@ impl TransportClient {
             clone_self.engineio_connected.swap(false, Ordering::Relaxed);
             if let Some(function) = clone_self.get_event_callback(&Event::Close) {
                 let mut lock = function.1.write().unwrap();
-                lock(String::from("Connection is closed"));
+                lock(Payload::String(String::from("Connection is closed")));
                 drop(lock)
             }
         };
@@ -296,6 +377,30 @@ impl TransportClient {
             .on_data(move |data| Self::handle_new_message(&data, &clone_self))
     }
 
+    /// Handles a binary event.
+    #[inline]
+    fn handle_binary_event(
+        socket_packet: SocketPacket,
+        clone_self: &TransportClient,
+    ) -> Result<()> {
+        let event = if let Some(string_data) = socket_packet.data {
+            string_data.replace("\"", "").into()
+        } else {
+            Event::Message
+        };
+
+        if let Some(binary_payload) = socket_packet.binary_data {
+            if let Some(function) = clone_self.get_event_callback(&event) {
+                spawn_scoped!({
+                    let mut lock = function.1.write().unwrap();
+                    lock(Payload::Binary(binary_payload));
+                    drop(lock);
+                });
+            }
+        }
+        Ok(())
+    }
+
     /// A method for handling the Event Socket Packets.
     // this could only be called with an event
     fn handle_event(socket_packet: SocketPacket, clone_self: &TransportClient) -> Result<()> {
@@ -313,7 +418,7 @@ impl TransportClient {
                                 if let Some(function) = clone_self.get_event_callback(&Event::Custom(event.to_owned())) {
                                     spawn_scoped!({
                                         let mut lock = function.1.write().unwrap();
-                                        lock(contents[1].to_string());
+                                        lock(Payload::String(contents[1].to_string()));
                                         drop(lock);
                                     });
                                 }
@@ -321,7 +426,7 @@ impl TransportClient {
                         } else if let Some(function) = clone_self.get_event_callback(&Event::Message) {
                             spawn_scoped!({
                                 let mut lock = function.1.write().unwrap();
-                                lock(contents[0].to_string());
+                                lock(Payload::String(contents[0].to_string()));
                                 drop(lock);
                             });
                         }
@@ -334,7 +439,7 @@ impl TransportClient {
 
     /// A convenient method for finding a callback for a certain event.
     #[inline]
-    fn get_event_callback(&self, event: &Event) -> Option<&(Event, Callback<String>)> {
+    fn get_event_callback(&self, event: &Event) -> Option<&(Event, Callback<Payload>)> {
         self.on.iter().find(|item| item.0 == *event)
     }
 }
@@ -375,19 +480,35 @@ mod test {
 
         assert!(socket
             .on("test".into(), |s| {
-                println!("{}", s);
+                if let Payload::String(st) = s {
+                    println!("{}", st)
+                }
             })
             .is_ok());
 
         assert!(socket.connect().is_ok());
 
-        let ack_callback = |message: String| {
+        let ack_callback = |message: Payload| {
             println!("Yehaa! My ack got acked?");
-            println!("Ack data: {}", message);
+            match message {
+                Payload::String(str) => {
+                    println!("Received string ack");
+                    println!("Ack data: {}", str);
+                }
+                Payload::Binary(bin) => {
+                    println!("Received binary ack");
+                    println!("Ack data: {:#?}", bin);
+                }
+            }
         };
 
         assert!(socket
-            .emit_with_ack("test".into(), "123", Duration::from_secs(10), ack_callback)
+            .emit_with_ack(
+                "test".into(),
+                Payload::String("123".to_owned()),
+                Duration::from_secs(10),
+                ack_callback
+            )
             .is_ok());
     }
 }


Test patch: