Api improve

CHANGELOG.md

@@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Add `Peer::to_route_table` API
+- Add `Peer:send_to_peers` API
 - Add `max_udp_len` configuration parameter
 - Add range checks to MTU (between 1296 and 8192)
 - Add network version to handshake messages
@@ -20,6 +22,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Fix raptorQ cache default config
 - Fix ObjectTransmissionInformation deserialization
 - Fix duplicate processing for messages with different RaptorQ configurations
+- Fix idle nodes removal on maintainance
 
 ### Changed
 

Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "kadcast"
 authors = ["herr-seppia <seppia@dusk.network>"]
-version = "0.7.0-rc.8"
+version = "0.7.0-rc.10"
 edition = "2018"
 description = "Implementation of the Kadcast Network Protocol."
 categories = ["network-programming"]

src/kbucket.rs

@@ -99,6 +99,15 @@ impl<V> Tree<V> {
             .take(ITEM_COUNT)
     }
 
+    pub(crate) fn buckets(
+        &self,
+    ) -> impl Iterator<Item = (BucketHeight, impl Iterator<Item = &Node<V>>)>
+    {
+        self.buckets
+            .iter()
+            .map(|(&height, bucket)| (height, bucket.peers()))
+    }
+
     pub(crate) fn all_sorted(
         &self,
     ) -> impl Iterator<Item = (BucketHeight, impl Iterator<Item = &Node<V>>)>

src/kbucket/bucket.rs

@@ -264,7 +264,7 @@ impl<V> Bucket<V> {
     /// Get idle nodes from the bucket.
     pub(crate) fn idle_nodes(&self) -> impl Iterator<Item = &Node<V>> {
         let ttl = self.bucket_config.node_ttl;
-        self.nodes.iter().filter(move |n| n.is_alive(ttl))
+        self.nodes.iter().filter(move |n| !n.is_alive(ttl))
     }
 
     /// Checks if the bucket contains a node with the given peer key.

src/kbucket/node.rs

@@ -78,4 +78,18 @@ impl<TValue> Node<TValue> {
     pub(super) fn is_alive(&self, duration: Duration) -> bool {
         self.seen_at.elapsed() < duration
     }
+
+    /// Returns the time when the node was last seen.
+    ///
+    /// This function provides a reference to the `Instant` representing the
+    /// last time the node was observed as active in the network. It can be
+    /// used to determine the recency of the node's activity or
+    /// availability.
+    ///
+    /// # Returns
+    /// A reference to an `Instant` indicating the last observed activity of the
+    /// node.
+    pub fn seen_at(&self) -> &Instant {
+        &self.seen_at
+    }
 }

src/lib.rs

@@ -4,9 +4,11 @@
 //
 // Copyright (c) DUSK NETWORK. All rights reserved.
 
+use std::collections::BTreeMap;
 use std::collections::HashSet;
 use std::net::AddrParseError;
 use std::net::SocketAddr;
+use std::time::Instant;
 
 use config::Config;
 use encoding::message::{Header, Message};
@@ -182,6 +184,28 @@ impl Peer {
         });
     }
 
+    /// Returns the current routing table.
+    ///
+    /// # Returns
+    /// A `BTreeMap<u8, Vec<(SocketAddr, Instant)>>` where each key is the
+    /// bucket height and each value is a vector of tuples containing a
+    /// node's address and last seen time.
+    pub async fn to_route_table(
+        &self,
+    ) -> BTreeMap<u8, Vec<(SocketAddr, Instant)>> {
+        let mut route_table = BTreeMap::new();
+
+        let table_read = self.ktable.read().await;
+        table_read.buckets().for_each(|(h, nodes)| {
+            let nodes = nodes
+                .map(|p| (*p.value().address(), *p.seen_at()))
+                .collect::<Vec<_>>();
+            route_table.insert(h, nodes);
+        });
+
+        route_table
+    }
+
     /// Broadcast a message to the network
     ///
     /// # Arguments
@@ -239,6 +263,25 @@ impl Peer {
     /// The function returns just after the message is put on the internal queue
     /// system. It **does not guarantee** the message will be broadcasted
     pub async fn send(&self, message: &[u8], target: SocketAddr) {
+        self.send_to_peers(message, vec![target]).await
+    }
+
+    /// Send a message to multiple peers in the network
+    ///
+    /// # Arguments
+    ///
+    /// * `message` - Byte array containing the message to be sent
+    /// * `targets` - Vector of receiver addresses (`Vec<SocketAddr>`)
+    ///
+    /// Note:
+    /// The function returns just after the message is put on the internal queue
+    /// system. It **does not guarantee** the message will be broadcasted to
+    /// all.
+    pub async fn send_to_peers(
+        &self,
+        message: &[u8],
+        targets: Vec<SocketAddr>,
+    ) {
         if message.is_empty() {
             return;
         }
@@ -248,10 +291,9 @@ impl Peer {
             self.header,
             BroadcastPayload {
                 height: 0,
-                gossip_frame: message.to_vec(), //FIX_ME: avoid clone
+                gossip_frame: message.to_vec(),
             },
         );
-        let targets = vec![target];
         self.outbound_sender
             .send((msg, targets))
             .await