Merge pull request #14 from rsocket/feature/messaging

Feature/messaging

Author: jjeffcaii

.gitignore

@@ -1,3 +1,5 @@
-/target
+target/
 **/*.rs.bk
 Cargo.lock
+
+

Cargo.toml

@@ -1,12 +1,16 @@
 [workspace]
 
 members = [
+# core
 "rsocket",
+# transports
 "rsocket-transport-tcp",
 "rsocket-transport-websocket",
 "rsocket-transport-wasm",
-
-# Internal
+# extra
+"rsocket-messaging",
+# internal
 "examples",
 "rsocket-test",
+"rsocket-benchmark",
 ]

examples/Cargo.toml

@@ -8,20 +8,20 @@ publish = false
 [dev-dependencies]
 log = "0.4.8"
 env_logger = "0.7.1"
-futures = "0.3.4"
-clap = "2.33.0"
+futures = "0.3.5"
+clap = "2.33.1"
 
 [dev-dependencies.rsocket_rust]
-path = "../rsocket"
+version = "0.5.2"
 
 [dev-dependencies.rsocket_rust_transport_tcp]
-path = "../rsocket-transport-tcp"
+version = "0.5.2"
 
 [dev-dependencies.rsocket_rust_transport_websocket]
-path = "../rsocket-transport-websocket"
+version = "0.5.2"
 
 [dev-dependencies.tokio]
-version = "0.2.16"
+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

@@ -0,0 +1,40 @@
+[package]
+name = "rsocket_rust_benchmark"
+version = "0.0.0"
+authors = ["Jeffsky <[email protected]>"]
+edition = "2018"
+publish = false
+
+[dev-dependencies]
+log = "0.4"
+futures = "0.3.5"
+env_logger = "0.7.1"
+bytes = "0.5.4"
+hex = "0.4.2"
+rand = "0.7.3"
+serde = "1.0.110"
+serde_derive = "1.0.110"
+criterion = "0.3.2"
+
+[dev-dependencies.rsocket_rust]
+version = "0.5.2"
+features = ["frame"]
+
+[dev-dependencies.rsocket_rust_transport_tcp]
+version = "0.5.2"
+
+[dev-dependencies.rsocket_rust_transport_websocket]
+version = "0.5.2"
+
+[dev-dependencies.rsocket_rust_messaging]
+version = "0.5.2"
+
+[dev-dependencies.tokio]
+version = "0.2.21"
+default-features = false
+features = ["full"]
+
+[[bench]]
+name = "bench_main"
+path = "benchmark_main.rs"
+harness = false

rsocket-benchmark/benchmark_main.rs

@@ -0,0 +1,8 @@
+use criterion::criterion_main;
+
+mod benchmarks;
+
+criterion_main! {
+    benchmarks::fibonaccis::benches,
+    benchmarks::frames::benches,
+}

rsocket-benchmark/benchmarks/fibonaccis.rs

@@ -0,0 +1,15 @@
+use criterion::{black_box, criterion_group, Criterion};
+
+fn fibonacci(n: u64) -> u64 {
+    match n {
+        0 => 1,
+        1 => 1,
+        n => fibonacci(n - 1) + fibonacci(n - 2),
+    }
+}
+
+fn fibonaccis_benchmark(c: &mut Criterion) {
+    c.bench_function("fib 20", |b| b.iter(|| fibonacci(black_box(20))));
+}
+
+criterion_group!(benches, fibonaccis_benchmark);

rsocket-benchmark/benchmarks/frames.rs

@@ -0,0 +1,19 @@
+use bytes::{Bytes, BytesMut};
+use criterion::{criterion_group, Criterion};
+use rsocket_rust::frame::*;
+use rsocket_rust::utils::Writeable;
+
+fn bench_unmarshal_request_response(c: &mut Criterion) {
+    c.bench_function("unmarshal request_response", |b| {
+        b.iter(|| {
+            let f = RequestResponse::builder(1234, 0)
+                .set_data(Bytes::from("Hello World"))
+                .set_metadata(Bytes::from("Foobar"))
+                .build();
+            let mut bf = BytesMut::with_capacity(f.len() as usize);
+            f.write_to(&mut bf);
+        })
+    });
+}
+
+criterion_group!(benches, bench_unmarshal_request_response);

rsocket-benchmark/benchmarks/mod.rs

@@ -0,0 +1,2 @@
+pub mod fibonaccis;
+pub mod frames;

rsocket-messaging/Cargo.toml

@@ -0,0 +1,29 @@
+[package]
+name = "rsocket_rust_messaging"
+version = "0.5.2"
+authors = ["Jeffsky <[email protected]>"]
+edition = "2018"
+license = "Apache-2.0"
+readme = "README.md"
+repository = "https://github.com/rsocket/rsocket-rust"
+homepage = "https://github.com/rsocket/rsocket-rust"
+description = "Communicate with Spring RSocket Messaging."
+
+[dependencies]
+futures = "0.3.5"
+bytes = "0.5.4"
+serde = "1.0.110"
+serde_json = "1.0.53"
+serde_cbor = "0.11.1"
+hex = "0.4.2"
+url = "2.1.1"
+
+[dependencies.rsocket_rust]
+version = "0.5.2"
+features = ["frame"]
+
+[dependencies.rsocket_rust_transport_tcp]
+version = "0.5.2"
+
+[dependencies.rsocket_rust_transport_websocket]
+version = "0.5.2"