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

Issue:
Asynchronous Client: "Error while parsing an incomplete packet socketio" on first heartbeat killing the connection
Hi! Thanks for spending your time working on this library. It's great to see support for an asynchronous client.

## Problem Description

Running the asynchronous client a bit longer (25 seconds) seems to crash it with the error "Error while parsing an incomplete packet socketio".
This also results in one CPU core being 100% utilized, suggesting that there is some busy waiting or similar.

While debugging this, I found that this appears to be the heartbeat which causes the crash.

## 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:

```rust
use futures_util::FutureExt;
use log::{error, info};
use rust_socketio::{
    asynchronous::{Client, ClientBuilder},
    Event, Payload, TransportType,
};
use serde_json::json;
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");

    // define a callback which is called when a payload is received
    // this callback gets the payload as well as an instance of the
    // socket to communicate with the server
    let callback = |payload: Payload, socket: Client| {
        async move {
            match payload {
                Payload::String(str) => println!("Received: {}", str),
                Payload::Binary(bin_data) => println!("Received bytes: {:#?}", bin_data),
            }
            socket
                .emit("test", json!({"got ack": true}))
                .await
                .expect("Server unreachable");
        }
        .boxed()
    };

    // 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 {
                    info!("Connected!");
                }
                .boxed()
            }
        })
        .on(Event::Close, |_payload, _client| {
            {
                async move {
                    info!("Connection closed!");
                }
                .boxed()
            }
        })
        .on(Event::Error, |err, _| {
            async move { error!("Error: {:#?}", err) }.boxed()
        })
        .on("test", callback)
        .connect()
        .await
        .expect("Connection failed");

    // emit to the "foo" event
    let json_payload = json!({"token": 123});
    socket
        .emit("foo", json_payload)
        .await
        .expect("Server unreachable");

    info!("Sent foo event");

    info!("Waiting 30 seconds");
    sleep(Duration::from_secs(30)).await;

    let json_payload = json!({"token": 123});
    socket
        .emit("foo", json_payload)
        .await
        .expect("Server unreachable");

    info!("Sent foo event");

    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"
```

Notice that after 25 seconds the error is printed. While the event can still be sent out, waiting for some more time will disconnect the client as shown in the output of the Python server.
Moreover, notice how the CPU load increases to 100% for one core.

## Possible Solution

Digging through the code, I found that the asynchronous implementation of the packet handling in `Socket::stream`:

https://github.com/1c3t3a/rust-socketio/blob/4a2ecd2bf8b60a8696d725053f00b21a82b7fd6b/socketio/src/asynchronous/socket.rs#L132-L146

is missing a check for `packet.packet_id == EnginePacketId::Message || packet.packet_id == EnginePacketId::MessageBinary` similar to the one in the synchronous version:

https://github.com/1c3t3a/rust-socketio/blob/4a2ecd2bf8b60a8696d725053f00b21a82b7fd6b/socketio/src/socket.rs#L123-L143

Adding that check appears to be working as expected.

I created #312 which solves that.


Discussion / hints:

Fix patch:
diff --git a/socketio/src/asynchronous/socket.rs b/socketio/src/asynchronous/socket.rs
index 7ab4c7f5..6bf0db21 100644
--- a/socketio/src/asynchronous/socket.rs
+++ b/socketio/src/asynchronous/socket.rs
@@ -137,10 +137,14 @@ impl Socket {
                 for await received_data in client.clone() {
                     let packet = received_data?;
 
-                    let packet = Self::handle_engineio_packet(packet, client.clone()).await?;
-                    Self::handle_socketio_packet(&packet, is_connected.clone());
-
-                    yield packet;
+                    if packet.packet_id == EnginePacketId::Message
+                        || packet.packet_id == EnginePacketId::MessageBinary
+                    {
+                        let packet = Self::handle_engineio_packet(packet, client.clone()).await?;
+                        Self::handle_socketio_packet(&packet, is_connected.clone());
+
+                        yield packet;
+                    }
                 }
         })
     }


Test patch: