fix: adapt to ttrpc 0.9.0 async API changes

Signed-off-by: Maksym Pavlenko <mxpv@users.noreply.github.com>

Author: mxpv

crates/shim-protos/examples/connect-async.rs

@@ -32,7 +32,9 @@ async fn main() {
     let pid = args.get(2).map(|str| str.to_owned()).unwrap_or_default();
 
     println!("Connecting to {}...", socket_path);
-    let client = Client::connect(socket_path).expect("Failed to connect to shim");
+    let client = Client::connect(socket_path)
+        .await
+        .expect("Failed to connect to shim");
 
     let task_client = TaskClient::new(client);
 

crates/shim-protos/examples/ttrpc-client-async.rs

@@ -30,7 +30,9 @@ fn default_ctx() -> Context {
 
 #[tokio::main]
 async fn main() {
-    let c = Client::connect("unix:///tmp/shim-proto-ttrpc-001").unwrap();
+    let c = Client::connect("unix:///tmp/shim-proto-ttrpc-001")
+        .await
+        .unwrap();
     let task = TaskClient::new(c);
     let now = std::time::Instant::now();
 

crates/shim/src/asynchronous/mod.rs

@@ -349,11 +349,11 @@ pub async fn spawn(opts: StartOpts, grouping: &str, vars: Vec<(&str, &str)>) ->
 
 #[cfg_attr(feature = "tracing", tracing::instrument(skip_all, level = "info"))]
 async fn create_server(flags: &args::Flags) -> Result<Server> {
-    use std::os::fd::IntoRawFd;
+    use containerd_shim_protos::ttrpc::r#async::transport::Listener;
     let listener = start_listener(&flags.socket).await?;
-    let mut server = Server::new();
-    server = server.add_listener(listener.into_raw_fd())?;
-    server = server.set_domain_unix();
+    let listener =
+        Listener::try_from(listener).map_err(io_error!(e, "creating ttrpc listener"))?;
+    let server = Server::new().add_listener(listener);
     Ok(server)
 }
 
@@ -543,7 +543,7 @@ async fn start_listener(address: &str) -> Result<UnixListener> {
 #[cfg_attr(feature = "tracing", tracing::instrument(level = "info"))]
 async fn wait_socket_working(address: &str, interval_in_ms: u64, count: u32) -> Result<()> {
     for _i in 0..count {
-        match Client::connect(address) {
+        match Client::connect(address).await {
             Ok(_) => {
                 return Ok(());
             }

crates/shim/src/asynchronous/publisher.rs

@@ -134,8 +134,9 @@ impl RemotePublisher {
         })
         .await?;
 
-        // Client::new() takes ownership of the RawFd.
-        Ok(Client::new(fd))
+        // Safety: `fd` is a unix socket returned by `connect()`.
+        // `from_raw_unix_socket_fd` takes ownership of the RawFd.
+        Ok(unsafe { Client::from_raw_unix_socket_fd(fd) })
     }
 
     /// Publish a new event.
@@ -195,17 +196,14 @@ impl Events for RemotePublisher {
 
 #[cfg(test)]
 mod tests {
-    use std::{
-        os::unix::{io::AsRawFd, net::UnixListener},
-        sync::Arc,
-    };
+    use std::{os::unix::net::UnixListener, sync::Arc};
 
     use async_trait::async_trait;
     use containerd_shim_protos::{
         api::{Empty, ForwardRequest},
         events::task::TaskOOM,
         shim_async::{create_events, Events},
-        ttrpc::asynchronous::Server,
+        ttrpc::asynchronous::{transport::Listener, Server},
     };
     use tokio::sync::{
         mpsc::{channel, Sender},
@@ -247,13 +245,11 @@ mod tests {
         let barrier2 = barrier.clone();
         let server_thread = tokio::spawn(async move {
             let listener = UnixListener::bind(&path1).unwrap();
+            let listener = Listener::try_from(listener).unwrap();
             let service = create_events(Arc::new(server));
             let mut server = Server::new()
-                .set_domain_unix()
-                .add_listener(listener.as_raw_fd())
-                .unwrap()
+                .add_listener(listener)
                 .register_service(service);
-            std::mem::forget(listener);
             server.start().await.unwrap();
             barrier2.wait().await;