examples/rust: add echo-stream-nats-{client,server}

Cargo.toml

@@ -1,7 +1,7 @@
 [package]
+description = "WebAssembly component-native RPC framework based on WIT"
 name = "wrpc"
 version = "0.5.0"
-description = "WebAssembly component-native RPC framework based on WIT"
 
 authors.workspace = true
 categories.workspace = true
@@ -80,22 +80,23 @@ wrpc-wasmtime-nats-cli = { workspace = true, optional = true }
 anyhow = { workspace = true }
 bytes = { workspace = true }
 futures = { workspace = true, features = ["async-await"] }
-rcgen = { workspace = true, features = ["crypto", "ring", "zeroize"] }
-rustls = { workspace = true, features = ["logging", "ring"] }
-test-log = { workspace = true, features = ["color", "log", "trace"] }
-tokio = { workspace = true, features = ["process", "rt-multi-thread"] }
 quinn = { workspace = true, features = [
     "log",
     "platform-verifier",
     "ring",
     "runtime-tokio",
     "rustls",
 ] }
+rcgen = { workspace = true, features = ["crypto", "ring", "zeroize"] }
+rustls = { workspace = true, features = ["logging", "ring"] }
+test-log = { workspace = true, features = ["color", "log", "trace"] }
+tokio = { workspace = true, features = ["process", "rt-multi-thread"] }
 wrpc-cli = { workspace = true }
 
 [workspace.dependencies]
 anyhow = { version = "1", default-features = false }
 async-nats = { package = "async-nats-wrpc", version = "0.35.1", default-features = false }
+async-stream = { version = "0.3", default-features = false }
 bitflags = { version = "2", default-features = false }
 bytes = { version = "1", default-features = false }
 clap = { version = "4", default-features = false }
@@ -121,8 +122,8 @@ tokio-stream = { version = "0.1", default-features = false }
 tokio-util = { version = "0.7", default-features = false }
 tracing = { version = "0.1", default-features = false }
 tracing-subscriber = { version = "0.3", default-features = false }
-uuid = { version = "1", default-features = false }
 url = { version = "2", default-features = false }
+uuid = { version = "1", default-features = false }
 wasi-preview1-component-adapter-provider = { version = "23.0.2", default-features = false }
 wasm-tokio = { version = "0.5.16", default-features = false }
 wasmparser = { version = "0.214", default-features = false }

examples/rust/echo-stream-nats-client/Cargo.toml

@@ -0,0 +1,28 @@
+[package]
+name = "echo-stream-nats-client"
+version = "0.1.0"
+
+authors.workspace = true
+categories.workspace = true
+edition.workspace = true
+license.workspace = true
+repository.workspace = true
+
+[dependencies]
+anyhow = { workspace = true }
+async-nats = { workspace = true }
+async-stream = { workspace = true }
+clap = { workspace = true, features = [
+    "color",
+    "derive",
+    "error-context",
+    "help",
+    "std",
+    "suggestions",
+    "usage",
+] }
+tokio = { workspace = true, features = ["rt-multi-thread"] }
+tracing-subscriber = { workspace = true, features = ["ansi", "fmt"] }
+url = { workspace = true }
+wit-bindgen-wrpc = { workspace = true }
+wrpc-transport-nats = { workspace = true }

examples/rust/echo-stream-nats-client/src/main.rs

@@ -0,0 +1,99 @@
+use std::time::Duration;
+
+use anyhow::Context as _;
+use async_stream::stream;
+use bindings::wrpc_examples::echo_stream::handler::Req;
+use clap::Parser;
+use tokio::{sync::mpsc, time::sleep};
+use url::Url;
+use wit_bindgen_wrpc::futures::StreamExt;
+
+mod bindings {
+    wit_bindgen_wrpc::generate!({
+        with: {
+            "wrpc-examples:echo-stream/handler": generate
+        }
+    });
+}
+
+#[derive(Parser, Debug)]
+#[command(author, version, about, long_about = None)]
+struct Args {
+    /// NATS.io URL to connect to
+    #[arg(short, long, default_value = "nats://127.0.0.1:4222")]
+    nats: Url,
+
+    /// Prefixes to invoke `wrpc-examples:hello/handler.hello` on
+    prefixes: Vec<String>,
+}
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    tracing_subscriber::fmt().init();
+
+    let Args { nats, prefixes } = Args::parse();
+
+    let nats = connect(nats)
+        .await
+        .context("failed to connect to NATS.io")?;
+    for prefix in prefixes {
+        let input_stream = Box::pin(stream! {
+            for i in 1..=10 {
+                yield vec![i];
+                sleep(Duration::from_secs(1)).await;
+            }
+        });
+        let wrpc = wrpc_transport_nats::Client::new(nats.clone(), prefix.clone(), None);
+        let (mut output_stream, res) = bindings::wrpc_examples::echo_stream::handler::echo(
+            &wrpc,
+            None,
+            Req {
+                input: input_stream,
+            },
+        )
+        .await
+        .context("failed to invoke `wrpc-examples.hello/handler.hello`")?;
+        let task = tokio::spawn(async move {
+            match res {
+                Some(fut) => Some(fut.await),
+                None => None,
+            }
+        });
+        while let Some(item) = output_stream.next().await {
+            eprintln!("got {item:?}");
+        }
+        if let Some(res) = task.await? {
+            res?;
+        }
+    }
+    Ok(())
+}
+
+/// Connect to NATS.io server and ensure that the connection is fully established before
+/// returning the resulting [`async_nats::Client`]
+async fn connect(url: Url) -> anyhow::Result<async_nats::Client> {
+    let (conn_tx, mut conn_rx) = mpsc::channel(1);
+    let client = async_nats::connect_with_options(
+        String::from(url),
+        async_nats::ConnectOptions::new()
+            .retry_on_initial_connect()
+            .event_callback(move |event| {
+                let conn_tx = conn_tx.clone();
+                async move {
+                    if let async_nats::Event::Connected = event {
+                        conn_tx
+                            .send(())
+                            .await
+                            .expect("failed to send NATS.io server connection notification");
+                    }
+                }
+            }),
+    )
+    .await
+    .context("failed to connect to NATS.io server")?;
+    conn_rx
+        .recv()
+        .await
+        .context("failed to await NATS.io server connection to be established")?;
+    Ok(client)
+}

examples/rust/echo-stream-nats-client/wit/deps.lock

@@ -0,0 +1,4 @@
+[echo-stream]
+path = "../../../wit/echo-stream"
+sha256 = "3747db334a26bd9fa6438fba616ed862bff5b5055e48337f91b7636bd16412d5"
+sha512 = "6519ce1910672726e00ca8b0b87406ccf7304d0e54605e1a8651a8949df70f7c12f8238a4166589655638d0b2142d1897d9780b248e491f2545b05f7332726b1"

examples/rust/echo-stream-nats-client/wit/deps.toml

@@ -0,0 +1 @@
+echo-stream = "../../../wit/echo-stream"

examples/rust/echo-stream-nats-client/wit/deps/echo-stream/echo-stream.wit

@@ -0,0 +1,16 @@
+package wrpc-examples:echo-stream;
+
+interface handler {
+    record req {
+        input: stream<u64>,
+    }
+    echo: func(r: req) -> stream<u64>;
+}
+
+world client {
+    import handler;
+}
+
+world server {
+    export handler;
+}

examples/rust/echo-stream-nats-client/wit/world.wit

@@ -0,0 +1,5 @@
+package wrpc-examples:echo-stream-rust-client;
+
+world client {
+    include wrpc-examples:echo-stream/client;
+}

examples/rust/echo-stream-nats-server/Cargo.toml

@@ -0,0 +1,30 @@
+[package]
+name = "echo-stream-nats-server"
+version = "0.1.0"
+
+authors.workspace = true
+categories.workspace = true
+edition.workspace = true
+license.workspace = true
+repository.workspace = true
+
+[dependencies]
+anyhow = { workspace = true }
+async-nats = { workspace = true }
+async-stream = { workspace = true }
+clap = { workspace = true, features = [
+    "color",
+    "derive",
+    "error-context",
+    "help",
+    "std",
+    "suggestions",
+    "usage",
+] }
+futures = { workspace = true }
+tokio = { workspace = true, features = ["rt-multi-thread", "signal"] }
+tracing = { workspace = true }
+tracing-subscriber = { workspace = true, features = ["ansi", "fmt"] }
+url = { workspace = true }
+wit-bindgen-wrpc = { workspace = true }
+wrpc-transport-nats = { workspace = true }