feat(socket): expose is_closed and run disconnect handlers when the connect packet fails

crates/socketioxide/src/ns.rs

@@ -138,7 +138,9 @@ impl<A: Adapter> Namespace<A> {
             #[cfg(feature = "tracing")]
             tracing::debug!(%sid, "error sending connect packet: {_e}, closing conn");
 
-            self.remove_socket(sid);
+            // Through `Socket::close` rather than `remove_socket`, so a disconnect handler
+            // registered by a middleware still runs for a client that dropped mid-connect.
+            socket.close(DisconnectReason::TransportError);
             esocket.close(engineioxide::DisconnectReason::TransportError);
             return Err(ConnectFail);
         }

crates/socketioxide/src/socket.rs

@@ -607,6 +607,17 @@ impl<A: Adapter> Socket<A> {
         self.connected.load(Ordering::SeqCst)
     }
 
+    /// # Return true if the underlying engine.io connection is closed.
+    ///
+    /// Unlike [`connected`](Self::connected), this is meaningful inside a
+    /// [connect middleware](crate::handler::connect#middlewares): it tells whether the client
+    /// dropped the connection while the middleware was running, for example during an
+    /// authentication lookup, so the middleware can stop early instead of accepting a socket
+    /// that can no longer be reached.
+    pub fn is_closed(&self) -> bool {
+        self.esocket.is_closed()
+    }
+
     // Socket operators
 
     #[doc = include_str!("../docs/operators/to.md")]

crates/socketioxide/tests/connect.rs

@@ -1,13 +1,19 @@
+mod fixture;
 mod utils;
 
 use bytes::Bytes;
 use engineioxide::Packet::*;
 use serde::Serialize;
-use socketioxide::{SendError, SocketError, SocketIo, extract::SocketRef, handler::ConnectHandler};
+use socketioxide::{
+    SendError, SocketError, SocketIo, extract::SocketRef, handler::ConnectHandler,
+    socket::DisconnectReason,
+};
 use socketioxide_core::{Value, packet::Packet, parser::Parse};
 use socketioxide_parser_common::CommonParser;
 use tokio::sync::mpsc;
 
+use fixture::{create_server, create_ws_connection};
+
 fn create_msg(ns: &'static str, event: &str, data: impl Serialize) -> engineioxide::Packet {
     let packet = Packet::event(ns, CommonParser.encode_value(&data, Some(event)).unwrap());
     match CommonParser.encode(packet) {
@@ -230,3 +236,58 @@ async fn remove_ns_from_disconnect_handler() {
     let elapsed = tokio::time::timeout(std::time::Duration::from_millis(200), rx.recv()).await;
     assert!(elapsed.is_err() || elapsed.unwrap().is_none());
 }
+
+#[tokio::test]
+async fn is_closed_reports_a_client_dropped_during_the_middleware() {
+    let (svc, io) = create_server().await;
+    let (tx, mut rx) = mpsc::channel::<bool>(2);
+    let client_dropped = std::sync::Arc::new(tokio::sync::Notify::new());
+
+    io.ns(
+        "/",
+        { async || () }.with({
+            let client_dropped = client_dropped.clone();
+            async move |s: SocketRef| {
+                tx.try_send(s.is_closed()).unwrap();
+                // The test drops the client while the middleware is still running.
+                client_dropped.notified().await;
+                tx.try_send(s.is_closed()).unwrap();
+                Ok::<_, std::convert::Infallible>(())
+            }
+        }),
+    );
+
+    let ws = create_ws_connection(&svc).await;
+    assert!(!timeout_rcv(&mut rx).await);
+    drop(ws);
+    // Let the engine notice the closed transport and close the session.
+    tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+    client_dropped.notify_one();
+    assert!(timeout_rcv(&mut rx).await);
+}
+
+#[tokio::test]
+async fn disconnect_handler_runs_when_the_client_drops_before_the_connect_packet() {
+    let (svc, io) = create_server().await;
+    let (tx, mut rx) = mpsc::channel::<DisconnectReason>(1);
+    let client_dropped = std::sync::Arc::new(tokio::sync::Notify::new());
+
+    io.ns(
+        "/",
+        { async || () }.with({
+            let client_dropped = client_dropped.clone();
+            async move |s: SocketRef| {
+                let tx = tx.clone();
+                s.on_disconnect(async move |reason: DisconnectReason| tx.try_send(reason).unwrap());
+                client_dropped.notified().await;
+                Ok::<_, std::convert::Infallible>(())
+            }
+        }),
+    );
+
+    let ws = create_ws_connection(&svc).await;
+    drop(ws);
+    tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+    client_dropped.notify_one();
+    assert_eq!(timeout_rcv(&mut rx).await, DisconnectReason::TransportError);
+}