feat: Introduce configurable bundle directories and enhance routing peer management

- Added constants for advanced and custom routing bundle directories in `consts.rs`.
- Updated example files to utilize these constants for improved path management.
- Refactored `DtnNode` to include a `ClaManager` for better peer handling and connectivity checks.
- Implemented `ClaPeer` trait for TCP and BLE peers, enhancing the routing algorithm's ability to select reachable peers.
- Improved epidemic routing algorithm to support asynchronous peer selection based on connectivity.
- Updated tests to reflect changes in peer management and routing logic.

These enhancements improve the flexibility and reliability of the DTN node's routing capabilities.

Author: Ray-Gee

examples/advanced.rs

@@ -1,3 +1,4 @@
+use sdtn::consts::BUNDLES_ADVANCED_DIR;
 use sdtn::{convenience, BundleStatus, DtnNode};
 
 #[tokio::main]
@@ -18,13 +19,15 @@ async fn main() -> anyhow::Result<()> {
 
     for (i, msg) in messages.iter().enumerate() {
         println!("  Inserting bundle {}: {}", i + 1, msg);
-        node.insert_bundle(msg.to_string())?;
+        node.insert_bundle(msg.to_string()).await?;
     }
 
     // Method 1b: Using DtnNode API with custom store path
     println!("\n📋 Method 1b: Using DtnNode API (custom store path)");
-    let custom_node = DtnNode::with_store_path("./advanced_bundles")?;
-    custom_node.insert_bundle("Message in custom store".to_string())?;
+    let custom_node = DtnNode::with_store_path(BUNDLES_ADVANCED_DIR)?;
+    custom_node
+        .insert_bundle("Message in custom store".to_string())
+        .await?;
 
     // Get detailed status for a specific bundle
     let bundles = node.list_bundles()?;
@@ -49,7 +52,7 @@ async fn main() -> anyhow::Result<()> {
     println!("\n📋 Method 2: Using convenience functions");
 
     // Quick insert
-    convenience::insert_bundle_quick("Quick message from convenience function")?;
+    convenience::insert_bundle_quick("Quick message from convenience function").await?;
 
     // Quick list
     let quick_bundles = convenience::list_bundles_quick()?;
@@ -123,7 +126,9 @@ async fn main() -> anyhow::Result<()> {
     // Method 5: Using Default trait
     println!("\n📋 Method 5: Using Default trait");
     let default_node = DtnNode::default();
-    default_node.insert_bundle("Message from default instance".to_string())?;
+    default_node
+        .insert_bundle("Message from default instance".to_string())
+        .await?;
 
     println!("\n✅ Advanced usage example completed successfully!");
     Ok(())

examples/basic.rs

@@ -7,7 +7,8 @@ async fn main() -> anyhow::Result<()> {
 
     // Insert a bundle
     println!("📦 Inserting a new bundle...");
-    node.insert_bundle("Hello from SpaceArth DTN!".to_string())?;
+    node.insert_bundle("Hello from SpaceArth DTN!".to_string())
+        .await?;
 
     // List all bundles
     println!("📋 Listing all bundles...");

examples/routing.rs

@@ -1,3 +1,4 @@
+use sdtn::consts::BUNDLES_CUSTOM_ROUTING_DIR;
 use sdtn::routing::algorithm::{RouteEntry, RoutingAlgorithmType, RoutingConfig};
 use sdtn::{bpv7::EndpointId, DtnNode};
 
@@ -51,7 +52,8 @@ async fn main() -> anyhow::Result<()> {
     }
 
     // Insert a test bundle
-    node.insert_bundle("Test message for real routing".to_string())?;
+    node.insert_bundle("Test message for real routing".to_string())
+        .await?;
 
     // Get the bundle and test routing
     let bundles = node.list_bundles()?;
@@ -61,7 +63,7 @@ async fn main() -> anyhow::Result<()> {
         println!("  Destination: {}", bundle.primary.destination);
 
         // Test the new routing method using routing table
-        match node.select_routes_for_forwarding(&bundle) {
+        match node.select_routes_for_forwarding(&bundle).await {
             Ok(routes) => {
                 println!("  Selected {} routes for forwarding:", routes.len());
                 for (i, route) in routes.iter().enumerate() {
@@ -102,7 +104,7 @@ async fn main() -> anyhow::Result<()> {
     // Method 2: Using custom routing algorithm with routing table
     println!("\n📋 Method 2: Using Prophet routing with routing table");
     let routing_config = RoutingConfig::new(RoutingAlgorithmType::Prophet);
-    let custom_node = DtnNode::with_routing_algorithm("./custom_routing_bundles", routing_config)?;
+    let custom_node = DtnNode::with_routing_algorithm(BUNDLES_CUSTOM_ROUTING_DIR, routing_config)?;
 
     // Add routes to custom node
     custom_node.add_route(RouteEntry {
@@ -122,15 +124,17 @@ async fn main() -> anyhow::Result<()> {
     })?;
 
     // Insert a test bundle with custom routing
-    custom_node.insert_bundle("Message with Prophet routing".to_string())?;
+    custom_node
+        .insert_bundle("Message with Prophet routing".to_string())
+        .await?;
 
     // Test routing with custom algorithm
     let bundles = custom_node.list_bundles()?;
     if let Some(id) = bundles.first() {
         let bundle = custom_node.show_bundle(id)?;
         println!("  Testing Prophet routing for bundle: {}", id);
 
-        match custom_node.select_routes_for_forwarding(&bundle) {
+        match custom_node.select_routes_for_forwarding(&bundle).await {
             Ok(routes) => {
                 println!("  Selected {} routes for forwarding:", routes.len());
                 for (i, route) in routes.iter().enumerate() {

src/api/convenience.rs

@@ -2,9 +2,9 @@ use super::DtnNode;
 use crate::bpv7::bundle::Bundle;
 
 /// Quick bundle insertion using default settings
-pub fn insert_bundle_quick(message: &str) -> anyhow::Result<()> {
+pub async fn insert_bundle_quick(message: &str) -> anyhow::Result<()> {
     let node = DtnNode::new()?;
-    node.insert_bundle(message.to_string())
+    node.insert_bundle(message.to_string()).await
 }
 
 /// Quick bundle listing using default settings

src/api/node.rs

@@ -1,21 +1,25 @@
 use crate::bpv7::bundle::*;
 use crate::cla::manager::ClaManager;
+use crate::cla::TcpPeer;
 use crate::config::{generate_creation_timestamp, Config};
+use crate::consts::BUNDLES_DIR;
 use crate::routing::algorithm::{
-    ConvergenceSender, RouteEntry, RoutingAlgorithm, RoutingConfig, RoutingTable, TcpSender,
+    ClaPeer, RouteEntry, RoutingAlgorithm, RoutingConfig, RoutingTable,
 };
 use crate::store::bundle_descriptor::BundleDescriptor;
 use crate::store::BundleStore;
 use std::sync::{Arc, Mutex};
+use tokio::sync::Mutex as TokioMutex;
 
 use super::BundleStatus;
 
 /// DTN Node API for managing DTN bundles and network operations
 pub struct DtnNode {
     store: BundleStore,
     store_path: String,
-    routing_algorithm: Arc<Mutex<Box<dyn RoutingAlgorithm>>>,
+    routing_algorithm: Arc<TokioMutex<Box<dyn RoutingAlgorithm>>>,
     routing_table: Arc<Mutex<RoutingTable>>,
+    cla_manager: Arc<ClaManager>,
 }
 
 impl DtnNode {
@@ -26,7 +30,7 @@ impl DtnNode {
             Ok(path) => path,
             Err(_) => match Config::load() {
                 Ok(config) => config.storage.path,
-                Err(_) => "./bundles".to_string(),
+                Err(_) => BUNDLES_DIR.to_string(),
             },
         };
 
@@ -38,20 +42,22 @@ impl DtnNode {
         let store = BundleStore::new(store_path)?;
         let config = Config::load()?;
         let routing_config = RoutingConfig::new(config.get_routing_algorithm_type());
-        let routing_algorithm = Arc::new(Mutex::new(routing_config.create_algorithm()));
+        let routing_algorithm = Arc::new(TokioMutex::new(routing_config.create_algorithm()));
         let routing_table = Arc::new(Mutex::new(RoutingTable::new()));
+        let cla_manager = Arc::new(ClaManager::new(|_bundle| {}));
 
         Ok(Self {
             store,
             store_path: store_path.to_string(),
             routing_algorithm,
             routing_table,
+            cla_manager,
         })
     }
 
     /// Create a new DTN CLI instance with custom configuration
     pub fn with_config(store_path: Option<&str>) -> anyhow::Result<Self> {
-        let path = store_path.unwrap_or("./bundles");
+        let path = store_path.unwrap_or(BUNDLES_DIR);
         Self::with_store_path(path)
     }
 
@@ -61,14 +67,16 @@ impl DtnNode {
         routing_config: RoutingConfig,
     ) -> anyhow::Result<Self> {
         let store = BundleStore::new(store_path)?;
-        let routing_algorithm = Arc::new(Mutex::new(routing_config.create_algorithm()));
+        let routing_algorithm = Arc::new(TokioMutex::new(routing_config.create_algorithm()));
         let routing_table = Arc::new(Mutex::new(RoutingTable::new()));
+        let cla_manager = Arc::new(ClaManager::new(|_bundle| {}));
 
         Ok(Self {
             store,
             store_path: store_path.to_string(),
             routing_algorithm,
             routing_table,
+            cla_manager,
         })
     }
 
@@ -82,6 +90,11 @@ impl DtnNode {
         }
     }
 
+    /// Get access to the routing table for advanced operations
+    pub fn get_routing_table(&self) -> Arc<Mutex<RoutingTable>> {
+        Arc::clone(&self.routing_table)
+    }
+
     /// Get all routes from the routing table
     pub fn get_all_routes(&self) -> anyhow::Result<Vec<RouteEntry>> {
         if let Ok(table) = self.routing_table.lock() {
@@ -104,7 +117,7 @@ impl DtnNode {
     }
 
     /// Insert a new bundle with the given message
-    pub fn insert_bundle(&self, message: String) -> anyhow::Result<()> {
+    pub async fn insert_bundle(&self, message: String) -> anyhow::Result<()> {
         #[cfg(test)]
         let config = {
             // In tests, use a slightly different timestamp each time to avoid duplicates
@@ -130,58 +143,51 @@ impl DtnNode {
 
         // Notify routing algorithm about new bundle
         let descriptor = BundleDescriptor::new(bundle);
-        if let Ok(mut algorithm) = self.routing_algorithm.lock() {
-            algorithm.notify_new_bundle(&descriptor);
-        }
+        let mut algorithm = self.routing_algorithm.lock().await;
+        algorithm.notify_new_bundle(&descriptor);
 
         Ok(())
     }
 
     /// Select peers for forwarding a bundle (legacy method)
-    pub fn select_peers_for_forwarding(
+    pub async fn select_peers_for_forwarding(
         &self,
         bundle: &Bundle,
-    ) -> anyhow::Result<Vec<Box<dyn ConvergenceSender>>> {
+    ) -> anyhow::Result<Vec<Box<dyn ClaPeer>>> {
         let descriptor = BundleDescriptor::new(bundle.clone());
 
-        // For now, create some dummy senders for demonstration
-        // In a real implementation, this would come from the CLA manager
-        let senders: Vec<Box<dyn ConvergenceSender>> = vec![
-            Box::new(TcpSender::new(crate::bpv7::EndpointId::from("dtn://peer1"))),
-            Box::new(TcpSender::new(crate::bpv7::EndpointId::from("dtn://peer2"))),
-        ];
-
-        if let Ok(algorithm) = self.routing_algorithm.lock() {
-            let selected_refs = algorithm.select_peers_for_forwarding(&descriptor, &senders);
-
-            // Convert references back to owned boxes (this is a bit awkward, but necessary for the trait)
-            let result = selected_refs
-                .into_iter()
-                .map(|sender_ref| {
-                    let eid = sender_ref.get_peer_endpoint_id();
-                    Box::new(TcpSender::new(eid)) as Box<dyn ConvergenceSender>
-                })
-                .collect();
+        // Get peers from CLA manager (dummy for now)
+        let peers = self.cla_manager.list_peers().await;
 
-            Ok(result)
-        } else {
-            anyhow::bail!("Failed to lock routing algorithm")
-        }
+        let algorithm = self.routing_algorithm.lock().await;
+        let selected_refs = algorithm.select_peers_for_forwarding(&descriptor, &peers);
+
+        // Convert references back to owned boxes (this is a bit awkward, but necessary for the trait)
+        let result = selected_refs
+            .into_iter()
+            .map(|peer_ref| {
+                let eid = peer_ref.get_peer_endpoint_id();
+                let address = peer_ref.get_connection_address();
+                Box::new(TcpPeer::new(eid, address)) as Box<dyn ClaPeer>
+            })
+            .collect();
+
+        Ok(result)
     }
 
     /// Select routes for forwarding a bundle (new method using routing table)
-    pub fn select_routes_for_forwarding(&self, bundle: &Bundle) -> anyhow::Result<Vec<RouteEntry>> {
+    pub async fn select_routes_for_forwarding(
+        &self,
+        bundle: &Bundle,
+    ) -> anyhow::Result<Vec<RouteEntry>> {
         let descriptor = BundleDescriptor::new(bundle.clone());
 
-        if let Ok(algorithm) = self.routing_algorithm.lock() {
-            if let Ok(table) = self.routing_table.lock() {
-                let routes = algorithm.select_routes_for_forwarding(&descriptor, &table);
-                Ok(routes)
-            } else {
-                anyhow::bail!("Failed to lock routing table")
-            }
+        let algorithm = self.routing_algorithm.lock().await;
+        if let Ok(table) = self.routing_table.lock() {
+            let routes = algorithm.select_routes_for_forwarding(&descriptor, &table);
+            Ok(routes)
         } else {
-            anyhow::bail!("Failed to lock routing algorithm")
+            anyhow::bail!("Failed to lock routing table")
         }
     }
 
@@ -261,7 +267,7 @@ impl DtnNode {
 
     /// Start a TCP dialer daemon
     pub async fn start_tcp_dialer(&self, target_addr: String) -> anyhow::Result<()> {
-        let cla = Arc::new(crate::cla::TcpClaDialer { target_addr });
+        let cla = Arc::new(crate::cla::TcpClaClient { target_addr });
         let manager = ClaManager::new(|bundle| {
             println!("📤 Should not receive here (Dialer): {:?}", bundle);
         });
@@ -270,6 +276,34 @@ impl DtnNode {
         tokio::time::sleep(tokio::time::Duration::from_secs(2)).await;
         Ok(())
     }
+
+    /// Select peers for forwarding a bundle with connectivity check (async version)
+    pub async fn select_peers_for_forwarding_async(
+        &self,
+        bundle: &Bundle,
+    ) -> anyhow::Result<Vec<Box<dyn ClaPeer>>> {
+        let descriptor = BundleDescriptor::new(bundle.clone());
+
+        // Get peers from CLA manager (dummy for now)
+        let peers = self.cla_manager.list_peers().await;
+
+        let algorithm = self.routing_algorithm.lock().await;
+        let selected_refs = algorithm
+            .select_peers_for_forwarding_async(&descriptor, &peers)
+            .await;
+
+        // Convert references back to owned boxes
+        let result = selected_refs
+            .into_iter()
+            .map(|peer_ref| {
+                let eid = peer_ref.get_peer_endpoint_id();
+                let address = peer_ref.get_connection_address();
+                Box::new(TcpPeer::new(eid, address)) as Box<dyn ClaPeer>
+            })
+            .collect();
+
+        Ok(result)
+    }
 }
 
 /// Default implementation for DtnNode