1. Optimize some codes.

2. Add qps benchmark.

Author: jjeffcaii

examples/Cargo.toml

@@ -9,7 +9,7 @@ publish = false
 log = "0.4.8"
 env_logger = "0.7.1"
 futures = "0.3.5"
-clap = "2.33.0"
+clap = "2.33.1"
 
 [dev-dependencies.rsocket_rust]
 path = "../rsocket"
@@ -21,7 +21,7 @@ path = "../rsocket-transport-tcp"
 path = "../rsocket-transport-websocket"
 
 [dev-dependencies.tokio]
-version = "0.2.20"
+version = "0.2.21"
 default-features = false
 features = ["full"]
 
@@ -35,4 +35,8 @@ path = "proxy.rs"
 
 [[example]]
 name = "cli"
-path = "cli.rs"
+path = "cli.rs"
+
+[[example]]
+name = "qps"
+path = "qps.rs"

examples/qps.rs

@@ -0,0 +1,61 @@
+#[macro_use]
+extern crate log;
+
+use rsocket_rust::prelude::*;
+use rsocket_rust_transport_tcp::TcpClientTransport;
+use std::error::Error;
+use std::sync::{
+    atomic::{AtomicU32, Ordering},
+    Arc,
+};
+use std::time::SystemTime;
+use tokio::runtime::Runtime;
+use tokio::sync::Notify;
+
+const TOTAL: u32 = 1_000_000;
+
+fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
+    env_logger::builder().format_timestamp_millis().init();
+
+    let mut rt = Runtime::new()?;
+    let client = rt.block_on(async {
+        RSocketFactory::connect()
+            .transport(TcpClientTransport::from("127.0.0.1:7878"))
+            .start()
+            .await
+    })?;
+    // simulate 1KB payload.
+    let req = Payload::builder()
+        .set_data_utf8("X".repeat(1024).as_ref())
+        .build();
+    let counter = Arc::new(AtomicU32::new(0));
+    let start_time = SystemTime::now();
+    let notify = Arc::new(Notify::new());
+    for _ in 0..TOTAL {
+        let client = client.clone();
+        let counter = counter.clone();
+        let notify = notify.clone();
+        let req = req.clone();
+        rt.spawn(async move {
+            client.request_response(req).await.expect("Request failed");
+            let current = counter.fetch_add(1, Ordering::SeqCst) + 1;
+            if current >= TOTAL {
+                notify.notify();
+            }
+        });
+    }
+    rt.block_on(async move {
+        notify.notified().await;
+    });
+    let costs = SystemTime::now()
+        .duration_since(start_time)
+        .unwrap()
+        .as_millis();
+    info!(
+        "total={}, cost={}ms, qps={}",
+        TOTAL,
+        costs,
+        1000f64 * (TOTAL as f64) / (costs as f64)
+    );
+    Ok(())
+}

rsocket-benchmark/Cargo.toml

@@ -30,7 +30,7 @@ path = "../rsocket-transport-websocket"
 path = "../rsocket-messaging"
 
 [dev-dependencies.tokio]
-version = "0.2.20"
+version = "0.2.21"
 default-features = false
 features = ["full"]
 

rsocket-messaging/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "rsocket_rust_messaging"
-version = "0.1.0"
+version = "0.5.2"
 authors = ["Jeffsky <[email protected]>"]
 edition = "2018"
 

rsocket-test/tests/test_clients.rs

@@ -146,7 +146,7 @@ async fn test_request_response_err() {
 
 async fn exec_request_response<R>(socket: &Client<R>)
 where
-    R: Send + Sync + Clone + Spawner + 'static,
+    R: Send + Sync + Copy + Spawner + 'static,
 {
     // request response
     let sending = Payload::builder()
@@ -159,7 +159,7 @@ where
 
 async fn exec_metadata_push<R>(socket: &Client<R>)
 where
-    R: Send + Sync + Clone + Spawner + 'static,
+    R: Send + Sync + Copy + Spawner + 'static,
 {
     let pa = Payload::builder().set_metadata_utf8("Hello World!").build();
     // metadata push
@@ -168,7 +168,7 @@ where
 
 async fn exec_fire_and_forget<R>(socket: &Client<R>)
 where
-    R: Send + Sync + Clone + Spawner + 'static,
+    R: Send + Sync + Copy + Spawner + 'static,
 {
     // request fnf
     let fnf = Payload::from("Hello World!");
@@ -177,7 +177,7 @@ where
 
 async fn exec_request_stream<R>(socket: &Client<R>)
 where
-    R: Send + Sync + Clone + Spawner + 'static,
+    R: Send + Sync + Copy + Spawner + 'static,
 {
     // request stream
     let sending = Payload::builder()
@@ -197,7 +197,7 @@ where
 
 async fn exec_request_channel<R>(socket: &Client<R>)
 where
-    R: Send + Sync + Clone + Spawner + 'static,
+    R: Send + Sync + Copy + Spawner + 'static,
 {
     let sends: Vec<_> = (0..10)
         .map(|n| {

rsocket-transport-tcp/Cargo.toml

@@ -19,7 +19,7 @@ path = "../rsocket"
 features = ["frame"]
 
 [dependencies.tokio]
-version = "0.2.20"
+version = "0.2.21"
 default-features = false
 features = [ "rt-core", "rt-threaded", "tcp", "sync", "stream" ]
 

rsocket-transport-tcp/src/client.rs

@@ -4,8 +4,8 @@ use rsocket_rust::error::RSocketError;
 use rsocket_rust::frame::Frame;
 use rsocket_rust::runtime::{DefaultSpawner, Spawner};
 use rsocket_rust::transport::{ClientTransport, Rx, Tx, TxOnce};
-use std::net::SocketAddr;
-use std::net::TcpStream as StdTcpStream;
+use std::net::{AddrParseError, SocketAddr, TcpStream as StdTcpStream};
+use std::str::FromStr;
 use tokio::net::TcpStream;
 use tokio_util::codec::Framed;
 
@@ -75,6 +75,19 @@ impl ClientTransport for TcpClientTransport {
     }
 }
 
+impl FromStr for TcpClientTransport {
+    type Err = AddrParseError;
+
+    fn from_str(addr: &str) -> Result<Self, Self::Err> {
+        let socket_addr = if addr.starts_with("tcp://") || addr.starts_with("TCP://") {
+            addr.chars().skip(6).collect::<String>().parse()?
+        } else {
+            addr.parse()?
+        };
+        Ok(TcpClientTransport::new(Connector::Lazy(socket_addr)))
+    }
+}
+
 impl From<SocketAddr> for TcpClientTransport {
     fn from(addr: SocketAddr) -> TcpClientTransport {
         TcpClientTransport::new(Connector::Lazy(addr))
@@ -83,12 +96,12 @@ impl From<SocketAddr> for TcpClientTransport {
 
 impl From<&str> for TcpClientTransport {
     fn from(addr: &str) -> TcpClientTransport {
-        let socket_addr: SocketAddr = if addr.starts_with("tcp://") {
-            let ss: String = addr.chars().skip(6).collect();
-            ss.parse().unwrap()
+        let socket_addr: SocketAddr = if addr.starts_with("tcp://") || addr.starts_with("TCP://") {
+            addr.chars().skip(6).collect::<String>().parse()
         } else {
-            addr.parse().unwrap()
-        };
+            addr.parse()
+        }
+        .expect("Invalid transport string!");
         TcpClientTransport::new(Connector::Lazy(socket_addr))
     }
 }

rsocket-transport-wasm/src/runtime.rs

@@ -2,7 +2,7 @@ use rsocket_rust::runtime::Spawner;
 use std::future::Future;
 use wasm_bindgen_futures::spawn_local;
 
-#[derive(Clone)]
+#[derive(Clone, Copy, Debug)]
 pub struct WASMSpawner;
 
 impl Spawner for WASMSpawner {

rsocket-transport-websocket/Cargo.toml

@@ -21,6 +21,6 @@ path = "../rsocket"
 features = ["frame"]
 
 [dependencies.tokio]
-version = "0.2.20"
+version = "0.2.21"
 default-features = false
 features = [ "rt-core", "rt-threaded", "tcp", "sync", "stream" ]

rsocket/Cargo.toml

@@ -16,7 +16,7 @@ futures = "0.3.5"
 lazy_static = "1.4.0"
 
 [dependencies.tokio]
-version = "0.2.20"
+version = "0.2.21"
 default-features = false
 features = [ "rt-core", "rt-threaded", "sync", "stream" ]