[WIP] custom packet serializer implementation

engineio/src/asynchronous/async_socket.rs

@@ -15,7 +15,7 @@ use tokio::{runtime::Handle, sync::Mutex, time::Instant};
 use crate::{
     asynchronous::{callback::OptionalCallback, transport::AsyncTransportType},
     error::Result,
-    packet::{HandshakePacket, Payload},
+    packet::{HandshakePacket, PacketSerializer},
     Error, Packet, PacketId,
 };
 
@@ -24,6 +24,7 @@ pub struct Socket {
     handle: Handle,
     transport: Arc<Mutex<AsyncTransportType>>,
     transport_raw: AsyncTransportType,
+    serializer: Arc<PacketSerializer>,
     on_close: OptionalCallback<()>,
     on_data: OptionalCallback<Bytes>,
     on_error: OptionalCallback<String>,
@@ -39,6 +40,7 @@ pub struct Socket {
 impl Socket {
     pub(crate) fn new(
         transport: AsyncTransportType,
+        serializer: Arc<PacketSerializer>,
         handshake: HandshakePacket,
         on_close: OptionalCallback<()>,
         on_data: OptionalCallback<Bytes>,
@@ -57,6 +59,7 @@ impl Socket {
             on_packet,
             transport: Arc::new(Mutex::new(transport.clone())),
             transport_raw: transport,
+            serializer,
             connected: Arc::new(AtomicBool::default()),
             last_ping: Arc::new(Mutex::new(Instant::now())),
             last_pong: Arc::new(Mutex::new(Instant::now())),
@@ -117,9 +120,13 @@ impl Socket {
     }
 
     /// Helper method that parses bytes and returns an iterator over the elements.
-    fn parse_payload(bytes: Bytes) -> impl Stream<Item = Result<Packet>> {
+    fn parse_payload(
+        bytes: Bytes,
+        serializer: Arc<PacketSerializer>,
+    ) -> impl Stream<Item = Result<Packet>> {
         try_stream! {
-            let payload = Payload::try_from(bytes);
+            // let payload = Payload::try_from(bytes);
+            let payload = serializer.decode_payload(bytes);
 
             for elem in payload?.into_iter() {
                 yield elem;
@@ -131,12 +138,13 @@ impl Socket {
     /// underlying transport types.
     fn stream(
         mut transport: AsyncTransportType,
+        serialzer: Arc<PacketSerializer>,
     ) -> Pin<Box<impl Stream<Item = Result<Packet>> + 'static + Send>> {
         // map the byte stream of the underlying transport
         // to a packet stream
         Box::pin(try_stream! {
             for await payload in transport.as_pin_box() {
-                for await packet in Self::parse_payload(payload?) {
+                for await packet in Self::parse_payload(payload?, serialzer.clone()) {
                     yield packet?;
                 }
             }
@@ -172,7 +180,8 @@ impl Socket {
         let data: Bytes = if is_binary {
             packet.data
         } else {
-            packet.into()
+            // packet.into()
+            self.serializer.encode(packet)
         };
 
         let lock = self.transport.lock().await;
@@ -249,7 +258,7 @@ impl Socket {
         &'a self,
     ) -> Pin<Box<dyn Stream<Item = Result<Packet>> + Send + 'a>> {
         stream::unfold(
-            Self::stream(self.transport_raw.clone()),
+            Self::stream(self.transport_raw.clone(), self.serializer.clone()),
             |mut stream| async {
                 // Wait for the next payload or until we should have received the next ping.
                 match tokio::time::timeout(

engineio/src/asynchronous/async_transports/polling.rs

@@ -1,8 +1,7 @@
 use adler32::adler32;
 use async_stream::try_stream;
 use async_trait::async_trait;
-use base64::{engine::general_purpose, Engine as _};
-use bytes::{BufMut, Bytes, BytesMut};
+use bytes::Bytes;
 use futures_util::{Stream, StreamExt};
 use http::HeaderMap;
 use native_tls::TlsConnector;
@@ -102,23 +101,10 @@ impl Stream for PollingTransport {
 #[async_trait]
 impl AsyncTransport for PollingTransport {
     async fn emit(&self, data: Bytes, is_binary_att: bool) -> Result<()> {
-        let data_to_send = if is_binary_att {
-            // the binary attachment gets `base64` encoded
-            let mut packet_bytes = BytesMut::with_capacity(data.len() + 1);
-            packet_bytes.put_u8(b'b');
-
-            let encoded_data = general_purpose::STANDARD.encode(data);
-            packet_bytes.put(encoded_data.as_bytes());
-
-            packet_bytes.freeze()
-        } else {
-            data
-        };
-
         let status = self
             .client
             .post(self.address().await?)
-            .body(data_to_send)
+            .body(data)
             .send()
             .await?
             .status()

engineio/src/asynchronous/async_transports/websocket.rs

@@ -4,6 +4,7 @@ use std::sync::Arc;
 
 use crate::asynchronous::transport::AsyncTransport;
 use crate::error::Result;
+use crate::PacketSerializer;
 use async_trait::async_trait;
 use bytes::Bytes;
 use futures_util::stream::StreamExt;
@@ -27,7 +28,11 @@ pub struct WebsocketTransport {
 
 impl WebsocketTransport {
     /// Creates a new instance over a request that might hold additional headers and an URL.
-    pub async fn new(base_url: Url, headers: Option<HeaderMap>) -> Result<Self> {
+    pub async fn new(
+        base_url: Url,
+        headers: Option<HeaderMap>,
+        serializer: Arc<PacketSerializer>,
+    ) -> Result<Self> {
         let mut url = base_url;
         url.query_pairs_mut().append_pair("transport", "websocket");
         url.set_scheme("ws").unwrap();
@@ -41,7 +46,7 @@ impl WebsocketTransport {
         let (ws_stream, _) = connect_async(req).await?;
         let (sen, rec) = ws_stream.split();
 
-        let inner = AsyncWebsocketGeneralTransport::new(sen, rec).await;
+        let inner = AsyncWebsocketGeneralTransport::new(sen, rec, serializer).await;
         Ok(WebsocketTransport {
             inner,
             base_url: Arc::new(RwLock::new(url)),
@@ -118,7 +123,12 @@ mod test {
         let url = crate::test::engine_io_server()?.to_string()
             + "engine.io/?EIO="
             + &ENGINE_IO_VERSION.to_string();
-        WebsocketTransport::new(Url::from_str(&url[..])?, None).await
+        WebsocketTransport::new(
+            Url::from_str(&url[..])?,
+            None,
+            PacketSerializer::default_arc(),
+        )
+        .await
     }
 
     #[tokio::test]

engineio/src/asynchronous/async_transports/websocket_general.rs

@@ -1,6 +1,6 @@
 use std::{borrow::Cow, str::from_utf8, sync::Arc, task::Poll};
 
-use crate::{error::Result, Error, Packet, PacketId};
+use crate::{error::Result, Error, Packet, PacketId, PacketSerializer};
 use bytes::{BufMut, Bytes, BytesMut};
 use futures_util::{
     ready,
@@ -22,16 +22,19 @@ type AsyncWebsocketReceiver = SplitStream<WebSocketStream<MaybeTlsStream<TcpStre
 pub(crate) struct AsyncWebsocketGeneralTransport {
     sender: Arc<Mutex<AsyncWebsocketSender>>,
     receiver: Arc<Mutex<AsyncWebsocketReceiver>>,
+    serializer: Arc<PacketSerializer>,
 }
 
 impl AsyncWebsocketGeneralTransport {
     pub(crate) async fn new(
         sender: SplitSink<WebSocketStream<MaybeTlsStream<TcpStream>>, Message>,
         receiver: SplitStream<WebSocketStream<MaybeTlsStream<TcpStream>>>,
+        serializer: Arc<PacketSerializer>,
     ) -> Self {
         AsyncWebsocketGeneralTransport {
             sender: Arc::new(Mutex::new(sender)),
             receiver: Arc::new(Mutex::new(receiver)),
+            serializer,
         }
     }
 
@@ -41,25 +44,30 @@ impl AsyncWebsocketGeneralTransport {
         let mut receiver = self.receiver.lock().await;
         let mut sender = self.sender.lock().await;
 
+        let ping_packet = Packet::new(PacketId::Ping, Bytes::from("probe"));
+        let ping_packet = self.serializer.encode(ping_packet);
+
         sender
-            .send(Message::text(Cow::Borrowed(from_utf8(&Bytes::from(
-                Packet::new(PacketId::Ping, Bytes::from("probe")),
-            ))?)))
+            .send(Message::text(Cow::Borrowed(from_utf8(&ping_packet)?)))
             .await?;
 
         let msg = receiver
             .next()
             .await
             .ok_or(Error::IllegalWebsocketUpgrade())??;
 
-        if msg.into_data() != Bytes::from(Packet::new(PacketId::Pong, Bytes::from("probe"))) {
+        let pong_packet = Packet::new(PacketId::Pong, Bytes::from("probe"));
+        let pong_packet = self.serializer.encode(pong_packet);
+
+        if msg.into_data() != pong_packet {
             return Err(Error::InvalidPacket());
         }
 
+        let upgrade_packet = Packet::new(PacketId::Upgrade, Bytes::from(""));
+        let upgrade_packet = self.serializer.encode(upgrade_packet);
+
         sender
-            .send(Message::text(Cow::Borrowed(from_utf8(&Bytes::from(
-                Packet::new(PacketId::Upgrade, Bytes::from("")),
-            ))?)))
+            .send(Message::text(Cow::Borrowed(from_utf8(&upgrade_packet)?)))
             .await?;
 
         Ok(())

engineio/src/asynchronous/async_transports/websocket_secure.rs

@@ -4,6 +4,7 @@ use std::sync::Arc;
 
 use crate::asynchronous::transport::AsyncTransport;
 use crate::error::Result;
+use crate::PacketSerializer;
 use async_trait::async_trait;
 use bytes::Bytes;
 use futures_util::Stream;
@@ -34,6 +35,7 @@ impl WebsocketSecureTransport {
         base_url: Url,
         tls_config: Option<TlsConnector>,
         headers: Option<HeaderMap>,
+        serializer: Arc<PacketSerializer>,
     ) -> Result<Self> {
         let mut url = base_url;
         url.query_pairs_mut().append_pair("transport", "websocket");
@@ -61,7 +63,7 @@ impl WebsocketSecureTransport {
         .await?;
 
         let (sen, rec) = ws_stream.split();
-        let inner = AsyncWebsocketGeneralTransport::new(sen, rec).await;
+        let inner = AsyncWebsocketGeneralTransport::new(sen, rec, serializer).await;
 
         Ok(WebsocketSecureTransport {
             inner,
@@ -143,6 +145,7 @@ mod test {
             Url::from_str(&url[..])?,
             Some(crate::test::tls_connector()?),
             None,
+            PacketSerializer::default_arc(),
         )
         .await
     }

engineio/src/asynchronous/client/builder.rs

@@ -1,3 +1,5 @@
+use std::sync::Arc;
+
 use crate::{
     asynchronous::{
         async_socket::Socket as InnerSocket,
@@ -7,7 +9,7 @@ use crate::{
     },
     error::Result,
     header::HeaderMap,
-    packet::HandshakePacket,
+    packet::{HandshakePacket, PacketSerializer},
     Error, Packet, ENGINE_IO_VERSION,
 };
 use bytes::Bytes;
@@ -22,6 +24,7 @@ pub struct ClientBuilder {
     url: Url,
     tls_config: Option<TlsConnector>,
     headers: Option<HeaderMap>,
+    serializer: Arc<PacketSerializer>,
     handshake: Option<HandshakePacket>,
     on_error: OptionalCallback<String>,
     on_open: OptionalCallback<()>,
@@ -45,6 +48,7 @@ impl ClientBuilder {
             headers: None,
             tls_config: None,
             handshake: None,
+            serializer: Arc::new(PacketSerializer::default()),
             on_close: OptionalCallback::default(),
             on_data: OptionalCallback::default(),
             on_error: OptionalCallback::default(),
@@ -53,6 +57,13 @@ impl ClientBuilder {
         }
     }
 
+    /// Specify Packet Serializer
+    pub fn packet_serializer(mut self, packet_serializer: Arc<PacketSerializer>) -> Self {
+        self.serializer = packet_serializer;
+
+        self
+    }
+
     /// Specify transport's tls config
     pub fn tls_config(mut self, tls_config: TlsConnector) -> Self {
         self.tls_config = Some(tls_config);
@@ -127,9 +138,10 @@ impl ClientBuilder {
 
         let mut url = self.url.clone();
 
-        let handshake: HandshakePacket =
-            Packet::try_from(transport.next().await.ok_or(Error::IncompletePacket())??)?
-                .try_into()?;
+        let handshake: HandshakePacket = self
+            .serializer
+            .decode(transport.next().await.ok_or(Error::IncompletePacket())??)?
+            .try_into()?;
 
         // update the base_url with the new sid
         url.query_pairs_mut().append_pair("sid", &handshake.sid[..]);
@@ -184,6 +196,7 @@ impl ClientBuilder {
         // SAFETY: handshake function called previously.
         Ok(Client::new(InnerSocket::new(
             transport.into(),
+            self.serializer,
             self.handshake.unwrap(),
             self.on_close,
             self.on_data,
@@ -214,7 +227,9 @@ impl ClientBuilder {
 
         match self.url.scheme() {
             "http" | "ws" => {
-                let mut transport = WebsocketTransport::new(self.url.clone(), headers).await?;
+                let mut transport =
+                    WebsocketTransport::new(self.url.clone(), headers, self.serializer.clone())
+                        .await?;
 
                 if self.handshake.is_some() {
                     transport.upgrade().await?;
@@ -225,6 +240,7 @@ impl ClientBuilder {
                 // SAFETY: handshake function called previously.
                 Ok(Client::new(InnerSocket::new(
                     transport.into(),
+                    self.serializer,
                     self.handshake.unwrap(),
                     self.on_close,
                     self.on_data,
@@ -238,6 +254,7 @@ impl ClientBuilder {
                     self.url.clone(),
                     self.tls_config.clone(),
                     headers,
+                    self.serializer.clone(),
                 )
                 .await?;
 
@@ -250,6 +267,7 @@ impl ClientBuilder {
                 // SAFETY: handshake function called previously.
                 Ok(Client::new(InnerSocket::new(
                     transport.into(),
+                    self.serializer,
                     self.handshake.unwrap(),
                     self.on_close,
                     self.on_data,