feat(nats): demultiplex a single client subscriber

Signed-off-by: Roman Volosatovs <[email protected]>

Author: rvolosatovs

Cargo.lock

@@ -121,6 +121,7 @@ criterion = { version = "0.5", default-features = false }
 futures = { version = "0.3", default-features = false }
 heck = { version = "0.5", default-features = false }
 humantime = { version = "2.1", default-features = false }
+nuid = { version = "0.5", default-features = false }
 pin-project-lite = { version = "0.2", default-features = false }
 prettyplease = { version = "0.2.25", default-features = false }
 proc-macro2 = { version = "1", default-features = false }
@@ -163,7 +164,7 @@ wrpc-introspect = { version = "0.5", default-features = false, path = "./crates/
 wrpc-runtime-wasmtime = { version = "0.24.1", path = "./crates/runtime-wasmtime", default-features = false }
 wrpc-test = { path = "./crates/test", default-features = false }
 wrpc-transport = { version = "0.28.2", path = "./crates/transport", default-features = false }
-wrpc-transport-nats = { version = "0.26", path = "./crates/transport-nats", default-features = false }
+wrpc-transport-nats = { version = "0.27", path = "./crates/transport-nats", default-features = false }
 wrpc-transport-quic = { version = "0.3", path = "./crates/transport-quic", default-features = false }
 wrpc-wasi-keyvalue = { version = "0.1", path = "./crates/wasi-keyvalue", default-features = false }
 wrpc-wasi-keyvalue-mem = { version = "0.1", path = "./crates/wasi-keyvalue-mem", default-features = false }

Cargo.toml

@@ -204,8 +204,8 @@ fn bench_wasm_ping_direct(
             handler
                 .ping(None::<async_nats::HeaderMap>)
                 .await
-                .expect("failed to call handler")
-        })
+                .expect("failed to call handler");
+        });
     });
     Ok(())
 }
@@ -225,7 +225,7 @@ fn bench_wasm_greet_direct(
                 .await
                 .expect("failed to call handler");
             assert_eq!(greeting, "Hello, test");
-        })
+        });
     });
     Ok(())
 }
@@ -272,7 +272,7 @@ where
                     .expect("failed to send message");
                 assert!(status.is_none());
                 assert_eq!(payload, expect);
-            })
+            });
         });
         stop_tx.send(()).expect("failed to stop server");
         rt.block_on(async { srv.await.context("server task panicked")? })?;
@@ -435,7 +435,9 @@ fn bench_wasm_ping_nats_wrpc(
         .block_on(WasmHandler::new(wasm))
         .context("failed to construct a Wasm handler")?;
     with_nats(&rt, |nats| {
-        let wrpc = wrpc_transport_nats::Client::new(nats, "", None);
+        let wrpc = rt
+            .block_on(wrpc_transport_nats::Client::new(nats, "", None))
+            .context("failed to construct client")?;
 
         let invocations = rt
             .block_on(ping_bindings_wrpc::serve(&wrpc, handler))
@@ -468,7 +470,7 @@ fn bench_wasm_ping_nats_wrpc(
                 ping_bindings_wrpc::wrpc_bench::bench::ping::ping(&wrpc, None)
                     .await
                     .expect("failed to call `ping`");
-            })
+            });
         });
         stop_tx.send(()).expect("failed to stop server");
         rt.block_on(async { srv.await.context("server task panicked")? })?;
@@ -485,7 +487,9 @@ fn bench_wasm_greet_nats_wrpc(
         .block_on(WasmHandler::new(wasm))
         .context("failed to construct a Wasm handler")?;
     with_nats(&rt, |nats| {
-        let wrpc = wrpc_transport_nats::Client::new(nats, "", None);
+        let wrpc = rt
+            .block_on(wrpc_transport_nats::Client::new(nats, "", None))
+            .context("failed to construct client")?;
 
         let invocations = rt
             .block_on(greet_bindings_wrpc::serve(&wrpc, handler))
@@ -518,7 +522,7 @@ fn bench_wasm_greet_nats_wrpc(
                 greet_bindings_wrpc::wrpc_bench::bench::greet::greet(&wrpc, None, "test")
                     .await
                     .expect("failed to call `greet`");
-            })
+            });
         });
         stop_tx.send(()).expect("failed to stop server");
         rt.block_on(async { srv.await.context("server task panicked")? })?;
@@ -529,7 +533,9 @@ fn bench_wasm_greet_nats_wrpc(
 fn bench_nats_wrpc_ping(g: &mut BenchmarkGroup<impl Measurement>) -> anyhow::Result<()> {
     let rt = tokio::runtime::Runtime::new().context("failed to build Tokio runtime")?;
     with_nats(&rt, |nats| {
-        let wrpc = wrpc_transport_nats::Client::new(nats, "", None);
+        let wrpc = rt
+            .block_on(wrpc_transport_nats::Client::new(nats, "", None))
+            .context("failed to construct client")?;
 
         let invocations = rt
             .block_on(ping_bindings_wrpc::serve(&wrpc, NativeHandler))
@@ -562,7 +568,7 @@ fn bench_nats_wrpc_ping(g: &mut BenchmarkGroup<impl Measurement>) -> anyhow::Res
                 ping_bindings_wrpc::wrpc_bench::bench::ping::ping(&wrpc, None)
                     .await
                     .expect("failed to call `ping`");
-            })
+            });
         });
         stop_tx.send(()).expect("failed to stop server");
         rt.block_on(async { srv.await.context("server task panicked")? })?;
@@ -573,7 +579,9 @@ fn bench_nats_wrpc_ping(g: &mut BenchmarkGroup<impl Measurement>) -> anyhow::Res
 fn bench_nats_wrpc_greet(g: &mut BenchmarkGroup<impl Measurement>) -> anyhow::Result<()> {
     let rt = tokio::runtime::Runtime::new().context("failed to build Tokio runtime")?;
     with_nats(&rt, |nats| {
-        let wrpc = wrpc_transport_nats::Client::new(nats, "", None);
+        let wrpc = rt
+            .block_on(wrpc_transport_nats::Client::new(nats, "", None))
+            .context("failed to construct client")?;
 
         let invocations = rt
             .block_on(greet_bindings_wrpc::serve(&wrpc, NativeHandler))
@@ -607,7 +615,7 @@ fn bench_nats_wrpc_greet(g: &mut BenchmarkGroup<impl Measurement>) -> anyhow::Re
                         .await
                         .expect("failed to call `greet`");
                 assert_eq!(greeting, "Hello, test");
-            })
+            });
         });
         stop_tx.send(()).expect("failed to stop server");
         rt.block_on(async { srv.await.context("server task panicked")? })?;

benches/bench.rs

@@ -1,6 +1,6 @@
 [package]
 name = "wrpc-transport-nats"
-version = "0.26.0"
+version = "0.27.0"
 description = "wRPC NATS transport"
 
 authors.workspace = true
@@ -22,7 +22,9 @@ async-nats-0_36 = { package = "async-nats", version = "0.36", default-features =
 ], optional = true }
 bytes = { workspace = true }
 futures = { workspace = true, features = ["async-await"] }
+nuid = { workspace = true }
 tokio = { workspace = true, features = ["io-util", "rt-multi-thread"] }
+tokio-stream = { workspace = true, features = ["sync"] }
 tokio-util = { workspace = true, features = ["codec", "io"] }
 tracing = { workspace = true, features = ["attributes"] }
 wasm-tokio = { workspace = true }