Repository: 1c3t3a/rust-socketio (language: rust)

Issue:
Async client is not notified of (unexpected) websocket transport close
## Problem

Using the async client without reconnect, when the transport websocket closes properly but without an engineio message indicating the close, the client doesn't fire an event and as such the user of the client doesn't notice the closure.

This has the problem that we can't easily react to the closure and keep thinking that we are connected.

## Steps to reproduce

1. Run a server like this:

    ```javascript
    const server = require('http').createServer();
    const io = require('socket.io')(server);

    console.log('Started');
    var callback = client => {
        console.log('Connected!');

        // Close underlying websocket connection
        client.client.conn.close();
    };
    io.on('connection', callback);
    // the socket.io client runs on port 4201
    server.listen(4200);
    ```

1. Connect with a Rust client like the following. Make sure to set `reconnect` to false. To get some more logging, you can run `export RUST_LOG="info,tungstenite::protocol=trace,tungstenite::protocol::frame::frame=debug,rust_socketio=trace,rust_engineio=trace"` first.

    ```rust
    use futures_util::FutureExt;
    use log::{error, info};
    use rust_socketio::{asynchronous::ClientBuilder, Event, TransportType,};
    use tokio::signal::{self,unix::{signal, SignalKind}};

    #[tokio::main]
    async fn main() {
        env_logger::Builder::new().parse_default_env().init();

        info!("Starting");

        // get a socket that is connected to the admin namespace
        let socket = ClientBuilder::new("http://localhost:4200/")
            .transport_type(TransportType::Websocket)
            .reconnect(false)
            .on(Event::Connect, |_payload, _client| {{ async move { info!("Connected!"); }.boxed() }})
            .on(Event::Close, |_payload, _client| {{ async move { info!("Connection closed!"); }.boxed() }})
            .on(Event::Error, |err, _| { async move { error!("Error: {:#?}", err) }.boxed() })
            .connect()
            .await
            .expect("Connection failed");

        let mut sigterm = signal(SignalKind::terminate()).expect("unable to create sigterm signal");
        tokio::select! {
            _ = signal::ctrl_c() => { info!("Received SIGINT, shutting down."); },
            _ = sigterm.recv() => { info!("Received SIGTERM, shutting down."); },
        }

        socket.disconnect().await.expect("Disconnect failed");
    }
    ```

1. Notice that the tungstenite websocket closes by receiving an sending an `opecode: CLOSE` frame, but the client isn't notified.
    
## Reason

When the async client connects to the server, it spawns a new thread to handle the arriving messages asynchronously, which is immediately detached with no chance of awaiting its completion.

This may happen if the underlying tungstenite websocket shuts down ('Received close frame: None') but there are no engineio/socketio close frames. Hence, since the stream terminates, the message handling task stops without a Close or Error event being fired.

https://github.com/1c3t3a/rust-socketio/blob/64d2a133560aa0b31b2263a820645f982e7276b3/socketio/src/asynchronous/client/client.rs#L151-L159

This was kind of introduced by me in a9a81e9 when I added handling for the stream closure to avoid the panic. As it says in the commit message *However, the program which uses the socket.io client currently isn't notified of the unexpected closure.*, so time to clean that up :)

## Potential Solution

Looking at the official JavaScript [`disconnect` specification](https://socket.io/docs/v4/client-socket-instance/#disconnect) there are reasons specified for the disconnection which include a `transport close`.

Thus, once we disconnect for an unknown reason (e.g. the transport closed), we could fire a `CLOSE` event to signal the closure to the user.

Alternatively, we could also fire this event only after the loop terminates, before the thread itself terminates.

See #449 for an implementation of that and feel free to discuss the proposed changes.


Discussion / hints:

Fix patch:
diff --git a/ci/socket-io.js b/ci/socket-io.js
index 37649712..8f609b33 100644
--- a/ci/socket-io.js
+++ b/ci/socket-io.js
@@ -36,6 +36,14 @@ var callback = client => {
             ack(Buffer.from([1, 2, 3]));
         }
     });
+
+    // This event allows the test framework to arbitrarily close the underlying connection
+    client.on('close_transport', data => {
+        console.log(['close_transport', 'Request to close transport received'])
+        // Close underlying websocket connection
+        client.client.conn.close();
+    })
+
     client.emit('Hello from the message event!');
     client.emit('test', 'Hello from the test event!');
     client.emit(Buffer.from([4, 5, 6]));
diff --git a/socketio/src/asynchronous/client/client.rs b/socketio/src/asynchronous/client/client.rs
index 67feb7db..da1d5f09 100644
--- a/socketio/src/asynchronous/client/client.rs
+++ b/socketio/src/asynchronous/client/client.rs
@@ -2,7 +2,7 @@ use std::{ops::DerefMut, pin::Pin, sync::Arc};
 
 use backoff::{backoff::Backoff, ExponentialBackoffBuilder};
 use futures_util::{future::BoxFuture, stream, Stream, StreamExt};
-use log::trace;
+use log::{error, trace};
 use rand::{thread_rng, Rng};
 use serde_json::Value;
 use tokio::{
@@ -19,7 +19,7 @@ use crate::{
     asynchronous::socket::Socket as InnerSocket,
     error::{Error, Result},
     packet::{Packet, PacketId},
-    Event, Payload,
+    CloseReason, Event, Payload,
 };
 
 #[derive(Default)]
@@ -162,7 +162,21 @@ impl Client {
                 drop(stream);
 
                 let should_reconnect = match *(client_clone.disconnect_reason.read().await) {
-                    DisconnectReason::Unknown => reconnect,
+                    DisconnectReason::Unknown => {
+                        // If we disconnected for an unknown reason, the client might not have noticed
+                        // the closure yet. Hence, fire a transport close event to notify it.
+                        // We don't need to do that in the other cases, since proper server close
+                        // and manual client close are handled explicitly.
+                        if let Some(err) = client_clone
+                            .callback(&Event::Close, CloseReason::TransportClose.as_str())
+                            .await
+                            .err()
+                        {
+                            error!("Error while notifying client of transport close: {err}")
+                        }
+
+                        reconnect
+                    }
                     DisconnectReason::Manual => false,
                     DisconnectReason::Server => reconnect_on_disconnect,
                 };
@@ -510,7 +524,8 @@ impl Client {
                 }
                 PacketId::Disconnect => {
                     *(self.disconnect_reason.write().await) = DisconnectReason::Server;
-                    self.callback(&Event::Close, "").await?;
+                    self.callback(&Event::Close, CloseReason::IOServerDisconnect.as_str())
+                        .await?;
                 }
                 PacketId::ConnectError => {
                     self.callback(
@@ -590,7 +605,7 @@ mod test {
         },
         error::Result,
         packet::{Packet, PacketId},
-        Payload, TransportType,
+        CloseReason, Event, Payload, TransportType,
     };
 
     #[tokio::test]
@@ -926,6 +941,50 @@ mod test {
         Ok(())
     }
 
+    #[tokio::test]
+    async fn socket_io_transport_close() -> Result<()> {
+        let url = crate::test::socket_io_server();
+
+        let (tx, mut rx) = mpsc::channel(1);
+
+        let notify = Arc::new(tokio::sync::Notify::new());
+        let notify_clone = notify.clone();
+
+        let socket = ClientBuilder::new(url)
+            .on(Event::Connect, move |_, _| {
+                let cl = notify_clone.clone();
+                async move {
+                    cl.notify_one();
+                }
+                .boxed()
+            })
+            .on(Event::Close, move |payload, _| {
+                let clone_tx = tx.clone();
+                async move { clone_tx.send(payload).await.unwrap() }.boxed()
+            })
+            .connect()
+            .await?;
+
+        // Wait until socket is connected
+        let connect_timeout = timeout(Duration::from_secs(1), notify.notified()).await;
+        assert!(connect_timeout.is_ok());
+
+        // Instruct server to close transport
+        let result = socket.emit("close_transport", Payload::from("")).await;
+        assert!(result.is_ok());
+
+        // Wait for Event::Close
+        let rx_timeout = timeout(Duration::from_secs(1), rx.recv()).await;
+        assert!(rx_timeout.is_ok());
+
+        assert_eq!(
+            rx_timeout.unwrap(),
+            Some(Payload::from(CloseReason::TransportClose.as_str()))
+        );
+
+        Ok(())
+    }
+
     #[tokio::test]
     async fn socketio_polling_integration() -> Result<()> {
         let url = crate::test::socket_io_server();
diff --git a/socketio/src/client/raw_client.rs b/socketio/src/client/raw_client.rs
index 0686683f..865a9ec6 100644
--- a/socketio/src/client/raw_client.rs
+++ b/socketio/src/client/raw_client.rs
@@ -1,7 +1,7 @@
 use super::callback::Callback;
 use crate::packet::{Packet, PacketId};
 use crate::Error;
-pub(crate) use crate::{event::Event, payload::Payload};
+pub(crate) use crate::{event::CloseReason, event::Event, payload::Payload};
 use rand::{thread_rng, Rng};
 use serde_json::Value;
 
@@ -149,7 +149,7 @@ impl RawClient {
         let _ = self.socket.send(disconnect_packet);
         self.socket.disconnect()?;
 
-        let _ = self.callback(&Event::Close, ""); // trigger on_close
+        let _ = self.callback(&Event::Close, CloseReason::IOClientDisconnect.as_str()); // trigger on_close
         Ok(())
     }
 
@@ -372,7 +372,7 @@ impl RawClient {
                     self.callback(&Event::Connect, "")?;
                 }
                 PacketId::Disconnect => {
-                    self.callback(&Event::Close, "")?;
+                    self.callback(&Event::Close, CloseReason::IOServerDisconnect.as_str())?;
                 }
                 PacketId::ConnectError => {
                     self.callback(
diff --git a/socketio/src/event.rs b/socketio/src/event.rs
index 2d2ed1f9..ef1c65a8 100644
--- a/socketio/src/event.rs
+++ b/socketio/src/event.rs
@@ -57,3 +57,38 @@ impl Display for Event {
         f.write_str(self.as_str())
     }
 }
+
+/// A `CloseReason` is the payload of the [`Event::Close`] and specifies the reason for
+/// why it was fired.
+/// These are aligned with the official Socket.IO disconnect reasons, see
+/// https://socket.io/docs/v4/client-socket-instance/#disconnect
+#[derive(Debug, PartialEq, PartialOrd, Clone, Eq, Hash)]
+pub enum CloseReason {
+    IOServerDisconnect,
+    IOClientDisconnect,
+    TransportClose,
+}
+
+impl CloseReason {
+    pub fn as_str(&self) -> &str {
+        match self {
+            // Inspired by https://github.com/socketio/socket.io/blob/d0fc72042068e7eaef448941add617f05e1ec236/packages/socket.io-client/lib/socket.ts#L865
+            CloseReason::IOServerDisconnect => "io server disconnect",
+            // Inspired by https://github.com/socketio/socket.io/blob/d0fc72042068e7eaef448941add617f05e1ec236/packages/socket.io-client/lib/socket.ts#L911
+            CloseReason::IOClientDisconnect => "io client disconnect",
+            CloseReason::TransportClose => "transport close",
+        }
+    }
+}
+
+impl From<CloseReason> for String {
+    fn from(event: CloseReason) -> Self {
+        Self::from(event.as_str())
+    }
+}
+
+impl Display for CloseReason {
+    fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult {
+        f.write_str(self.as_str())
+    }
+}
diff --git a/socketio/src/lib.rs b/socketio/src/lib.rs
index b913eb4d..1445cc82 100644
--- a/socketio/src/lib.rs
+++ b/socketio/src/lib.rs
@@ -191,7 +191,7 @@ pub mod asynchronous;
 
 pub use error::Error;
 
-pub use {event::Event, payload::Payload};
+pub use {event::CloseReason, event::Event, payload::Payload};
 
 pub use client::{ClientBuilder, RawClient, TransportType};
 


Test patch: