feat(socketio): notify client of transport completion

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.

For long-running programs (e.g. a client program that never really disconnects
from the server) this can be problematic, as unexpected stream completions
would go unnoticed. 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.

Thus, we now fire an additional Event::Close when the stream terminates,
to signal the (potentially unexpected) close to the user.

Author: sirkrypt0

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]));

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::{
@@ -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, "transport close")
+                            .await
+                            .err()
+                        {
+                            error!("Error while notifying client of transport close: {err}")
+                        }
+
+                        reconnect
+                    }
                     DisconnectReason::Manual => false,
                     DisconnectReason::Server => reconnect_on_disconnect,
                 };
@@ -590,6 +604,7 @@ mod test {
         },
         error::Result,
         packet::{Packet, PacketId},
+        Event,
         Payload, TransportType,
     };
 
@@ -926,6 +941,47 @@ 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("transport close")));
+
+        Ok(())
+    }
+
     #[tokio::test]
     async fn socketio_polling_integration() -> Result<()> {
         let url = crate::test::socket_io_server();