Enable Mercury to be shut down and all pending requests being cancelled.

core/src/mercury/mod.rs

@@ -20,6 +20,7 @@ component! {
         sequence: SeqGenerator<u64> = SeqGenerator::new(0),
         pending: HashMap<Vec<u8>, MercuryPending> = HashMap::new(),
         subscriptions: Vec<(String, mpsc::UnboundedSender<MercuryResponse>)> = Vec::new(),
+        is_shutdown: bool = false,
     }
 }
 
@@ -61,7 +62,11 @@ impl MercuryManager {
         };
 
         let seq = self.next_seq();
-        self.lock(|inner| inner.pending.insert(seq.clone(), pending));
+        self.lock(|inner| {
+            if !inner.is_shutdown {
+                inner.pending.insert(seq.clone(), pending);
+            }
+        });
 
         let cmd = req.method.command();
         let data = req.encode(&seq);
@@ -109,6 +114,7 @@ impl MercuryManager {
             let (tx, rx) = mpsc::unbounded();
 
             manager.lock(move |inner| {
+                if !inner.is_shutdown {
                     debug!("subscribed uri={} count={}", uri, response.payload.len());
                     if response.payload.len() > 0 {
                         // Old subscription protocol, watch the provided list of URIs
@@ -125,6 +131,7 @@ impl MercuryManager {
                         // New subscription protocol, watch the requested URI
                         inner.subscriptions.push((uri, tx));
                     }
+                }
             });
 
             rx
@@ -222,4 +229,13 @@ impl MercuryManager {
             }
         }
     }
+
+    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.pending.clear();
+            inner.subscriptions.clear();
+        });
+    }
 }

core/src/session.rs

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