Async client is not notified of (unexpected) websocket transport close

[sirkrypt0]

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:

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

2. 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.

   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");
   }

3. 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.

rust-socketio/socketio/src/asynchronous/client/client.rs

Lines 151 to 159 in 64d2a13

	tokio::runtime::Handle::current().spawn(async move {
	loop {
	let mut stream = client_clone.as_stream().await;
	// Consume the stream until it returns None and the stream is closed.
	while let Some(item) = stream.next().await {
	if let Err(e) = item {
	trace!("Network error occurred: {}", e);
	}
	}

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 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.