From 00474821c9f431c2006c3bfff4750380c3b2cf49 Mon Sep 17 00:00:00 2001 From: malkerdi Date: Mon, 12 Oct 2026 19:08:30 +0000 Subject: [PATCH] feat(socket): expose is_closed and run disconnect handlers when the connect packet fails --- crates/socketioxide/src/ns.rs | 4 +- crates/socketioxide/src/socket.rs | 11 +++++ crates/socketioxide/tests/connect.rs | 63 +++++++++++++++++++++++++++- 3 files changed, 76 insertions(+), 2 deletions(-) diff --git a/crates/socketioxide/src/ns.rs b/crates/socketioxide/src/ns.rs index 5ea4fc9c..4a493c09 100644 --- a/crates/socketioxide/src/ns.rs +++ b/crates/socketioxide/src/ns.rs @@ -138,7 +138,9 @@ impl Namespace { #[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); } diff --git a/crates/socketioxide/src/socket.rs b/crates/socketioxide/src/socket.rs index 57a3ce88..390cac3c 100644 --- a/crates/socketioxide/src/socket.rs +++ b/crates/socketioxide/src/socket.rs @@ -607,6 +607,17 @@ impl Socket { 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")] diff --git a/crates/socketioxide/tests/connect.rs b/crates/socketioxide/tests/connect.rs index 9e40558f..da9f3b83 100644 --- a/crates/socketioxide/tests/connect.rs +++ b/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::(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::(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); +}