implement observeQuery on memory cache

packages/@orbit/data/src/operation.ts

@@ -455,3 +455,55 @@ export function recordDiffs(
 
   return diffs;
 }
+
+export interface RecordChange extends RecordIdentity {
+  keys: string[];
+  attributes: string[];
+  relationships: string[];
+  remove: boolean;
+}
+
+export function recordOperationChange(
+  operation: RecordOperation
+): RecordChange {
+  const record = operation.record as Record;
+  const change: RecordChange = {
+    ...cloneRecordIdentity(record),
+    remove: false,
+    keys: [],
+    attributes: [],
+    relationships: []
+  };
+
+  switch (operation.op) {
+    case 'addRecord':
+    case 'updateRecord':
+      if (record.keys) {
+        change.keys = Object.keys(record.keys);
+      }
+      if (record.attributes) {
+        change.attributes = Object.keys(record.attributes);
+      }
+      if (record.relationships) {
+        change.relationships = Object.keys(record.relationships);
+      }
+      break;
+    case 'replaceAttribute':
+      change.attributes = [operation.attribute];
+      break;
+    case 'replaceKey':
+      change.keys = [operation.key];
+      break;
+    case 'replaceRelatedRecord':
+    case 'replaceRelatedRecords':
+    case 'addToRelatedRecords':
+    case 'removeFromRelatedRecords':
+      change.relationships = [operation.relationship];
+      break;
+    case 'removeRecord':
+      change.remove = true;
+      break;
+  }
+
+  return change;
+}

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

@@ -1,14 +1,22 @@
 /* eslint-disable valid-jsdoc */
 import Orbit from '@orbit/core';
 import { clone, Dict } from '@orbit/utils';
-import { Record, RecordIdentity, equalRecordIdentities } from '@orbit/data';
+import {
+  Record,
+  RecordIdentity,
+  equalRecordIdentities,
+  QueryOrExpressions,
+  buildQuery
+} from '@orbit/data';
 import {
   RecordRelationshipIdentity,
   SyncRecordCache,
   SyncRecordCacheSettings
 } from '@orbit/record-cache';
 import { ImmutableMap } from '@orbit/immutable';
 
+import QueryObserver from './query-observer';
+
 const { deprecate } = Orbit;
 
 export interface MemoryCacheSettings extends SyncRecordCacheSettings {
@@ -157,6 +165,24 @@ export default class MemoryCache extends SyncRecordCache {
     });
   }
 
+  observeQuery(
+    queryOrExpressions: QueryOrExpressions,
+    options?: object,
+    id?: string
+  ): QueryObserver {
+    const query = buildQuery(
+      queryOrExpressions,
+      options,
+      id,
+      this.queryBuilder
+    );
+
+    return new QueryObserver({
+      cache: this,
+      query
+    });
+  }
+
   /**
    * Resets the cache's state to be either empty or to match the state of
    * another cache.

packages/@orbit/memory/src/query-observer.ts

@@ -0,0 +1,172 @@
+import { Listener } from '@orbit/core';
+import {
+  RecordChange,
+  QueryExpression,
+  FindRecord,
+  FindRecords,
+  FindRelatedRecord,
+  FindRelatedRecords,
+  equalRecordIdentities,
+  Query,
+  RecordNotFoundException
+} from '@orbit/data';
+import {
+  QueryResult,
+  QueryResultData,
+  SyncRecordCache
+} from '@orbit/record-cache';
+
+export interface QueryObserverSettings {
+  cache: SyncRecordCache;
+  query: Query;
+}
+
+export default class QueryObserver {
+  cache: SyncRecordCache;
+  query: Query;
+
+  constructor(settings: QueryObserverSettings) {
+    this.cache = settings.cache;
+    this.query = settings.query;
+  }
+
+  match(change: RecordChange): boolean {
+    return !!this.query.expressions.find(expression =>
+      this._queryExpressionMatchChange(expression, change)
+    );
+  }
+
+  execute(): QueryResult {
+    try {
+      return this.cache.query(this.query);
+    } catch (error) {
+      if (error instanceof RecordNotFoundException) {
+        if (this.query.expressions.length === 1) {
+          return this._emptyQueryResult(this.query.expressions[0]);
+        } else {
+          return this.query.expressions.map(expression =>
+            this._emptyQueryResult(expression)
+          );
+        }
+      }
+      throw error;
+    }
+  }
+
+  on(listener: Listener): () => void {
+    const unsubscribeChange = this.cache.on(
+      'change',
+      (change: RecordChange) => {
+        if (this.match(change)) {
+          listener(this.execute());
+        }
+      }
+    );
+
+    const unsubscribeReset = this.cache.on('reset', () => {
+      listener(this.execute());
+    });
+
+    listener(this.execute());
+
+    return function unsubscribe() {
+      unsubscribeChange();
+      unsubscribeReset();
+    };
+  }
+
+  protected _emptyQueryResult(expression: QueryExpression): QueryResultData {
+    switch (expression.op) {
+      case 'findRecords':
+      case 'findRelatedRecords':
+        return [];
+      default:
+        return null;
+    }
+  }
+
+  protected _queryExpressionMatchChange(
+    expression: QueryExpression,
+    change: RecordChange
+  ): boolean {
+    switch (expression.op) {
+      case 'findRecord':
+        return this._findRecordQueryExpressionMatchChange(
+          expression as FindRecord,
+          change
+        );
+      case 'findRecords':
+        return this._findRecordsQueryExpressionMatchChange(
+          expression as FindRecords,
+          change
+        );
+      case 'findRelatedRecord':
+        return this._findRelatedRecordQueryExpressionMatchChange(
+          expression as FindRelatedRecord,
+          change
+        );
+      case 'findRelatedRecords':
+        return this._findRelatedRecordsQueryExpressionMatchChange(
+          expression as FindRelatedRecords,
+          change
+        );
+      default:
+        return true;
+    }
+  }
+
+  protected _findRecordQueryExpressionMatchChange(
+    expression: FindRecord,
+    change: RecordChange
+  ): boolean {
+    return equalRecordIdentities(expression.record, change);
+  }
+
+  protected _findRecordsQueryExpressionMatchChange(
+    expression: FindRecords,
+    change: RecordChange
+  ): boolean {
+    if (expression.type) {
+      return expression.type === change.type;
+    } else if (expression.records) {
+      for (let record of expression.records) {
+        if (record.type === change.type) {
+          return true;
+        }
+      }
+      return false;
+    }
+    return true;
+  }
+
+  protected _findRelatedRecordQueryExpressionMatchChange(
+    expression: FindRelatedRecord,
+    change: RecordChange
+  ): boolean {
+    return (
+      equalRecordIdentities(expression.record, change) &&
+      (change.relationships.includes(expression.relationship) || change.remove)
+    );
+  }
+
+  protected _findRelatedRecordsQueryExpressionMatchChange(
+    expression: FindRelatedRecords,
+    change: RecordChange
+  ): boolean {
+    const { type } = this.cache.schema.getRelationship(
+      expression.record.type,
+      expression.relationship
+    );
+
+    if (Array.isArray(type) && type.find(type => type === change.type)) {
+      return true;
+    } else if (type === change.type) {
+      return true;
+    }
+
+    return (
+      equalRecordIdentities(expression.record, change) &&
+      (change.relationships.includes(expression.relationship) || change.remove)
+    );
+  }
+}