implement observeQuery on memory cache

packages/@orbit/memory/test/memory-cache-test.ts

@@ -2010,4 +2010,193 @@ module('MemoryCache', function(hooks) {
       jupiter
     );
   });
+
+  test('#observeQuery', async function(assert) {
+    let cache = new MemoryCache({ schema, keyMap });
+
+    const jupiter: Record = {
+      id: 'jupiter',
+      type: 'planet',
+      attributes: { name: 'Jupiter' }
+    };
+
+    const jupiter2 = {
+      ...jupiter,
+      attributes: { name: 'Jupiter 2' }
+    };
+
+    const callisto: Record = {
+      id: 'callisto',
+      type: 'moon',
+      attributes: { name: 'Callisto' },
+      relationships: { planet: { data: { type: 'planet', id: 'jupiter' } } }
+    };
+
+    const jupiterWithCallisto = {
+      ...jupiter2,
+      relationships: { moons: { data: [{ type: 'moon', id: 'callisto' }] } }
+    };
+
+    const livePlanet = cache.observeQuery(q =>
+      q.findRecord({ type: 'planet', id: 'jupiter' })
+    );
+    const livePlanets = cache.observeQuery(q => q.findRecords('planet'));
+    const livePlanetMoons = cache.observeQuery(q =>
+      q.findRelatedRecords(jupiter, 'moons')
+    );
+    const liveMoonPlanet = cache.observeQuery(q =>
+      q.findRelatedRecord(callisto, 'planet')
+    );
+
+    interface Deferred {
+      promise?: Promise<any>;
+      resolve?: () => void;
+      reject?: () => void;
+    }
+    function defer() {
+      let defer: Deferred = {};
+      defer.promise = new Promise((resolve, reject) => {
+        defer.resolve = resolve;
+        defer.reject = reject;
+      });
+      return defer;
+    }
+
+    let jupiterAdded = defer();
+    let jupiterUpdated = defer();
+    let callistoAdded = defer();
+    let jupiterRemoved = defer();
+
+    function next() {
+      if (n === 1 && i === 1) {
+        jupiterAdded.resolve();
+      }
+      if (n === 2 && i === 2) {
+        jupiterUpdated.resolve();
+      }
+      if (n === 3 && i === 3 && j === 2 && k === 1) {
+        callistoAdded.resolve();
+      }
+      if (n === 5 && i === 5 && j === 5 && k === 2) {
+        jupiterRemoved.resolve();
+      }
+    }
+
+    let n = 0;
+    let offLivePlanet = livePlanet.on(result => {
+      if (n === 0) {
+        assert.deepEqual(result, null);
+      } else if (n === 1) {
+        assert.deepEqual(result, jupiter);
+      } else if (n === 2) {
+        assert.deepEqual(result, jupiter2);
+      } else if (n === 3) {
+        assert.deepEqual(result, jupiterWithCallisto);
+      } else if (n === 4) {
+        assert.deepEqual(result, null);
+      } else {
+        assert.ok(false, 'findRecord should not execute');
+      }
+      next();
+      n++;
+    });
+
+    let i = 0;
+    let offLivePlanets = livePlanets.on(result => {
+      if (i === 0) {
+        assert.deepEqual(result, []);
+      } else if (i === 1) {
+        assert.deepEqual(result, [jupiter]);
+      } else if (i === 2) {
+        assert.deepEqual(result, [jupiter2]);
+      } else if (i === 3) {
+        assert.deepEqual(result, [jupiterWithCallisto]);
+      } else if (i === 4) {
+        assert.deepEqual(result, []);
+      } else {
+        assert.ok(false, 'findRecords should not execute');
+      }
+      next();
+      i++;
+    });
+
+    let j = 0;
+    let offLivePlanetMoons = livePlanetMoons.on(result => {
+      if (j === 0 || j === 1) {
+        assert.deepEqual(result, [], 'findRelatedRecords jupiter.moons => []');
+      } else if (j === 2) {
+        assert.deepEqual(
+          result,
+          [callisto],
+          'findRelatedRecords jupiter.moons => [callisto]'
+        );
+      } else if (j === 3 || j === 4) {
+        assert.deepEqual(result, [], 'findRelatedRecords jupiter.moons => []');
+      } else {
+        assert.ok(false, 'findRelatedRecords should not execute');
+      }
+      next();
+      j++;
+    });
+
+    let k = 0;
+    let offLiveMoonPlanet = liveMoonPlanet.on(result => {
+      if (k === 0) {
+        assert.deepEqual(
+          result,
+          null,
+          'findRelatedRecord callisto.planet => null'
+        );
+      } else if (k === 1) {
+        assert.deepEqual(
+          result,
+          jupiter2,
+          'findRelatedRecord callisto.planet => jupiter'
+        );
+      } else if (k === 2) {
+        assert.deepEqual(
+          result,
+          null,
+          'findRelatedRecord callisto.planet => null'
+        );
+      } else {
+        assert.ok(false, 'findRelatedRecord should not execute');
+      }
+      next();
+      k++;
+    });
+
+    cache.patch(t => t.addRecord(jupiter));
+    await jupiterAdded.promise;
+
+    cache.patch(t => t.updateRecord(jupiter2));
+    await jupiterUpdated.promise;
+
+    cache.patch(t => t.addRecord(callisto));
+    await callistoAdded.promise;
+
+    cache.patch(t => t.removeRecord(jupiter));
+    await jupiterRemoved.promise;
+
+    assert.expect(22);
+    assert.equal(n, 5, 'findRecord should run 5 times');
+    assert.equal(i, 5, 'findRecords should run 5 times');
+    assert.equal(j, 5, 'findRelatedRecords should run 5 times');
+    assert.equal(k, 3, 'findRelatedRecord should run 3 times');
+
+    offLivePlanet();
+    offLivePlanets();
+    offLivePlanetMoons();
+    offLiveMoonPlanet();
+
+    cache.patch(t =>
+      t.addRecord({
+        type: 'planet',
+        id: 'mercury',
+        attributes: {
+          name: 'Mercury'
+        }
+      })
+    );
+  });
 });

packages/@orbit/record-cache/src/async-record-cache.ts

@@ -38,6 +38,7 @@ import {
 } from './record-accessor';
 import { PatchResult } from './patch-result';
 import { QueryResult, QueryResultData } from './query-result';
+import AsyncQueryObserver from './query-observer/async-query-observer';
 
 const { assert } = Orbit;
 
@@ -246,6 +247,24 @@ export abstract class AsyncRecordCache implements Evented, AsyncRecordAccessor {
     return result;
   }
 
+  observeQuery(
+    queryOrExpressions: QueryOrExpressions,
+    options?: object,
+    id?: string
+  ): AsyncQueryObserver {
+    const query = buildQuery(
+      queryOrExpressions,
+      options,
+      id,
+      this.queryBuilder
+    );
+
+    return new AsyncQueryObserver({
+      cache: this,
+      query
+    });
+  }
+
   /////////////////////////////////////////////////////////////////////////////
   // Protected methods
   /////////////////////////////////////////////////////////////////////////////

packages/@orbit/record-cache/src/query-observer/async-query-observer.ts

@@ -0,0 +1,28 @@
+import { AsyncRecordCache } from '../async-record-cache';
+import { QueryResult } from '../query-result';
+import QueryObserver, { QueryObserverSettings } from './query-observer';
+
+export interface AsyncQueryObserverSettings extends QueryObserverSettings {
+  cache: AsyncRecordCache;
+}
+
+export default class AsyncQueryObserver extends QueryObserver {
+  cache: AsyncRecordCache;
+
+  constructor(settings: AsyncQueryObserverSettings) {
+    super(settings);
+    this.cache = settings.cache;
+  }
+
+  get schema() {
+    return this.cache.schema;
+  }
+
+  async execute(): Promise<QueryResult> {
+    try {
+      return this.cache.query(this.query);
+    } catch (error) {
+      return this._handleQueryError(error);
+    }
+  }
+}