Implement socketio onAny (1c3t3a#169)

marekvospel · web-flow · commit d29086b2dd9c · 2022-03-18T18:38:33.000+01:00
* test: auth use iter instead of emit_with_ack

* feat: implement socketio onAny

* fix: only call on_any on Message / Custom event

* docs: add on_any documentation

* test: rename on_any integration

* test: add mpsc to on_any integration

* docs: add sleep doc to on_any integration
diff --git a/ci/socket-io-auth.js b/ci/socket-io-auth.js
@@ -3,14 +3,9 @@ const io = require('socket.io')(server);
 
 console.log('Started');
 var callback = client => {
-    console.log('Connected!');;
-    client.on('test', function (arg, ack) {
-        console.log('Ack received')
-        if (ack) {
-            const payload = client.handshake.auth.password === '123' ? '456' : '789'
-            ack(payload);
-        }
-    });
+    console.log('Connected!');
+
+    client.emit('auth', client.handshake.auth.password === '123' ? 'success' : 'failed')
 };
 io.on('connection', callback);
 io.of('/admin').on('connection', callback);
diff --git a/socketio/src/client/builder.rs b/socketio/src/client/builder.rs
@@ -6,6 +6,7 @@ use rust_engineio::client::ClientBuilder as EngineIoClientBuilder;
 use rust_engineio::header::{HeaderMap, HeaderValue};
 use url::Url;
 
+use crate::client::callback::{SocketAnyCallback, SocketCallback};
 use crate::error::{Error, Result};
 use std::collections::HashMap;
 use std::thread;
@@ -31,7 +32,8 @@ pub enum TransportType {
 /// acts the `build` method and returns a connected [`Client`].
 pub struct ClientBuilder {
     address: String,
-    on: HashMap<Event, Callback>,
+    on: HashMap<Event, Callback<SocketCallback>>,
+    on_any: Option<Callback<SocketAnyCallback>>,
     namespace: String,
     tls_config: Option<TlsConnector>,
     opening_headers: Option<HeaderMap>,
@@ -74,6 +76,7 @@ impl ClientBuilder {
         Self {
             address: address.into(),
             on: HashMap::new(),
+            on_any: None,
             namespace: "/".to_owned(),
             tls_config: None,
             opening_headers: None,
@@ -117,7 +120,32 @@ impl ClientBuilder {
     where
         F: for<'a> FnMut(Payload, Client) + 'static + Sync + Send,
     {
-        self.on.insert(event.into(), Callback::new(callback));
+        self.on
+            .insert(event.into(), Callback::<SocketCallback>::new(callback));
+        self
+    }
+
+    /// Registers a Callback for all [`crate::event::Event::Custom`] and [`crate::event::Event::Message`].
+    ///
+    /// # Example
+    /// ```rust
+    /// use rust_socketio::{ClientBuilder, Payload};
+    ///
+    /// let client = ClientBuilder::new("http://localhost:4200/")
+    ///     .namespace("/admin")
+    ///     .on_any(|event, payload, _client| {
+    ///         if let Payload::String(str) = payload {
+    ///           println!("{} {}", String::from(event), str);
+    ///         }
+    ///     })
+    ///     .connect();
+    ///
+    /// ```
+    pub fn on_any<F>(mut self, callback: F) -> Self
+    where
+        F: for<'a> FnMut(Event, Payload, Client) + 'static + Sync + Send,
+    {
+        self.on_any = Some(Callback::<SocketAnyCallback>::new(callback));
         self
     }
 
@@ -283,7 +311,13 @@ impl ClientBuilder {
 
         let inner_socket = InnerSocket::new(engine_client)?;
 
-        let socket = Client::new(inner_socket, &self.namespace, self.on, self.auth)?;
+        let socket = Client::new(
+            inner_socket,
+            &self.namespace,
+            self.on,
+            self.on_any,
+            self.auth,
+        )?;
         socket.connect()?;
 
         Ok(socket)
diff --git a/socketio/src/client/callback.rs b/socketio/src/client/callback.rs
@@ -4,35 +4,39 @@ use std::{
 };
 
 use super::Client;
-use crate::Payload;
+use crate::{Event, Payload};
 
-type InnerCallback = Box<dyn for<'a> FnMut(Payload, Client) + 'static + Sync + Send>;
+pub(crate) type SocketCallback = Box<dyn for<'a> FnMut(Payload, Client) + 'static + Sync + Send>;
+pub(crate) type SocketAnyCallback =
+    Box<dyn for<'a> FnMut(Event, Payload, Client) + 'static + Sync + Send>;
 
-pub(crate) struct Callback {
-    inner: InnerCallback,
+pub(crate) struct Callback<T> {
+    inner: T,
 }
 
-impl Debug for Callback {
+// SocketCallback implementations
+
+impl Debug for Callback<SocketCallback> {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.write_str("Callback")
     }
 }
 
-impl Deref for Callback {
+impl Deref for Callback<SocketCallback> {
     type Target = dyn for<'a> FnMut(Payload, Client) + 'static + Sync + Send;
 
     fn deref(&self) -> &Self::Target {
         self.inner.as_ref()
     }
 }
 
-impl DerefMut for Callback {
+impl DerefMut for Callback<SocketCallback> {
     fn deref_mut(&mut self) -> &mut Self::Target {
         self.inner.as_mut()
     }
 }
 
-impl Callback {
+impl Callback<SocketCallback> {
     pub(crate) fn new<T>(callback: T) -> Self
     where
         T: for<'a> FnMut(Payload, Client) + 'static + Sync + Send,
@@ -42,3 +46,36 @@ impl Callback {
         }
     }
 }
+
+// SocketAnyCallback implementations
+
+impl Debug for Callback<SocketAnyCallback> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str("Callback")
+    }
+}
+
+impl Deref for Callback<SocketAnyCallback> {
+    type Target = dyn for<'a> FnMut(Event, Payload, Client) + 'static + Sync + Send;
+
+    fn deref(&self) -> &Self::Target {
+        self.inner.as_ref()
+    }
+}
+
+impl DerefMut for Callback<SocketAnyCallback> {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        self.inner.as_mut()
+    }
+}
+
+impl Callback<SocketAnyCallback> {
+    pub(crate) fn new<T>(callback: T) -> Self
+    where
+        T: for<'a> FnMut(Event, Payload, Client) + 'static + Sync + Send,
+    {
+        Callback {
+            inner: Box::new(callback),
+        }
+    }
+}
diff --git a/socketio/src/client/client.rs b/socketio/src/client/client.rs
@@ -3,6 +3,7 @@ use super::callback::Callback;
 use crate::packet::{Packet, PacketId};
 use rand::{thread_rng, Rng};
 
+use crate::client::callback::{SocketAnyCallback, SocketCallback};
 use crate::error::Result;
 use std::collections::HashMap;
 use std::ops::DerefMut;
@@ -21,7 +22,7 @@ pub struct Ack {
     pub id: i32,
     timeout: Duration,
     time_started: Instant,
-    callback: Callback,
+    callback: Callback<SocketCallback>,
 }
 
 /// A socket which handles communication with the server. It's initialized with
@@ -31,7 +32,8 @@ pub struct Ack {
 pub struct Client {
     /// The inner socket client to delegate the methods to.
     socket: InnerSocket,
-    on: Arc<RwLock<HashMap<Event, Callback>>>,
+    on: Arc<RwLock<HashMap<Event, Callback<SocketCallback>>>>,
+    on_any: Arc<RwLock<Option<Callback<SocketAnyCallback>>>>,
     outstanding_acks: Arc<RwLock<Vec<Ack>>>,
     // namespace, for multiplexing messages
     nsp: String,
@@ -47,13 +49,15 @@ impl Client {
     pub(crate) fn new<T: Into<String>>(
         socket: InnerSocket,
         namespace: T,
-        on: HashMap<Event, Callback>,
+        on: HashMap<Event, Callback<SocketCallback>>,
+        on_any: Option<Callback<SocketAnyCallback>>,
         auth: Option<serde_json::Value>,
     ) -> Result<Self> {
         Ok(Client {
             socket,
             nsp: namespace.into(),
             on: Arc::new(RwLock::new(on)),
+            on_any: Arc::new(RwLock::new(on_any)),
             outstanding_acks: Arc::new(RwLock::new(Vec::new())),
             auth,
         })
@@ -202,7 +206,7 @@ impl Client {
             id,
             time_started: Instant::now(),
             timeout,
-            callback: Callback::new(callback),
+            callback: Callback::<SocketCallback>::new(callback),
         };
 
         // add the ack to the tuple of outstanding acks
@@ -238,11 +242,25 @@ impl Client {
 
     fn callback<P: Into<Payload>>(&self, event: &Event, payload: P) -> Result<()> {
         let mut on = self.on.write()?;
+        let mut on_any = self.on_any.write()?;
         let lock = on.deref_mut();
+        let on_any_lock = on_any.deref_mut();
+
+        let payload = payload.into();
+
         if let Some(callback) = lock.get_mut(event) {
-            callback(payload.into(), self.clone());
+            callback(payload.clone(), self.clone());
+        }
+        match event.clone() {
+            Event::Message | Event::Custom(_) => {
+                if let Some(callback) = on_any_lock {
+                    callback(event.clone(), payload, self.clone())
+                }
+            }
+            _ => {}
         }
         drop(on);
+        drop(on_any);
         Ok(())
     }
 
@@ -530,33 +548,69 @@ mod test {
         test_socketio_socket(socket, "/admin".to_owned())
     }
 
+    #[test]
+    fn socket_io_on_any_integration() -> Result<()> {
+        let url = crate::test::socket_io_server();
+
+        let (tx, rx) = mpsc::sync_channel(1);
+
+        let _socket = ClientBuilder::new(url)
+            .namespace("/")
+            .auth(json!({ "password": "123" }))
+            .on("auth", |payload, _client| {
+                if let Payload::String(msg) = payload {
+                    println!("{}", msg);
+                }
+            })
+            .on_any(move |event, payload, _client| {
+                if let Payload::String(str) = payload {
+                    println!("{} {}", String::from(event.clone()), str);
+                }
+                tx.send(String::from(event)).unwrap();
+            })
+            .connect()?;
+
+        // Sleep to give server enough time to send 2 events
+        sleep(Duration::from_secs(2));
+
+        let event = rx.recv().unwrap();
+        assert_eq!(event, "message");
+        let event = rx.recv().unwrap();
+        assert_eq!(event, "test");
+
+        Ok(())
+    }
+
     #[test]
     fn socket_io_auth_builder_integration() -> Result<()> {
         let url = crate::test::socket_io_auth_server();
+        let nsp = String::from("/admin");
         let socket = ClientBuilder::new(url)
-            .namespace("/admin")
+            .namespace(nsp.clone())
             .auth(json!({ "password": "123" }))
-            .connect()?;
+            .connect_manual()?;
 
-        let (tx, rx) = mpsc::sync_channel(0);
+        let mut iter = socket
+            .iter()
+            .map(|packet| packet.unwrap())
+            .filter(|packet| packet.packet_type != PacketId::Connect);
 
-        // Send emit with ack after 1s, so socketio server has enough time to register it's listeners
-        sleep(Duration::from_secs(1));
+        let packet: Option<Packet> = iter.next();
+        assert!(packet.is_some());
 
-        assert!(socket
-            .emit_with_ack(
-                "test",
-                json!({ "msg": "1" }),
-                Duration::from_secs(1),
-                move |payload, _| {
-                    println!("Got ack");
-                    tx.send(Payload::from(json!(["456"])) == payload).unwrap();
-                }
-            )
-            .is_ok());
+        let packet = packet.unwrap();
 
-        let received = rx.recv();
-        assert!(received.is_ok() && received.unwrap());
+        assert_eq!(
+            packet,
+            Packet::new(
+                PacketId::Event,
+                nsp.clone(),
+                Some("[\"auth\",\"success\"]".to_owned()),
+                None,
+                0,
+                None
+            )
+        );
 
         Ok(())
     }
