net: channel::subscribe_stop return ChannelStopped error if the channel

is stopped

src/net/channel.rs

@@ -114,9 +114,9 @@ impl Channel {
     /// the channel has been closed.
     pub async fn stop(&self) {
         debug!(target: "net", "Channel::stop() [START, address={}]", self.address());
-        let mut stopped = self.stopped.lock().await;
-        if !*stopped {
-            *stopped = true;
+        let mut stopped = *self.stopped.lock().await;
+        if !stopped {
+            stopped = true;
             drop(stopped);
 
             self.stop_subscriber.notify(Error::ChannelStopped).await;
@@ -127,19 +127,26 @@ impl Channel {
     }
 
     /// Creates a subscription to a stopped signal.
-    pub async fn subscribe_stop(&self) -> Subscription<Error> {
+    pub async fn subscribe_stop(&self) -> Result<Subscription<Error>> {
         debug!(target: "net",
          "Channel::subscribe_stop() [START, address={}]",
          self.address()
         );
-        // TODO: this should check the stopped status
-        // Call to receive should return ChannelStopped on newly created sub
+
+        {
+            let stopped = *self.stopped.lock().await;
+            if stopped {
+                return Err(Error::ChannelStopped)
+            }
+        }
+
         let sub = self.stop_subscriber.clone().subscribe().await;
         debug!(target: "net",
          "Channel::subscribe_stop() [END, address={}]",
          self.address()
         );
-        sub
+
+        Ok(sub)
     }
 
     /// Sends a message across a channel. Calls function 'send_message' that
@@ -152,7 +159,6 @@ impl Channel {
          self.address()
         );
 
-        // TODO can we use RwLock here instead of Mutex
         {
             let stopped = *self.stopped.lock().await;
             if stopped {

src/net/protocol/protocol_jobs_manager.rs

@@ -46,9 +46,10 @@ impl ProtocolJobsManager {
     async fn handle_stop(self: Arc<Self>) {
         let stop_sub = self.channel.clone().subscribe_stop().await;
 
-        // Wait for the stop signal
-        // Not interested in the exact error
-        let _ = stop_sub.receive().await;
+        if stop_sub.is_ok() {
+            // Wait for the stop signal
+            stop_sub.unwrap().receive().await;
+        }
 
         self.close_all_tasks().await
     }

src/net/session/inbound_session.rs

@@ -128,7 +128,10 @@ impl InboundSession {
             .insert(key.clone(), InboundInfo { channel: channel.clone() });
 
         let stop_sub = channel.subscribe_stop().await;
-        stop_sub.receive().await;
+
+        if stop_sub.is_ok() {
+            stop_sub.unwrap().receive().await;
+        }
 
         self.connect_infos.lock().await.remove(&key);
     }

src/net/session/manual_session.rs

@@ -82,6 +82,10 @@ impl ManualSession {
 
                     let stop_sub = channel.subscribe_stop().await;
 
+                    if stop_sub.is_err() {
+                        continue
+                    }
+
                     self.clone().register_channel(channel.clone(), executor.clone()).await?;
 
                     // Channel is now connected but not yet setup
@@ -92,7 +96,7 @@ impl ManualSession {
                     //self.clone().attach_protocols(channel, executor.clone()).await?;
 
                     // Wait for channel to close
-                    stop_sub.receive().await;
+                    stop_sub.unwrap().receive().await;
                 }
                 Err(err) => {
                     info!(target: "net", "Unable to connect to manual outbound [{}]: {}", addr, err);

src/net/session/mod.rs

@@ -52,8 +52,12 @@ async fn remove_sub_on_stop(p2p: P2pPtr, channel: ChannelPtr) {
     debug!(target: "net", "remove_sub_on_stop() [START]");
     // Subscribe to stop events
     let stop_sub = channel.clone().subscribe_stop().await;
-    // Wait for a stop event
-    let _ = stop_sub.receive().await;
+
+    if stop_sub.is_ok() {
+        // Wait for a stop event
+        stop_sub.unwrap().receive().await;
+    }
+
     debug!(target: "net",
         "remove_sub_on_stop(): received stop event. Removing channel {}",
         channel.address()

src/net/session/outbound_session.rs

@@ -153,6 +153,10 @@ impl OutboundSession {
 
                     let stop_sub = channel.subscribe_stop().await;
 
+                    if stop_sub.is_err() {
+                        continue
+                    }
+
                     self.clone().register_channel(channel.clone(), executor.clone()).await?;
 
                     // Channel is now connected but not yet setup
@@ -166,7 +170,7 @@ impl OutboundSession {
                     }
 
                     // Wait for channel to close
-                    stop_sub.receive().await;
+                    stop_sub.unwrap().receive().await;
                 }
                 Err(err) => {
                     info!(target: "net", "Unable to connect to outbound [{}]: {}", &addr, err);