net/outbound_session: flag to toggle channel_subscriber notifications, <br> to avoid subscriber.notify() on closed channel

src/net/p2p.rs

@@ -195,6 +195,9 @@ impl P2p {
             let self_inbound_addr = self.settings().external_addr.clone();
             let addrs = self.hosts().load_all().await;
 
+            // Enable outbound channel subscriber notifications
+            self.session_outbound().await.clone().enable_notify().await;
+
             // Retrieve outbound channel subscriber ptr
             let outbound_sub =
                 self.session_outbound.lock().await.as_ref().unwrap().subscribe_channel().await;
@@ -213,6 +216,9 @@ impl P2p {
                     );
                 }
             }
+
+            // Disable outbound channel subscriber notifications
+            self.session_outbound().await.disable_notify().await;
         }
 
         debug!(target: "net", "P2p::wait_for_outbound() [END]");

src/net/session/outbound_session.rs

@@ -78,7 +78,10 @@ pub struct OutboundSession {
     p2p: Weak<P2p>,
     connect_slots: Mutex<Vec<StoppableTaskPtr>>,
     slot_info: Mutex<Vec<OutboundInfo>>,
+    /// Subscriber used to signal channels processing
     channel_subscriber: SubscriberPtr<Result<ChannelPtr>>,
+    /// Flag to toggle channel_subscriber notifications
+    notify: Mutex<bool>,
 }
 
 impl OutboundSession {
@@ -89,6 +92,7 @@ impl OutboundSession {
             connect_slots: Mutex::new(Vec::new()),
             slot_info: Mutex::new(Vec::new()),
             channel_subscriber: Subscriber::new(),
+            notify: Mutex::new(false),
         })
     }
 
@@ -175,7 +179,9 @@ impl OutboundSession {
                     }
 
                     // Notify that channel processing has been finished
-                    self.channel_subscriber.notify(Ok(channel)).await;
+                    if *self.notify.lock().await {
+                        self.channel_subscriber.notify(Ok(channel)).await;
+                    }
 
                     // Wait for channel to close
                     stop_sub.unwrap().receive().await;
@@ -190,7 +196,9 @@ impl OutboundSession {
                     }
 
                     // Notify that channel processing has been finished
-                    self.channel_subscriber.notify(Err(err)).await;
+                    if *self.notify.lock().await {
+                        self.channel_subscriber.notify(Err(err)).await;
+                    }
                 }
             }
         }
@@ -242,6 +250,16 @@ impl OutboundSession {
     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;
+    }
 }
 
 #[async_trait]