chore(deps): upgrade dependencies, fix hyper io traits issue (#252)

crates/krata/Cargo.toml

@@ -15,6 +15,7 @@ bytes = { workspace = true }
 libc = { workspace = true }
 log = { workspace = true }
 once_cell = { workspace = true }
+pin-project-lite = { workspace = true }
 prost = { workspace = true }
 prost-reflect = { workspace = true }
 prost-types = { workspace = true }
@@ -27,6 +28,8 @@ tower = { workspace = true }
 url = { workspace = true }
 
 [target.'cfg(unix)'.dependencies]
+hyper = { workspace = true }
+hyper-util = { workspace = true }
 nix = { workspace = true, features = ["term"] }
 
 [build-dependencies]

crates/krata/src/client.rs

@@ -1,14 +1,10 @@
+#[cfg(unix)]
+use crate::unix::HyperUnixConnector;
 use crate::{dial::ControlDialAddress, v1::control::control_service_client::ControlServiceClient};
 #[cfg(not(unix))]
 use anyhow::anyhow;
 use anyhow::Result;
-#[cfg(unix)]
-use tokio::net::UnixStream;
-#[cfg(unix)]
-use tonic::transport::Uri;
 use tonic::transport::{Channel, ClientTlsConfig, Endpoint};
-#[cfg(unix)]
-use tower::service_fn;
 
 pub struct ControlClientProvider {}
 
@@ -52,10 +48,7 @@ impl ControlClientProvider {
     async fn dial_unix_socket(path: String) -> Result<Channel> {
         // This URL is not actually used but is required to be specified.
         Ok(Endpoint::try_from(format!("unix://localhost/{}", path))?
-            .connect_with_connector(service_fn(|uri: Uri| {
-                let path = uri.path().to_string();
-                UnixStream::connect(path)
-            }))
+            .connect_with_connector(HyperUnixConnector {})
             .await?)
     }
 }

crates/krata/src/lib.rs

@@ -12,6 +12,9 @@ pub mod launchcfg;
 #[cfg(target_os = "linux")]
 pub mod ethtool;
 
+#[cfg(unix)]
+pub mod unix;
+
 pub static DESCRIPTOR_POOL: Lazy<DescriptorPool> = Lazy::new(|| {
     DescriptorPool::decode(
         include_bytes!(concat!(env!("OUT_DIR"), "/file_descriptor_set.bin")).as_ref(),

crates/krata/src/unix.rs

@@ -0,0 +1,73 @@
+use std::future::Future;
+use std::io::Error;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+use hyper::rt::ReadBufCursor;
+use hyper_util::rt::TokioIo;
+use pin_project_lite::pin_project;
+use tokio::io::AsyncWrite;
+use tokio::net::UnixStream;
+use tonic::transport::Uri;
+use tower::Service;
+
+pin_project! {
+    #[derive(Debug)]
+    pub struct HyperUnixStream {
+        #[pin]
+        pub stream: UnixStream,
+    }
+}
+
+impl hyper::rt::Read for HyperUnixStream {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: ReadBufCursor<'_>,
+    ) -> Poll<Result<(), Error>> {
+        let mut tokio = TokioIo::new(self.project().stream);
+        Pin::new(&mut tokio).poll_read(cx, buf)
+    }
+}
+
+impl hyper::rt::Write for HyperUnixStream {
+    fn poll_write(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<Result<usize, Error>> {
+        self.project().stream.poll_write(cx, buf)
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
+        self.project().stream.poll_flush(cx)
+    }
+
+    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
+        self.project().stream.poll_shutdown(cx)
+    }
+}
+
+pub struct HyperUnixConnector;
+
+impl Service<Uri> for HyperUnixConnector {
+    type Response = HyperUnixStream;
+    type Error = Error;
+    #[allow(clippy::type_complexity)]
+    type Future =
+        Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static>>;
+
+    fn call(&mut self, req: Uri) -> Self::Future {
+        let fut = async move {
+            let path = req.path().to_string();
+            let stream = UnixStream::connect(path).await?;
+            Ok(HyperUnixStream { stream })
+        };
+
+        Box::pin(fut)
+    }
+
+    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Poll::Ready(Ok(()))
+    }
+}