Generalize listening API for reactive variables.

src/cache/inmemory/reactiveVars.ts

@@ -3,7 +3,12 @@ import { dep } from "optimism";
 import { InMemoryCache } from "./inMemoryCache";
 import { ApolloCache } from '../../core';
 
-export type ReactiveVar<T> = (newValue?: T) => T;
+export interface ReactiveVar<T> {
+  (newValue?: T): T;
+  onNextChange(listener: ReactiveListener<T>): () => void;
+}
+
+export type ReactiveListener<T> = (value: T) => any;
 
 const varDep = dep<ReactiveVar<any>>();
 
@@ -12,34 +17,48 @@ const varDep = dep<ReactiveVar<any>>();
 export const cacheSlot = new Slot<ApolloCache<any>>();
 
 export function makeVar<T>(value: T): ReactiveVar<T> {
-  const caches = new Set<ApolloCache<any>>();
+  const listeners = new Set<ReactiveListener<T>>();
 
-  return function rv(newValue) {
+  const rv: ReactiveVar<T> = function (newValue) {
     if (arguments.length > 0) {
       if (value !== newValue) {
         value = newValue!;
         varDep.dirty(rv);
-        // Trigger broadcast for any caches that were previously involved
-        // in reading this variable.
-        caches.forEach(broadcast);
+        listeners.forEach(listener => {
+          // Listener functions listen only for the next update, not all
+          // future updates.
+          listeners.delete(listener);
+          listener(value);
+        });
       }
     } else {
       // When reading from the variable, obtain the current cache from
       // context via cacheSlot. This isn't entirely foolproof, but it's
       // the same system that powers varDep.
       const cache = cacheSlot.getValue();
-      if (cache) caches.add(cache);
+      if (cache && (cache as any).broadcastWatches) {
+        listeners.add(() => broadcast(cache));
+      }
       varDep(rv);
     }
 
     return value;
   };
+
+  rv.onNextChange = listener => {
+    listeners.add(listener);
+    return () => {
+      listeners.delete(listener);
+    };
+  };
+
+  return rv;
 }
 
 type Broadcastable = ApolloCache<any> & {
   // This method is protected in InMemoryCache, which we are ignoring, but
   // we still want some semblance of type safety when we call it.
-  broadcastWatches: InMemoryCache["broadcastWatches"];
+  broadcastWatches?: InMemoryCache["broadcastWatches"];
 };
 
 function broadcast(cache: Broadcastable) {