snapshot listeners source from cache

common/api-review/firestore.api.md

@@ -350,6 +350,12 @@ export function limit(limit: number): QueryLimitConstraint;
 // @public
 export function limitToLast(limit: number): QueryLimitConstraint;
 
+// @public
+export const enum ListenSource {
+    Cache = 1,
+    Default = 0
+}
+
 // @public
 export function loadBundle(firestore: Firestore, bundleData: ReadableStream<Uint8Array> | ArrayBuffer | string): LoadBundleTask;
 
@@ -651,6 +657,7 @@ export function snapshotEqual<AppModelType, DbModelType extends DocumentData>(le
 // @public
 export interface SnapshotListenOptions {
     readonly includeMetadataChanges?: boolean;
+    readonly source?: ListenSource;
 }
 
 // @public

docs-devsite/firestore_.md

@@ -157,6 +157,12 @@ https://github.com/firebase/firebase-js-sdk
 |  [Transaction](./firestore_.transaction.md#transaction_class) | A reference to a transaction.<!-- -->The <code>Transaction</code> object passed to a transaction's <code>updateFunction</code> provides the methods to read and write data within the transaction context. See [runTransaction()](./firestore_.md#runtransaction_6f03ec4)<!-- -->. |
 |  [WriteBatch](./firestore_.writebatch.md#writebatch_class) | A write batch, used to perform multiple writes as a single atomic unit.<!-- -->A <code>WriteBatch</code> object can be acquired by calling [writeBatch()](./firestore_.md#writebatch_231a8e0)<!-- -->. It provides methods for adding writes to the write batch. None of the writes will be committed (or visible locally) until [WriteBatch.commit()](./firestore_.writebatch.md#writebatchcommit) is called. |
 
+## Enumerations
+
+|  Enumeration | Description |
+|  --- | --- |
+|  [ListenSource](./firestore_.md#listensource) | Describe the source a query listens to. |
+
 ## Interfaces
 
 |  Interface | Description |
@@ -2716,3 +2722,21 @@ export declare type WithFieldValue<T> = T | (T extends Primitive ? T : T extends
     [K in keyof T]: WithFieldValue<T[K]> | FieldValue;
 } : never);
 ```
+
+## ListenSource
+
+Describe the source a query listens to.
+
+<b>Signature:</b>
+
+```typescript
+export declare const enum ListenSource 
+```
+
+## Enumeration Members
+
+|  Member | Value | Description |
+|  --- | --- | --- |
+|  Cache | <code>1</code> | Listen to changes in cache only |
+|  Default | <code>0</code> | Listen to both cache and server changes |
+

docs-devsite/firestore_.snapshotlistenoptions.md

@@ -23,6 +23,7 @@ export declare interface SnapshotListenOptions
 |  Property | Type | Description |
 |  --- | --- | --- |
 |  [includeMetadataChanges](./firestore_.snapshotlistenoptions.md#snapshotlistenoptionsincludemetadatachanges) | boolean | Include a change even if only the metadata of the query or of a document changed. Default is false. |
+|  [source](./firestore_.snapshotlistenoptions.md#snapshotlistenoptionssource) | [ListenSource](./firestore_.md#listensource) | Set the source the query listens to. Default to ListenSource.Default, which listens to both cache and server. |
 
 ## SnapshotListenOptions.includeMetadataChanges
 
@@ -33,3 +34,13 @@ Include a change even if only the metadata of the query or of a document changed
 ```typescript
 readonly includeMetadataChanges?: boolean;
 ```
+
+## SnapshotListenOptions.source
+
+Set the source the query listens to. Default to ListenSource.Default, which listens to both cache and server.
+
+<b>Signature:</b>
+
+```typescript
+readonly source?: ListenSource;
+```

packages/firestore/src/api.ts

@@ -139,7 +139,11 @@ export {
   WhereFilterOp
 } from './api/filter';
 
-export { SnapshotListenOptions, Unsubscribe } from './api/reference_impl';
+export {
+  ListenSource,
+  SnapshotListenOptions,
+  Unsubscribe
+} from './api/reference_impl';
 
 export { TransactionOptions } from './api/transaction_options';
 

packages/firestore/src/api/reference_impl.ts

@@ -78,6 +78,21 @@ export interface SnapshotListenOptions {
    * changed. Default is false.
    */
   readonly includeMetadataChanges?: boolean;
+
+  /**
+   * Set the source the query listens to. Default to ListenSource.Default, which
+   * listens to both cache and server.
+   */
+  readonly source?: ListenSource;
+}
+
+/** Describe the source a query listens to. */
+export const enum ListenSource {
+  /** Listen to both cache and server changes */
+  Default,
+
+  /** Listen to changes in cache only */
+  Cache
 }
 
 /**
@@ -668,7 +683,8 @@ export function onSnapshot<AppModelType, DbModelType extends DocumentData>(
   reference = getModularInstance(reference);
 
   let options: SnapshotListenOptions = {
-    includeMetadataChanges: false
+    includeMetadataChanges: false,
+    source: ListenSource.Default
   };
   let currArg = 0;
   if (typeof args[currArg] === 'object' && !isPartialObserver(args[currArg])) {
@@ -677,7 +693,8 @@ export function onSnapshot<AppModelType, DbModelType extends DocumentData>(
   }
 
   const internalOptions = {
-    includeMetadataChanges: options.includeMetadataChanges
+    includeMetadataChanges: options.includeMetadataChanges,
+    source: options.source
   };
 
   if (isPartialObserver(args[currArg])) {

packages/firestore/src/core/event_manager.ts

@@ -15,6 +15,7 @@
  * limitations under the License.
  */
 
+import { ListenSource } from '../api/reference_impl';
 import { debugAssert, debugCast } from '../util/assert';
 import { wrapInUserErrorIfRecoverable } from '../util/async_queue';
 import { FirestoreError } from '../util/error';
@@ -32,6 +33,11 @@ import { ChangeType, DocumentViewChange, ViewSnapshot } from './view_snapshot';
 class QueryListenersInfo {
   viewSnap: ViewSnapshot | undefined = undefined;
   listeners: QueryListener[] = [];
+
+  // Helper methods to filter listeners that listens to watch changes.
+  getRemoteListeners(): QueryListener[] {
+    return this.listeners.filter(l => l.options.source !== ListenSource.Cache);
+  }
 }
 
 /**
@@ -52,8 +58,13 @@ export interface Observer<T> {
  * allows users to tree-shake the Watch logic.
  */
 export interface EventManager {
-  onListen?: (query: Query) => Promise<ViewSnapshot>;
-  onUnlisten?: (query: Query) => Promise<void>;
+  onListen?: (
+    query: Query,
+    enableRemoteListen: boolean
+  ) => Promise<ViewSnapshot>;
+  onUnlisten?: (query: Query, disableRemoteListen: boolean) => Promise<void>;
+  onRemoteStoreListen?: (query: Query) => Promise<void>;
+  onRemoteStoreUnlisten?: (query: Query) => Promise<void>;
 }
 
 export function newEventManager(): EventManager {
@@ -71,18 +82,49 @@ export class EventManagerImpl implements EventManager {
   snapshotsInSyncListeners: Set<Observer<void>> = new Set();
 
   /** Callback invoked when a Query is first listen to. */
-  onListen?: (query: Query) => Promise<ViewSnapshot>;
+  onListen?: (
+    query: Query,
+    enableRemoteListen: boolean
+  ) => Promise<ViewSnapshot>;
   /** Callback invoked once all listeners to a Query are removed. */
-  onUnlisten?: (query: Query) => Promise<void>;
+  onUnlisten?: (query: Query, disableRemoteListen: boolean) => Promise<void>;
+
+  /**
+   * Callback invoked when a Query starts listening to the remote store, while
+   * already listening to the cache.
+   */
+  onRemoteStoreListen?: (query: Query) => Promise<void>;
+  /**
+   * Callback invoked when a Query stops listening to the remote store, while
+   * still listening to the cache.
+   */
+  onRemoteStoreUnlisten?: (query: Query) => Promise<void>;
+}
+
+function listensToRemoteStore(listener: QueryListener): boolean {
+  return listener.options.source !== ListenSource.Cache;
+}
+
+function validateEventManager(eventManagerImpl: EventManagerImpl): void {
+  debugAssert(!!eventManagerImpl.onListen, 'onListen not set');
+  debugAssert(
+    !!eventManagerImpl.onRemoteStoreListen,
+    'onRemoteStoreListen not set'
+  );
+  debugAssert(!!eventManagerImpl.onUnlisten, 'onUnlisten not set');
+  debugAssert(
+    !!eventManagerImpl.onRemoteStoreUnlisten,
+    'onRemoteStoreUnlisten not set'
+  );
 }
 
 export async function eventManagerListen(
   eventManager: EventManager,
   listener: QueryListener
 ): Promise<void> {
   const eventManagerImpl = debugCast(eventManager, EventManagerImpl);
+  validateEventManager(eventManagerImpl);
 
-  debugAssert(!!eventManagerImpl.onListen, 'onListen not set');
   const query = listener.query;
   let firstListen = false;
 
@@ -92,9 +134,18 @@ export async function eventManagerListen(
     queryInfo = new QueryListenersInfo();
   }
 
+  const firstListenToRemoteStore =
+    listensToRemoteStore(listener) &&
+    queryInfo.getRemoteListeners().length === 0;
+
   if (firstListen) {
+    // When listening to a query for the first time, it may or may not establish
+    // watch connection based on the source the query is listening to.
     try {
-      queryInfo.viewSnap = await eventManagerImpl.onListen(query);
+      queryInfo.viewSnap = await eventManagerImpl.onListen!(
+        query,
+        firstListenToRemoteStore
+      );
     } catch (e) {
       const firestoreError = wrapInUserErrorIfRecoverable(
         e as Error,
@@ -103,6 +154,21 @@ export async function eventManagerListen(
       listener.onError(firestoreError);
       return;
     }
+  } else if (firstListenToRemoteStore) {
+    // A query might have listened to previously, but no watch connection is created
+    // as it has been listening to cache only.
+    try {
+      await eventManagerImpl.onRemoteStoreListen!(query);
+    } catch (e) {
+      const firestoreError = wrapInUserErrorIfRecoverable(
+        e as Error,
+        `Initialization of remote connection for query '${stringifyQuery(
+          listener.query
+        )}' failed`
+      );
+      listener.onError(firestoreError);
+      return;
+    }
   }
 
   eventManagerImpl.queries.set(query, queryInfo);
@@ -130,23 +196,34 @@ export async function eventManagerUnlisten(
   listener: QueryListener
 ): Promise<void> {
   const eventManagerImpl = debugCast(eventManager, EventManagerImpl);
+  validateEventManager(eventManagerImpl);
 
-  debugAssert(!!eventManagerImpl.onUnlisten, 'onUnlisten not set');
   const query = listener.query;
   let lastListen = false;
+  let lastListenToRemoteStore = false;
 
   const queryInfo = eventManagerImpl.queries.get(query);
   if (queryInfo) {
     const i = queryInfo.listeners.indexOf(listener);
     if (i >= 0) {
       queryInfo.listeners.splice(i, 1);
+
       lastListen = queryInfo.listeners.length === 0;
+      lastListenToRemoteStore =
+        listensToRemoteStore(listener) &&
+        queryInfo.getRemoteListeners().length === 0;
     }
   }
 
   if (lastListen) {
     eventManagerImpl.queries.delete(query);
-    return eventManagerImpl.onUnlisten(query);
+    // When removing the last listener, trigger remote store un-listen based
+    // on the source it is listening to. If it is cache, watch connection might
+    // have not been established previously or already been closed.
+    return eventManagerImpl.onUnlisten!(query, lastListenToRemoteStore);
+  } else if (lastListenToRemoteStore) {
+    // Un-listen to the remote store if there are no listeners sourced from watch left.
+    return eventManagerImpl.onRemoteStoreUnlisten!(query);
   }
 }
 
@@ -250,6 +327,9 @@ export interface ListenOptions {
    * offline.
    */
   readonly waitForSyncWhenOnline?: boolean;
+
+  /** Set the source events raised from. */
+  readonly source?: ListenSource;
 }
 
 /**
@@ -265,7 +345,7 @@ export class QueryListener {
    */
   private raisedInitialEvent = false;
 
-  private options: ListenOptions;
+  readonly options: ListenOptions;
 
   private snap: ViewSnapshot | null = null;
 
@@ -359,6 +439,11 @@ export class QueryListener {
       return true;
     }
 
+    // Always raise event if listening to cache
+    if (this.options.source === ListenSource.Cache) {
+      return true;
+    }
+
     // NOTE: We consider OnlineState.Unknown as online (it should become Offline
     // or Online if we wait long enough).
     const maybeOnline = onlineState !== OnlineState.Offline;

packages/firestore/src/core/firestore_client.ts

@@ -86,7 +86,9 @@ import {
   syncEngineLoadBundle,
   syncEngineRegisterPendingWritesCallback,
   syncEngineUnlisten,
-  syncEngineWrite
+  syncEngineWrite,
+  triggerRemoteStoreListen,
+  triggerRemoteStoreUnlisten
 } from './sync_engine_impl';
 import { Transaction } from './transaction';
 import { TransactionOptions } from './transaction_options';
@@ -397,6 +399,14 @@ export async function getEventManager(
     null,
     onlineComponentProvider.syncEngine
   );
+  eventManager.onRemoteStoreListen = triggerRemoteStoreListen.bind(
+    null,
+    onlineComponentProvider.syncEngine
+  );
+  eventManager.onRemoteStoreUnlisten = triggerRemoteStoreUnlisten.bind(
+    null,
+    onlineComponentProvider.syncEngine
+  );
   return eventManager;
 }