type1 compiles

Author: wu-hui

packages/firestore/src/api/pipeline.ts

@@ -11,12 +11,14 @@ import { AbstractUserDataWriter } from '../lite-api/user_data_writer';
 import { DocumentKey } from '../model/document_key';
 
 import { ensureFirestoreConfigured, Firestore } from './database';
-import { DocumentSnapshot, PipelineSnapshot } from './snapshot';
+import { DocumentSnapshot, PipelineSnapshot, QuerySnapshot } 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 { CompleteFn, ErrorFn, NextFn } from './observer';
+import { ViewSnapshot } from '../core/view_snapshot';
 
 export class Pipeline<
   AppModelType = DocumentData
@@ -140,6 +142,13 @@ export class Pipeline<
     this.stages.push(new Sort([Field.of('__name__').ascending()]));
 
     const client = ensureFirestoreConfigured(this.db);
+    const observer = {
+      next: (snapshot: ViewSnapshot) => {
+        new PipelineSnapshot(this, snapshot);
+      },
+      error: error,
+      complete: complete
+    };
     // TODO(pipeline) hook up options
     firestoreClientListen(client, this, {}, observer);
 

packages/firestore/src/api/snapshot.ts

@@ -41,7 +41,9 @@ import { Code, FirestoreError } from '../util/error';
 import { Firestore } from './database';
 import { SnapshotListenOptions } from './reference_impl';
 import { Pipeline } from './pipeline';
-import { PipelineResult } from '../lite-api/pipeline-result';
+import { PipelineResult, toPipelineResult } from '../lite-api/pipeline-result';
+import { isPipeline } from '../core/pipeline-util';
+import { newPipelineComparator } from '../core/pipeline_run';
 
 /**
  * Converter used by `withConverter()` to transform user objects of type
@@ -673,12 +675,11 @@ export function changesFromSnapshot<
         change.type === ChangeType.Added,
         'Invalid event type for first snapshot'
       );
+      const comparator = isPipeline(querySnapshot._snapshot.query)
+        ? newPipelineComparator(querySnapshot._snapshot.query)
+        : newQueryComparator(querySnapshot.query._query);
       debugAssert(
-        !lastDoc ||
-          newQueryComparator(querySnapshot._snapshot.query)(
-            lastDoc,
-            change.doc
-          ) < 0,
+        !lastDoc || comparator(lastDoc, change.doc) < 0,
         'Got added events in wrong order'
       );
       const doc = new QueryDocumentSnapshot<AppModelType, DbModelType>(
@@ -800,16 +801,30 @@ export class PipelineSnapshot<AppModelType = DocumentData> {
    */
   readonly pipeline: Pipeline<AppModelType>;
 
+  /**
+   * Metadata about this snapshot, concerning its source and if it has local
+   * modifications.
+   */
+  readonly metadata: SnapshotMetadata;
+
   /** @hideconstructor */
   constructor(
     pipeline: Pipeline<AppModelType>,
-    readonly _snapshot: PipelineResult<AppModelType>[]
+    readonly _snapshot: ViewSnapshot
   ) {
+    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>> {
-    return this._snapshot;
+    const result: Array<PipelineResult<AppModelType>> = [];
+    this._snapshot.docs.forEach(doc =>
+      result.push(toPipelineResult(doc, this.pipeline))
+    );
+    return result;
   }
 }

packages/firestore/src/core/event_manager.ts

@@ -21,13 +21,18 @@ import { Code, FirestoreError } from '../util/error';
 import { EventHandler } from '../util/misc';
 import { ObjectMap } from '../util/obj_map';
 
-import { canonifyQuery, Query, queryEquals, stringifyQuery } from './query';
+import { Query, stringifyQuery } from './query';
 import { OnlineState } from './types';
 import { ChangeType, DocumentViewChange, ViewSnapshot } from './view_snapshot';
 import { Pipeline } from '../api/pipeline';
 import { PipelineSnapshot } from '../api/snapshot';
-import { PipelineResultView } from './sync_engine_impl';
-import { canonifyPipeline, pipelineEq } from './pipeline-util';
+import {
+  canonifyPipeline,
+  canonifyQueryOrPipeline,
+  isPipeline,
+  QueryOrPipeline,
+  queryOrPipelineEqual
+} from './pipeline-util';
 
 /**
  * Holds the listeners and the last received ViewSnapshot for a query being
@@ -51,12 +56,6 @@ export interface Observer<T> {
   error: EventHandler<FirestoreError>;
 }
 
-export type QueryOrPipeline = Query | Pipeline;
-
-export function isPipeline(q: QueryOrPipeline): q is Pipeline {
-  return q instanceof Pipeline;
-}
-
 /**
  * EventManager is responsible for mapping queries to query event emitters.
  * It handles "fan-out". -- Identical queries will re-use the same watch on the
@@ -122,39 +121,6 @@ export class EventManagerImpl implements EventManager {
   }
 }
 
-export function stringifyQueryOrPipeline(q: QueryOrPipeline): string {
-  if (isPipeline(q)) {
-    return canonifyPipeline(q);
-  }
-
-  return stringifyQuery(q);
-}
-
-export function canonifyQueryOrPipeline(q: QueryOrPipeline): string {
-  if (isPipeline(q)) {
-    return canonifyPipeline(q);
-  }
-
-  return canonifyQuery(q);
-}
-
-export function queryOrPipelineEqual(
-  left: QueryOrPipeline,
-  right: QueryOrPipeline
-): boolean {
-  if (left instanceof Pipeline && right instanceof Pipeline) {
-    return pipelineEq(left, right);
-  }
-  if (
-    (left instanceof Pipeline && !(right instanceof Pipeline)) ||
-    (!(left instanceof Pipeline) && right instanceof Pipeline)
-  ) {
-    return false;
-  }
-
-  return queryEquals(left as Query, right as Query);
-}
-
 function newQueriesObjectMap(): ObjectMap<QueryOrPipeline, QueryListenersInfo> {
   return new ObjectMap<QueryOrPipeline, QueryListenersInfo>(
     q => canonifyQueryOrPipeline(q),
@@ -621,22 +587,3 @@ export class QueryListener {
     return this.options.source !== ListenerDataSource.Cache;
   }
 }
-
-export class PipelineListener {
-  private view: PipelineResultView | null = null;
-
-  constructor(
-    readonly pipeline: Pipeline,
-    private queryObserver: Observer<PipelineSnapshot>
-  ) {}
-
-  onViewSnapshot(view: PipelineResultView): boolean {
-    this.view = view;
-    this.queryObserver.next(view.toPipelineSnapshot());
-    return true;
-  }
-
-  onError(error: FirestoreError): void {
-    this.queryObserver.error(error);
-  }
-}

packages/firestore/src/core/firestore_client.ts

@@ -84,14 +84,12 @@ import {
   ListenOptions,
   Observer,
   QueryListener,
-  QueryOrPipeline,
   removeSnapshotsInSyncListener
 } from './event_manager';
 import { newQueryForPath, Query } from './query';
 import { SyncEngine } from './sync_engine';
 import {
   syncEngineListen,
-  syncEngineListenPipeline,
   syncEngineLoadBundle,
   syncEngineRegisterPendingWritesCallback,
   syncEngineUnlisten,
@@ -106,6 +104,7 @@ import { View } from './view';
 import { ViewSnapshot } from './view_snapshot';
 import { Unsubscribe } from '../api/reference_impl';
 import { PipelineSnapshot } from '../api/snapshot';
+import { QueryOrPipeline } from './pipeline-util';
 
 const LOG_TAG = 'FirestoreClient';
 export const MAX_CONCURRENT_LIMBO_RESOLUTIONS = 100;

packages/firestore/src/core/pipeline-util.ts

@@ -37,7 +37,7 @@ import {
   Timestamp as ProtoTimestamp,
   Value as ProtoValue
 } from '../protos/firestore_proto_api';
-import { fail } from '../util/assert';
+import { debugAssert, fail } from '../util/assert';
 import { isPlainObject } from '../util/input_validation';
 
 import {
@@ -47,7 +47,6 @@ import {
   Filter as FilterInternal,
   Operator
 } from './filter';
-import { Pipeline } from '../lite-api/pipeline';
 import {
   AddFields,
   Aggregate,
@@ -64,6 +63,15 @@ import {
   Stage,
   Where
 } from '../lite-api/stage';
+import { Pipeline } from '../api/pipeline';
+import { Pipeline as LitePipeline } from '../lite-api/pipeline';
+import { canonifyQuery, Query, queryEquals, stringifyQuery } from './query';
+import {
+  canonifyTarget,
+  Target,
+  targetEquals,
+  targetIsPipelineTarget
+} from './target';
 
 /* eslint @typescript-eslint/no-explicit-any: 0 */
 
@@ -357,12 +365,17 @@ function canonifyExprMap(map: Map<string, Expr>): string {
     .join(',')}`;
 }
 
-export function canonifyPipeline(p: Pipeline): string {
+export function canonifyPipeline(p: LitePipeline): string;
+export function canonifyPipeline(p: Pipeline): string;
+export function canonifyPipeline(p: Pipeline | LitePipeline): string {
   return p.stages.map(s => canonifyStage(s)).join('|');
 }
 
 // TODO(pipeline): do a proper implementation for eq.
-export function pipelineEq(left: Pipeline, right: Pipeline): boolean {
+export function pipelineEq(
+  left: Pipeline | LitePipeline,
+  right: Pipeline | LitePipeline
+): boolean {
   return canonifyPipeline(left) === canonifyPipeline(right);
 }
 
@@ -389,3 +402,120 @@ export function getPipelineFlavor(p: Pipeline): PipelineFlavor {
 
   return flavor;
 }
+
+export type PipelineSourceType =
+  | 'collection'
+  | 'collection-group'
+  | 'database'
+  | 'documents';
+
+export function getPipelineSourceType(
+  p: Pipeline
+): PipelineSourceType | 'unknown' {
+  debugAssert(p.stages.length > 0, 'Pipeline must have at least one stage');
+  const source = p.stages[0];
+
+  if (
+    source.name === CollectionSource.name ||
+    source.name === CollectionGroupSource.name ||
+    source.name === DatabaseSource.name ||
+    source.name === DocumentsSource.name
+  ) {
+    return source.name as PipelineSourceType;
+  }
+
+  return 'unknown';
+}
+
+export function getPipelineCollection(p: Pipeline): string | undefined {
+  if (getPipelineSourceType(p) === 'collection') {
+    return (p.stages[0] as CollectionSource).collectionPath;
+  }
+  return undefined;
+}
+
+export function getPipelineCollectionGroup(p: Pipeline): string | undefined {
+  if (getPipelineSourceType(p) === 'collection-group') {
+    return (p.stages[0] as CollectionGroupSource).collectionId;
+  }
+  return undefined;
+}
+
+export function getPipelineDocuments(p: Pipeline): string[] | undefined {
+  if (getPipelineSourceType(p) === 'documents') {
+    return (p.stages[0] as DocumentsSource).docPaths;
+  }
+  return undefined;
+}
+
+export type QueryOrPipeline = Query | Pipeline;
+
+export function isPipeline(q: QueryOrPipeline): q is Pipeline {
+  return q instanceof Pipeline;
+}
+
+export function stringifyQueryOrPipeline(q: QueryOrPipeline): string {
+  if (isPipeline(q)) {
+    return canonifyPipeline(q);
+  }
+
+  return stringifyQuery(q);
+}
+
+export function canonifyQueryOrPipeline(q: QueryOrPipeline): string {
+  if (isPipeline(q)) {
+    return canonifyPipeline(q);
+  }
+
+  return canonifyQuery(q);
+}
+
+export function queryOrPipelineEqual(
+  left: QueryOrPipeline,
+  right: QueryOrPipeline
+): boolean {
+  if (left instanceof Pipeline && right instanceof Pipeline) {
+    return pipelineEq(left, right);
+  }
+  if (
+    (left instanceof Pipeline && !(right instanceof Pipeline)) ||
+    (!(left instanceof Pipeline) && right instanceof Pipeline)
+  ) {
+    return false;
+  }
+
+  return queryEquals(left as Query, right as Query);
+}
+
+export type TargetOrPipeline = Target | Pipeline;
+
+export function canonifyTargetOrPipeline(q: TargetOrPipeline): string {
+  if (targetIsPipelineTarget(q)) {
+    return canonifyPipeline(q);
+  }
+
+  return canonifyTarget(q as Target);
+}
+
+export function targetOrPipelineEqual(
+  left: TargetOrPipeline,
+  right: TargetOrPipeline
+): boolean {
+  if (left instanceof Pipeline && right instanceof Pipeline) {
+    return pipelineEq(left, right);
+  }
+  if (
+    (left instanceof Pipeline && !(right instanceof Pipeline)) ||
+    (!(left instanceof Pipeline) && right instanceof Pipeline)
+  ) {
+    return false;
+  }
+
+  return targetEquals(left as Target, right as Target);
+}
+
+export function pipelineHasRanges(pipeline: Pipeline): boolean {
+  return pipeline.stages.some(
+    stage => stage.name === Limit.name || stage.name === Offset.name
+  );
+}