Implement shared custom data (re-worked) based on the previous work from kitgxrl

Cargo.toml

@@ -1,2 +1,3 @@
 [workspace]
 members = ["engineio", "socketio"]
+resolver = "2"

engineio/src/asynchronous/client/async_client.rs

@@ -88,7 +88,7 @@ impl Debug for Client {
     }
 }
 
-#[cfg(all(test))]
+#[cfg(test)]
 mod test {
 
     use super::*;

engineio/src/asynchronous/mod.rs

@@ -1,7 +1,7 @@
 pub mod async_transports;
 pub mod transport;
 
-pub(self) mod async_socket;
+mod async_socket;
 #[cfg(feature = "async-callbacks")]
 mod callback;
 #[cfg(feature = "async")]

engineio/src/client/client.rs

@@ -1,11 +1,10 @@
 use super::super::socket::Socket as InnerSocket;
 use crate::callback::OptionalCallback;
-use crate::socket::DEFAULT_MAX_POLL_TIMEOUT;
-use crate::transport::Transport;
-
 use crate::error::{Error, Result};
 use crate::header::HeaderMap;
 use crate::packet::{HandshakePacket, Packet, PacketId};
+use crate::socket::DEFAULT_MAX_POLL_TIMEOUT;
+use crate::transport::Transport;
 use crate::transports::{PollingTransport, WebsocketSecureTransport, WebsocketTransport};
 use crate::ENGINE_IO_VERSION;
 use bytes::Bytes;

engineio/src/client/mod.rs

@@ -1,3 +1,4 @@
+#![allow(clippy::module_inception)]
 mod client;
 pub use client::Iter;
 pub use {client::Client, client::ClientBuilder, client::Iter as SocketIter};

engineio/src/lib.rs

@@ -85,7 +85,7 @@ pub mod client;
 /// Generic header map
 pub mod header;
 pub mod packet;
-pub(self) mod socket;
+mod socket;
 pub mod transport;
 pub mod transports;
 

socketio/Cargo.toml

@@ -48,3 +48,8 @@ async = ["async-callbacks", "rust_engineio/async", "tokio", "futures-util", "asy
 name = "async"
 path = "examples/async.rs"
 required-features = ["async"]
+
+[[example]]
+name = "async-transmitter"
+path = "examples/async_transmitter.rs"
+required-features = ["async"]

socketio/examples/async_transmitter.rs

@@ -0,0 +1,112 @@
+use futures_util::future::{BoxFuture, FutureExt};
+use rust_socketio::{
+    asynchronous::{Client as SocketIOClient, ClientBuilder as SocketIOClientBuilder},
+    Error as SocketIOError, Payload,
+};
+use serde_json::{json, Value};
+use std::sync::{mpsc, Arc};
+use std::time::Duration;
+use tokio::time::sleep;
+
+type JsonValues = Vec<Value>;
+
+fn test_event_handler<'event>(payload: Payload, socket: SocketIOClient) -> BoxFuture<'event, ()> {
+    async move {
+        if let Payload::Text(values) = payload {
+            match socket.try_transmitter::<mpsc::Sender<JsonValues>>() {
+                Ok(tx) => {
+                    tx.send(values.to_owned()).map_or_else(
+                        |err| eprintln!("{}", err),
+                        |_| println!("Data transmitted successfully"),
+                    );
+                }
+                Err(err) => {
+                    eprintln!("{}", err);
+                }
+            }
+        }
+    }
+    .boxed()
+}
+
+fn error_event_handler<'event>(payload: Payload, _: SocketIOClient) -> BoxFuture<'event, ()> {
+    async move { eprintln!("Error: {:#?}", payload) }.boxed()
+}
+
+struct ComplexData {
+    /// There should be many more fields below in real life,
+    /// probaly wrapped in Arc<Mutex<T>> if you're writing a more serious client.
+    data: String,
+}
+
+struct TransmitterClient {
+    receiver: mpsc::Receiver<JsonValues>,
+    complex: ComplexData,
+    client: SocketIOClient,
+}
+
+impl TransmitterClient {
+    async fn connect(url: &str) -> Result<Self, SocketIOError> {
+        let (sender, receiver) = mpsc::channel::<JsonValues>();
+
+        let client = SocketIOClientBuilder::new(url)
+            .namespace("/admin")
+            .on("test", test_event_handler)
+            .on("error", error_event_handler)
+            .transmitter(Arc::new(sender))
+            .connect()
+            .await?;
+
+        Ok(Self {
+            client,
+            receiver,
+            complex: ComplexData {
+                data: String::from(""),
+            },
+        })
+    }
+
+    async fn get_test(&mut self) -> Option<String> {
+        match self.client.emit("test", json!({"got ack": true})).await {
+            Ok(_) => {
+                match self.receiver.recv() {
+                    Ok(values) => {
+                        // Json deserialization and parsing business logic should be implemented
+                        // here to avoid over-complicating the handler callbacks.
+                        if let Some(value) = values.first() {
+                            if value.is_string() {
+                                self.complex.data = String::from(value.as_str().unwrap());
+                                return Some(self.complex.data.clone());
+                            }
+                        }
+                        None
+                    }
+                    Err(err) => {
+                        eprintln!("Transmission buffer is probably full: {}", err);
+                        None
+                    }
+                }
+            }
+            Err(err) => {
+                eprintln!("Server unreachable: {}", err);
+                None
+            }
+        }
+    }
+}
+
+#[tokio::main]
+async fn main() {
+    match TransmitterClient::connect("http://localhost:4200/").await {
+        Ok(mut client) => {
+            if let Some(test_data) = client.get_test().await {
+                println!("test event data from internal transmitter: {}", test_data);
+            }
+        }
+        Err(err) => {
+            eprintln!("Failed to connect to server: {}", err);
+        }
+    }
+
+    sleep(Duration::from_secs(2)).await;
+}

socketio/examples/readme.rs

@@ -1,5 +1,6 @@
 use rust_socketio::{ClientBuilder, Payload, RawClient};
 use serde_json::json;
+use std::thread::sleep;
 use std::time::Duration;
 
 fn main() {
@@ -45,5 +46,7 @@ fn main() {
         .emit_with_ack("test", json_payload, Duration::from_secs(2), ack_callback)
         .expect("Server unreachable");
 
+    sleep(Duration::from_secs(2));
+
     socket.disconnect().expect("Disconnect failed")
 }

socketio/examples/sync_transmitter.rs

@@ -0,0 +1,106 @@
+use rust_socketio::{
+    client::Client as SocketIOClient, ClientBuilder as SocketIOClientBuilder,
+    Error as SocketIOError, Payload, RawClient,
+};
+use serde_json::{json, Value};
+use std::sync::{mpsc, Arc};
+use std::thread::sleep;
+use std::time::Duration;
+
+type JsonValues = Vec<Value>;
+
+fn test_event_handler(payload: Payload, socket: RawClient) {
+    if let Payload::Text(values) = payload {
+        match socket.try_transmitter::<mpsc::Sender<JsonValues>>() {
+            Ok(tx) => {
+                tx.send(values.to_owned()).map_or_else(
+                    |err| eprintln!("{}", err),
+                    |_| println!("Data transmitted successfully"),
+                );
+            }
+            Err(err) => {
+                eprintln!("{}", err);
+            }
+        }
+    }
+}
+
+fn error_event_handler(payload: Payload, _: RawClient) {
+    eprintln!("Error: {:#?}", payload);
+}
+
+struct ComplexData {
+    /// There should be many more fields below in real life,
+    /// probaly wrapped in Arc<Mutex<T>> if you're writing a more serious client.
+    data: String,
+}
+
+struct TransmitterClient {
+    client: SocketIOClient,
+    receiver: mpsc::Receiver<JsonValues>,
+    complex: ComplexData,
+}
+
+impl TransmitterClient {
+    fn connect(url: &str) -> Result<Self, SocketIOError> {
+        let (sender, receiver) = mpsc::channel::<JsonValues>();
+
+        let client = SocketIOClientBuilder::new(url)
+            .namespace("/admin")
+            .on("test", test_event_handler)
+            .on("error", error_event_handler)
+            .transmitter(Arc::new(sender))
+            .connect()?;
+
+        Ok(Self {
+            client,
+            receiver,
+            complex: ComplexData {
+                data: "".to_string(),
+            },
+        })
+    }
+
+    fn get_test(&mut self) -> Option<String> {
+        match self.client.emit("test", json!({"got ack": true})) {
+            Ok(_) => {
+                match self.receiver.recv() {
+                    Ok(values) => {
+                        // Json deserialization and parsing business logic should be implemented
+                        // here to avoid over-complicating the handler callbacks.
+                        if let Some(value) = values.first() {
+                            if value.is_string() {
+                                self.complex.data = String::from(value.as_str().unwrap());
+                                return Some(self.complex.data.clone());
+                            }
+                        }
+                        None
+                    }
+                    Err(err) => {
+                        eprintln!("Transmission buffer is probably full: {}", err);
+                        None
+                    }
+                }
+            }
+            Err(err) => {
+                eprintln!("Server unreachable: {}", err);
+                None
+            }
+        }
+    }
+}
+
+fn main() {
+    match TransmitterClient::connect("http://localhost:4200/") {
+        Ok(mut client) => {
+            if let Some(test_data) = client.get_test() {
+                println!("test event data from internal transmitter: {}", test_data);
+            }
+        }
+        Err(err) => {
+            eprintln!("Failed to connect to server: {}", err);
+        }
+    }
+
+    sleep(Duration::from_secs(2));
+}