refactor(plaintext): use quick-protobuf-codec

Instead of depending on `unsigned-varint`, we can implement the handshake using `quick-protobuf-codec`. This is a lot shorter and allows us to reuse more code.

Pull-Request: #4783.

transports/plaintext/Cargo.toml

@@ -18,7 +18,7 @@ libp2p-core = { workspace = true }
 libp2p-identity = { workspace = true }
 log = "0.4.20"
 quick-protobuf = "0.8"
-unsigned-varint = { version = "0.7", features = ["asynchronous_codec"] }
+quick-protobuf-codec = { workspace = true }
 
 [dev-dependencies]
 env_logger = "0.10.0"

transports/plaintext/src/error.rs

@@ -41,7 +41,7 @@ pub enum Error {
 }
 
 #[derive(Debug)]
-pub struct DecodeError(pub(crate) quick_protobuf::Error);
+pub struct DecodeError(pub(crate) quick_protobuf_codec::Error);
 
 impl fmt::Display for DecodeError {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {

transports/plaintext/src/handshake.rs

@@ -21,106 +21,54 @@
 use crate::error::{DecodeError, Error};
 use crate::proto::Exchange;
 use crate::Config;
-
 use asynchronous_codec::{Framed, FramedParts};
-use bytes::{Bytes, BytesMut};
+use bytes::Bytes;
 use futures::prelude::*;
 use libp2p_identity::{PeerId, PublicKey};
 use log::{debug, trace};
-use quick_protobuf::{BytesReader, MessageRead, MessageWrite, Writer};
 use std::io::{Error as IoError, ErrorKind as IoErrorKind};
-use unsigned_varint::codec::UviBytes;
-
-struct HandshakeContext<T> {
-    config: Config,
-    state: T,
-}
-
-// HandshakeContext<()> --with_local-> HandshakeContext<Local>
-struct Local {
-    // Our local exchange's raw bytes:
-    exchange_bytes: Vec<u8>,
-}
-
-// HandshakeContext<Local> --with_remote-> HandshakeContext<Remote>
-pub(crate) struct Remote {
-    // The remote's peer ID:
-    pub(crate) peer_id: PeerId, // The remote's public key:
-    pub(crate) public_key: PublicKey,
-}
-
-impl HandshakeContext<Local> {
-    fn new(config: Config) -> Self {
-        #[allow(deprecated)]
-        let exchange = Exchange {
-            id: Some(config.local_public_key.to_peer_id().to_bytes()),
-            pubkey: Some(config.local_public_key.encode_protobuf()),
-        };
-        let mut buf = Vec::with_capacity(exchange.get_size());
-        let mut writer = Writer::new(&mut buf);
-        exchange
-            .write_message(&mut writer)
-            .expect("Encoding to succeed");
-
-        Self {
-            config,
-            state: Local {
-                exchange_bytes: buf,
-            },
-        }
-    }
 
-    fn with_remote(self, exchange_bytes: BytesMut) -> Result<HandshakeContext<Remote>, Error> {
-        let mut reader = BytesReader::from_bytes(&exchange_bytes);
-        let prop = Exchange::from_reader(&mut reader, &exchange_bytes).map_err(DecodeError)?;
-
-        let public_key = PublicKey::try_decode_protobuf(&prop.pubkey.unwrap_or_default())?;
-        let peer_id = PeerId::from_bytes(&prop.id.unwrap_or_default())?;
-
-        // Check the validity of the remote's `Exchange`.
-        if peer_id != public_key.to_peer_id() {
-            return Err(Error::PeerIdMismatch);
-        }
-
-        Ok(HandshakeContext {
-            config: self.config,
-            state: Remote {
-                peer_id,
-                public_key,
-            },
-        })
-    }
-}
-
-pub(crate) async fn handshake<S>(socket: S, config: Config) -> Result<(S, Remote, Bytes), Error>
+pub(crate) async fn handshake<S>(socket: S, config: Config) -> Result<(S, PublicKey, Bytes), Error>
 where
     S: AsyncRead + AsyncWrite + Send + Unpin,
 {
     // The handshake messages all start with a variable-length integer indicating the size.
-    let mut framed_socket = Framed::new(socket, UviBytes::default());
-
-    trace!("starting handshake");
-    let context = HandshakeContext::new(config);
+    let mut framed_socket = Framed::new(socket, quick_protobuf_codec::Codec::<Exchange>::new(100));
 
     trace!("sending exchange to remote");
     framed_socket
-        .send(BytesMut::from(&context.state.exchange_bytes[..]))
-        .await?;
+        .send(Exchange {
+            id: Some(config.local_public_key.to_peer_id().to_bytes()),
+            pubkey: Some(config.local_public_key.encode_protobuf()),
+        })
+        .await
+        .map_err(DecodeError)?;
 
     trace!("receiving the remote's exchange");
-    let context = match framed_socket.next().await {
-        Some(p) => context.with_remote(p?)?,
+    let public_key = match framed_socket
+        .next()
+        .await
+        .transpose()
+        .map_err(DecodeError)?
+    {
+        Some(remote) => {
+            let public_key = PublicKey::try_decode_protobuf(&remote.pubkey.unwrap_or_default())?;
+            let peer_id = PeerId::from_bytes(&remote.id.unwrap_or_default())?;
+
+            if peer_id != public_key.to_peer_id() {
+                return Err(Error::PeerIdMismatch);
+            }
+
+            public_key
+        }
         None => {
             debug!("unexpected eof while waiting for remote's exchange");
             let err = IoError::new(IoErrorKind::BrokenPipe, "unexpected eof");
             return Err(err.into());
         }
     };
 
-    trace!(
-        "received exchange from remote; pubkey = {:?}",
-        context.state.public_key
-    );
+    trace!("received exchange from remote; pubkey = {:?}", public_key);
 
     let FramedParts {
         io,
@@ -129,5 +77,5 @@ where
         ..
     } = framed_socket.into_parts();
     assert!(write_buffer.is_empty());
-    Ok((io, context.state, read_buffer.freeze()))
+    Ok((io, public_key, read_buffer.freeze()))
 }

transports/plaintext/src/lib.rs

@@ -102,14 +102,14 @@ impl Config {
         T: AsyncRead + AsyncWrite + Send + Unpin + 'static,
     {
         debug!("Starting plaintext handshake.");
-        let (socket, remote, read_buffer) = handshake::handshake(socket, self).await?;
+        let (socket, remote_key, read_buffer) = handshake::handshake(socket, self).await?;
         debug!("Finished plaintext handshake.");
 
         Ok((
-            remote.peer_id,
+            remote_key.to_peer_id(),
             Output {
                 socket,
-                remote_key: remote.public_key,
+                remote_key,
                 read_buffer,
             },
         ))