feat(kad): report get_providers call event based

Author: dignifiedquire

examples/distributed-key-value-store.rs

@@ -53,6 +53,7 @@ use libp2p::{
     swarm::{NetworkBehaviourEventProcess, SwarmEvent},
     NetworkBehaviour, PeerId, Swarm,
 };
+use libp2p_kad::{GetProvidersProgress, QueryProgress};
 use std::error::Error;
 
 #[async_std::main]
@@ -89,16 +90,16 @@ async fn main() -> Result<(), Box<dyn Error>> {
         // Called when `kademlia` produces an event.
         fn inject_event(&mut self, message: KademliaEvent) {
             match message {
-                KademliaEvent::OutboundQueryCompleted { result, .. } => match result {
-                    QueryResult::GetProviders(Ok(ok)) => {
-                        for peer in ok.providers {
-                            println!(
-                                "Peer {:?} provides key {:?}",
-                                peer,
-                                std::str::from_utf8(ok.key.as_ref()).unwrap()
-                            );
-                        }
+                KademliaEvent::OutboundQueryProgressed { result, .. } => match result {
+                    QueryProgress::GetProviders(GetProvidersProgress { key, provider, .. }) => {
+                        println!(
+                            "Peer {:?} provides key {:?}",
+                            provider,
+                            std::str::from_utf8(key.as_ref()).unwrap()
+                        );
                     }
+                },
+                KademliaEvent::OutboundQueryCompleted { result, .. } => match result {
                     QueryResult::GetProviders(Err(err)) => {
                         eprintln!("Failed to get providers: {:?}", err);
                     }

examples/file-sharing.rs

@@ -211,14 +211,15 @@ mod network {
     use libp2p::identity;
     use libp2p::identity::ed25519;
     use libp2p::kad::record::store::MemoryStore;
-    use libp2p::kad::{GetProvidersOk, Kademlia, KademliaEvent, QueryId, QueryResult};
+    use libp2p::kad::{Kademlia, KademliaEvent, QueryId, QueryResult};
     use libp2p::multiaddr::Protocol;
     use libp2p::request_response::{
         ProtocolSupport, RequestId, RequestResponse, RequestResponseCodec, RequestResponseEvent,
         RequestResponseMessage, ResponseChannel,
     };
     use libp2p::swarm::{ConnectionHandlerUpgrErr, SwarmBuilder, SwarmEvent};
     use libp2p::{NetworkBehaviour, Swarm};
+    use libp2p_kad::{GetProvidersProgress, QueryProgress};
     use std::collections::{HashMap, HashSet};
     use std::iter;
 
@@ -324,12 +325,12 @@ mod network {
 
         /// Find the providers for the given file on the DHT.
         pub async fn get_providers(&mut self, file_name: String) -> HashSet<PeerId> {
-            let (sender, receiver) = oneshot::channel();
+            let (sender, receiver) = mpsc::channel(0);
             self.sender
                 .send(Command::GetProviders { file_name, sender })
                 .await
                 .expect("Command receiver not to be dropped.");
-            receiver.await.expect("Sender not to be dropped.")
+            receiver.collect().await
         }
 
         /// Request the content of the given file from the given peer.
@@ -365,7 +366,7 @@ mod network {
         event_sender: mpsc::Sender<Event>,
         pending_dial: HashMap<PeerId, oneshot::Sender<Result<(), Box<dyn Error + Send>>>>,
         pending_start_providing: HashMap<QueryId, oneshot::Sender<()>>,
-        pending_get_providers: HashMap<QueryId, oneshot::Sender<HashSet<PeerId>>>,
+        pending_get_providers: HashMap<QueryId, mpsc::Sender<PeerId>>,
         pending_request_file:
             HashMap<RequestId, oneshot::Sender<Result<String, Box<dyn Error + Send>>>>,
     }
@@ -421,18 +422,31 @@ mod network {
                         .expect("Completed query to be previously pending.");
                     let _ = sender.send(());
                 }
+                SwarmEvent::Behaviour(ComposedEvent::Kademlia(
+                    KademliaEvent::OutboundQueryProgressed {
+                        id,
+                        result: QueryProgress::GetProviders(GetProvidersProgress { provider, .. }),
+                        ..
+                    },
+                )) => {
+                    let _ = self
+                        .pending_get_providers
+                        .get_mut(&id)
+                        .expect("Completed query to be previously pending.")
+                        .send(provider);
+                }
                 SwarmEvent::Behaviour(ComposedEvent::Kademlia(
                     KademliaEvent::OutboundQueryCompleted {
                         id,
-                        result: QueryResult::GetProviders(Ok(GetProvidersOk { providers, .. })),
+                        result: QueryResult::GetProviders(..),
                         ..
                     },
                 )) => {
+                    // Drop channel to signal query is complete.
                     let _ = self
                         .pending_get_providers
                         .remove(&id)
-                        .expect("Completed query to be previously pending.")
-                        .send(providers);
+                        .expect("Completed query to be previously pending.");
                 }
                 SwarmEvent::Behaviour(ComposedEvent::Kademlia(_)) => {}
                 SwarmEvent::Behaviour(ComposedEvent::RequestResponse(
@@ -620,7 +634,7 @@ mod network {
         },
         GetProviders {
             file_name: String,
-            sender: oneshot::Sender<HashSet<PeerId>>,
+            sender: mpsc::Sender<PeerId>,
         },
         RequestFile {
             file_name: String,

misc/metrics/src/kad.rs

@@ -208,10 +208,7 @@ impl super::Recorder<libp2p_kad::KademliaEvent> for super::Metrics {
                         }
                     },
                     libp2p_kad::QueryResult::GetProviders(result) => match result {
-                        Ok(ok) => self
-                            .kad
-                            .query_result_get_providers_ok
-                            .observe(ok.providers.len() as f64),
+                        Ok(_) => {}
                         Err(error) => {
                             self.kad
                                 .query_result_get_providers_error
@@ -222,6 +219,11 @@ impl super::Recorder<libp2p_kad::KademliaEvent> for super::Metrics {
                     _ => {}
                 }
             }
+            libp2p_kad::KademliaEvent::OutboundQueryProgressed { result, .. } => match result {
+                libp2p_kad::QueryProgress::GetProviders(_) => {
+                    self.kad.query_result_get_providers_ok.observe(1.);
+                }
+            },
             libp2p_kad::KademliaEvent::RoutingUpdated {
                 is_new_peer,
                 old_peer,