Update Observers of all existing cache entries upon registration (#27)

* update observers of all existing cache entries upon registration

query/src/query_cache.rs

@@ -33,7 +33,7 @@ slotmap::new_key_type! {
 struct CacheEntry<K, V>(HashMap<K, Query<K, V>>);
 
 // Trait to enable cache introspection among distinct cache entry maps.
-trait CacheEntryTrait: CacheSize + CacheInvalidate + CacheEntryClear {
+trait CacheEntryTrait: CacheSize + CacheInvalidate + CacheClear + CacheUpdateObserver {
     fn as_any(&self) -> &dyn Any;
     fn as_any_mut(&mut self) -> &mut dyn Any;
 }
@@ -79,11 +79,11 @@ where
     }
 }
 
-trait CacheEntryClear {
+trait CacheClear {
     fn clear(&mut self, cache: &QueryCache);
 }
 
-impl<K, V> CacheEntryClear for CacheEntry<K, V>
+impl<K, V> CacheClear for CacheEntry<K, V>
 where
     K: QueryKey + 'static,
     V: QueryValue + 'static,
@@ -96,6 +96,24 @@ where
     }
 }
 
+// Update an observer with all existing cache entries, upon subscription.
+trait CacheUpdateObserver {
+    fn update_observer(&self, observer: &dyn CacheObserver);
+}
+
+impl<K, V> CacheUpdateObserver for CacheEntry<K, V>
+where
+    K: QueryKey + 'static,
+    V: QueryValue + 'static,
+{
+    fn update_observer(&self, observer: &dyn CacheObserver) {
+        for (_, query) in self.0.iter() {
+            let event = CacheEvent::created(query.clone());
+            observer.process_cache_event(event);
+        }
+    }
+}
+
 impl QueryCache {
     pub fn new(owner: Owner) -> Self {
         Self {
@@ -374,6 +392,11 @@ impl QueryCache {
     }
 
     pub fn register_observer(&self, observer: impl CacheObserver + 'static) -> CacheObserverKey {
+        // Update all existing cache entries with the new observer.
+        self.cache.borrow().values().for_each(|cache| {
+            cache.update_observer(&observer);
+        });
+
         self.observers
             .try_borrow_mut()
             .expect("register_query_observer borrow mut")