chore(communication): bump libp2p version

communication/Cargo.toml

@@ -12,7 +12,7 @@ async-trait = "0.1.40"
 bytes = "0.5.6"
 clap = { version = "3.0.0-beta.1", features = ["yaml"] }
 futures = "0.3.1"
-libp2p = {version = "0.28.1", default-features = false, features = ["dns", "identify", "mdns-async-std", "mplex", "noise", "request-response", "tcp-async-std", "yamux", "websocket"]}
+libp2p = {version = "0.32", default-features = false, features = ["dns", "identify", "mdns", "mplex", "noise", "request-response", "tcp-async-std", "yamux", "websocket"]}
 prost = {version = "0.6.1", default-features = false, features = ["prost-derive"] }
 regex = "1.3.9"
 thiserror = "1.0.21"

communication/src/actor.rs

@@ -377,8 +377,9 @@ mod test {
             {
                 self.has_received_response = true;
             } else if let CommunicationEvent::ConnectPeerResult { addr: _, result } = msg {
+                let peer_id = result.expect("Panic due to no network connection");
                 let request = CommunicationEvent::<Request, Response>::Message(P2PReqResEvent::Req {
-                    peer_id: result.unwrap(),
+                    peer_id,
                     request_id: None,
                     request: Request::Ping,
                 });

communication/src/behaviour.rs

@@ -4,6 +4,7 @@
 pub mod error;
 pub mod message;
 mod protocol;
+use async_std::task;
 use core::{
     iter,
     marker::PhantomData,
@@ -26,7 +27,7 @@ use libp2p::{
     swarm::{NetworkBehaviourAction, NetworkBehaviourEventProcess, PollParameters, Swarm},
     tcp::TcpConfig,
     websocket::WsConfig,
-    yamux::Config as YamuxConfig,
+    yamux::YamuxConfig,
     NetworkBehaviour, Transport,
 };
 #[cfg(feature = "mdns")]
@@ -103,12 +104,13 @@ impl<T: MessageEvent, U: MessageEvent> P2PNetworkBehaviour<T, U> {
             .or_transport(WsConfig::new(dns_transport))
             .upgrade(upgrade::Version::V1)
             .authenticate(noise)
-            .multiplex(YamuxConfig::default());
+            .multiplex(YamuxConfig::default())
+            .boxed();
 
         // multicast DNS for peer discovery within a local network
         #[cfg(feature = "mdns")]
-        let mdns =
-            Mdns::new().map_err(|e| QueryError::ConnectionError(format!("Could not build mdns behaviour: {:?}", e)))?;
+        let mdns = task::block_on(Mdns::new())
+            .map_err(|e| QueryError::ConnectionError(format!("Could not build mdns behaviour: {:?}", e)))?;
         // Identify protocol to receive identifying information of a remote peer once a connection
         // was established
         let identify = Identify::new(
@@ -165,21 +167,20 @@ impl<T: MessageEvent, U: MessageEvent> P2PNetworkBehaviour<T, U> {
         self.peers.get(peer_id)
     }
 
-    pub fn get_all_peers(&self) -> &BTreeMap<PeerId, Multiaddr> {
-        &self.peers
+    pub fn get_all_peers(&self) -> Vec<(&PeerId, &Multiaddr)> {
+        self.peers.iter().collect()
     }
 
     pub fn send_request(&mut self, peer_id: &PeerId, request: T) -> RequestId {
         self.msg_proto.send_request(peer_id, request)
     }
 
-    pub fn send_response(&mut self, response: U, request_id: RequestId) -> QueryResult<()> {
+    pub fn send_response(&mut self, response: U, request_id: RequestId) -> Result<(), U> {
         let channel = self
             .response_channels
             .remove(&request_id.to_string())
-            .ok_or_else(|| QueryError::MissingChannelError(request_id.to_string()))?;
-        self.msg_proto.send_response(channel, response);
-        Ok(())
+            .ok_or_else(|| response.clone())?;
+        self.msg_proto.send_response(channel, response)
     }
     #[cfg(feature = "mdns")]
     /// Get the peers discovered by mdns
@@ -306,10 +307,10 @@ mod test {
         let addr = mock_addr();
         swarm.add_peer(peer_id.clone(), addr.clone());
         assert!(swarm.get_peer_addr(&peer_id).is_some());
-        assert!(swarm.get_all_peers().contains_key(&peer_id));
+        assert!(swarm.get_all_peers().contains(&(&peer_id, &addr)));
         assert_eq!(swarm.remove_peer(&peer_id).unwrap(), addr);
         assert!(swarm.get_peer_addr(&peer_id).is_none());
-        assert!(!swarm.get_all_peers().contains_key(&peer_id));
+        assert!(!swarm.get_all_peers().contains(&(&peer_id, &addr)));
     }
 
     #[test]

communication/src/behaviour/error.rs

@@ -16,9 +16,6 @@ pub enum QueryError {
 
     #[error("IO error: `{0}`")]
     IOError(String),
-
-    #[error("Missing response channel: `{0}`")]
-    MissingChannelError(String),
 }
 
 pub type QueryResult<T> = std::result::Result<T, QueryError>;

communication/src/behaviour/message.rs

@@ -96,8 +96,10 @@ pub enum P2PInboundFailure {
     Timeout,
     /// The local peer supports none of the requested protocols.
     UnsupportedProtocols,
-    /// The connection closed before a response was delivered.
-    ConnectionClosed,
+    /// The local peer failed to respond to an inbound request
+    /// due to the [`ResponseChannel`] being dropped instead of
+    /// being passed to [`RequestResponse::send_response`].
+    ResponseOmission,
 }
 
 /// Event emitted  by the `RequestResponse` behaviour.
@@ -135,6 +137,10 @@ pub enum P2PReqResEvent<T, U> {
         request_id: RequestId,
         error: P2POutboundFailure,
     },
+    ResponseSent {
+        peer_id: PeerId,
+        request_id: RequestId,
+    },
 }
 
 /// Event that was emitted by one of the protocols of the `P2PNetwokBehaviour`
@@ -236,7 +242,7 @@ impl<T, U> From<RequestResponseEvent<T, U>> for P2PEvent<T, U> {
             } => {
                 let error = match error {
                     InboundFailure::Timeout => P2PInboundFailure::Timeout,
-                    InboundFailure::ConnectionClosed => P2PInboundFailure::ConnectionClosed,
+                    InboundFailure::ResponseOmission => P2PInboundFailure::ResponseOmission,
                     InboundFailure::UnsupportedProtocols => P2PInboundFailure::UnsupportedProtocols,
                 };
                 P2PEvent::RequestResponse(Box::new(P2PReqResEvent::InboundFailure {
@@ -245,6 +251,12 @@ impl<T, U> From<RequestResponseEvent<T, U>> for P2PEvent<T, U> {
                     error,
                 }))
             }
+            RequestResponseEvent::ResponseSent { peer, request_id } => {
+                P2PEvent::RequestResponse(Box::new(P2PReqResEvent::ResponseSent {
+                    peer_id: peer,
+                    request_id,
+                }))
+            }
         }
     }
 }