feat(examples): add UDS KV server

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

Author: rvolosatovs

Cargo.lock

@@ -0,0 +1,39 @@
+[package]
+name = "wasi-keyvalue-unix-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 = [
+    "fs",
+    "rt-multi-thread",
+    "net",
+    "signal",
+] }
+tracing = { workspace = true }
+tracing-subscriber = { workspace = true, features = [
+    "ansi",
+    "env-filter",
+    "fmt",
+] }
+wrpc-transport = { workspace = true, features = ["net"] }
+wrpc-wasi-keyvalue = { workspace = true }
+wrpc-wasi-keyvalue-mem = { workspace = true }

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

@@ -0,0 +1,98 @@
+use core::pin::pin;
+use std::path::PathBuf;
+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::{fs, select, signal};
+use tracing::{debug, error, info, warn};
+
+#[derive(Parser, Debug)]
+#[command(author, version, about, long_about = None)]
+struct Args {
+    /// Path to serve `wasi:keyvalue` on
+    #[arg(default_value = "/tmp/wrpc/wasi/keyvalue.sock")]
+    path: PathBuf,
+}
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    tracing_subscriber::fmt().init();
+
+    let Args { path } = Args::parse();
+
+    if let Some(dir) = path.parent() {
+        if !dir.exists() {
+            fs::create_dir_all(dir)
+                .await
+                .with_context(|| format!("failed to create `{}`", dir.display()))?
+        }
+    }
+    let lis = tokio::net::UnixListener::bind(&path)
+        .with_context(|| format!("failed to bind Unix listener on `{}`", path.display()))?;
+    let srv = Arc::new(wrpc_transport::Server::default());
+    let accept = tokio::spawn({
+        let srv = Arc::clone(&srv);
+        async move {
+            loop {
+                if let Err(err) = srv.accept(&lis).await {
+                    error!(?err, "failed to accept Unix connection");
+                }
+            }
+        }
+    });
+
+    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-unix-server/src/main.rs

@@ -248,10 +248,7 @@ impl<C: Send + Sync> store::Handler<C> for Handler {
                 }
             }
             #[cfg(unix)]
-            "wrpc+uds" => {
-                if url.set_scheme("file").is_err() {
-                    return Ok(Err(store::Error::Other("invalid URL".to_string())));
-                }
+            "wrpc+unix" => {
                 let Ok(path) = url.to_file_path() else {
                     return Ok(Err(store::Error::Other(
                         "failed to get filesystem path from URL".to_string(),