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

Issue:
Asynchronous Client: Cannot use await in callbacks
Hi! Thanks for spending your time working on this library. It's great to see support for an asynchronous client.

## Problem Description

Running the given example for the asynchronous client only works when *not* having an `await` statement within the callback. Code after the first `await` in the callback is not executed.
Thus, calling async functions is not possible.

When debugging, I found out that this is caused by the `Stream` implementation for the async client.

##  Steps To Reproduce

Run this Python server:

```python
import socketio
import eventlet

sio = socketio.Server(async_mode='eventlet')
app = socketio.WSGIApp(sio)

@sio.event
def connect(sid, environ):
    print('Client connected:', sid)

@sio.event
def disconnect(sid):
    print('Client disconnected:', sid)

@sio.event
def foo(sid, data):
    print('foo event received:', data)

if __name__ == '__main__':
    eventlet.wsgi.server(eventlet.listen(('localhost', 4200)), app)
```

Connect with this Rust client and see that there is no output (note that "Connected!" will not be printed):

```rust
use futures_util::FutureExt;
use log::info;
use rust_socketio::{asynchronous::ClientBuilder, Event, TransportType};
use std::time::Duration;
use tokio::{
    signal::{
        self,
        unix::{signal, SignalKind},
    },
    time::sleep,
};

#[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)
        .on(Event::Connect, |_payload, _client| {
            {
                async move {
                    sleep(Duration::from_secs(1)).await;
                    info!("Connected!");
                }
                .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");
}
```

Here's the corresponding Cargo.toml:

```toml
[package]
name = "client"
version = "0.1.0"
edition = "2021"

[dependencies]
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"
rust_socketio = { version = "0.4.1-alpha.1", features = ["async"] }
env_logger = { version = "0.10" }
log = "0.4"
tokio = { version = "1", features = ["macros", "rt", "signal", "time"] }
futures-util = "0.3"
```

It seems like the callback is aborted on the first `await`.

## Possible Solution

Debugging around, I found that in the client's stream implementation, the `ready!` macro is used. This macro only returns the results of futures if they return immediately:

https://github.com/1c3t3a/rust-socketio/blob/10d5a4633c4e88cea2ae795de85daf161193f9fe/socketio/src/asynchronous/client/client.rs#L410-L441

One possible solution I found was to use `futures_util::stream::unfold` in a new function on the `Client` instead of implementing stream (something like that):

```rust
pub(crate) fn stream(&mut self) -> impl Stream<Item = Result<Packet>> + '_ {
    stream::unfold(self.socket.clone(), | mut socket | async {
          let packet: Option<std::result::Result<Packet, Error>> = socket.clone().next().await;
          match packet {
              None => None,
              Some(Err(err)) => {
                  let res = self.callback(&Event::Error, err.to_string()).await;
                  Some((Err(err), socket))
              },
              Some(Ok(packet)) => {
                  self.handle_socketio_packet(&packet).await;
                  Some((Ok(packet), socket))
              }
          }
      })
}
```

I created #314 as potential fix.




Discussion / hints:

Fix patch:
diff --git a/socketio/src/asynchronous/client/builder.rs b/socketio/src/asynchronous/client/builder.rs
index c25db585..ba85b8eb 100644
--- a/socketio/src/asynchronous/client/builder.rs
+++ b/socketio/src/asynchronous/client/builder.rs
@@ -1,5 +1,3 @@
-use std::collections::HashMap;
-
 use futures_util::{future::BoxFuture, StreamExt};
 use log::trace;
 use native_tls::TlsConnector;
@@ -7,6 +5,7 @@ use rust_engineio::{
     asynchronous::ClientBuilder as EngineIoClientBuilder,
     header::{HeaderMap, HeaderValue},
 };
+use std::collections::HashMap;
 use url::Url;
 
 use crate::{error::Result, Error, Event, Payload, TransportType};
@@ -339,17 +338,17 @@ impl ClientBuilder {
     /// ```
     pub async fn connect(self) -> Result<Client> {
         let socket = self.connect_manual().await?;
-        let mut socket_clone = socket.clone();
+        let socket_clone = socket.clone();
 
         // Use thread to consume items in iterator in order to call callbacks
         tokio::runtime::Handle::current().spawn(async move {
+            let mut stream = socket_clone.as_stream();
             loop {
                 // tries to restart a poll cycle whenever a 'normal' error occurs,
                 // it just logs on network errors, in case the poll cycle returned
                 // `Result::Ok`, the server receives a close frame so it's safe to
                 // terminate
-                if let Some(e @ Err(Error::IncompleteResponseFromEngineIo(_))) =
-                    socket_clone.next().await
+                if let Some(e @ Err(Error::IncompleteResponseFromEngineIo(_))) = stream.next().await
                 {
                     trace!("Network error occured: {}", e.unwrap_err());
                 }
diff --git a/socketio/src/asynchronous/client/client.rs b/socketio/src/asynchronous/client/client.rs
index a6c0f771..cc7557ee 100644
--- a/socketio/src/asynchronous/client/client.rs
+++ b/socketio/src/asynchronous/client/client.rs
@@ -1,6 +1,6 @@
-use std::{collections::HashMap, ops::DerefMut, pin::Pin, sync::Arc, task::Poll};
+use std::{collections::HashMap, ops::DerefMut, pin::Pin, sync::Arc};
 
-use futures_util::{future::BoxFuture, ready, FutureExt, Stream, StreamExt};
+use futures_util::{future::BoxFuture, stream, Stream, StreamExt};
 use log::trace;
 use rand::{thread_rng, Rng};
 use serde_json::{from_str, Value};
@@ -405,51 +405,47 @@ impl Client {
         }
         Ok(())
     }
-}
 
-impl Stream for Client {
-    type Item = Result<Packet>;
-
-    fn poll_next(
-        mut self: Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-    ) -> std::task::Poll<Option<Self::Item>> {
-        loop {
-            // poll for the next payload
-            let next = ready!(self.socket.poll_next_unpin(cx));
-
-            match next {
-                None => {
-                    // end the stream if the underlying one is closed
-                    return Poll::Ready(None);
-                }
+    /// Returns the packet stream for the client.
+    pub(crate) fn as_stream<'a>(
+        &'a self,
+    ) -> Pin<Box<dyn Stream<Item = Result<Packet>> + Send + 'a>> {
+        stream::unfold(self.socket.clone(), |mut socket| async {
+            // wait for the next payload
+            let packet: Option<std::result::Result<Packet, Error>> = socket.next().await;
+            match packet {
+                // end the stream if the underlying one is closed
+                None => None,
                 Some(Err(err)) => {
                     // call the error callback
-                    ready!(Box::pin(self.callback(&Event::Error, err.to_string())).poll_unpin(cx))?;
-                    return Poll::Ready(Some(Err(err)));
-                }
-                Some(Ok(packet)) => {
-                    // if this packet is not meant for the current namespace, skip it an poll for the next one
-                    if packet.nsp == self.nsp {
-                        ready!(Box::pin(self.handle_socketio_packet(&packet)).poll_unpin(cx))?;
-                        return Poll::Ready(Some(Ok(packet)));
+                    match self.callback(&Event::Error, err.to_string()).await {
+                        Err(callback_err) => Some((Err(callback_err), socket)),
+                        Ok(_) => Some((Err(err), socket)),
                     }
                 }
+                Some(Ok(packet)) => match self.handle_socketio_packet(&packet).await {
+                    Err(callback_err) => Some((Err(callback_err), socket)),
+                    Ok(_) => Some((Ok(packet), socket)),
+                },
             }
-        }
+        })
+        .boxed()
     }
 }
 
 #[cfg(test)]
 mod test {
 
-    use std::time::Duration;
+    use std::{sync::Arc, time::Duration};
 
     use bytes::Bytes;
     use futures_util::{FutureExt, StreamExt};
     use native_tls::TlsConnector;
     use serde_json::json;
-    use tokio::{sync::mpsc, time::sleep};
+    use tokio::{
+        sync::mpsc,
+        time::{sleep, timeout},
+    };
 
     use crate::{
         asynchronous::client::{builder::ClientBuilder, client::Client},
@@ -516,6 +512,43 @@ mod test {
         Ok(())
     }
 
+    #[tokio::test]
+    async fn socket_io_async_callback() -> Result<()> {
+        // Test whether asynchronous callbacks are fully executed.
+        let url = crate::test::socket_io_server();
+
+        // This synchronization mechanism is used to let the test know that the end of the
+        // async callback was reached.
+        let notify = Arc::new(tokio::sync::Notify::new());
+        let notify_clone = notify.clone();
+
+        let socket = ClientBuilder::new(url)
+            .on("test", move |_, _| {
+                let cl = notify_clone.clone();
+                async move {
+                    sleep(Duration::from_secs(1)).await;
+                    // The async callback should be awaited and not aborted.
+                    // Thus, the notification should be called.
+                    cl.notify_one();
+                }
+                .boxed()
+            })
+            .connect()
+            .await?;
+
+        let payload = json!({"token": 123_i32});
+        let result = socket
+            .emit("test", Payload::String(payload.to_string()))
+            .await;
+
+        assert!(result.is_ok());
+        // If the timeout did not trigger, the async callback was fully executed.
+        let timeout = timeout(Duration::from_secs(5), notify.notified()).await;
+        assert!(timeout.is_ok());
+
+        Ok(())
+    }
+
     #[tokio::test]
     async fn socket_io_builder_integration() -> Result<()> {
         let url = crate::test::socket_io_server();
@@ -651,17 +684,18 @@ mod test {
     async fn socket_io_auth_builder_integration() -> Result<()> {
         let url = crate::test::socket_io_auth_server();
         let nsp = String::from("/admin");
-        let mut socket = ClientBuilder::new(url)
+        let socket = ClientBuilder::new(url)
             .namespace(nsp.clone())
             .auth(json!({ "password": "123" }))
             .connect_manual()
             .await?;
 
         // open packet
-        let _ = socket.next().await.unwrap()?;
+        let mut socket_stream = socket.as_stream();
+        let _ = socket_stream.next().await.unwrap()?;
 
         println!("Here12");
-        let packet = socket.next().await.unwrap()?;
+        let packet = socket_stream.next().await.unwrap()?;
         assert_eq!(
             packet,
             Packet::new(
@@ -717,11 +751,12 @@ mod test {
         test_socketio_socket(socket, "/".to_owned()).await
     }
 
-    async fn test_socketio_socket(mut socket: Client, nsp: String) -> Result<()> {
+    async fn test_socketio_socket(socket: Client, nsp: String) -> Result<()> {
         // open packet
-        let _: Option<Packet> = Some(socket.next().await.unwrap()?);
+        let mut socket_stream = socket.as_stream();
+        let _: Option<Packet> = Some(socket_stream.next().await.unwrap()?);
 
-        let packet: Option<Packet> = Some(socket.next().await.unwrap()?);
+        let packet: Option<Packet> = Some(socket_stream.next().await.unwrap()?);
 
         assert!(packet.is_some());
 
@@ -739,7 +774,7 @@ mod test {
             )
         );
 
-        let packet: Option<Packet> = Some(socket.next().await.unwrap()?);
+        let packet: Option<Packet> = Some(socket_stream.next().await.unwrap()?);
 
         assert!(packet.is_some());
 
@@ -756,7 +791,7 @@ mod test {
                 None
             )
         );
-        let packet: Option<Packet> = Some(socket.next().await.unwrap()?);
+        let packet: Option<Packet> = Some(socket_stream.next().await.unwrap()?);
 
         assert!(packet.is_some());
 
@@ -773,7 +808,7 @@ mod test {
             )
         );
 
-        let packet: Option<Packet> = Some(socket.next().await.unwrap()?);
+        let packet: Option<Packet> = Some(socket_stream.next().await.unwrap()?);
 
         assert!(packet.is_some());
 
@@ -811,7 +846,7 @@ mod test {
             .await
             .is_ok());
 
-        let packet: Option<Packet> = Some(socket.next().await.unwrap()?);
+        let packet: Option<Packet> = Some(socket_stream.next().await.unwrap()?);
 
         assert!(packet.is_some());
         let packet = packet.unwrap();
@@ -827,7 +862,7 @@ mod test {
             )
         );
 
-        let packet: Option<Packet> = Some(socket.next().await.unwrap()?);
+        let packet: Option<Packet> = Some(socket_stream.next().await.unwrap()?);
 
         assert!(packet.is_some());
         let packet = packet.unwrap();


Test patch: