feat: add WebTransport keyvalue server example

Signed-off-by: Roman Volosatovs <rvolosatovs@riseup.net>

Author: rvolosatovs

Cargo.lock

@@ -0,0 +1,36 @@
+[package]
+name = "wasi-keyvalue-web-server"
+version = "0.1.0"
+
+authors.workspace = true
+categories.workspace = true
+edition.workspace = true
+homepage.workspace = true
+license.workspace = true
+repository.workspace = true
+
+[dependencies]
+anyhow = { workspace = true, features = ["std"] }
+bytes = { 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",
+    "env-filter",
+    "fmt",
+] }
+wrpc-transport = { workspace = true }
+wrpc-transport-web = { workspace = true }
+wrpc-wasi-keyvalue = { workspace = true }
+wrpc-wasi-keyvalue-mem = { workspace = true }
+wtransport = { workspace = true, features = ["self-signed"] }

examples/rust/wasi-keyvalue-web-server/Cargo.toml

@@ -0,0 +1,133 @@
+use core::net::SocketAddr;
+use core::pin::pin;
+use core::time::Duration;
+
+use std::sync::Arc;
+
+use anyhow::Context as _;
+use clap::Parser;
+use futures::stream::select_all;
+use futures::StreamExt as _;
+use tokio::task::JoinSet;
+use tokio::{select, signal};
+use tracing::{debug, error, info, warn};
+use wtransport::{Endpoint, Identity, ServerConfig};
+
+#[derive(Parser, Debug)]
+#[command(author, version, about, long_about = None)]
+struct Args {
+    /// Address to serve `wasi:keyvalue` on
+    #[arg(default_value = "[::1]:4433")]
+    addr: SocketAddr,
+}
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    tracing_subscriber::fmt().init();
+
+    let Args { addr } = Args::parse();
+
+    let id = Identity::self_signed(["localhost", "127.0.0.1", "::1"])
+        .context("failed to generate server certificate")?;
+
+    let ep = Endpoint::server(
+        ServerConfig::builder()
+            .with_bind_address(addr)
+            .with_identity(id)
+            .keep_alive_interval(Some(Duration::from_secs(3)))
+            .build(),
+    )
+    .context("failed to create server endpoint")?;
+
+    let srv = Arc::new(wrpc_transport_web::Server::new());
+    let accept = tokio::spawn({
+        let mut tasks = JoinSet::<anyhow::Result<()>>::new();
+        let srv = Arc::clone(&srv);
+        async move {
+            loop {
+                select! {
+                    conn = ep.accept() => {
+                        let srv = Arc::clone(&srv);
+                        tasks.spawn(async move {
+                            let req = conn
+                                .await
+                                .context("failed to accept WebTransport connection")?;
+                            let conn = req
+                                .accept()
+                                .await
+                                .context("failed to establish WebTransport connection")?;
+                            let wrpc = wrpc_transport_web::Client::from(conn);
+                            loop {
+                                srv.accept(&wrpc)
+                                    .await
+                                    .context("failed to accept wRPC connection")?;
+                            }
+                        });
+                    }
+                    Some(res) = tasks.join_next() => {
+                        match res {
+                            Ok(Ok(())) => {}
+                            Ok(Err(err)) => {
+                                warn!(?err, "failed to serve connection")
+                            }
+                            Err(err) => {
+                                error!(?err, "failed to join task")
+                            }
+                        }
+                    }
+                }
+            }
+        }
+    });
+
+    let invocations =
+        wrpc_wasi_keyvalue::serve(srv.as_ref(), wrpc_wasi_keyvalue_mem::Handler::default())
+            .await
+            .context("failed to serve `wasi:keyvalue`")?;
+    // NOTE: This will conflate all invocation streams into a single stream via `futures::stream::SelectAll`,
+    // to customize this, iterate over the returned `invocations` and set up custom handling per export
+    let mut invocations = select_all(
+        invocations
+            .into_iter()
+            .map(|(instance, name, invocations)| invocations.map(move |res| (instance, name, res))),
+    );
+    let shutdown = signal::ctrl_c();
+    let mut shutdown = pin!(shutdown);
+    let mut tasks = JoinSet::new();
+    loop {
+        select! {
+            Some((instance, name, res)) = invocations.next() => {
+                match res {
+                    Ok(fut) => {
+                        debug!(instance, name, "invocation accepted");
+                        tasks.spawn(async move {
+                            if let Err(err) = fut.await {
+                                warn!(?err, "failed to handle invocation");
+                            } else {
+                                info!(instance, name, "invocation successfully handled");
+                            }
+                        });
+                    }
+                    Err(err) => {
+                        warn!(?err, instance, name, "failed to accept invocation");
+                    }
+                }
+            }
+            Some(res) = tasks.join_next() => {
+                if let Err(err) = res {
+                    error!(?err, "failed to join task")
+                }
+            }
+            res = &mut shutdown => {
+                accept.abort();
+                // wait for all invocations to complete
+                while let Some(res) = tasks.join_next().await {
+                    if let Err(err) = res {
+                        error!(?err, "failed to join task")
+                    }
+                }
+                return res.context("failed to listen for ^C")
+            }
+        }
+    }
+}

examples/rust/wasi-keyvalue-web-server/src/main.rs

@@ -31,7 +31,7 @@ quinn = { workspace = true, features = [
     "runtime-tokio",
     "rustls",
 ] }
-rustls = { workspace = true }
+rustls = { workspace = true, features = ["ring"] }
 tokio = { workspace = true, features = [
     "macros",
     "rt-multi-thread",

examples/web/rust/Cargo.toml

@@ -30,7 +30,7 @@ use url::Url;
 use uuid::Uuid;
 use wrpc_transport::{ResourceBorrow, ResourceOwn};
 use wrpc_wasi_keyvalue::exports::wasi::keyvalue::store;
-use wtransport::tls::Sha256DigestFmt;
+use wtransport::tls::{Sha256DigestFmt, WEBTRANSPORT_ALPN};
 use wtransport::{Endpoint, Identity, ServerConfig};
 
 pub type Result<T, E = store::Error> = core::result::Result<T, E>;
@@ -164,7 +164,7 @@ impl<C: Send + Sync> store::Handler<C> for Handler {
             return Ok(Ok(ResourceOwn::from(id)));
         }
         let (url, suffix) = identifier.split_once(';').unwrap_or((&identifier, ""));
-        let mut url = match Url::parse(url) {
+        let url = match Url::parse(url) {
             Ok(url) => url,
             Err(err) => return Ok(Err(store::Error::Other(err.to_string()))),
         };
@@ -260,14 +260,12 @@ impl<C: Send + Sync> store::Handler<C> for Handler {
                 }
             }
             "wrpc+web" => {
-                if url.set_scheme("https").is_err() {
-                    return Ok(Err(store::Error::Other("invalid URL".to_string())));
-                }
-
+                let mut conf = client_tls_config();
+                conf.alpn_protocols.push(WEBTRANSPORT_ALPN.to_vec());
                 let ep = match wtransport::Endpoint::client(
                     wtransport::ClientConfig::builder()
                         .with_bind_default()
-                        .with_custom_tls(client_tls_config())
+                        .with_custom_tls(conf)
                         .build(),
                 )
                 .context("failed to create WebTransport endpoint")
@@ -276,7 +274,7 @@ impl<C: Send + Sync> store::Handler<C> for Handler {
                     Err(err) => return Ok(Err(store::Error::Other(format!("{err:#}")))),
                 };
                 let conn = match ep
-                    .connect(url)
+                    .connect(format!("https://{}", url.authority()))
                     .await
                     .context("failed to establish connection")
                 {