Two-stage connectivity with handshake

radicle-node/src/service.rs

@@ -501,13 +501,15 @@ where
         peer.attempted();
     }
 
-    // TODO: Split into two functions: `connected` and `negotiated`
-    pub fn connected(
+    pub fn connecting(
         &mut self,
-        addr: std::net::SocketAddr,
-        _local_addr: &std::net::SocketAddr,
-        link: Link,
+        _addr: net::SocketAddr,
+        _local_addr: &net::SocketAddr,
+        _link: Link,
     ) {
+    }
+
+    pub fn connected(&mut self, addr: net::SocketAddr, link: Link) {
         let ip = addr.ip();
         let address = addr.into();
 

radicle-node/src/test/peer.rs

@@ -210,7 +210,8 @@ where
         let local = net::SocketAddr::new(self.ip, self.rng.u16(..));
 
         self.initialize();
-        self.service.connected(remote, &local, Link::Inbound);
+        self.service.connecting(remote, &local, Link::Inbound);
+        self.service.connected(remote, Link::Inbound);
         self.receive(
             &remote,
             Message::init(peer.node_id(), vec![Address::from(remote)]),
@@ -237,7 +238,8 @@ where
         self.initialize();
         self.service.attempted(&remote);
         self.service
-            .connected(remote, &self.local_addr, Link::Outbound);
+            .connecting(remote, &self.local_addr, Link::Outbound);
+        self.service.connected(remote, Link::Outbound);
 
         let mut msgs = self.messages(&remote);
         msgs.find(|m| matches!(m, Message::Initialize { .. }))

radicle-node/src/test/simulator.rs

@@ -385,7 +385,8 @@ impl<S: WriteStorage + 'static, G: Signer> Simulation<S, G> {
                         let attempted = link.is_outbound() && self.attempts.remove(&conn);
                         if attempted || link.is_inbound() {
                             if self.connections.insert(conn, local_addr.port()).is_none() {
-                                p.connected(addr, &local_addr, link);
+                                p.connecting(addr, &local_addr, link);
+                                p.connected(addr, link);
                             }
                         }
                     }

radicle-node/src/wire.rs

@@ -482,8 +482,8 @@ where
     }
 
     fn connected(&mut self, addr: net::SocketAddr, local_addr: &net::SocketAddr, link: Link) {
-        self.handshakes.insert(addr, H::new());
-        self.inner.connected(addr, local_addr, link)
+        self.handshakes.insert(addr, H::new(link));
+        self.inner.connecting(addr, local_addr, link)
     }
 
     fn disconnected(
@@ -509,7 +509,7 @@ where
                     }
                     return;
                 }
-                HandshakeResult::Complete(transcoder, reply) => {
+                HandshakeResult::Complete(transcoder, reply, link) => {
                     log::debug!("handshake with peer {} is complete", addr);
                     if !reply.is_empty() {
                         self.inner_queue
@@ -522,6 +522,7 @@ where
                             deserializer: Deserializer::new(256),
                         },
                     );
+                    self.inner.connected(*addr, link);
                 }
                 HandshakeResult::Error(err) => {
                     log::error!("invalid handshake input. Details: {}", err);

radicle-node/src/wire/transcoder.rs

@@ -1,3 +1,4 @@
+use nakamoto_net::Link;
 use std::convert::Infallible;
 
 // TODO: Implement Try trait once stabilized
@@ -6,7 +7,7 @@ pub enum HandshakeResult<H: Handshake, T: Transcode> {
     /// Handshake is not completed; we proceed to the next handshake stage.
     Next(H, Vec<u8>),
     /// Handshake is completed; we now can communicate in a secure way.
-    Complete(T, Vec<u8>),
+    Complete(T, Vec<u8>, Link),
     /// Handshake has failed with some error.
     Error(H::Error),
 }
@@ -21,25 +22,31 @@ pub trait Handshake: Sized {
     type Error: std::error::Error;
 
     /// Create a new handshake state-machine.
-    fn new() -> Self;
+    fn new(link: Link) -> Self;
     /// Advance the state-machine to the next state.
     fn step(self, input: &[u8]) -> HandshakeResult<Self, Self::Transcoder>;
+    /// Returns direction of the handshake protocol.
+    fn link(&self) -> Link;
 }
 
 /// Dumb handshake structure which runs void protocol.
-#[derive(Debug, Default)]
-pub struct NoHandshake;
+#[derive(Debug)]
+pub struct NoHandshake(Link);
 
 impl Handshake for NoHandshake {
     type Transcoder = PlainTranscoder;
     type Error = Infallible;
 
-    fn new() -> Self {
-        NoHandshake
+    fn new(link: Link) -> Self {
+        NoHandshake(link)
     }
 
     fn step(self, _input: &[u8]) -> HandshakeResult<Self, Self::Transcoder> {
-        HandshakeResult::Complete(PlainTranscoder, vec![])
+        HandshakeResult::Complete(PlainTranscoder, vec![], self.0)
+    }
+
+    fn link(&self) -> Link {
+        self.0
     }
 }