Unfinished

distant-net/src/client/channel.rs

@@ -141,8 +141,8 @@ mod tests {
 
     #[tokio::test]
     async fn mail_should_return_mailbox_that_receives_responses_until_transport_closes() {
-        let (t1, mut t2) = FramedTransport::pair(100);
-        let session: TestClient = Client::from_framed_transport(t1).unwrap();
+        let (t1, mut t2) = FramedTransport::test_pair(100);
+        let session = TestClient::new(t1);
         let mut channel = session.clone_channel();
 
         let req = Request::new(0);
@@ -151,13 +151,13 @@ mod tests {
         let mut mailbox = channel.mail(req).await.unwrap();
 
         // Get first response
-        match tokio::join!(mailbox.next(), t2.write(res.clone())) {
+        match tokio::join!(mailbox.next(), t2.write_frame(res.to_vec().unwrap())) {
             (Some(actual), _) => assert_eq!(actual, res),
             x => panic!("Unexpected response: {:?}", x),
         }
 
         // Get second response
-        match tokio::join!(mailbox.next(), t2.write(res.clone())) {
+        match tokio::join!(mailbox.next(), t2.write_frame(res.to_vec().unwrap())) {
             (Some(actual), _) => assert_eq!(actual, res),
             x => panic!("Unexpected response: {:?}", x),
         }
@@ -176,14 +176,14 @@ mod tests {
 
     #[tokio::test]
     async fn send_should_wait_until_response_received() {
-        let (t1, mut t2) = FramedTransport::pair(100);
-        let session: TestClient = Client::from_framed_transport(t1).unwrap();
+        let (t1, mut t2) = FramedTransport::test_pair(100);
+        let session = TestClient::new(t1);
         let mut channel = session.clone_channel();
 
         let req = Request::new(0);
         let res = Response::new(req.id.clone(), 1);
 
-        let (actual, _) = tokio::join!(channel.send(req), t2.write(res.clone()));
+        let (actual, _) = tokio::join!(channel.send(req), t2.write_frame(res.to_vec().unwrap()));
         match actual {
             Ok(actual) => assert_eq!(actual, res),
             x => panic!("Unexpected response: {:?}", x),
@@ -192,8 +192,8 @@ mod tests {
 
     #[tokio::test]
     async fn send_timeout_should_fail_if_response_not_received_in_time() {
-        let (t1, mut t2) = FramedTransport::pair(100);
-        let session: TestClient = Client::from_framed_transport(t1).unwrap();
+        let (t1, mut t2) = FramedTransport::test_pair(100);
+        let session: TestClient = Client::new(t1);
         let mut channel = session.clone_channel();
 
         let req = Request::new(0);
@@ -208,8 +208,8 @@ mod tests {
 
     #[tokio::test]
     async fn fire_should_send_request_and_not_wait_for_response() {
-        let (t1, mut t2) = FramedTransport::pair(100);
-        let session: TestClient = Client::from_framed_transport(t1).unwrap();
+        let (t1, mut t2) = FramedTransport::test_pair(100);
+        let session: TestClient = Client::new(t1);
         let mut channel = session.clone_channel();
 
         let req = Request::new(0);

distant-net/src/client/ext/tcp.rs

@@ -1,9 +1,10 @@
-use crate::{BoxedCodec, Client, FramedTransport, TcpTransport};
+use crate::{Client, FramedTransport, TcpTransport};
 use async_trait::async_trait;
 use serde::{de::DeserializeOwned, Serialize};
 use std::{convert, net::SocketAddr};
 use tokio::{io, time::Duration};
 
+/// Interface that provides ability to connect to a TCP server
 #[async_trait]
 pub trait TcpClientExt<T, U>
 where
@@ -11,16 +12,11 @@ where
     U: DeserializeOwned + Send + Sync,
 {
     /// Connect to a remote TCP server using the provided information
-    async fn connect(addr: SocketAddr, codec: impl Into<BoxedCodec>) -> io::Result<Client<T, U>>;
+    async fn connect(addr: SocketAddr) -> io::Result<Client<T, U>>;
 
     /// Connect to a remote TCP server, timing out after duration has passed
-    async fn connect_timeout<C>(
-        addr: SocketAddr,
-        codec: impl Into<BoxedCodec> + Send,
-        duration: Duration,
-    ) -> io::Result<Client<T, U>> {
-        let codec = codec.into();
-        tokio::time::timeout(duration, Self::connect(addr, codec))
+    async fn connect_timeout<C>(addr: SocketAddr, duration: Duration) -> io::Result<Client<T, U>> {
+        tokio::time::timeout(duration, Self::connect(addr))
             .await
             .map_err(|x| io::Error::new(io::ErrorKind::TimedOut, x))
             .and_then(convert::identity)
@@ -34,9 +30,14 @@ where
     U: Send + Sync + DeserializeOwned + 'static,
 {
     /// Connect to a remote TCP server using the provided information
-    async fn connect(addr: SocketAddr, codec: impl Into<BoxedCodec>) -> io::Result<Client<T, U>> {
+    async fn connect(addr: SocketAddr) -> io::Result<Client<T, U>> {
         let transport = TcpTransport::connect(addr).await?;
-        let transport = FramedTransport::new(transport, codec);
-        Self::new(transport)
+
+        // Establish our framed transport and perform a handshake to set the codec
+        // NOTE: Using default capacity
+        let mut transport = FramedTransport::<_>::plain(transport);
+        transport.client_handshake().await?;
+
+        Ok(Self::new(transport))
     }
 }

distant-net/src/client/ext/unix.rs

@@ -1,4 +1,4 @@
-use crate::{Client, Codec, FramedTransport, UnixSocketTransport};
+use crate::{Client, FramedTransport, UnixSocketTransport};
 use async_trait::async_trait;
 use serde::{de::DeserializeOwned, Serialize};
 use std::{convert, path::Path};
@@ -11,22 +11,16 @@ where
     U: DeserializeOwned + Send + Sync,
 {
     /// Connect to a proxy unix socket
-    async fn connect<P, C>(path: P, codec: C) -> io::Result<Client<T, U>>
+    async fn connect<P>(path: P) -> io::Result<Client<T, U>>
     where
-        P: AsRef<Path> + Send,
-        C: Codec + Send + 'static;
+        P: AsRef<Path> + Send;
 
     /// Connect to a proxy unix socket, timing out after duration has passed
-    async fn connect_timeout<P, C>(
-        path: P,
-        codec: C,
-        duration: Duration,
-    ) -> io::Result<Client<T, U>>
+    async fn connect_timeout<P>(path: P, duration: Duration) -> io::Result<Client<T, U>>
     where
         P: AsRef<Path> + Send,
-        C: Codec + Send + 'static,
     {
-        tokio::time::timeout(duration, Self::connect(path, codec))
+        tokio::time::timeout(duration, Self::connect(path))
             .await
             .map_err(|x| io::Error::new(io::ErrorKind::TimedOut, x))
             .and_then(convert::identity)
@@ -40,15 +34,18 @@ where
     U: Send + Sync + DeserializeOwned + 'static,
 {
     /// Connect to a proxy unix socket
-    async fn connect<P, C>(path: P, codec: C) -> io::Result<Client<T, U>>
+    async fn connect<P>(path: P) -> io::Result<Client<T, U>>
     where
         P: AsRef<Path> + Send,
-        C: Codec + Send + 'static,
     {
         let p = path.as_ref();
         let transport = UnixSocketTransport::connect(p).await?;
-        let transport = FramedTransport::new(transport, codec);
-        let (writer, reader) = transport.into_split();
-        Ok(Client::new(writer, reader)?)
+
+        // Establish our framed transport and perform a handshake to set the codec
+        // NOTE: Using default capacity
+        let mut transport = FramedTransport::<_>::plain(transport);
+        transport.client_handshake().await?;
+
+        Ok(Client::new(transport))
     }
 }

distant-net/src/client/ext/windows.rs

@@ -1,4 +1,4 @@
-use crate::{Client, Codec, FramedTransport, WindowsPipeTransport};
+use crate::{Client, FramedTransport, WindowsPipeTransport};
 use async_trait::async_trait;
 use serde::{de::DeserializeOwned, Serialize};
 use std::{
@@ -15,54 +15,42 @@ where
 {
     /// Connect to a server listening on a Windows pipe at the specified address
     /// using the given codec
-    async fn connect<A, C>(addr: A, codec: C) -> io::Result<Client<T, U>>
+    async fn connect<A>(addr: A) -> io::Result<Client<T, U>>
     where
-        A: AsRef<OsStr> + Send,
-        C: Codec + Send + 'static;
+        A: AsRef<OsStr> + Send;
 
     /// Connect to a server listening on a Windows pipe at the specified address
     /// via `\\.\pipe\{name}` using the given codec
-    async fn connect_local<N, C>(name: N, codec: C) -> io::Result<Client<T, U>>
+    async fn connect_local<N>(name: N) -> io::Result<Client<T, U>>
     where
         N: AsRef<OsStr> + Send,
-        C: Codec + Send + 'static,
     {
         let mut addr = OsString::from(r"\\.\pipe\");
         addr.push(name.as_ref());
-        Self::connect(addr, codec).await
+        Self::connect(addr).await
     }
 
     /// Connect to a server listening on a Windows pipe at the specified address
     /// using the given codec, timing out after duration has passed
-    async fn connect_timeout<A, C>(
-        addr: A,
-        codec: C,
-        duration: Duration,
-    ) -> io::Result<Client<T, U>>
+    async fn connect_timeout<A>(addr: A, duration: Duration) -> io::Result<Client<T, U>>
     where
         A: AsRef<OsStr> + Send,
-        C: Codec + Send + 'static,
     {
-        tokio::time::timeout(duration, Self::connect(addr, codec))
+        tokio::time::timeout(duration, Self::connect(addr))
             .await
             .map_err(|x| io::Error::new(io::ErrorKind::TimedOut, x))
             .and_then(convert::identity)
     }
 
     /// Connect to a server listening on a Windows pipe at the specified address
-    /// via `\\.\pipe\{name}` using the given codec, timing out after duration has passed
-    async fn connect_local_timeout<N, C>(
-        name: N,
-        codec: C,
-        duration: Duration,
-    ) -> io::Result<Client<T, U>>
+    /// via `\\.\pipe\{name}`, timing out after duration has passed
+    async fn connect_local_timeout<N>(name: N, duration: Duration) -> io::Result<Client<T, U>>
     where
         N: AsRef<OsStr> + Send,
-        C: Codec + Send + 'static,
     {
         let mut addr = OsString::from(r"\\.\pipe\");
         addr.push(name.as_ref());
-        Self::connect_timeout(addr, codec, duration).await
+        Self::connect_timeout(addr, duration).await
     }
 }
 
@@ -72,15 +60,18 @@ where
     T: Send + Sync + Serialize + 'static,
     U: Send + Sync + DeserializeOwned + 'static,
 {
-    async fn connect<A, C>(addr: A, codec: C) -> io::Result<Client<T, U>>
+    async fn connect<A>(addr: A) -> io::Result<Client<T, U>>
     where
         A: AsRef<OsStr> + Send,
-        C: Codec + Send + 'static,
     {
         let a = addr.as_ref();
         let transport = WindowsPipeTransport::connect(a).await?;
-        let transport = FramedTransport::new(transport, codec);
-        let (writer, reader) = transport.into_split();
-        Ok(Client::new(writer, reader)?)
+
+        // Establish our framed transport and perform a handshake to set the codec
+        // NOTE: Using default capacity
+        let mut transport = FramedTransport::<_>::plain(transport);
+        transport.client_handshake().await?;
+
+        Ok(Client::new(transport))
     }
 }

distant-net/src/server/ext.rs

@@ -1,7 +1,7 @@
 use crate::{
-    utils::Timer, ConnectionId, FramedTransport, GenericServerRef, Interest, Listener, PlainCodec,
-    Response, Server, ServerConnection, ServerCtx, ServerRef, ServerReply, ServerState, Shutdown,
-    Transport, UntypedRequest,
+    utils::Timer, ConnectionId, FramedTransport, GenericServerRef, Interest, Listener, Response,
+    Server, ServerConnection, ServerCtx, ServerRef, ServerReply, ServerState, Shutdown, Transport,
+    UntypedRequest,
 };
 use log::*;
 use serde::{de::DeserializeOwned, Serialize};
@@ -207,17 +207,11 @@ where
         let (tx, mut rx) = mpsc::channel::<Response<S::Response>>(1);
 
         // Perform a handshake to ensure that the connection is properly established
-        let mut transport = match self
-            .server
-            .on_handshake(FramedTransport::new(self.transport, Box::new(PlainCodec)))
-            .await
-        {
-            Ok(transport) => transport,
-            Err(x) => {
-                error!("[Conn {connection_id}] Handshake failed: {x}");
-                return;
-            }
-        };
+        let mut transport: FramedTransport<T> = FramedTransport::plain(self.transport);
+        if let Err(x) = transport.server_handshake().await {
+            error!("[Conn {connection_id}] Handshake failed: {x}");
+            return;
+        }
 
         loop {
             let ready = transport

distant-net/src/server/ext/tcp.rs

@@ -41,7 +41,7 @@ where
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::{Client, PlainCodec, Request, ServerCtx, TcpClientExt};
+    use crate::{Client, Request, ServerCtx, TcpClientExt};
     use std::net::{Ipv6Addr, SocketAddr};
 
     pub struct TestServer;
@@ -67,12 +67,10 @@ mod tests {
             .await
             .expect("Failed to start TCP server");
 
-        let mut client: Client<String, String> = Client::connect(
-            SocketAddr::from((server.ip_addr(), server.port())),
-            PlainCodec,
-        )
-        .await
-        .expect("Client failed to connect");
+        let mut client: Client<String, String> =
+            Client::connect(SocketAddr::from((server.ip_addr(), server.port())))
+                .await
+                .expect("Client failed to connect");
 
         let response = client
             .send(Request::new("hello".to_string()))

distant-net/src/transport/framed.rs

@@ -122,6 +122,129 @@ where
 
         Ok(())
     }
+}
+
+impl<T, const CAPACITY: usize> FramedTransport<T, CAPACITY>
+where
+    T: Transport,
+{
+    /// Reads a frame of bytes by using the [`Codec`] tied to this transport. Returns
+    /// `Ok(Some(frame))` upon reading a frame, or `Ok(None)` if the underlying transport has
+    /// closed.
+    ///
+    /// This call may return an error with [`ErrorKind::WouldBlock`] in the case that the transport
+    /// is not ready to read data or has not received a full frame before waiting.
+    ///
+    /// [`ErrorKind::WouldBlock`]: io::ErrorKind::WouldBlock
+    pub fn try_read_frame(&mut self) -> io::Result<Option<OwnedFrame>> {
+        // Continually read bytes into the incoming queue and then attempt to tease out a frame
+        let mut buf = [0; CAPACITY];
+
+        loop {
+            match self.inner.try_read(&mut buf) {
+                // Getting 0 bytes on read indicates the channel has closed. If we were still
+                // expecting more bytes for our frame, then this is an error, otherwise if we
+                // have nothing remaining if our queue then this is an expected end and we
+                // return None
+                Ok(0) if self.incoming.is_empty() => return Ok(None),
+                Ok(0) => return Err(io::Error::from(io::ErrorKind::UnexpectedEof)),
+
+                // Got some additional bytes, which we will add to our queue and then attempt to
+                // decode into a frame
+                Ok(n) => {
+                    self.incoming.extend_from_slice(&buf[..n]);
+
+                    // Attempt to read a frame, returning the decoded frame if we get one,
+                    // continuing to try to read more bytes if we don't find a frame, and returning
+                    // any error that is encountered from reading frames or failing to decode
+                    let frame = match Frame::read(&mut self.incoming) {
+                        Ok(Some(frame)) => frame,
+                        Ok(None) => continue,
+                        Err(x) => return Err(x),
+                    };
+
+                    return Ok(Some(self.codec.decode(frame)?.into_owned()));
+                }
+
+                // Any error (including WouldBlock) will get bubbled up
+                Err(x) => return Err(x),
+            }
+        }
+    }
+
+    /// Continues to invoke [`try_read_frame`] until a frame is successfully read, an error is
+    /// encountered that is not [`ErrorKind::WouldBlock`], or the underlying transport has closed.
+    ///
+    /// [`try_read_frame`]: FramedTransport::try_read_frame
+    /// [`ErrorKind::WouldBlock`]: io::ErrorKind::WouldBlock
+    pub async fn read_frame(&mut self) -> io::Result<Option<OwnedFrame>> {
+        loop {
+            self.readable().await?;
+
+            match self.try_read_frame() {
+                Err(x) if x.kind() == io::ErrorKind::WouldBlock => continue,
+                x => return x,
+            }
+        }
+    }
+
+    /// Writes a `frame` of bytes by using the [`Codec`] tied to this transport.
+    ///
+    /// This is accomplished by continually calling the inner transport's `try_write`. If 0 is
+    /// returned from a call to `try_write`, this will fail with [`ErrorKind::WriteZero`].
+    ///
+    /// This call may return an error with [`ErrorKind::WouldBlock`] in the case that the transport
+    /// is not ready to write data or has not written the entire frame before waiting.
+    ///
+    /// [`ErrorKind::WriteZero`]: io::ErrorKind::WriteZero
+    /// [`ErrorKind::WouldBlock`]: io::ErrorKind::WouldBlock
+    pub fn try_write_frame<'a>(&mut self, frame: impl Into<Frame<'a>>) -> io::Result<()> {
+        // Encode the frame and store it in our outgoing queue
+        let frame = self.codec.encode(frame.into())?;
+        frame.write(&mut self.outgoing)?;
+
+        // Attempt to write everything in our queue
+        self.try_flush()
+    }
+
+    /// Invokes [`try_write_frame`] followed by a continuous calls to [`try_flush`] until a frame
+    /// is successfully written, an error is encountered that is not [`ErrorKind::WouldBlock`], or
+    /// the underlying transport has closed.
+    ///
+    /// [`try_write_frame`]: FramedTransport::try_write_frame
+    /// [`try_flush`]: FramedTransport::try_flush
+    /// [`ErrorKind::WouldBlock`]: io::ErrorKind::WouldBlock
+    pub async fn write_frame<'a>(&mut self, frame: impl Into<Frame<'a>>) -> io::Result<()> {
+        self.writeable().await?;
+
+        match self.try_write_frame(frame) {
+            // Would block, so continually try to flush until good to go
+            Err(x) if x.kind() == io::ErrorKind::WouldBlock => loop {
+                self.writeable().await?;
+                match self.try_flush() {
+                    Err(x) if x.kind() == io::ErrorKind::WouldBlock => continue,
+                    x => return x,
+                }
+            },
+
+            // Already fully succeeded or failed
+            x => x,
+        }
+    }
+
+    /// Perform the client-side of a handshake. See [`handshake`] for more details.
+    ///
+    /// [`handshake`]: FramedTransport::handshake
+    pub async fn client_handshake(&mut self) -> io::Result<()> {
+        self.handshake(Handshake::client()).await
+    }
+
+    /// Perform the server-side of a handshake. See [`handshake`] for more details.
+    ///
+    /// [`handshake`]: FramedTransport::handshake
+    pub async fn server_handshake(&mut self) -> io::Result<()> {
+        self.handshake(Handshake::server()).await
+    }
 
     /// Performs a handshake in order to establish a new codec to use between this transport and
     /// the other side. The parameter `handshake` defines how the transport will handle the
@@ -158,10 +281,7 @@ where
     /// If a failure happens, the codec will be reset to what it was prior to the handshake
     /// request, and all internal buffers will be cleared to avoid corruption.
     ///
-    pub async fn handshake(&mut self, handshake: Handshake) -> io::Result<()>
-    where
-        T: Transport,
-    {
+    pub async fn handshake(&mut self, handshake: Handshake) -> io::Result<()> {
         // Place transport in plain text communication mode for start of handshake, and clear any
         // data that is lingering within internal buffers
         //
@@ -300,9 +420,17 @@ where
 
         // Bundle our compression and encryption codecs into a single, chained codec
         let codec: BoxedCodec = match (compression_codec, encryption_codec) {
-            (Some(c), Some(e)) => Box::new(ChainCodec::new(c, e)),
+            // If we have both encryption and compression, do the encryption first and then
+            // compress in order to get smallest result
+            (Some(c), Some(e)) => Box::new(ChainCodec::new(e, c)),
+
+            // If we just have compression, pass along the compression codec
             (Some(c), None) => Box::new(c),
+
+            // If we just have encryption, pass along the encryption codec
             (None, Some(e)) => Box::new(e),
+
+            // If we have neither compression nor encryption, use a plaintext codec
             (None, None) => Box::new(PlainCodec::new()),
         };
 
@@ -310,115 +438,6 @@ where
     }
 }
 
-impl<T, const CAPACITY: usize> FramedTransport<T, CAPACITY>
-where
-    T: Transport,
-{
-    /// Reads a frame of bytes by using the [`Codec`] tied to this transport. Returns
-    /// `Ok(Some(frame))` upon reading a frame, or `Ok(None)` if the underlying transport has
-    /// closed.
-    ///
-    /// This call may return an error with [`ErrorKind::WouldBlock`] in the case that the transport
-    /// is not ready to read data or has not received a full frame before waiting.
-    ///
-    /// [`ErrorKind::WouldBlock`]: io::ErrorKind::WouldBlock
-    pub fn try_read_frame(&mut self) -> io::Result<Option<OwnedFrame>> {
-        // Continually read bytes into the incoming queue and then attempt to tease out a frame
-        let mut buf = [0; CAPACITY];
-
-        loop {
-            match self.inner.try_read(&mut buf) {
-                // Getting 0 bytes on read indicates the channel has closed. If we were still
-                // expecting more bytes for our frame, then this is an error, otherwise if we
-                // have nothing remaining if our queue then this is an expected end and we
-                // return None
-                Ok(0) if self.incoming.is_empty() => return Ok(None),
-                Ok(0) => return Err(io::Error::from(io::ErrorKind::UnexpectedEof)),
-
-                // Got some additional bytes, which we will add to our queue and then attempt to
-                // decode into a frame
-                Ok(n) => {
-                    self.incoming.extend_from_slice(&buf[..n]);
-
-                    // Attempt to read a frame, returning the decoded frame if we get one,
-                    // continuing to try to read more bytes if we don't find a frame, and returning
-                    // any error that is encountered from reading frames or failing to decode
-                    let frame = match Frame::read(&mut self.incoming) {
-                        Ok(Some(frame)) => frame,
-                        Ok(None) => continue,
-                        Err(x) => return Err(x),
-                    };
-
-                    return Ok(Some(self.codec.decode(frame)?.into_owned()));
-                }
-
-                // Any error (including WouldBlock) will get bubbled up
-                Err(x) => return Err(x),
-            }
-        }
-    }
-
-    /// Continues to invoke [`try_read_frame`] until a frame is successfully read, an error is
-    /// encountered that is not [`ErrorKind::WouldBlock`], or the underlying transport has closed.
-    ///
-    /// [`try_read_frame`]: FramedTransport::try_read_frame
-    /// [`ErrorKind::WouldBlock`]: io::ErrorKind::WouldBlock
-    pub async fn read_frame(&mut self) -> io::Result<Option<OwnedFrame>> {
-        loop {
-            self.readable().await?;
-
-            match self.try_read_frame() {
-                Err(x) if x.kind() == io::ErrorKind::WouldBlock => continue,
-                x => return x,
-            }
-        }
-    }
-
-    /// Writes a `frame` of bytes by using the [`Codec`] tied to this transport.
-    ///
-    /// This is accomplished by continually calling the inner transport's `try_write`. If 0 is
-    /// returned from a call to `try_write`, this will fail with [`ErrorKind::WriteZero`].
-    ///
-    /// This call may return an error with [`ErrorKind::WouldBlock`] in the case that the transport
-    /// is not ready to write data or has not written the entire frame before waiting.
-    ///
-    /// [`ErrorKind::WriteZero`]: io::ErrorKind::WriteZero
-    /// [`ErrorKind::WouldBlock`]: io::ErrorKind::WouldBlock
-    pub fn try_write_frame<'a>(&mut self, frame: impl Into<Frame<'a>>) -> io::Result<()> {
-        // Encode the frame and store it in our outgoing queue
-        let frame = self.codec.encode(frame.into())?;
-        frame.write(&mut self.outgoing)?;
-
-        // Attempt to write everything in our queue
-        self.try_flush()
-    }
-
-    /// Invokes [`try_write_frame`] followed by a continuous calls to [`try_flush`] until a frame
-    /// is successfully written, an error is encountered that is not [`ErrorKind::WouldBlock`], or
-    /// the underlying transport has closed.
-    ///
-    /// [`try_write_frame`]: FramedTransport::try_write_frame
-    /// [`try_flush`]: FramedTransport::try_flush
-    /// [`ErrorKind::WouldBlock`]: io::ErrorKind::WouldBlock
-    pub async fn write_frame<'a>(&mut self, frame: impl Into<Frame<'a>>) -> io::Result<()> {
-        self.writeable().await?;
-
-        match self.try_write_frame(frame) {
-            // Would block, so continually try to flush until good to go
-            Err(x) if x.kind() == io::ErrorKind::WouldBlock => loop {
-                self.writeable().await?;
-                match self.try_flush() {
-                    Err(x) if x.kind() == io::ErrorKind::WouldBlock => continue,
-                    x => return x,
-                }
-            },
-
-            // Already fully succeeded or failed
-            x => x,
-        }
-    }
-}
-
 #[async_trait]
 impl<T, const CAPACITY: usize> Reconnectable for FramedTransport<T, CAPACITY>
 where
@@ -447,6 +466,19 @@ impl<const CAPACITY: usize> FramedTransport<super::InmemoryTransport, CAPACITY>
     }
 }
 
+#[cfg(test)]
+impl FramedTransport<super::InmemoryTransport> {
+    /// Generates a test pair with default capacity
+    pub fn test_pair(
+        buffer: usize,
+    ) -> (
+        FramedTransport<super::InmemoryTransport>,
+        FramedTransport<super::InmemoryTransport>,
+    ) {
+        Self::pair(buffer)
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

distant-net/src/transport/framed/exchange.rs

@@ -1,4 +1,4 @@
-use super::HeapSecretKey;
+use super::SecretKey32;
 use p256::{ecdh::EphemeralSecret, PublicKey};
 use rand::rngs::OsRng;
 use sha2::Sha256;
@@ -43,7 +43,7 @@ impl KeyExchange {
         &self,
         public_key: PublicKeyBytes,
         salt: Salt,
-    ) -> io::Result<HeapSecretKey> {
+    ) -> io::Result<SecretKey32> {
         // Decode the public key of the other side
         let decoded_public_key = PublicKey::try_from(public_key)?;
 
@@ -59,7 +59,7 @@ impl KeyExchange {
         // Derive a shared key (32 bytes)
         let mut shared_key = [0u8; 32];
         match hkdf.expand(&[], &mut shared_key) {
-            Ok(_) => Ok(HeapSecretKey::from(shared_key)),
+            Ok(_) => Ok(SecretKey32::from(shared_key)),
             Err(x) => Err(io::Error::new(io::ErrorKind::InvalidData, x.to_string())),
         }
     }

distant-net/src/transport/stateful.rs

@@ -96,23 +96,19 @@ where
     T: Transport + Send + Sync,
 {
     async fn reconnect(&mut self) -> io::Result<()> {
-        match self.state {
+        // If authenticated, we perform a reconnect followed by re-authentication using our
+        // previously-acquired key to skip the need to do another authentication. Note that
+        // this can still change the underlying codec used by the transport if an alternative
+        // compression or encryption codec is picked.
+        if let State::Authenticated { key } = &self.state {
+            Reconnectable::reconnect(&mut self.inner).await?;
+
+            todo!("do handshake with key");
+        } else {
             // If not authenticated or in the process of authenticating, we simply perform a raw
             // reconnect and reset to not being authenticated
-            State::NotAuthenticated | State::Authenticating => {
-                self.state = State::NotAuthenticated;
-                Reconnectable::reconnect(&mut self.inner).await
-            }
-
-            // If authenticated, we perform a reconnect followed by re-authentication using our
-            // previously-acquired key to skip the need to do another authentication. Note that
-            // this can still change the underlying codec used by the transport if an alternative
-            // compression or encryption codec is picked.
-            State::Authenticated { key, .. } => {
-                Reconnectable::reconnect(&mut self.inner).await?;
-
-                todo!("do handshake with key");
-            }
+            self.state = State::NotAuthenticated;
+            Reconnectable::reconnect(&mut self.inner).await
         }
     }
 }