feat: implement AsyncRead/Write

Author: plaflamme

Cargo.lock

@@ -65,7 +65,7 @@ async-client = [
 ## Data structures implement `serde::Serialize` and `serde::Deserialize`.
 serde = ["dep:serde"]
 
-russh = ["dep:russh"]
+russh = ["dep:russh", "dep:tokio"]
 
 [[test]]
 name = "blocking-transport"
@@ -124,6 +124,7 @@ async-std = { version = "1.12.0", optional = true }
 document-features = { version = "0.2.0", optional = true }
 
 russh = { version = "0.53", optional = true }
+tokio = { version = "1", optional = true, default-features = false }
 
 [dev-dependencies]
 gix-pack = { path = "../gix-pack", default-features = false, features = [

gix-transport/Cargo.toml

@@ -1,4 +1,4 @@
-use std::sync::Arc;
+use std::{ops::DerefMut, sync::Arc, task::ready};
 
 use russh::{
     client::Handle,
@@ -9,6 +9,7 @@ pub enum AuthMode {
     UsernamePassword { username: String, password: String },
 }
 
+#[derive(Clone)]
 pub struct Client {
     handle: Arc<Handle<ClientHandler>>,
 }
@@ -37,10 +38,90 @@ impl Client {
             }
         }
     }
+
+    pub async fn open_session(&mut self) -> Result<Session, super::Error> {
+        let channel = self.handle.channel_open_session().await?;
+        let stream = channel.into_stream();
+        Ok(Session {
+            stream: Arc::new(std::sync::Mutex::new(stream)),
+        })
+    }
+}
+
+#[derive(Clone)]
+pub struct Session {
+    stream: Arc<std::sync::Mutex<russh::ChannelStream<russh::client::Msg>>>,
+}
+
+impl Session {
+    fn poll_fn<F, R>(self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>, poll_fn: F) -> std::task::Poll<R>
+    where
+        F: FnOnce(
+            std::pin::Pin<&mut russh::ChannelStream<russh::client::Msg>>,
+            &mut std::task::Context<'_>,
+        ) -> std::task::Poll<R>,
+    {
+        match self.stream.try_lock() {
+            Ok(mut inner) => {
+                let pinned = std::pin::Pin::new(inner.deref_mut());
+                (poll_fn)(pinned, cx)
+            }
+            Err(_) => {
+                cx.waker().wake_by_ref();
+                std::task::Poll::Pending
+            }
+        }
+    }
+}
+
+impl futures_io::AsyncRead for Session {
+    fn poll_read(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+        slice: &mut [u8],
+    ) -> std::task::Poll<std::io::Result<usize>> {
+        self.poll_fn(cx, |pinned, cx| {
+            let mut buf = tokio::io::ReadBuf::new(slice);
+            ready!(tokio::io::AsyncRead::poll_read(pinned, cx, &mut buf))?;
+            std::task::Poll::Ready(Ok(buf.filled().len()))
+        })
+    }
+}
+
+impl futures_io::AsyncWrite for Session {
+    fn poll_write(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+        buf: &[u8],
+    ) -> std::task::Poll<std::io::Result<usize>> {
+        self.poll_fn(cx, |pinned, cx| tokio::io::AsyncWrite::poll_write(pinned, cx, buf))
+    }
+
+    fn poll_flush(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<std::io::Result<()>> {
+        self.poll_fn(cx, tokio::io::AsyncWrite::poll_flush)
+    }
+
+    fn poll_close(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<std::io::Result<()>> {
+        self.poll_fn(cx, tokio::io::AsyncWrite::poll_shutdown)
+    }
 }
 
 struct ClientHandler;
 
 impl Handler for ClientHandler {
     type Error = super::Error;
+
+    async fn check_server_key(
+        &mut self,
+        _server_public_key: &russh::keys::ssh_key::PublicKey,
+    ) -> Result<bool, Self::Error> {
+        // TODO: configurable
+        Ok(true)
+    }
 }

gix-transport/src/client/async_io/ssh/client.rs

@@ -16,7 +16,7 @@ pub struct NativeSsh {
 
     identity: Option<gix_sec::identity::Account>,
     client: Option<client::Client>,
-    connection: Option<crate::client::git::Connection<&'static [u8], Vec<u8>>>,
+    connection: Option<crate::client::git::Connection<client::Session, client::Session>>,
 }
 
 impl TransportWithoutIO for NativeSsh {
@@ -31,7 +31,11 @@ impl TransportWithoutIO for NativeSsh {
         on_into_read: crate::client::MessageKind,
         trace: bool,
     ) -> Result<super::RequestWriter<'_>, crate::client::Error> {
-        todo!()
+        if let Some(connection) = &mut self.connection {
+            connection.request(write_mode, on_into_read, trace)
+        } else {
+            Err(crate::client::Error::MissingHandshake)
+        }
     }
 
     fn to_url(&self) -> std::borrow::Cow<'_, bstr::BStr> {
@@ -44,7 +48,7 @@ impl TransportWithoutIO for NativeSsh {
 
     fn configure(
         &mut self,
-        config: &dyn std::any::Any,
+        _config: &dyn std::any::Any,
     ) -> Result<(), Box<dyn std::error::Error + Send + Sync + 'static>> {
         Ok(())
     }
@@ -72,11 +76,12 @@ impl Transport for NativeSsh {
             None => return Err(crate::client::Error::AuthenticationUnsupported),
         };
 
-        let client = client::Client::connect(host, port, auth_mode).await?;
+        let mut client = client::Client::connect(host, port, auth_mode).await?;
+        let session = client.open_session().await?;
 
         let connection = crate::client::git::Connection::new(
-            [0u8].as_slice(), // TODO
-            vec![],           // TODO
+            session.clone(),
+            session,
             self.desired_version,
             self.url.path.clone(),
             None::<(String, _)>,