review: separate Info details provided by the behaviour,

provided on new_handler constructor Everything that doesn't change over
the lifetime.

Author: jxs

protocols/identify/src/behaviour.rs

@@ -18,7 +18,7 @@
 // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 // DEALINGS IN THE SOFTWARE.
 
-use crate::handler::{self, InEvent, Proto};
+use crate::handler::{self, BehaviourInfo, InEvent, Proto};
 use crate::protocol::{Info, UpgradeError};
 use libp2p_core::{
     connection::ConnectionId, multiaddr::Protocol, ConnectedPoint, Multiaddr, PeerId, PublicKey,
@@ -48,8 +48,8 @@ pub struct Behaviour {
     config: Config,
     /// For each peer we're connected to, the observed address to send back to it.
     connected: HashMap<PeerId, HashMap<ConnectionId, Multiaddr>>,
-    /// Pending requests to respond.
-    requests: VecDeque<Request>,
+    /// Information requests from the handlers to be fullfiled.
+    requests: VecDeque<PeerId>,
     /// Pending events to be emitted when polled.
     events: VecDeque<NetworkBehaviourAction<Event, Proto>>,
     /// Peers to which an active push with current information about
@@ -59,12 +59,6 @@ pub struct Behaviour {
     discovered_peers: PeerCache,
 }
 
-/// An inbound identification request.
-struct Request {
-    peer: PeerId,
-    observed: Multiaddr,
-}
-
 /// Configuration for the [`identify::Behaviour`](Behaviour).
 #[non_exhaustive]
 #[derive(Debug, Clone)]
@@ -228,13 +222,19 @@ impl NetworkBehaviour for Behaviour {
     type OutEvent = Event;
 
     fn new_handler(&mut self) -> Self::ConnectionHandler {
-        Proto::new(self.config.initial_delay, self.config.interval)
+        Proto::new(
+            self.config.initial_delay,
+            self.config.interval,
+            self.config.local_public_key.clone(),
+            self.config.protocol_version.clone(),
+            self.config.agent_version.clone(),
+        )
     }
 
     fn on_connection_handler_event(
         &mut self,
         peer_id: PeerId,
-        connection: ConnectionId,
+        _connection: ConnectionId,
         event: <<Self::ConnectionHandler as IntoConnectionHandler>::Handler as ConnectionHandler>::OutEvent,
     ) {
         match event {
@@ -272,19 +272,7 @@ impl NetworkBehaviour for Behaviour {
                     }));
             }
             handler::Event::Identify => {
-                let observed = self
-                    .connected
-                    .get(&peer_id)
-                    .and_then(|addrs| addrs.get(&connection))
-                    .expect(
-                        "`on_connection_handler_event` is only called \
-                        with an established connection and calling `NetworkBehaviour::on_event` \
-                        with `FromSwarm::ConnectionEstablished ensures there is an entry; qed",
-                    );
-                self.requests.push_back(Request {
-                    peer: peer_id,
-                    observed: observed.clone(),
-                });
+                self.requests.push_back(peer_id);
             }
             handler::Event::IdentificationError(error) => {
                 self.events
@@ -339,15 +327,11 @@ impl NetworkBehaviour for Behaviour {
             });
         }
 
-        // Check for pending requests to send back to the handler for reply.
-        if let Some(Request { peer, observed }) = self.requests.pop_front() {
-            let info = Info {
+        // Check for information requests from the handlers.
+        if let Some(peer) = self.requests.pop_front() {
+            let info = BehaviourInfo {
                 listen_addrs: listen_addrs(params),
                 protocols: supported_protocols(params),
-                public_key: self.config.local_public_key.clone(),
-                protocol_version: self.config.protocol_version.clone(),
-                agent_version: self.config.agent_version.clone(),
-                observed_addr: observed,
             };
             return Poll::Ready(NetworkBehaviourAction::NotifyHandler {
                 peer_id: peer,

protocols/identify/src/handler.rs

@@ -26,7 +26,7 @@ use futures::prelude::*;
 use futures_timer::Delay;
 use libp2p_core::either::{EitherError, EitherOutput};
 use libp2p_core::upgrade::{EitherUpgrade, SelectUpgrade};
-use libp2p_core::{ConnectedPoint, PeerId};
+use libp2p_core::{ConnectedPoint, Multiaddr, PeerId, PublicKey};
 use libp2p_swarm::handler::{
     ConnectionEvent, DialUpgradeError, FullyNegotiatedInbound, FullyNegotiatedOutbound,
 };
@@ -42,22 +42,47 @@ use std::{io, pin::Pin, task::Context, task::Poll, time::Duration};
 pub struct Proto {
     initial_delay: Duration,
     interval: Duration,
+    public_key: PublicKey,
+    protocol_version: String,
+    agent_version: String,
 }
 
 impl Proto {
-    pub fn new(initial_delay: Duration, interval: Duration) -> Self {
+    pub fn new(
+        initial_delay: Duration,
+        interval: Duration,
+        public_key: PublicKey,
+        protocol_version: String,
+        agent_version: String,
+    ) -> Self {
         Proto {
             initial_delay,
             interval,
+            public_key,
+            protocol_version,
+            agent_version,
         }
     }
 }
 
 impl IntoConnectionHandler for Proto {
     type Handler = Handler;
 
-    fn into_handler(self, remote_peer_id: &PeerId, _endpoint: &ConnectedPoint) -> Self::Handler {
-        Handler::new(self.initial_delay, self.interval, *remote_peer_id)
+    fn into_handler(self, remote_peer_id: &PeerId, endpoint: &ConnectedPoint) -> Self::Handler {
+        let observed_addr = match endpoint {
+            ConnectedPoint::Dialer { address, .. } => address,
+            ConnectedPoint::Listener { send_back_addr, .. } => send_back_addr,
+        };
+
+        Handler::new(
+            self.initial_delay,
+            self.interval,
+            *remote_peer_id,
+            self.public_key,
+            self.protocol_version,
+            self.agent_version,
+            observed_addr.clone(),
+        )
     }
 
     fn inbound_protocol(&self) -> <Self::Handler as ConnectionHandler>::InboundProtocol {
@@ -94,9 +119,6 @@ pub struct Handler {
         >; 4],
     >,
 
-    /// Identify request information.
-    info: Option<Info>,
-
     /// Pending replies to send.
     pending_replies: VecDeque<Pending>,
 
@@ -108,6 +130,33 @@ pub struct Handler {
 
     /// The interval of `trigger_next_identify`, i.e. the recurrent delay.
     interval: Duration,
+
+    /// The public key of the local peer.
+    public_key: PublicKey,
+
+    /// Application-specific version of the protocol family used by the peer,
+    /// e.g. `ipfs/1.0.0` or `polkadot/1.0.0`.
+    protocol_version: String,
+
+    /// Name and version of the peer, similar to the `User-Agent` header in
+    /// the HTTP protocol.
+    agent_version: String,
+
+    /// Address observed by or for the remote.
+    observed_addr: Multiaddr,
+
+    /// Information provided by the `Behaviour` upon requesting.
+    behaviour_info: Option<BehaviourInfo>,
+}
+
+/// Information provided by the `Behaviour` upon requesting.
+#[derive(Debug)]
+pub struct BehaviourInfo {
+    /// The addresses that the peer is listening on.
+    pub listen_addrs: Vec<Multiaddr>,
+
+    /// The list of protocols supported by the peer, e.g. `/ipfs/ping/1.0.0`.
+    pub protocols: Vec<String>,
 }
 
 /// Event produced by the `Handler`.
@@ -132,21 +181,33 @@ pub enum InEvent {
     /// Identifying information of the local node that is pushed to a remote.
     Push(Info),
     /// Identifying information requested from this node.
-    Identify(Info),
+    Identify(BehaviourInfo),
 }
 
 impl Handler {
     /// Creates a new `Handler`.
-    pub fn new(initial_delay: Duration, interval: Duration, remote_peer_id: PeerId) -> Self {
+    pub fn new(
+        initial_delay: Duration,
+        interval: Duration,
+        remote_peer_id: PeerId,
+        public_key: PublicKey,
+        protocol_version: String,
+        agent_version: String,
+        observed_addr: Multiaddr,
+    ) -> Self {
         Self {
             remote_peer_id,
             inbound_identify_push: Default::default(),
             events: SmallVec::new(),
-            info: None,
             pending_replies: VecDeque::new(),
             trigger_next_identify: Delay::new(initial_delay),
             keep_alive: KeepAlive::Yes,
             interval,
+            public_key,
+            protocol_version,
+            agent_version,
+            observed_addr,
+            behaviour_info: None,
         }
     }
 
@@ -161,9 +222,9 @@ impl Handler {
     ) {
         match output {
             EitherOutput::First(substream) => {
-                // If we already have `Info` we can proceed responding to the Identify request,
-                // if not, we request `Info` from the behaviour.
-                if self.info.is_none() {
+                // If we already have `BehaviourInfo` we can proceed responding to the Identify request,
+                // if not, we request it .
+                if self.behaviour_info.is_none() {
                     self.events
                         .push(ConnectionHandlerEvent::Custom(Event::Identify));
                 }
@@ -259,7 +320,7 @@ impl ConnectionHandler for Handler {
                     });
             }
             InEvent::Identify(info) => {
-                self.info = Some(info);
+                self.behaviour_info = Some(info);
             }
         }
     }
@@ -303,11 +364,19 @@ impl ConnectionHandler for Handler {
         }
 
         // Check for pending replies to send.
-        if let Some(ref info) = self.info {
+        if let Some(ref info) = self.behaviour_info {
             if let Some(mut pending) = self.pending_replies.pop_front() {
                 loop {
                     match pending {
                         Pending::Queued(io) => {
+                            let info = Info {
+                                public_key: self.public_key.clone(),
+                                protocol_version: self.protocol_version.clone(),
+                                agent_version: self.agent_version.clone(),
+                                listen_addrs: info.listen_addrs.clone(),
+                                protocols: info.protocols.clone(),
+                                observed_addr: self.observed_addr.clone(),
+                            };
                             let io = Box::pin(io.send(info.clone()));
                             pending = Pending::Sending {
                                 peer: self.remote_peer_id,