dht: add `DhtEvent`

src/dht/event.rs

@@ -0,0 +1,61 @@
+/* This file is part of DarkFi (https://dark.fi)
+ *
+ * Copyright (C) 2020-2025 Dyne.org foundation
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+use std::fmt::Debug;
+
+use crate::{dht::DhtNode, net::ChannelPtr, Result};
+
+type K = blake3::Hash;
+
+#[derive(Clone, Debug)]
+pub enum DhtEvent<N: DhtNode, V: Clone + Debug> {
+    BootstrapStarted,
+    BootstrapCompleted,
+    PingReceived { from: ChannelPtr, result: Result<K> },
+    PingSent { to: ChannelPtr, result: Result<()> },
+    ValueFound { key: K, value: V },
+    NodesFound { key: K, nodes: Vec<N> },
+    ValueLookupStarted { key: K },
+    NodesLookupStarted { key: K },
+    ValueLookupCompleted { key: K, nodes: Vec<N>, values: Vec<V> },
+    NodesLookupCompleted { key: K, nodes: Vec<N> },
+}
+
+impl<N: DhtNode, V: Clone + Debug> DhtEvent<N, V> {
+    pub fn key(&self) -> Option<&blake3::Hash> {
+        match self {
+            DhtEvent::BootstrapStarted => None,
+            DhtEvent::BootstrapCompleted => None,
+            DhtEvent::PingReceived { .. } => None,
+            DhtEvent::PingSent { .. } => None,
+            DhtEvent::ValueFound { key, .. } => Some(key),
+            DhtEvent::NodesFound { key, .. } => Some(key),
+            DhtEvent::ValueLookupStarted { key } => Some(key),
+            DhtEvent::NodesLookupStarted { key } => Some(key),
+            DhtEvent::ValueLookupCompleted { key, .. } => Some(key),
+            DhtEvent::NodesLookupCompleted { key, .. } => Some(key),
+        }
+    }
+
+    pub fn into_value(self) -> Option<V> {
+        match self {
+            DhtEvent::ValueFound { value, .. } => Some(value),
+            _ => None,
+        }
+    }
+}

src/dht/mod.rs

@@ -36,12 +36,13 @@ use tracing::{debug, info, warn};
 use url::Url;
 
 use crate::{
+    dht::event::DhtEvent,
     net::{
         connector::Connector,
         session::{Session, SESSION_REFINE, SESSION_SEED},
         ChannelPtr, Message, P2pPtr,
     },
-    system::{timeout::timeout, ExecutorPtr, PublisherPtr},
+    system::{msleep, ExecutorPtr, Publisher, PublisherPtr, Subscription},
     Error, Result,
 };
 
@@ -53,6 +54,8 @@ pub use handler::DhtHandler;
 
 pub mod tasks;
 
+pub mod event;
+
 pub trait DhtNode: Debug + Clone + Send + Sync + PartialEq + Eq + Hash {
     fn id(&self) -> blake3::Hash;
     fn addresses(&self) -> Vec<Url>;
@@ -128,6 +131,8 @@ pub struct Dht<H: DhtHandler> {
     pub channel_cache: Arc<RwLock<HashMap<u32, ChannelCacheItem<H::Node>>>>,
     /// DHT settings
     pub settings: DhtSettings,
+    /// DHT event publisher
+    pub event_publisher: PublisherPtr<DhtEvent<H::Node, H::Value>>,
     /// P2P network pointer
     pub p2p: P2pPtr,
     /// Global multithreaded executor reference
@@ -150,6 +155,8 @@ impl<H: DhtHandler> Dht<H> {
             bootstrapped: Arc::new(RwLock::new(false)),
             channel_cache: Arc::new(RwLock::new(HashMap::new())),
 
+            event_publisher: Publisher::new(),
+
             settings: settings.clone(),
 
             p2p: p2p.clone(),
@@ -171,6 +178,10 @@ impl<H: DhtHandler> Dht<H> {
         *bootstrapped = value;
     }
 
+    pub async fn subscribe(&self) -> Subscription<DhtEvent<H::Node, H::Value>> {
+        self.event_publisher.clone().subscribe().await
+    }
+
     /// Get the distance between `key_1` and `key_2`
     pub fn distance(&self, key_1: &blake3::Hash, key_2: &blake3::Hash) -> [u8; 32] {
         let bytes1 = key_1.as_bytes();
@@ -274,12 +285,11 @@ impl<H: DhtHandler> Dht<H> {
         }
 
         self.handler().await.add_value(key, value).await;
-        let nodes = self.lookup_nodes(key).await?;
-        info!(target: "dht::announce()", "Announcing {} to {} nodes", H::key_to_string(key), nodes.len());
+        let nodes = self.lookup_nodes(key).await;
+        info!(target: "dht::announce()", "[DHT] Announcing {} to {} nodes", H::key_to_string(key), nodes.len());
 
         for node in nodes {
-            let channel_res = self.get_channel(&node, None).await;
-            if let Ok(channel) = channel_res {
+            if let Ok((channel, _)) = self.get_channel(&node).await {
                 let _ = channel.send(message).await;
                 self.cleanup_channel(channel).await;
             }
@@ -288,16 +298,20 @@ impl<H: DhtHandler> Dht<H> {
         Ok(())
     }
 
-    /// Lookup our own node id to bootstrap our DHT
+    /// Lookup our own node id
     pub async fn bootstrap(&self) {
         self.set_bootstrapped(true).await;
 
+        info!(target: "dht::bootstrap()", "[DHT] Bootstrapping");
+        self.event_publisher.notify(DhtEvent::BootstrapStarted).await;
+
         let self_node_id = self.handler().await.node().await.id();
-        debug!(target: "dht::bootstrap()", "DHT bootstrapping {}", H::key_to_string(&self_node_id));
         let nodes = self.lookup_nodes(&self_node_id).await;
 
-        if nodes.is_err() || nodes.map_or(true, |v| v.is_empty()) {
+        if nodes.is_empty() {
             self.set_bootstrapped(false).await;
+        } else {
+            self.event_publisher.notify(DhtEvent::BootstrapCompleted).await;
         }
     }