feat(wit-bindgen-rust): add async support

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

Author: rvolosatovs

crates/introspect/src/lib.rs

@@ -163,10 +163,14 @@ pub fn async_paths_tyid(resolve: &Resolve, id: TypeId) -> (BTreeSet<VecDeque<Opt
         TypeDefKind::Future(ty) => {
             let mut paths = BTreeSet::default();
             if let Some(ty) = ty {
-                let (nested, _) = async_paths_ty(resolve, ty);
-                for path in nested {
+                let (nested, fut) = async_paths_ty(resolve, ty);
+                for mut path in nested {
+                    path.push_front(Some(0));
                     paths.insert(path);
                 }
+                if fut {
+                    paths.insert(vec![Some(0)].into());
+                }
             }
             (paths, true)
         }

crates/transport-nats/src/lib.rs

@@ -18,7 +18,7 @@ use futures::{Stream, StreamExt};
 use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
 use tokio::sync::oneshot;
 use tokio::try_join;
-use tracing::{instrument, trace, warn};
+use tracing::{debug, instrument, trace, warn};
 use wrpc_transport::Index as _;
 
 pub const PROTOCOL: &str = "wrpc.0.0.1";
@@ -314,11 +314,15 @@ pub struct Reader {
 impl wrpc_transport::Index<Self> for Reader {
     #[instrument(level = "trace", skip(self))]
     fn index(&self, path: &[usize]) -> anyhow::Result<Self> {
+        trace!("locking index tree");
         let mut nested = self
             .nested
             .lock()
             .map_err(|err| anyhow!(err.to_string()).context("failed to lock map"))?;
-        let incoming = nested.take(path).context("unknown subscription")?;
+        trace!("taking index subscription");
+        let incoming = nested
+            .take(path)
+            .with_context(|| format!("unknown subscription for path `{path:?}`"))?;
         Ok(Self {
             buffer: Bytes::default(),
             incoming,
@@ -391,7 +395,7 @@ impl SubjectWriter {
 impl wrpc_transport::Index<Self> for SubjectWriter {
     #[instrument(level = "trace", skip(self))]
     fn index(&self, path: &[usize]) -> anyhow::Result<Self> {
-        let tx: Subject = index_path(self.tx.as_str(), path).into();
+        let tx = Subject::from(index_path(self.tx.as_str(), path));
         let publisher = self.nats.publish_sink(tx.clone());
         Ok(Self {
             nats: Arc::clone(&self.nats),
@@ -402,7 +406,7 @@ impl wrpc_transport::Index<Self> for SubjectWriter {
 }
 
 impl AsyncWrite for SubjectWriter {
-    #[instrument(level = "trace", skip_all, ret, fields(subject = ?self.tx, buf = format!("{buf:02x?}")))]
+    #[instrument(level = "trace", skip_all, ret, fields(subject = self.tx.as_str(), buf = format!("{buf:02x?}")))]
     fn poll_write(
         mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
@@ -436,15 +440,15 @@ impl AsyncWrite for SubjectWriter {
         }
     }
 
-    #[instrument(level = "trace", skip_all, ret, fields(subject = ?self.tx))]
+    #[instrument(level = "trace", skip_all, ret, fields(subject = self.tx.as_str()))]
     fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
         trace!("flushing");
         self.publisher
             .poll_flush_unpin(cx)
             .map_err(|_| std::io::ErrorKind::BrokenPipe.into())
     }
 
-    #[instrument(level = "trace", skip_all, ret, fields(subject = ?self.tx))]
+    #[instrument(level = "trace", skip_all, ret, fields(subject = self.tx.as_str()))]
     fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
         trace!("writing empty buffer to shut down stream");
         ready!(self.as_mut().poll_write(cx, &[]))?;
@@ -822,16 +826,17 @@ impl wrpc_transport::Invoke for Client {
     type Incoming = Reader;
 
     #[instrument(level = "trace", skip(self, paths, params), fields(params = format!("{params:02x?}")))]
-    async fn invoke(
+    async fn invoke<P: AsRef<[Option<usize>]> + Send + Sync>(
         &self,
         cx: Self::Context,
         instance: &str,
         func: &str,
         mut params: Bytes,
-        paths: &[impl AsRef<[Option<usize>]> + Send + Sync],
+        paths: impl AsRef<[P]> + Send,
     ) -> anyhow::Result<(Self::Outgoing, Self::Incoming)> {
         let rx = Subject::from(self.nats.new_inbox());
         let result_rx = Subject::from(result_subject(&rx));
+        let paths = paths.as_ref();
         let (result_rx, handshake_rx, nested) = try_join!(
             async {
                 self.nats
@@ -930,15 +935,20 @@ async fn serve_connection(
     let tx = tx.context("peer did not specify a reply subject")?;
     let rx = nats.new_inbox();
     let param_rx = Subject::from(param_subject(&rx));
-    trace!("subscribing on subjects");
     let (param_rx, nested) = try_join!(
         async {
+            trace!(
+                subject = param_rx.as_str(),
+                "subscribing on parameter subject"
+            );
             nats.subscribe(param_rx.clone())
                 .await
                 .context("failed to subscribe on parameter subject")
         },
         try_join_all(paths.iter().map(|path| async {
-            nats.subscribe(Subject::from(subscribe_path(&param_rx, path.as_ref())))
+            let subject = subscribe_path(&param_rx, path.as_ref());
+            trace!(?subject, "subscribing on nested parameter subject");
+            nats.subscribe(Subject::from(subject))
                 .await
                 .context("failed to subscribe on nested parameter subject")
         }))
@@ -974,20 +984,22 @@ impl wrpc_transport::Serve for Client {
     type Incoming = Reader;
 
     #[instrument(level = "trace", skip(self, paths))]
-    async fn serve<P: AsRef<[Option<usize>]> + Send + Sync + 'static>(
+    async fn serve(
         &self,
         instance: &str,
         func: &str,
-        paths: impl Into<Arc<[P]>> + Send + Sync + 'static,
+        paths: impl Into<Arc<[Box<[Option<usize>]>]>> + Send,
     ) -> anyhow::Result<
         impl Stream<Item = anyhow::Result<(Self::Context, Self::Outgoing, Self::Incoming)>> + 'static,
     > {
         let subject = invocation_subject(&self.prefix, instance, func);
         let sub = if let Some(group) = &self.queue_group {
+            debug!(subject, ?group, "queue-subscribing on invocation subject");
             self.nats
                 .queue_subscribe(subject, group.to_string())
                 .await?
         } else {
+            debug!(subject, "subscribing on invocation subject");
             self.nats.subscribe(subject).await?
         };
         let paths = paths.into();

crates/transport-quic/src/lib.rs

@@ -404,6 +404,7 @@ impl wrpc_transport::Index<Self> for Incoming {
                 let mut lock = index.lock().map_err(|err| {
                     std::io::Error::new(std::io::ErrorKind::Other, err.to_string())
                 })?;
+                trace!("taking index subscription");
                 let rx = lock.take_rx(&path).ok_or_else(|| {
                     std::io::Error::new(
                         std::io::ErrorKind::NotFound,
@@ -674,14 +675,17 @@ impl wrpc_transport::Invoke for Client {
     type Incoming = Incoming;
 
     #[instrument(level = "trace", skip(self, paths, params), fields(params = format!("{params:02x?}")))]
-    async fn invoke(
+    async fn invoke<P>(
         &self,
         cx: Self::Context,
         instance: &str,
         func: &str,
         params: Bytes,
-        paths: &[impl AsRef<[Option<usize>]> + Send + Sync],
-    ) -> anyhow::Result<(Self::Outgoing, Self::Incoming)> {
+        paths: impl AsRef<[P]> + Send,
+    ) -> anyhow::Result<(Self::Outgoing, Self::Incoming)>
+    where
+        P: AsRef<[Option<usize>]> + Send + Sync,
+    {
         let san = san(instance, func);
         trace!(?san, "establishing connection");
         let conn = self
@@ -695,7 +699,7 @@ impl wrpc_transport::Invoke for Client {
             .open_bi()
             .await
             .context("failed to open parameter stream")?;
-        let index = Arc::new(std::sync::Mutex::new(paths.iter().collect()));
+        let index = Arc::new(std::sync::Mutex::new(paths.as_ref().iter().collect()));
         let mut io = JoinSet::new();
         io.spawn(demux_connection(Arc::clone(&index), conn.clone()).in_current_span());
         let io = Arc::new(io);
@@ -765,11 +769,11 @@ impl wrpc_transport::Serve for Server {
     type Incoming = Incoming;
 
     #[instrument(level = "trace", skip(self, paths))]
-    async fn serve<P: AsRef<[Option<usize>]> + Send + Sync + 'static>(
+    async fn serve(
         &self,
         instance: &str,
         func: &str,
-        paths: impl Into<Arc<[P]>> + Send + Sync + 'static,
+        paths: impl Into<Arc<[Box<[Option<usize>]>]>> + Send,
     ) -> anyhow::Result<
         impl Stream<Item = anyhow::Result<(Self::Context, Self::Outgoing, Self::Incoming)>> + 'static,
     > {

crates/transport-quic/tests/loopback.rs

@@ -66,7 +66,7 @@ async fn loopback() -> anyhow::Result<()> {
     let clt = Client::new(clt_ep, (Ipv4Addr::LOCALHOST, srv_addr.port()));
     let srv = Server::default();
     let invocations = srv
-        .serve("foo", "bar", [[Some(42), Some(0)]])
+        .serve("foo", "bar", [Box::from([Some(42), Some(0)])])
         .await
         .context("failed to serve `foo.bar`")?;
     let mut invocations = pin!(invocations);

crates/transport/Cargo.toml

@@ -10,16 +10,19 @@ license.workspace = true
 repository.workspace = true
 
 [features]
-default = ["frame"]
+default = ["frame", "fs", "net", "io-std"]
 frame = []
+fs = ["tokio/fs"]
+net = ["tokio/net"]
+io-std = ["tokio/io-std"]
 
 [dependencies]
 anyhow = { workspace = true, features = ["std"] }
 bytes = { workspace = true }
 futures = { workspace = true, features = ["std"] }
 tokio = { workspace = true }
 tokio-stream = { workspace = true }
-tokio-util = { workspace = true, features = ["codec"] }
+tokio-util = { workspace = true, features = ["codec", "io"] }
 tracing = { workspace = true, features = ["attributes"] }
 wasm-tokio = { workspace = true, features = ["tracing"] }