Gave route storage the abiliity to remove routes.

src/queue/data_director.rs

@@ -313,6 +313,23 @@ impl RouteStorage {
         }
     }
 
+    fn remove_route(&mut self, route: Route, sender: SenderID) {
+        let route_id: RouteID = route.into();
+        let mut remove = false;
+        match self.data.get_mut(&route_id) {
+            Some(store) => {
+                store.remove(&sender);
+                if store.len() == 0 {
+                    remove = true;
+                }
+            }
+            None => {}
+        }
+        if remove {
+            self.data.remove(&route_id);
+        }
+    }
+
     fn get(&self, route: Route) -> HashSet<SenderID> {
         let mut output = HashSet::new();
         for (route_id, set) in self.data.iter() {
@@ -439,6 +456,31 @@ mod route_storeage {
         routes.remove_sender_id(&id);
         assert_eq!(routes.data.len(), 0);
     }
+
+    #[test]
+    fn can_route_be_removed() {
+        let mut routes = RouteStorage::new();
+        let id = SenderID::new();
+        let route = Route::new(Include::Just(MessageID::new()), Include::All, Include::All);
+        routes.add(route.clone(), id.clone());
+        routes.remove_route(route.clone(), id);
+        assert_eq!(routes.data.len(), 0);
+    }
+
+    #[test]
+    fn can_shared_route_be_removed() {
+        let mut routes = RouteStorage::new();
+        let id1 = SenderID::new();
+        let id2 = SenderID::new();
+        let route = Route::new(Include::Just(MessageID::new()), Include::All, Include::All);
+        routes.add(route.clone(), id1.clone());
+        routes.add(route.clone(), id2.clone());
+        routes.remove_route(route.clone(), id1);
+        assert_eq!(routes.data.len(), 1);
+        let ids = routes.get(route.clone());
+        let id = ids.iter().last().unwrap();
+        assert_eq!(id, &id2);
+    }
 }
 
 pub struct DocRegistry {