feat(rust): Implement dht_provide() and dht_get_providers() (#666)

* Implement `dht_provide()` and `dht_get_providers()`

* Cleanup

* Fix lints

* `identity()` returns identity, not peer id

Author: rafal-ch

rust/hermes-ipfs/examples/provide-content-dht.rs

@@ -13,7 +13,7 @@ async fn connect_node_a_upload_and_provide(
     println!("");
     let peer_id_a = hermes_ipfs.identity(None).await?;
     let addresses = hermes_ipfs.listening_addresses().await?;
-    println!("* Peer ID: {peer_id_a}");
+    println!("* Peer ID: {}", peer_id_a.peer_id);
     for addr in addresses {
         println!("    * {addr}");
     }
@@ -34,7 +34,7 @@ async fn connect_node_a_upload_and_provide(
     println!("***************************************");
     println!("* Providing content to DHT:");
     println!("");
-    println!("* Providing {cid} as peer {peer_id_a}");
+    println!("* Providing {cid} as peer {}", peer_id_a.peer_id);
     println!("***************************************");
     println!("");
     Ok((hermes_ipfs, ipfs_path))
@@ -48,7 +48,7 @@ async fn connect_node_b_to_node_a(node_a: &HermesIpfs) -> anyhow::Result<HermesI
     println!("");
     let peer_id_b = hermes_ipfs_b.identity(None).await?;
     // node_b.connect(peer_id_a).await?;
-    println!("* Peer ID: {peer_id_b}");
+    println!("* Peer ID: {}", peer_id_b.peer_id);
     println!("* Listening addresses:");
     let addresses = hermes_ipfs_b.listening_addresses().await?;
     for addr in addresses {
@@ -63,7 +63,7 @@ async fn connect_node_b_to_node_a(node_a: &HermesIpfs) -> anyhow::Result<HermesI
     let node_a_addresses = node_a.listening_addresses().await?;
     let peer_a = node_a.identity(None).await?;
     for addr in node_a_addresses {
-        hermes_ipfs_b.add_peer(peer_a, addr.clone()).await?;
+        hermes_ipfs_b.add_peer(peer_a.peer_id, addr.clone()).await?;
         println!("    * {addr} - CONNECTED");
     }
     println!("***************************************");

rust/hermes-ipfs/examples/pubsub.rs

@@ -25,19 +25,19 @@ async fn start_bootstrapped_nodes() -> anyhow::Result<(HermesIpfs, HermesIpfs)>
     println!("***************************************");
     println!("* Hermes IPFS node A has started.");
     let peer_id_a = hermes_a.identity(None).await?;
-    println!("    Peer ID: {peer_id_a}");
+    println!("    Peer ID: {}", peer_id_a.peer_id);
     let addresses = hermes_a.listening_addresses().await?;
     let a_address = addresses[0].clone();
-    let a_p2p = a_address.with(rust_ipfs::Protocol::P2p(peer_id_a));
+    let a_p2p = a_address.with(rust_ipfs::Protocol::P2p(peer_id_a.peer_id));
     println!("    P2P addr: {a_p2p}");
     println!("***************************************");
     println!("* Hermes IPFS node B has started.");
     let hermes_b = HermesIpfs::start().await?;
     let peer_id_b = hermes_b.identity(None).await?;
-    println!("    Peer ID: {peer_id_b}");
+    println!("    Peer ID: {}", peer_id_b.peer_id);
     let addresses = hermes_b.listening_addresses().await?;
     let b_address = addresses[0].clone();
-    let b_p2p = b_address.with(rust_ipfs::Protocol::P2p(peer_id_b));
+    let b_p2p = b_address.with(rust_ipfs::Protocol::P2p(peer_id_b.peer_id));
     println!("    P2P addr: {b_p2p}");
     println!("***************************************");
     println!("* Bootstrapping node A.");

rust/hermes-ipfs/examples/put-get-dht.rs

@@ -9,7 +9,7 @@ async fn start_bootstrapped_nodes() -> anyhow::Result<(HermesIpfs, HermesIpfs)>
     let hermes_a = HermesIpfs::start().await?;
     println!("***************************************");
     println!("* Hermes IPFS node A has started.");
-    let peer_id_a = hermes_a.identity(None).await?;
+    let peer_id_a = hermes_a.identity(None).await?.peer_id;
     println!("    Peer ID: {peer_id_a}");
     let addresses = hermes_a.listening_addresses().await?;
     let a_address = addresses[0].clone();
@@ -18,7 +18,7 @@ async fn start_bootstrapped_nodes() -> anyhow::Result<(HermesIpfs, HermesIpfs)>
     println!("***************************************");
     println!("* Hermes IPFS node B has started.");
     let hermes_b = HermesIpfs::start().await?;
-    let peer_id_b = hermes_b.identity(None).await?;
+    let peer_id_b = hermes_b.identity(None).await?.peer_id;
     println!("    Peer ID: {peer_id_b}");
     let addresses = hermes_b.listening_addresses().await?;
     let b_address = addresses[0].clone();

rust/hermes-ipfs/src/lib.rs

@@ -2,10 +2,10 @@
 //!
 //! Provides support for storage, and `PubSub` functionality.
 
-use std::{convert::Infallible, str::FromStr};
+use std::{collections::HashSet, convert::Infallible, str::FromStr};
 
 use derive_more::{Display, From, Into};
-use futures::{StreamExt, pin_mut, stream::BoxStream};
+use futures::{StreamExt, TryStreamExt, pin_mut, stream::BoxStream};
 /// IPFS Content Identifier.
 pub use ipld_core::cid::Cid;
 /// IPLD
@@ -259,8 +259,8 @@ impl HermesIpfs {
     pub async fn identity(
         &self,
         peer_id: Option<PeerId>,
-    ) -> anyhow::Result<PeerId> {
-        self.node.identity(peer_id).await.map(|p| p.peer_id)
+    ) -> anyhow::Result<PeerInfo> {
+        self.node.identity(peer_id).await
     }
 
     /// Add peer to address book.
@@ -395,13 +395,67 @@ impl HermesIpfs {
     ) -> anyhow::Result<Vec<u8>> {
         let record_stream = self.node.dht_get(key).await?;
         pin_mut!(record_stream);
+        // TODO: We only ever return a single value from the stream. We might want to improve
+        // this.
         let record = record_stream
             .next()
             .await
             .ok_or(anyhow::anyhow!("No record found"))?;
         Ok(record.value)
     }
 
+    /// Announce this node as a provider for the given DHT key.
+    ///
+    /// ## Parameters
+    ///
+    /// * `key` - Key identifying the content or resource to provide on the DHT.
+    ///
+    /// ## Returns
+    ///
+    /// * `Result<()>` — Indicates whether the provider announcement succeeded.
+    ///
+    /// ## Errors
+    ///
+    /// Returns an error if announcing provider information to the DHT fails.
+    pub async fn dht_provide(
+        &self,
+        key: impl AsRef<[u8]> + ToRecordKey,
+    ) -> anyhow::Result<()> {
+        self.node.dht_provide(key).await
+    }
+
+    /// Retrieve all providers for the given DHT key.
+    ///
+    /// ## Parameters
+    ///
+    /// * `key` - Key identifying the content or resource in the DHT.
+    ///
+    /// ## Returns
+    ///
+    /// * `Result<HashSet<PeerId>>` — A set containing all `PeerId`s reported as providers
+    ///   for the given key.
+    ///
+    /// ## Errors
+    ///
+    /// Returns an error if the provider stream fails or if retrieving provider
+    /// information from the DHT encounters an underlying error.
+    pub async fn dht_get_providers(
+        &self,
+        key: impl AsRef<[u8]> + ToRecordKey,
+    ) -> anyhow::Result<HashSet<PeerId>> {
+        Ok(self
+            .node
+            .dht_get_providers(key)
+            .await?
+            .try_fold(HashSet::new(), |mut acc, set| {
+                async move {
+                    acc.extend(set);
+                    Ok(acc)
+                }
+            })
+            .await?)
+    }
+
     /// Add address to bootstrap nodes.
     ///
     /// ## Parameters