tests: Improve test bed

.github/workflows/github-ci.yml

@@ -18,7 +18,9 @@ jobs:
             toolchain: stable
             components: rustfmt
         - run: cargo fmt --all -- --check
-
+          env:
+            RUSTFLAGS: --cfg=socketioxide_test
+
   test:
     runs-on: ubuntu-latest
 
@@ -37,7 +39,8 @@ jobs:
             target/
           key: ${{ runner.os }}-cargo-${{ hashFiles('**/Cargo.lock') }}
       - run: cargo test --tests --all-features --workspace
-
+        env:
+          RUSTFLAGS: --cfg=socketioxide_test
   udeps:
     runs-on: ubuntu-latest
     steps:
@@ -87,6 +90,8 @@ jobs:
 
       - name: check crates
         run: cargo check -p socketioxide -p engineioxide --all-features
+        env:
+          RUSTFLAGS: --cfg=socketioxide_test
 
   feature_set:
     runs-on: ubuntu-latest
@@ -111,7 +116,25 @@ jobs:
           key: ${{ runner.os }}-cargo-${{ hashFiles('**/Cargo.lock') }}
 
       - name: check --feature-powerset
-        run: cargo hack check --feature-powerset --no-dev-deps --skip test-utils -p socketioxide -p engineioxide
+        run: cargo hack check --feature-powerset --no-dev-deps -p socketioxide -p engineioxide
+
+  examples:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: dtolnay/rust-toolchain@master
+        with:
+          toolchain: stable
+      - uses: actions/cache@v4
+        with:
+          path: |
+            ~/.cargo/bin/
+            ~/.cargo/registry/index/
+            ~/.cargo/registry/cache/
+            ~/.cargo/git/db/
+            examples/target/
+          key: ${{ runner.os }}-cargo-${{ hashFiles('**/Cargo.lock') }}-examples
+      - run: cd examples && cargo check --all-features
 
   doctest:
     runs-on: ubuntu-latest
@@ -165,6 +188,8 @@ jobs:
           --all-features
           --message-format=json | clippy-sarif | tee rust-clippy-results.sarif | sarif-fmt
         continue-on-error: true
+        env:
+          RUSTFLAGS: --cfg=socketioxide_test
 
       - name: Upload analysis results to GitHub
         uses: github/codeql-action/upload-sarif@v3

CONTRIBUTING.md

@@ -139,7 +139,11 @@ You will need [rustc and cargo](www.rust-lang.org/tools/install).
     ```shell
     git clone https://github.com/totodore/socketioxide
     ```
+2. To test socketioxide don't forget to enable the flag `socketioxide-test` through the `RUSTFLAGS` environment variable:
 
+    ```shell
+    export RUSTFLAGS="--cfg socketioxide-test"
+    ```
 2. Depending on what you want to change, clone the [socketio/engine.io-protocol](https://github.com/socketio/engine.io-protocol) repo or the [socketio/socket.io-protocol](https://github.com/socketio/socket.io-protocol) repo or both
     ```shell
     git clone https://github.com/socketio/engine.io-protocol

Cargo.toml

@@ -14,7 +14,7 @@ categories = [
 license = "MIT"
 
 [workspace]
-members = ["engineioxide", "socketioxide", "e2e/*", "examples/*"]
+members = ["engineioxide", "socketioxide", "e2e/*"]
 default-members = ["engineioxide", "socketioxide"]
 resolver = "2"
 
@@ -40,4 +40,3 @@ tracing-subscriber = { version = "0.3.17", features = ["env-filter"] }
 criterion = { version = "0.5.1", features = ["html_reports"] }
 axum = "0.7.2"
 salvo = { version = "0.66.0", features = ["tower-compat"] }
-rust_socketio = { version = "0.4.2", features = ["async"] }

engineioxide/Cargo.toml

@@ -54,7 +54,6 @@ hyper-util = { workspace = true, features = ["tokio", "client-legacy"] }
 
 [features]
 v3 = ["memchr", "unicode-segmentation"]
-test-utils = []
 tracing = ["dep:tracing"]
 
 [[bench]]

engineioxide/src/lib.rs

@@ -35,7 +35,7 @@
 pub use service::{ProtocolVersion, TransportType};
 pub use socket::{DisconnectReason, Socket};
 
-#[cfg(feature = "test-utils")]
+#[cfg(any(test, socketioxide_test))]
 pub use packet::*;
 
 pub mod config;

engineioxide/src/socket.rs

@@ -467,7 +467,7 @@ impl<D: Default + Send + Sync + 'static> std::fmt::Debug for Socket<D> {
     }
 }
 
-#[cfg(feature = "test-utils")]
+#[cfg(socketioxide_test)]
 impl<D> Drop for Socket<D>
 where
     D: Default + Send + Sync + 'static,
@@ -478,7 +478,7 @@ where
     }
 }
 
-#[cfg(feature = "test-utils")]
+#[cfg(any(socketioxide_test, test))]
 impl<D> Socket<D>
 where
     D: Default + Send + Sync + 'static,
@@ -487,11 +487,21 @@ where
     pub fn new_dummy(
         sid: Sid,
         close_fn: Box<dyn Fn(Sid, DisconnectReason) + Send + Sync>,
-    ) -> Socket<D> {
-        let (internal_tx, internal_rx) = mpsc::channel(200);
+    ) -> Arc<Socket<D>> {
+        Socket::new_dummy_piped(sid, close_fn, 1024).0
+    }
+
+    /// Create a dummy socket for testing purpose with a
+    /// receiver to get the packets sent to the client
+    pub fn new_dummy_piped(
+        sid: Sid,
+        close_fn: Box<dyn Fn(Sid, DisconnectReason) + Send + Sync>,
+        buffer_size: usize,
+    ) -> (Arc<Socket<D>>, tokio::sync::mpsc::Receiver<Packet>) {
+        let (internal_tx, internal_rx) = mpsc::channel(buffer_size);
         let (heartbeat_tx, heartbeat_rx) = mpsc::channel(1);
 
-        Self {
+        let sock = Self {
             id: sid,
             protocol: ProtocolVersion::V4,
             transport: AtomicU8::new(TransportType::Websocket as u8),
@@ -509,6 +519,20 @@ where
 
             #[cfg(feature = "v3")]
             supports_binary: true,
-        }
+        };
+        let sock = Arc::new(sock);
+
+        let (tx, rx) = mpsc::channel(buffer_size);
+        let sock_clone = sock.clone();
+        tokio::spawn(async move {
+            let mut internal_rx = sock_clone.internal_rx.try_lock().unwrap();
+            while let Some(packets) = internal_rx.recv().await {
+                for packet in packets {
+                    tx.send(packet).await.unwrap();
+                }
+            }
+        });
+
+        (sock, rx)
     }
 }

examples/Cargo.toml

@@ -0,0 +1,16 @@
+[workspace]
+members = ["*"]
+exclude = ["target"]
+resolver = "2"
+
+[workspace.dependencies]
+futures = "0.3.27"
+tokio = "1.35.0"
+serde = { version = "1.0.193", features = ["derive"] }
+serde_json = "1.0.108"
+tower = { version = "0.4.13", default-features = false }
+tracing = "0.1.37"
+tracing-subscriber = "0.3.18"
+axum = "0.7.5"
+hyper-util.version = "0.1.1"
+hyper = { version = "1.0.1", features = ["http1", "http2", "server"] }

examples/salvo-echo/Cargo.toml

@@ -8,13 +8,13 @@ rust-version = "1.67" # required by salvo
 
 [dependencies]
 socketioxide = { path = "../../socketioxide", features = ["tracing"] }
-salvo.workspace = true
 tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
 tracing-subscriber.workspace = true
 tracing.workspace = true
 serde_json.workspace = true
 tower.workspace = true
 tower-http = { version = "0.5.0", features = ["cors"] }
+salvo = { version = "0.66", features = ["tower-compat"] }
 
 [[bin]]
 name = "salvo-echo"

socketioxide/Cargo.toml

@@ -38,19 +38,13 @@ state = { version = "0.6.0", optional = true }
 
 [features]
 v4 = ["engineioxide/v3"]
-test-utils = []
 tracing = ["dep:tracing", "engineioxide/tracing"]
 extensions = ["dep:dashmap"]
 state = ["dep:state"]
 
 [dev-dependencies]
-engineioxide = { path = "../engineioxide", features = [
-    "v3",
-    "tracing",
-    "test-utils",
-] }
+engineioxide = { path = "../engineioxide", features = ["v3", "tracing"] }
 tokio-tungstenite.workspace = true
-rust_socketio.workspace = true
 axum.workspace = true
 salvo.workspace = true
 tokio = { workspace = true, features = [

socketioxide/src/client.rs

@@ -140,6 +140,54 @@ impl<A: Adapter> Client<A> {
         #[cfg(feature = "tracing")]
         tracing::debug!("all namespaces closed");
     }
+
+    #[cfg(socketioxide_test)]
+    pub async fn new_dummy_sock(
+        self: Arc<Self>,
+        ns: &'static str,
+        auth: impl serde::Serialize,
+    ) -> (
+        tokio::sync::mpsc::Sender<engineioxide::Packet>,
+        tokio::sync::mpsc::Receiver<engineioxide::Packet>,
+    ) {
+        let buffer_size = self.config.engine_config.max_buffer_size;
+        let sid = Sid::new();
+        let (esock, rx) = EIoSocket::new_dummy_piped(sid, Box::new(|_, _| {}), buffer_size);
+
+        let (tx1, mut rx1) = tokio::sync::mpsc::channel(buffer_size);
+        tokio::spawn({
+            let esock = esock.clone();
+            let client = self.clone();
+            async move {
+                while let Some(packet) = rx1.recv().await {
+                    match packet {
+                        engineioxide::Packet::Message(msg) => {
+                            client.on_message(msg, esock.clone());
+                        }
+                        engineioxide::Packet::Close => {
+                            client
+                                .on_disconnect(esock.clone(), EIoDisconnectReason::TransportClose);
+                        }
+                        engineioxide::Packet::Binary(bin) => {
+                            client.on_binary(bin, esock.clone());
+                        }
+                        _ => {}
+                    }
+                }
+            }
+        });
+        let p = Packet {
+            ns: ns.into(),
+            inner: PacketData::Connect(Some(serde_json::to_string(&auth).unwrap())),
+        }
+        .into();
+        self.on_message(p, esock.clone());
+
+        // wait for the socket to be connected to the namespace
+        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+
+        (tx1, rx)
+    }
 }
 
 #[derive(Debug, Default)]
@@ -290,6 +338,7 @@ fn apply_payload_on_packet(data: Vec<u8>, socket: &EIoSocket<SocketData>) -> boo
 
 #[cfg(test)]
 mod test {
+    use super::*;
     use tokio::sync::mpsc;
 
     use crate::adapter::LocalAdapter;
@@ -305,13 +354,12 @@ mod test {
         client
     }
 
-    use super::*;
     #[tokio::test]
     async fn connect_timeout_fail() {
         let client = create_client();
         let (tx, mut rx) = mpsc::channel(1);
         let close_fn = Box::new(move |_, _| tx.try_send(()).unwrap());
-        let sock = Arc::new(EIoSocket::new_dummy(Sid::new(), close_fn));
+        let sock = EIoSocket::new_dummy(Sid::new(), close_fn);
         client.on_connect(sock.clone());
         tokio::time::timeout(CONNECT_TIMEOUT * 2, rx.recv())
             .await
@@ -324,7 +372,7 @@ mod test {
         let client = create_client();
         let (tx, mut rx) = mpsc::channel(1);
         let close_fn = Box::new(move |_, _| tx.try_send(()).unwrap());
-        let sock = Arc::new(EIoSocket::new_dummy(Sid::new(), close_fn));
+        let sock = EIoSocket::new_dummy(Sid::new(), close_fn);
         client.on_connect(sock.clone());
         client.on_message("0".into(), sock.clone());
         tokio::time::timeout(CONNECT_TIMEOUT * 2, rx.recv())

socketioxide/src/io.rs

@@ -808,6 +808,22 @@ impl<A: Adapter> Clone for SocketIo<A> {
     }
 }
 
+#[cfg(any(test, socketioxide_test))]
+impl<A: Adapter> SocketIo<A> {
+    /// Create a dummy socket for testing purpose with a
+    /// receiver to get the packets sent to the client
+    pub async fn new_dummy_sock(
+        &self,
+        ns: &'static str,
+        auth: impl serde::Serialize,
+    ) -> (
+        tokio::sync::mpsc::Sender<engineioxide::Packet>,
+        tokio::sync::mpsc::Receiver<engineioxide::Packet>,
+    ) {
+        self.0.clone().new_dummy_sock(ns, auth).await
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -840,8 +856,7 @@ mod tests {
         let sid = Sid::new();
         let (_, io) = SocketIo::builder().build_svc();
         io.ns("/", || {});
-
-        let socket = Socket::new_dummy(sid, Box::new(|_, _| {})).into();
+        let socket = Socket::new_dummy(sid, Box::new(|_, _| {}));
         let config = SocketIoConfig::default().into();
         io.0.get_ns("/")
             .unwrap()

socketioxide/src/ns.rs

@@ -65,6 +65,9 @@ impl<A: Adapter> Namespace<A> {
         }
 
         self.sockets.write().unwrap().insert(sid, socket.clone());
+        #[cfg(feature = "tracing")]
+        tracing::trace!(?socket.id, ?self.path, "socket added to namespace");
+
         let protocol = esocket.protocol.into();
 
         if let Err(_e) = socket.send(Packet::connect(&self.path, socket.id, protocol)) {
@@ -129,7 +132,7 @@ impl<A: Adapter> Namespace<A> {
     }
 }
 
-#[cfg(test)]
+#[cfg(any(test, socketioxide_test))]
 impl<A: Adapter> Namespace<A> {
     pub fn new_dummy<const S: usize>(sockets: [Sid; S]) -> Arc<Self> {
         let ns = Namespace::new(Cow::Borrowed("/"), || {});

socketioxide/src/socket.rs

@@ -809,14 +809,15 @@ impl<A: Adapter> PartialEq for Socket<A> {
     }
 }
 
-#[cfg(test)]
+#[cfg(any(test, socketioxide_test))]
 impl<A: Adapter> Socket<A> {
+    /// Creates a dummy socket for testing purposes
     pub fn new_dummy(sid: Sid, ns: Arc<Namespace<A>>) -> Socket<A> {
         let close_fn = Box::new(move |_, _| ());
         let s = Socket::new(
             sid,
             ns,
-            engineioxide::Socket::new_dummy(sid, close_fn).into(),
+            engineioxide::Socket::new_dummy(sid, close_fn),
             Arc::new(SocketIoConfig::default()),
         );
         s.set_connected(true);
@@ -834,7 +835,7 @@ mod test {
         let ns = Namespace::<LocalAdapter>::new_dummy([sid]).into();
         let socket: Arc<Socket> = Socket::new_dummy(sid, ns).into();
         // Saturate the channel
-        for _ in 0..200 {
+        for _ in 0..1024 {
             socket
                 .send(Packet::event("test", "test", Value::Null))
                 .unwrap();