watch integration works.

Author: wu-hui

packages/firestore/src/api/pipeline.ts

@@ -5,7 +5,7 @@ import {
 import { Pipeline as LitePipeline } from '../lite-api/pipeline';
 import { PipelineResult } from '../lite-api/pipeline-result';
 import { DocumentData, DocumentReference } from '../lite-api/reference';
-import {AddFields, Sort, Stage, Where} from '../lite-api/stage';
+import { AddFields, Sort, Stage, Where } from '../lite-api/stage';
 import { UserDataReader } from '../lite-api/user_data_reader';
 import { AbstractUserDataWriter } from '../lite-api/user_data_writer';
 import { DocumentKey } from '../model/document_key';
@@ -15,8 +15,8 @@ import { DocumentSnapshot, PipelineSnapshot } from './snapshot';
 import { FirestoreError } from '../util/error';
 import { Unsubscribe } from './reference_impl';
 import { cast } from '../util/input_validation';
-import {Field, FilterCondition} from '../api';
-import {Expr} from '../lite-api/expressions';
+import { Field, FilterCondition } from '../api';
+import { Expr } from '../lite-api/expressions';
 
 export class Pipeline<
   AppModelType = DocumentData
@@ -32,7 +32,7 @@ export class Pipeline<
    * @param converter
    */
   constructor(
-    private db: Firestore,
+    readonly db: Firestore,
     userDataReader: UserDataReader,
     userDataWriter: AbstractUserDataWriter,
     documentReferenceFactory: (id: DocumentKey) => DocumentReference,
@@ -137,15 +137,10 @@ export class Pipeline<
     //   )
     // );
 
-    this.stages.push(
-      new Sort([
-        Field.of('__name__').ascending()
-        ]
-      )
-    );
+    this.stages.push(new Sort([Field.of('__name__').ascending()]));
 
     const client = ensureFirestoreConfigured(this.db);
-    firestoreClientListenPipeline(client, this, {next, error, complete});
+    firestoreClientListenPipeline(client, this, { next, error, complete });
 
     return () => {};
   }

packages/firestore/src/api/pipeline_source.ts

@@ -23,15 +23,15 @@ import {
   DatabaseSource,
   DocumentsSource
 } from '../lite-api/stage';
-import {PipelineSource as LitePipelineSource} from '../lite-api/pipeline-source';
+import { PipelineSource as LitePipelineSource } from '../lite-api/pipeline-source';
 import { UserDataReader } from '../lite-api/user_data_reader';
 import { AbstractUserDataWriter } from '../lite-api/user_data_writer';
 
 /**
  * Represents the source of a Firestore {@link Pipeline}.
  * @beta
  */
-export class PipelineSource extends LitePipelineSource{
+export class PipelineSource extends LitePipelineSource {
   /**
    * @internal
    * @private

packages/firestore/src/api/snapshot.ts

@@ -794,12 +794,6 @@ export function snapshotEqual<AppModelType, DbModelType extends DocumentData>(
 }
 
 export class PipelineSnapshot<AppModelType = DocumentData> {
-  /**
-   * Metadata about this snapshot, concerning its source and if it has local
-   * modifications.
-   */
-  readonly metadata: SnapshotMetadata;
-
   /**
    * The query on which you called `get` or `onSnapshot` in order to get this
    * `QuerySnapshot`.
@@ -808,21 +802,14 @@ export class PipelineSnapshot<AppModelType = DocumentData> {
 
   /** @hideconstructor */
   constructor(
-    readonly _firestore: Firestore,
-    readonly _userDataWriter: AbstractUserDataWriter,
     pipeline: Pipeline<AppModelType>,
-    readonly _snapshot: ViewSnapshot
+    readonly _snapshot: PipelineResult<AppModelType>[]
   ) {
-    this.metadata = new SnapshotMetadata(
-      _snapshot.hasPendingWrites,
-      _snapshot.fromCache
-    );
     this.pipeline = pipeline;
   }
 
   /** An array of all the documents in the `QuerySnapshot`. */
   get results(): Array<PipelineResult<AppModelType>> {
-    const result: Array<PipelineResult<AppModelType>> = [];
-    return result;
+    return this._snapshot;
   }
 }

packages/firestore/src/core/event_manager.ts

@@ -602,15 +602,16 @@ export class QueryListener {
 }
 
 export class PipelineListener {
-  private snap: PipelineResultView | null = null;
+  private view: PipelineResultView | null = null;
 
   constructor(
     readonly pipeline: Pipeline,
     private queryObserver: Observer<PipelineSnapshot>
   ) {}
 
-  onViewSnapshot(snap: PipelineResultView): boolean {
-    this.snap = snap;
+  onViewSnapshot(view: PipelineResultView): boolean {
+    this.view = view;
+    this.queryObserver.next(view.toPipelineSnapshot());
     return true;
   }
 

packages/firestore/src/core/sync_engine_impl.ts

@@ -121,6 +121,7 @@ import { ViewSnapshot } from './view_snapshot';
 import { Pipeline } from '../api/pipeline';
 import { PipelineSnapshot } from '../api/snapshot';
 import { PipelineResult } from '../lite-api/pipeline-result';
+import { doc } from '../lite-api/reference';
 
 const LOG_TAG = 'SyncEngine';
 
@@ -150,9 +151,12 @@ class QueryView {
 }
 
 export class PipelineResultView {
-  private keyToIndexMap: Map<DocumentKey, number>;
+  private keyToIndexMap: ObjectMap<DocumentKey, number>;
   constructor(public pipeline: Pipeline, public view: Array<MutableDocument>) {
-    this.keyToIndexMap = new Map<DocumentKey, number>();
+    this.keyToIndexMap = new ObjectMap<DocumentKey, number>(
+      key => key.toString(),
+      (a, b) => a.isEqual(b)
+    );
     this.buildKeyToIndexMap();
   }
 
@@ -197,6 +201,23 @@ export class PipelineResultView {
     }
     this.view[index] = doc;
   }
+
+  toPipelineSnapshot(): PipelineSnapshot {
+    return new PipelineSnapshot(
+      this.pipeline,
+      this.view.map(
+        d =>
+          new PipelineResult(
+            this.pipeline.userDataWriter,
+            doc(this.pipeline.db, d.key.toString()),
+            d.data,
+            d.readTime.toTimestamp(),
+            d.createTime.toTimestamp(),
+            d.version.toTimestamp()
+          )
+      )
+    );
+  }
 }
 
 /** Tracks a limbo resolution. */
@@ -1000,7 +1021,10 @@ function removeAndCleanupTarget(
   syncEngineImpl.sharedClientState.removeLocalQueryTarget(targetId);
 
   // TODO(pipeline): REMOVE this hack.
-  if(!syncEngineImpl.queriesByTarget.has(targetId)||syncEngineImpl.queriesByTarget.get(targetId)!.length !== 0){
+  if (
+    !syncEngineImpl.queriesByTarget.has(targetId) ||
+    syncEngineImpl.queriesByTarget.get(targetId)!.length !== 0
+  ) {
     return;
   }
 
@@ -1174,7 +1198,10 @@ export async function syncEngineEmitNewSnapsAndNotifyLocalStore(
     const change = remoteEvent?.targetChanges.get(targetId);
     if (!!change) {
       change.modifiedDocuments.forEach(key => {
-        results.updateResult(key, remoteEvent?.augmentedDocumentUpdates.get(key)!);
+        results.updateResult(
+          key,
+          remoteEvent?.augmentedDocumentUpdates.get(key)!
+        );
       });
       change.addedDocuments.forEach(key => {
         results.addResult(key, remoteEvent?.augmentedDocumentUpdates.get(key)!);
@@ -1319,17 +1346,21 @@ export function syncEngineGetRemoteKeysForTarget(
   } else {
     let keySet = documentKeySet();
     const queries = syncEngineImpl.queriesByTarget.get(targetId);
-    if (!queries) {
+    const pipelineView = syncEngineImpl.pipelineViewByTarget.get(targetId);
+    if (!queries && !pipelineView) {
       return keySet;
     }
-    for (const query of queries) {
+    for (const query of queries ?? []) {
       const queryView = syncEngineImpl.queryViewsByQuery.get(query);
       debugAssert(
         !!queryView,
         `No query view found for ${stringifyQuery(query)}`
       );
       keySet = keySet.unionWith(queryView.view.syncedDocuments);
     }
+    for (const doc of pipelineView?.view ?? []) {
+      keySet = keySet.add(doc.key);
+    }
     return keySet;
   }
 }

packages/firestore/src/lite-api/expressions.ts

@@ -2336,7 +2336,7 @@ export class Mod extends FirestoreFunction {
  */
 export class Eq extends FirestoreFunction implements FilterCondition {
   constructor(private left: Expr, private right: Expr) {
-    super('eq', [left, right]);
+    super('equals', [left, right]);
   }
   filterable = true as const;
 }

packages/firestore/src/lite-api/pipeline.ts

@@ -24,7 +24,8 @@ import {
 import { invokeExecutePipeline } from '../remote/datastore';
 import {
   getEncodedDatabaseId,
-  JsonProtoSerializer, ProtoSerializable
+  JsonProtoSerializer,
+  ProtoSerializable
 } from '../remote/serializer';
 
 import { getDatastore } from './components';
@@ -117,7 +118,9 @@ function isReadableUserData(value: any): value is ReadableUserData {
 /**
  * Base-class implementation
  */
-export class Pipeline<AppModelType = DocumentData> implements ProtoSerializable<ExecutePipelineRequest>{
+export class Pipeline<AppModelType = DocumentData>
+  implements ProtoSerializable<ExecutePipelineRequest>
+{
   /**
    * @internal
    * @private
@@ -135,7 +138,7 @@ export class Pipeline<AppModelType = DocumentData> implements ProtoSerializable<
      * @internal
      * @private
      */
-    protected userDataWriter: AbstractUserDataWriter,
+    readonly userDataWriter: AbstractUserDataWriter,
     /**
      * @internal
      * @private

packages/firestore/src/lite-api/stage.ts

@@ -19,7 +19,8 @@ import {
 } from '../protos/firestore_proto_api';
 import { toNumber } from '../remote/number_serializer';
 import {
-  JsonProtoSerializer, ProtoSerializable,
+  JsonProtoSerializer,
+  ProtoSerializable,
   toMapValue,
   toStringValue
 } from '../remote/serializer';
@@ -37,7 +38,7 @@ import { VectorValue } from './vector_value';
 /**
  * @beta
  */
-export interface Stage extends ProtoSerializable<ProtoStage>{
+export interface Stage extends ProtoSerializable<ProtoStage> {
   name: string;
 }
 

packages/firestore/src/local/local_store_impl.ts

@@ -1053,7 +1053,7 @@ export async function localStoreReleaseTarget(
 
   // TODO(pipeline): this is a hack that only works because pipelines are the only ones returning nulls here.
   // REMOVE ASAP.
-  if(targetData === null) {
+  if (targetData === null) {
     return;
   }
 

packages/firestore/src/protos/protos.json

@@ -2363,7 +2363,7 @@
                       "type": "DocumentsTarget",
                       "id": 3
                     },
-                    "pipeline_query": {
+                    "pipelineQuery": {
                       "type": "PipelineQueryTarget",
                       "id": 13
                     },