Enable proper shutdown of the channels.

core/src/channel.rs

@@ -14,6 +14,7 @@ component! {
         download_rate_estimate: usize = 0,
         download_measurement_start: Option<Instant> = None,
         download_measurement_bytes: usize = 0,
+        is_shutdown: bool = false,
     }
 }
 
@@ -46,7 +47,9 @@ impl ChannelManager {
 
         let seq = self.lock(|inner| {
             let seq = inner.sequence.get();
+            if !inner.is_shutdown {
                 inner.channels.insert(seq, tx);
+            }
             seq
         });
 
@@ -87,6 +90,14 @@ impl ChannelManager {
     pub fn get_download_rate_estimate(&self) -> usize {
         return self.lock(|inner| inner.download_rate_estimate);
     }
+
+    pub(crate) fn shutdown(&self) {
+        self.lock(|inner| {
+            inner.is_shutdown = true;
+            // destroy the sending halves of the channels to signal everyone who is waiting for something.
+            inner.channels.clear();
+        });
+    }
 }
 
 impl Channel {

core/src/session.rs

@@ -238,6 +238,7 @@ impl Session {
         debug!("Invalidating session[{}]", self.0.session_id);
         self.0.data.write().unwrap().invalid = true;
         self.mercury().shutdown();
+        self.channel().shutdown();
     }
 
     pub fn is_invalid(&self) -> bool {