net: gather unused channel_subscriber's into a single channel_subscriber in Hosts

Also remove redundant calls to notify() and provide a method to
subscribe to the subscriber.

src/net/hosts/store.rs

@@ -700,7 +700,7 @@ pub struct Hosts {
     store_subscriber: SubscriberPtr<usize>,
 
     /// Subscriber for notifications of new channels
-    channel_subscriber: SubscriberPtr<Result<ChannelPtr>>,
+    pub channel_subscriber: SubscriberPtr<Result<ChannelPtr>>,
 
     /// Keeps track of the last time a connection was made.
     pub last_connection: RwLock<Instant>,
@@ -857,6 +857,7 @@ impl Hosts {
 
         self.try_register(address.clone(), HostState::Connected(channel.clone())).await?;
 
+        // Notify that channel processing failed
         self.channel_subscriber.notify(Ok(channel.clone())).await;
 
         let mut last_online = self.last_connection.write().await;
@@ -869,6 +870,10 @@ impl Hosts {
         Ok(self.store_subscriber.clone().subscribe().await)
     }
 
+    pub async fn subscribe_channel(&self) -> Subscription<Result<ChannelPtr>> {
+        self.channel_subscriber.clone().subscribe().await
+    }
+
     // Verify whether a URL is local.
     // NOTE: This function is stateless and not specific to
     // `Hosts`. For this reason, it might make more sense

src/net/session/manual_session.rs

@@ -38,7 +38,6 @@ use url::Url;
 
 use super::{
     super::{
-        channel::ChannelPtr,
         connector::Connector,
         p2p::{P2p, P2pPtr},
     },
@@ -46,7 +45,7 @@ use super::{
 };
 use crate::{
     net::hosts::store::{HostColor, HostState},
-    system::{sleep, LazyWeak, StoppableTask, StoppableTaskPtr, Subscriber, SubscriberPtr},
+    system::{sleep, LazyWeak, StoppableTask, StoppableTaskPtr},
     Error, Result,
 };
 
@@ -56,18 +55,12 @@ pub type ManualSessionPtr = Arc<ManualSession>;
 pub struct ManualSession {
     pub(in crate::net) p2p: LazyWeak<P2p>,
     connect_slots: Mutex<Vec<StoppableTaskPtr>>,
-    /// Subscriber used to signal channels processing
-    channel_subscriber: SubscriberPtr<Result<ChannelPtr>>,
 }
 
 impl ManualSession {
     /// Create a new manual session.
     pub fn new() -> ManualSessionPtr {
-        Arc::new(Self {
-            p2p: LazyWeak::new(),
-            connect_slots: Mutex::new(Vec::new()),
-            channel_subscriber: Subscriber::new(),
-        })
+        Arc::new(Self { p2p: LazyWeak::new(), connect_slots: Mutex::new(Vec::new()) })
     }
 
     /// Stops the manual session.
@@ -138,9 +131,6 @@ impl ManualSession {
                     // Add this connection to the anchorlist
                     self.p2p().hosts().move_host(&addr, last_seen, HostColor::Gold).await;
 
-                    // Notify that channel processing has finished
-                    self.channel_subscriber.notify(Ok(channel)).await;
-
                     // Wait for channel to close
                     stop_sub.receive().await;
                     info!(
@@ -163,7 +153,7 @@ impl ManualSession {
             // Wait and try again.
             // TODO: Should we notify about the failure now, or after all attempts
             // have failed?
-            self.channel_subscriber.notify(Err(Error::ConnectFailed)).await;
+            self.p2p().hosts().channel_subscriber.notify(Err(Error::ConnectFailed)).await;
 
             remaining = if attempts == 0 { 1 } else { remaining - 1 };
             if remaining == 0 {

src/net/session/outbound_session.rs

@@ -51,10 +51,7 @@ use super::{
     Session, SessionBitFlag, SESSION_OUTBOUND,
 };
 use crate::{
-    system::{
-        sleep, timeout::timeout, CondVar, LazyWeak, StoppableTask, StoppableTaskPtr, Subscriber,
-        SubscriberPtr,
-    },
+    system::{sleep, timeout::timeout, CondVar, LazyWeak, StoppableTask, StoppableTaskPtr},
     Error, Result,
 };
 
@@ -64,9 +61,6 @@ pub type OutboundSessionPtr = Arc<OutboundSession>;
 pub struct OutboundSession {
     /// Weak pointer to parent p2p object
     pub(in crate::net) p2p: LazyWeak<P2p>,
-    /// Subscriber used to signal channels processing
-    channel_subscriber: SubscriberPtr<Result<ChannelPtr>>,
-
     /// Outbound connection slots
     slots: Mutex<Vec<Arc<Slot>>>,
     /// Peer discovery task
@@ -78,7 +72,6 @@ impl OutboundSession {
     pub(crate) fn new() -> OutboundSessionPtr {
         let self_ = Arc::new(Self {
             p2p: LazyWeak::new(),
-            channel_subscriber: Subscriber::new(),
             slots: Mutex::new(Vec::new()),
             peer_discovery: PeerDiscovery::new(),
         });
@@ -409,7 +402,9 @@ impl Slot {
 
             let stop_sub = channel.subscribe_stop().await.expect("Channel should not be stopped");
             // Setup new channel
-            if let Err(err) = self.setup_channel(channel.clone()).await {
+            if let Err(err) =
+                self.session().register_channel(channel.clone(), self.p2p().executor()).await
+            {
                 info!(
                     target: "net::outbound_session",
                     "[P2P] Outbound slot #{} disconnected: {}",
@@ -461,26 +456,13 @@ impl Slot {
                 self.p2p().hosts().move_host(&addr, last_seen, HostColor::Grey).await;
 
                 // Notify that channel processing failed
-                self.session().channel_subscriber.notify(Err(Error::ConnectFailed)).await;
+                self.p2p().hosts().channel_subscriber.notify(Err(Error::ConnectFailed)).await;
 
                 Err(Error::ConnectFailed)
             }
         }
     }
 
-    async fn setup_channel(&self, channel: ChannelPtr) -> Result<()> {
-        // Register the new channel
-        debug!(target: "net::outbound_session::setup_channel", "register_channel {}", channel.clone().address());
-        self.session().register_channel(channel.clone(), self.p2p().executor()).await?;
-
-        // Channel is now connected but not yet setup
-
-        // Notify that channel processing has been finished
-        self.session().channel_subscriber.notify(Ok(channel)).await;
-
-        Ok(())
-    }
-
     fn notify(&self) {
         self.wakeup_self.notify()
     }