p2pnet: manual sessions to support outbound transport enforcing, wait_for_outbound to include manual peers connections, minor fixes

src/net/p2p.rs

@@ -2,7 +2,7 @@ use async_std::sync::{Arc, Mutex};
 use std::fmt;
 
 use async_executor::Executor;
-use futures::{select, FutureExt};
+use futures::{select, try_join, FutureExt};
 use fxhash::{FxHashMap, FxHashSet};
 use log::{debug, warn};
 use serde_json::json;
@@ -193,9 +193,25 @@ impl P2p {
             self.settings.outbound_connections > 0
         {
             debug!(target: "net", "P2p::wait_for_outbound(): seeds are configured, waiting for outbound initialization...");
+            // Retrieve P2P network settings;
+            let settings = self.settings();
 
-            let self_inbound_addr = self.settings().external_addr.clone();
-            let addrs = self.hosts().load_all().await;
+            // Retrieve our own inbound addresses
+            let self_inbound_addr = &settings.external_addr;
+
+            // Retrieve timeout config
+            let timeout = settings.connect_timeout_seconds as u64;
+
+            // Retrieve outbound addresses to connect to (including manual peers)
+            let peers = &settings.peers;
+            let outbound = &self.hosts().load_all().await;
+
+            // Enable manual channel subscriber notifications
+            self.session_manual().await.clone().enable_notify().await;
+
+            // Retrieve manual channel subscriber ptr
+            let manual_sub =
+                self.session_manual.lock().await.as_ref().unwrap().subscribe_channel().await;
 
             // Enable outbound channel subscriber notifications
             self.session_outbound().await.clone().enable_notify().await;
@@ -204,46 +220,28 @@ impl P2p {
             let outbound_sub =
                 self.session_outbound.lock().await.as_ref().unwrap().subscribe_channel().await;
 
-            // Retrieve sto subscriber
-            let stop_sub = self.subscribe_stop().await;
-
-            // Retrieve timeout config
-            let timeout = self.settings().connect_timeout_seconds as u64;
-
-            // 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.
-                // We use a timeout to eliminate the following cases:
-                //  1. Network timeout
-                //  2. Thread reaching the receiver after peer has signal it
-                let (timeout_s, timeout_r) = async_channel::unbounded::<()>();
-                executor
-                    .spawn(async move {
-                        sleep(timeout).await;
-                        timeout_s.send(()).await.unwrap_or(());
-                    })
-                    .detach();
-
-                select! {
-                    msg = outbound_sub.receive().fuse() => {
-                            if let Err(e) = msg {
-                                warn!(
-                                    "P2p::wait_for_outbound(): Outbound connection failed [{}]: {}",
-                                    &addr, e
-                                );
-                            }
-                    },
-                    _ = stop_sub.receive().fuse() => debug!("P2p::wait_for_outbound(): stop signal received!"),
-                    _ = timeout_r.recv().fuse() => {
-                        warn!("P2p::wait_for_outbound(): Timeout on outbound connection: {}", &addr);
-                        continue
-                    },
-                }
-            }
+            // Create tasks for peers and outbound
+            let peers_task = Self::outbound_addr_loop(
+                self_inbound_addr,
+                timeout,
+                self.subscribe_stop().await,
+                peers,
+                manual_sub,
+                executor.clone(),
+            );
+            let outbound_task = Self::outbound_addr_loop(
+                self_inbound_addr,
+                timeout,
+                self.subscribe_stop().await,
+                outbound,
+                outbound_sub,
+                executor,
+            );
+            // Wait for both tasks completion
+            try_join!(peers_task, outbound_task)?;
+
+            // Disable manual channel subscriber notifications
+            self.session_manual().await.disable_notify().await;
 
             // Disable outbound channel subscriber notifications
             self.session_outbound().await.disable_notify().await;
@@ -253,6 +251,53 @@ impl P2p {
         Ok(())
     }
 
+    // Wait for the process for each of the provided addresses, excluding our own inbound addresses
+    async fn outbound_addr_loop(
+        self_inbound_addr: &Vec<Url>,
+        timeout: u64,
+        stop_sub: Subscription<()>,
+        addrs: &Vec<Url>,
+        subscriber: Subscription<Result<ChannelPtr>>,
+        executor: Arc<Executor<'_>>,
+    ) -> Result<()> {
+        // Process addresses
+        for addr in addrs {
+            if self_inbound_addr.contains(addr) {
+                continue
+            }
+
+            // Wait for address to be processed.
+            // We use a timeout to eliminate the following cases:
+            //  1. Network timeout
+            //  2. Thread reaching the receiver after peer has signal it
+            let (timeout_s, timeout_r) = async_channel::unbounded::<()>();
+            executor
+                .spawn(async move {
+                    sleep(timeout).await;
+                    timeout_s.send(()).await.unwrap_or(());
+                })
+                .detach();
+
+            select! {
+                msg = subscriber.receive().fuse() => {
+                        if let Err(e) = msg {
+                            warn!(
+                                "P2p::wait_for_outbound(): Outbound connection failed [{}]: {}",
+                                addr, e
+                            );
+                        }
+                },
+                _ = stop_sub.receive().fuse() => debug!("P2p::wait_for_outbound(): stop signal received!"),
+                _ = timeout_r.recv().fuse() => {
+                    warn!("P2p::wait_for_outbound(): Timeout on outbound connection: {}", addr);
+                    continue
+                },
+            }
+        }
+
+        Ok(())
+    }
+
     pub async fn stop(&self) {
         self.stop_subscriber.notify(()).await
     }
@@ -293,8 +338,22 @@ impl P2p {
     }
 
     /// Check whether a channel is stored in the list of connected channels.
-    pub async fn exists(&self, addr: &Url) -> bool {
-        self.channels.lock().await.contains_key(addr)
+    /// If key is not contained, we also check if we are connected with a different transport.
+    pub async fn exists(&self, addr: &Url) -> Result<bool> {
+        let channels = self.channels.lock().await;
+        if channels.contains_key(addr) {
+            return Ok(true)
+        }
+
+        let mut addr = addr.clone();
+        for transport in &self.settings.outbound_transports {
+            addr.set_scheme(&transport.to_scheme())?;
+            if channels.contains_key(&addr) {
+                return Ok(true)
+            }
+        }
+
+        Ok(false)
     }
 
     /// Add a channel to the list of pending channels.

src/net/session/manual_session.rs

@@ -7,25 +7,35 @@ use serde_json::json;
 use url::Url;
 
 use crate::{
-    system::{StoppableTask, StoppableTaskPtr},
+    net::TransportName,
+    system::{StoppableTask, StoppableTaskPtr, Subscriber, SubscriberPtr, Subscription},
     util::sleep,
     Error, Result,
 };
 
 use super::{
-    super::{Connector, P2p},
+    super::{ChannelPtr, Connector, P2p},
     Session, SessionBitflag, SESSION_MANUAL,
 };
 
 pub struct ManualSession {
     p2p: Weak<P2p>,
     connect_slots: Mutex<Vec<StoppableTaskPtr>>,
+    /// Subscriber used to signal channels processing
+    channel_subscriber: SubscriberPtr<Result<ChannelPtr>>,
+    /// Flag to toggle channel_subscriber notifications
+    notify: Mutex<bool>,
 }
 
 impl ManualSession {
     /// Create a new inbound session.
     pub fn new(p2p: Weak<P2p>) -> Arc<Self> {
-        Arc::new(Self { p2p, connect_slots: Mutex::new(Vec::new()) })
+        Arc::new(Self {
+            p2p,
+            connect_slots: Mutex::new(Vec::new()),
+            channel_subscriber: Subscriber::new(),
+            notify: Mutex::new(false),
+        })
     }
 
     /// Stop the outbound session.
@@ -57,13 +67,26 @@ impl ManualSession {
         executor: Arc<Executor<'_>>,
     ) -> Result<()> {
         let parent = Arc::downgrade(&self);
-        let connector = Connector::new(self.p2p().settings(), Arc::new(parent));
 
         let settings = self.p2p().settings();
 
+        let connector = Connector::new(settings.clone(), Arc::new(parent));
+
         let attempts = settings.manual_attempt_limit;
         let mut remaining = attempts;
 
+        // Retrieve preferent outbound transports
+        let outbound_transports = &settings.outbound_transports;
+
+        // Check that addr transport is in configured outbound transport
+        let addr_transport = TransportName::try_from(addr.clone())?;
+        let transports = if outbound_transports.contains(&addr_transport) {
+            vec![addr_transport]
+        } else {
+            warn!(target: "net", "Manual outbound address {} transport is not in accepted outbound transports, will try with: {:?}", addr, outbound_transports);
+            outbound_transports.clone()
+        };
+
         loop {
             // Loop forever if attempts is 0
             // Otherwise loop attempts number of times
@@ -74,38 +97,50 @@ impl ManualSession {
 
             self.p2p().add_pending(addr.clone()).await;
 
-            info!(target: "net", "Connecting to manual outbound [{}]", addr);
+            for transport in &transports {
+                // Replace addr transport
+                let mut transport_addr = addr.clone();
+                transport_addr.set_scheme(&transport.to_scheme())?;
+                info!(target: "net", "Connecting to manual outbound [{}]", transport_addr);
+                match connector.connect(transport_addr.clone()).await {
+                    Ok(channel) => {
+                        // Blacklist goes here
+                        info!(target: "net", "Connected to manual outbound [{}]", transport_addr);
 
-            match connector.connect(addr.clone()).await {
-                Ok(channel) => {
-                    // Blacklist goes here
+                        let stop_sub = channel.subscribe_stop().await;
+                        if stop_sub.is_err() {
+                            continue
+                        }
 
-                    info!(target: "net", "Connected to manual outbound [{}]", addr);
+                        self.clone().register_channel(channel.clone(), executor.clone()).await?;
 
-                    let stop_sub = channel.subscribe_stop().await;
+                        // Channel is now connected but not yet setup
 
-                    if stop_sub.is_err() {
-                        continue
-                    }
+                        // Remove pending lock since register_channel will add the channel to p2p
+                        self.p2p().remove_pending(&addr).await;
 
-                    self.clone().register_channel(channel.clone(), executor.clone()).await?;
+                        //self.clone().attach_protocols(channel, executor.clone()).await?;
 
-                    // Channel is now connected but not yet setup
+                        // Notify that channel processing has been finished
+                        if *self.notify.lock().await {
+                            self.channel_subscriber.notify(Ok(channel)).await;
+                        }
 
-                    // Remove pending lock since register_channel will add the channel to p2p
-                    self.p2p().remove_pending(&addr).await;
-
-                    //self.clone().attach_protocols(channel, executor.clone()).await?;
-
-                    // Wait for channel to close
-                    stop_sub.unwrap().receive().await;
+                        // Wait for channel to close
+                        stop_sub.unwrap().receive().await;
+                    }
+                    Err(err) => {
+                        info!(target: "net", "Unable to connect to manual outbound [{}]: {}", addr, err);
+                    }
                 }
-                Err(err) => {
-                    info!(target: "net", "Unable to connect to manual outbound [{}]: {}", addr, err);
+            }
 
-                    sleep(settings.connect_timeout_seconds.into()).await;
-                }
+            // Notify that channel processing has been finished (failed)
+            if *self.notify.lock().await {
+                self.channel_subscriber.notify(Err(Error::ConnectFailed)).await;
             }
+
+            sleep(settings.connect_timeout_seconds.into()).await;
         }
 
         warn!(
@@ -118,6 +153,21 @@ impl ManualSession {
         Ok(())
     }
 
+    /// Subscribe to a channel.
+    pub async fn subscribe_channel(&self) -> Subscription<Result<ChannelPtr>> {
+        self.channel_subscriber.clone().subscribe().await
+    }
+
+    /// Enable channel_subscriber notifications.
+    pub async fn enable_notify(self: Arc<Self>) {
+        *self.notify.lock().await = true;
+    }
+
+    /// Disable channel_subscriber notifications.
+    pub async fn disable_notify(self: Arc<Self>) {
+        *self.notify.lock().await = false;
+    }
+
     // Starts sending keep-alive and address messages across the channels.
     /*async fn attach_protocols(
     self: Arc<Self>,

src/net/session/outbound_session.rs

@@ -242,7 +242,12 @@ impl OutboundSession {
             addrs.shuffle(&mut rand::thread_rng());
 
             for addr in addrs {
-                if p2p.exists(&addr).await {
+                if p2p.exists(&addr).await? {
+                    continue
+                }
+
+                // Check if address is in peers list
+                if p2p.settings().peers.contains(&addr) {
                     continue
                 }