p2pnet: wait_for_outbound() implemented

script/research/fud/README.md

@@ -45,23 +45,9 @@ Run fud as follows:
 13:23:04 [INFO] Entry: seedd_config.toml
 13:23:04 [INFO] Starting 8 outbound connection slots.
 13:23:04 [INFO] Entry: lt.py
-13:23:04 [WARN] Hosts address pool is empty. Retrying connect slot #0
-13:23:04 [WARN] Hosts address pool is empty. Retrying connect slot #1
-13:23:04 [WARN] Hosts address pool is empty. Retrying connect slot #2
-13:23:04 [WARN] Hosts address pool is empty. Retrying connect slot #3
-13:23:04 [WARN] Hosts address pool is empty. Retrying connect slot #6
-13:23:04 [WARN] Hosts address pool is empty. Retrying connect slot #4
-13:23:04 [WARN] Hosts address pool is empty. Retrying connect slot #5
-13:23:04 [WARN] Hosts address pool is empty. Retrying connect slot #7
 13:23:07 [INFO] Caught termination signal, cleaning up and exiting...
 ```
 
-After daemon has been initialized, execute network syncing as follows:
-```
-% fu sync
-13:25:46 [INFO] Daemon synced successfully!
-```
-
 fu
 =======
 

script/research/fud/fud/src/main.rs

@@ -102,6 +102,12 @@ impl Fud {
         let entries = fs::read_dir(&self.folder).unwrap();
         {
             let mut lock = self.dht.write().await;
+            
+            // Sync lookup map with network
+            if let Err(e) = lock.sync_lookup_map().await {
+                error!("Failed to sync lookup map: {}", e);
+            }
+            
             for entry in entries {
                 let e = entry.unwrap();
                 let name = String::from(e.file_name().to_str().unwrap());
@@ -195,12 +201,7 @@ impl Fud {
             let mut lock = self.dht.write().await;
             let records = lock.map.clone();
             let mut entries_hashes = HashSet::new();
-
-            // Sync lookup map with network
-            if let Err(e) = lock.sync_lookup_map().await {
-                error!("Failed to sync lookup map: {}", e);
-            }
-
+            
             // We iterate files for new records
             for entry in entries {
                 let e = entry.unwrap();
@@ -404,6 +405,8 @@ async fn realmain(args: Args, ex: Arc<Executor<'_>>) -> Result<()> {
         }
     })
     .detach();
+    
+    p2p.wait_for_outbound().await?;
 
     fud.init().await?;
 

script/research/fud/localnet/fud_config0.toml

@@ -22,7 +22,7 @@ p2p_external = ["tls://127.0.0.1:13339"]
 #slots = 8
 
 # Seed nodes to connect to
-seeds = ["tls://127.0.0.1:13337"]
+seeds = ["tcp://127.0.0.1:13337"]
 
 # Peers to connect to
 #peers = []

script/research/fud/localnet/fud_config1.toml

@@ -22,7 +22,7 @@ p2p_external = ["tls://127.0.0.1:13341"]
 #slots = 8
 
 # Seed nodes to connect to
-seeds = ["tls://127.0.0.1:13337"]
+seeds = ["tcp://127.0.0.1:13337"]
 
 # Peers to connect to
 #peers = []

script/research/fud/localnet/lilith_config.toml

@@ -7,7 +7,7 @@
 ## uncommenting, or by using the command-line.
 
 # Daemon published url, common for all enabled networks
-url = ["tls://127.0.0.1"]
+url = ["tcp://127.0.0.1"]
 
 ## Per-network settings
 #[network."darkfid_sync"]

src/net/p2p.rs

@@ -182,6 +182,43 @@ impl P2p {
         Ok(())
     }
 
+    /// Wait for outbound connections to be established.
+    pub async fn wait_for_outbound(self: Arc<Self>) -> Result<()> {
+        debug!(target: "net", "P2p::wait_for_outbound() [BEGIN]");
+        // To verify that the network needs initialization, we check if we have seeds or peers configured,
+        // and have configured outbound slots.
+        if !(self.settings.seeds.is_empty() && self.settings.peers.is_empty()) &&
+            self.settings.outbound_connections > 0
+        {
+            debug!(target: "net", "P2p::wait_for_outbound(): seeds are configured, waiting for outbound initialization...");
+
+            let self_inbound_addr = self.settings().external_addr.clone();
+            let addrs = self.hosts().load_all().await;
+
+            // Retrieve outbound channel subscriber ptr
+            let outbound_sub =
+                self.session_outbound.lock().await.as_ref().unwrap().subscribe_channel().await;
+
+            // Wait for the result for each of the addresses, excluding our own inbound addresses
+            for addr in addrs {
+                if self_inbound_addr.contains(&addr) {
+                    continue
+                }
+
+                // Wait for address to be processed
+                if let Err(e) = outbound_sub.receive().await {
+                    debug!(
+                        "P2p::wait_for_outbound(): Outbound connection failed [{}]: {}",
+                        &addr, e
+                    );
+                }
+            }
+        }
+
+        debug!(target: "net", "P2p::wait_for_outbound() [END]");
+        Ok(())
+    }
+
     pub async fn stop(&self) {
         self.stop_subscriber.notify(()).await
     }
@@ -265,6 +302,7 @@ impl P2p {
         self.stop_subscriber.clone().subscribe().await
     }
 
+    /// Retrieve channels
     pub fn channels(&self) -> &ConnectedChannels {
         &self.channels
     }

src/net/session/outbound_session.rs

@@ -9,7 +9,7 @@ use serde_json::json;
 use url::Url;
 
 use crate::{
-    system::{StoppableTask, StoppableTaskPtr},
+    system::{StoppableTask, StoppableTaskPtr, Subscriber, SubscriberPtr, Subscription},
     util::async_util,
     Error, Result,
 };
@@ -78,6 +78,7 @@ pub struct OutboundSession {
     p2p: Weak<P2p>,
     connect_slots: Mutex<Vec<StoppableTaskPtr>>,
     slot_info: Mutex<Vec<OutboundInfo>>,
+    channel_subscriber: SubscriberPtr<Result<ChannelPtr>>,
 }
 
 impl OutboundSession {
@@ -87,6 +88,7 @@ impl OutboundSession {
             p2p,
             connect_slots: Mutex::new(Vec::new()),
             slot_info: Mutex::new(Vec::new()),
+            channel_subscriber: Subscriber::new(),
         })
     }
 
@@ -172,6 +174,9 @@ impl OutboundSession {
                         info.state = OutboundState::Connected;
                     }
 
+                    // Notify that channel processing has been finished
+                    self.channel_subscriber.notify(Ok(channel)).await;
+
                     // Wait for channel to close
                     stop_sub.unwrap().receive().await;
                 }
@@ -183,6 +188,9 @@ impl OutboundSession {
                         info.channel = None;
                         info.state = OutboundState::Open;
                     }
+
+                    // Notify that channel processing has been finished
+                    self.channel_subscriber.notify(Err(err)).await;
                 }
             }
         }
@@ -229,6 +237,11 @@ impl OutboundSession {
             async_util::sleep(p2p.settings().outbound_retry_seconds).await;
         }
     }
+
+    /// Subscribe to a channel.
+    pub async fn subscribe_channel(&self) -> Subscription<Result<ChannelPtr>> {
+        self.channel_subscriber.clone().subscribe().await
+    }
 }
 
 #[async_trait]