feat: implement QUIC transport

Cargo.toml

@@ -103,21 +103,26 @@ clap = { version = "4", default-features = false }
 futures = { version = "0.3", default-features = false }
 heck = { version = "0.5", default-features = false }
 leb128 = { version = "0.2", default-features = false }
+pin-project-lite = { version = "0.2", default-features = false }
 proc-macro2 = { version = "1", default-features = false }
+quinn = { version = "0.11", default-features = false }
 quote = { version = "1", default-features = false }
+rcgen = { version = "0.13", default-features = false }
 reqwest = { version = "0.11", default-features = false }
+rustls = { version = "0.23", default-features = false }
 serde = { version = "1", default-features = false }
 serde_json = { version = "1", default-features = false }
 syn = { version = "2", default-features = false, features = ["printing"] }
 test-helpers = { default-features = false, path = "./crates/test-helpers" }
+test-log = { version = "0.2", default-features = false }
 tokio = { version = "1", default-features = false }
 tokio-stream = { version = "0.1", default-features = false }
 tokio-util = { version = "0.7", default-features = false }
 tower = { version = "0.4", default-features = false }
 tracing = { version = "0.1", default-features = false }
 tracing-subscriber = { version = "0.3", default-features = false }
 url = { version = "2", default-features = false }
-wasm-tokio = { version = "0.4", default-features = false }
+wasm-tokio = { version = "0.5", default-features = false }
 wasmcloud-component-adapters = { version = "0.9", default-features = false }
 wasmparser = { version = "0.208", default-features = false }
 wasmtime = { version = "21", default-features = false }

crates/runtime-wasmtime/src/lib.rs

@@ -20,8 +20,8 @@ use tracing::{error, trace};
 use tracing::{instrument, warn};
 use wasm_tokio::cm::AsyncReadValue as _;
 use wasm_tokio::{
-    AsyncReadCore as _, AsyncReadLeb128 as _, AsyncReadUtf8 as _, CoreStringEncoder, Leb128Encoder,
-    Utf8Encoder,
+    AsyncReadCore as _, AsyncReadLeb128 as _, AsyncReadUtf8 as _, CoreNameEncoder, Leb128Encoder,
+    Utf8Codec,
 };
 use wasmtime::component::types::{self, Case, Field};
 use wasmtime::component::{Linker, ResourceType, Type, Val};
@@ -171,9 +171,9 @@ where
                 Ok(())
             }
             (Val::Char(v), Type::Char) => {
-                Utf8Encoder.encode(*v, dst).context("failed to encode char")
+                Utf8Codec.encode(*v, dst).context("failed to encode char")
             }
-            (Val::String(v), Type::String) => CoreStringEncoder
+            (Val::String(v), Type::String) => CoreNameEncoder
                 .encode(v.as_str(), dst)
                 .context("failed to encode string"),
             (Val::List(vs), Type::List(ty)) => {
@@ -502,14 +502,14 @@ where
                         let RemoteResource(id) = table
                             .delete(resource)
                             .context("failed to delete remote resource")?;
-                        CoreStringEncoder
+                        CoreNameEncoder
                             .encode(id, dst)
                             .context("failed to encode resource ID")
                     } else {
                         let RemoteResource(id) = table
                             .get(&resource)
                             .context("failed to get remote resource")?;
-                        CoreStringEncoder
+                        CoreNameEncoder
                             .encode(id.as_str(), dst)
                             .context("failed to encode resource ID")
                     }
@@ -604,7 +604,7 @@ where
         }
         Type::String => {
             let mut s = String::default();
-            r.read_core_string(&mut s).await?;
+            r.read_core_name(&mut s).await?;
             *val = Val::String(s);
             Ok(())
         }
@@ -787,7 +787,7 @@ where
             } else {
                 let mut store = store.as_context_mut();
                 let mut s = String::default();
-                r.read_core_string(&mut s).await?;
+                r.read_core_name(&mut s).await?;
                 let table = store.data_mut().table();
                 let resource = table
                     .push(RemoteResource(s))

crates/transport-nats/src/lib.rs

@@ -21,7 +21,7 @@ use tokio::try_join;
 use tokio_util::codec::Encoder;
 use tokio_util::io::StreamReader;
 use tracing::{instrument, trace, warn};
-use wasm_tokio::{AsyncReadCore as _, CoreStringEncoder};
+use wasm_tokio::{AsyncReadCore as _, CoreNameEncoder};
 use wrpc_transport::Index as _;
 
 pub const PROTOCOL: &str = "wrpc.0.0.1";
@@ -219,14 +219,14 @@ impl SubscriberTree {
         };
         match self {
             Self::Empty | Self::Leaf(..) => None,
+            Self::IndexNode { ref mut nested, .. } => nested
+                .get_mut(*i)
+                .and_then(|nested| nested.as_mut().and_then(|nested| nested.take(path))),
             Self::WildcardNode { .. } => None,
             // TODO: Demux the subscription
             //Self::WildcardNode { ref mut nested, .. } => {
             //    nested.as_mut().and_then(|nested| nested.take(path))
             //}
-            Self::IndexNode { ref mut nested, .. } => nested
-                .get_mut(*i)
-                .and_then(|nested| nested.as_mut().and_then(|nested| nested.take(path))),
         }
     }
 
@@ -852,10 +852,10 @@ impl<O: AsyncWrite + Send> wrpc_transport::Session for Session<O> {
     ) -> Result<Result<(), Self::Error>, Self::TransportError> {
         if let Err(err) = res {
             let mut buf = BytesMut::with_capacity(5 + err.len());
-            if let Err(err) = CoreStringEncoder.encode(err, &mut buf) {
+            if let Err(err) = CoreNameEncoder.encode(err, &mut buf) {
                 warn!(?err, "failed to encode error");
                 buf.clear();
-                if let Err(err) = CoreStringEncoder.encode(err.to_string(), &mut buf) {
+                if let Err(err) = CoreNameEncoder.encode(err.to_string(), &mut buf) {
                     warn!(?err, "failed to encode encoding error");
                     buf.clear();
                 }
@@ -875,7 +875,7 @@ impl<O: AsyncWrite + Send> wrpc_transport::Session for Session<O> {
         if (incoming.as_mut().peek().await).is_some() {
             let mut err = String::new();
             StreamReader::new(incoming)
-                .read_core_string(&mut err)
+                .read_core_name(&mut err)
                 .await
                 .context("failed to read error string")?;
             Ok(Err(err))

crates/transport-quic/Cargo.toml

@@ -0,0 +1,35 @@
+[package]
+name = "wrpc-transport-quic"
+version = "0.1.0"
+description = "wRPC QUIC transport"
+
+authors.workspace = true
+categories.workspace = true
+edition.workspace = true
+license.workspace = true
+repository.workspace = true
+
+[dependencies]
+anyhow = { workspace = true, features = ["std"] }
+bytes = { workspace = true }
+futures = { workspace = true }
+pin-project-lite = { workspace = true }
+quinn = { workspace = true, features = [
+    "log",
+    "platform-verifier",
+    "ring",
+    "runtime-tokio",
+    "rustls",
+] }
+tokio = { workspace = true, features = ["macros"] }
+tokio-stream = { workspace = true }
+tokio-util = { workspace = true, features = ["codec", "io"] }
+tracing = { workspace = true, features = ["attributes"] }
+wasm-tokio = { workspace = true }
+wrpc-transport = { workspace = true }
+
+[dev-dependencies]
+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 = ["rt-multi-thread"] }