Simplify Listen API (#3491)

Author: schmidt-sebastian

packages/firestore/exp/src/api/reference.ts

@@ -31,9 +31,6 @@ import { debugAssert } from '../../../src/util/assert';
 import { cast } from '../../../lite/src/api/util';
 import { DocumentSnapshot, QuerySnapshot } from './snapshot';
 import {
-  addDocSnapshotListener,
-  addQuerySnapshotListener,
-  addSnapshotsInSyncListener,
   applyFirestoreDataConverter,
   getDocsViaSnapshotListener,
   getDocViaSnapshotListener,
@@ -60,6 +57,7 @@ import {
   Unsubscribe
 } from '../../../src/api/observer';
 import { getFirestoreClient } from './components';
+import { newQueryForPath } from '../../../src/core/query';
 
 export function getDoc<T>(
   reference: firestore.DocumentReference<T>
@@ -369,7 +367,7 @@ export function onSnapshot<T>(
     args[currArg + 2] = userObserver.complete?.bind(userObserver);
   }
 
-  let asyncObserver: Promise<Unsubscribe>;
+  let asyncUnsubscribe: Promise<Unsubscribe>;
 
   if (ref instanceof DocumentReference) {
     const firestore = cast(ref.firestore, Firestore);
@@ -386,10 +384,9 @@ export function onSnapshot<T>(
       complete: args[currArg + 2] as CompleteFn
     };
 
-    asyncObserver = getFirestoreClient(firestore).then(firestoreClient =>
-      addDocSnapshotListener(
-        firestoreClient,
-        ref._key,
+    asyncUnsubscribe = getFirestoreClient(firestore).then(firestoreClient =>
+      firestoreClient.listen(
+        newQueryForPath(ref._key.path),
         internalOptions,
         observer
       )
@@ -412,21 +409,16 @@ export function onSnapshot<T>(
 
     validateHasExplicitOrderByForLimitToLast(query._query);
 
-    asyncObserver = getFirestoreClient(firestore).then(firestoreClient =>
-      addQuerySnapshotListener(
-        firestoreClient,
-        query._query,
-        internalOptions,
-        observer
-      )
+    asyncUnsubscribe = getFirestoreClient(firestore).then(firestoreClient =>
+      firestoreClient.listen(query._query, internalOptions, observer)
     );
   }
 
   // TODO(firestorexp): Add test that verifies that we don't raise a snapshot if
   // unsubscribe is called before `asyncObserver` resolves.
   return () => {
     // eslint-disable-next-line @typescript-eslint/no-floating-promises
-    asyncObserver.then(unsubscribe => unsubscribe());
+    asyncUnsubscribe.then(unsubscribe => unsubscribe());
   };
 }
 
@@ -458,7 +450,7 @@ export function onSnapshotsInSync(
   const asyncObserver = getFirestoreClient(
     firestoreImpl
   ).then(firestoreClient =>
-    addSnapshotsInSyncListener(firestoreClient, observer)
+    firestoreClient.addSnapshotsInSyncListener(observer)
   );
 
   // TODO(firestorexp): Add test that verifies that we don't raise a snapshot if

packages/firestore/src/api/database.ts

@@ -24,9 +24,9 @@ import { _FirebaseApp, FirebaseService } from '@firebase/app-types/private';
 import { DatabaseId, DatabaseInfo } from '../core/database_info';
 import { ListenOptions } from '../core/event_manager';
 import {
-  OnlineComponentProvider,
   MemoryOfflineComponentProvider,
-  OfflineComponentProvider
+  OfflineComponentProvider,
+  OnlineComponentProvider
 } from '../core/component_provider';
 import { FirestoreClient, PersistenceSettings } from '../core/firestore_client';
 import {
@@ -60,7 +60,6 @@ import { FieldPath, ResourcePath } from '../model/path';
 import { isServerTimestamp } from '../model/server_timestamps';
 import { refValue } from '../model/values';
 import { debugAssert, fail } from '../util/assert';
-import { AsyncObserver } from '../util/async_observer';
 import { AsyncQueue } from '../util/async_queue';
 import { Code, FirestoreError } from '../util/error';
 import {
@@ -485,16 +484,15 @@ export class Firestore implements firestore.FirebaseFirestore, FirebaseService {
     this.ensureClientConfigured();
 
     if (isPartialObserver(arg)) {
-      return addSnapshotsInSyncListener(
-        this._firestoreClient!,
+      return this._firestoreClient!.addSnapshotsInSyncListener(
         arg as PartialObserver<void>
       );
     } else {
       validateArgType('Firestore.onSnapshotsInSync', 'function', 1, arg);
       const observer: PartialObserver<void> = {
         next: arg as () => void
       };
-      return addSnapshotsInSyncListener(this._firestoreClient!, observer);
+      return this._firestoreClient!.addSnapshotsInSyncListener(observer);
     }
   }
 
@@ -686,29 +684,6 @@ export class Firestore implements firestore.FirebaseFirestore, FirebaseService {
   }
 }
 
-/** Registers the listener for onSnapshotsInSync() */
-export function addSnapshotsInSyncListener(
-  firestoreClient: FirestoreClient,
-  observer: PartialObserver<void>
-): Unsubscribe {
-  const errHandler = (err: Error): void => {
-    throw fail('Uncaught Error in onSnapshotsInSync');
-  };
-  const asyncObserver = new AsyncObserver<void>({
-    next: () => {
-      if (observer.next) {
-        observer.next();
-      }
-    },
-    error: errHandler
-  });
-  firestoreClient.addSnapshotsInSyncListener(asyncObserver);
-  return () => {
-    asyncObserver.mute();
-    firestoreClient.removeSnapshotsInSyncListener(asyncObserver);
-  };
-}
-
 /**
  * A reference to a transaction.
  */
@@ -1183,7 +1158,7 @@ export class DocumentReference<T = firestore.DocumentData>
       1,
       4
     );
-    let options: firestore.SnapshotListenOptions = {
+    let options: ListenOptions = {
       includeMetadataChanges: false
     };
     let currArg = 0;
@@ -1248,9 +1223,8 @@ export class DocumentReference<T = firestore.DocumentData>
       complete: args[currArg + 2] as CompleteFn
     };
 
-    return addDocSnapshotListener(
-      this._firestoreClient,
-      this._key,
+    return this._firestoreClient.listen(
+      newQueryForPath(this._key.path),
       internalOptions,
       observer
     );
@@ -1312,40 +1286,6 @@ export class DocumentReference<T = firestore.DocumentData>
   }
 }
 
-/** Registers an internal snapshot listener for `ref`. */
-export function addDocSnapshotListener(
-  firestoreClient: FirestoreClient,
-  key: DocumentKey,
-  options: ListenOptions,
-  observer: PartialObserver<ViewSnapshot>
-): Unsubscribe {
-  let errHandler = (err: Error): void => {
-    console.error('Uncaught Error in onSnapshot:', err);
-  };
-  if (observer.error) {
-    errHandler = observer.error.bind(observer);
-  }
-
-  const asyncObserver = new AsyncObserver<ViewSnapshot>({
-    next: snapshot => {
-      if (observer.next) {
-        observer.next(snapshot);
-      }
-    },
-    error: errHandler
-  });
-  const internalListener = firestoreClient.listen(
-    newQueryForPath(key.path),
-    asyncObserver,
-    options
-  );
-
-  return () => {
-    asyncObserver.mute();
-    firestoreClient.unlisten(internalListener);
-  };
-}
-
 /**
  * Retrieves a latency-compensated document from the backend via a
  * SnapshotListener.
@@ -1356,9 +1296,8 @@ export function getDocViaSnapshotListener(
   options?: firestore.GetOptions
 ): Promise<ViewSnapshot> {
   const result = new Deferred<ViewSnapshot>();
-  const unlisten = addDocSnapshotListener(
-    firestoreClient,
-    key,
+  const unlisten = firestoreClient.listen(
+    newQueryForPath(key.path),
     {
       includeMetadataChanges: true,
       waitForSyncWhenOnline: true
@@ -2159,7 +2098,7 @@ export class Query<T = firestore.DocumentData> implements firestore.Query<T> {
 
   onSnapshot(...args: unknown[]): Unsubscribe {
     validateBetweenNumberOfArgs('Query.onSnapshot', arguments, 1, 4);
-    let options: firestore.SnapshotListenOptions = {};
+    let options: ListenOptions = {};
     let currArg = 0;
     if (
       typeof args[currArg] === 'object' &&
@@ -2220,12 +2159,7 @@ export class Query<T = firestore.DocumentData> implements firestore.Query<T> {
 
     validateHasExplicitOrderByForLimitToLast(this._query);
     const firestoreClient = this.firestore.ensureClientConfigured();
-    return addQuerySnapshotListener(
-      firestoreClient,
-      this._query,
-      options,
-      observer
-    );
+    return firestoreClient.listen(this._query, options, observer);
   }
 
   get(options?: firestore.GetOptions): Promise<firestore.QuerySnapshot<T>> {
@@ -2254,8 +2188,7 @@ export function getDocsViaSnapshotListener(
   options?: firestore.GetOptions
 ): Promise<ViewSnapshot> {
   const result = new Deferred<ViewSnapshot>();
-  const unlisten = addQuerySnapshotListener(
-    firestore,
+  const unlisten = firestore.listen(
     query,
     {
       includeMetadataChanges: true,
@@ -2287,35 +2220,6 @@ export function getDocsViaSnapshotListener(
   return result.promise;
 }
 
-/** Registers an internal snapshot listener for `query`. */
-export function addQuerySnapshotListener(
-  firestore: FirestoreClient,
-  query: InternalQuery,
-  options: ListenOptions,
-  observer: PartialObserver<ViewSnapshot>
-): Unsubscribe {
-  let errHandler = (err: Error): void => {
-    console.error('Uncaught Error in onSnapshot:', err);
-  };
-  if (observer.error) {
-    errHandler = observer.error.bind(observer);
-  }
-  const asyncObserver = new AsyncObserver<ViewSnapshot>({
-    next: (result: ViewSnapshot): void => {
-      if (observer.next) {
-        observer.next(result);
-      }
-    },
-    error: errHandler
-  });
-
-  const internalListener = firestore.listen(query, asyncObserver, options);
-  return (): void => {
-    asyncObserver.mute();
-    firestore.unlisten(internalListener);
-  };
-}
-
 export class QuerySnapshot<T = firestore.DocumentData>
   implements firestore.QuerySnapshot<T> {
   private _cachedChanges: Array<firestore.DocumentChange<T>> | null = null;

packages/firestore/src/core/firestore_client.ts

@@ -35,7 +35,6 @@ import {
 } from './event_manager';
 import { SyncEngine } from './sync_engine';
 import { View } from './view';
-
 import { SharedClientState } from '../local/shared_client_state';
 import { AutoId } from '../util/misc';
 import { DatabaseId, DatabaseInfo } from './database_info';
@@ -47,6 +46,7 @@ import {
   MemoryOfflineComponentProvider,
   OfflineComponentProvider
 } from './component_provider';
+import { AsyncObserver } from '../util/async_observer';
 
 const LOG_TAG = 'FirestoreClient';
 const MAX_CONCURRENT_LIMBO_RESOLUTIONS = 100;
@@ -391,24 +391,17 @@ export class FirestoreClient {
 
   listen(
     query: Query,
-    observer: Observer<ViewSnapshot>,
-    options: ListenOptions
-  ): QueryListener {
+    options: ListenOptions,
+    observer: Partial<Observer<ViewSnapshot>>
+  ): () => void {
     this.verifyNotTerminated();
-    const listener = new QueryListener(query, observer, options);
+    const wrappedObserver = new AsyncObserver(observer);
+    const listener = new QueryListener(query, wrappedObserver, options);
     this.asyncQueue.enqueueAndForget(() => this.eventMgr.listen(listener));
-    return listener;
-  }
-
-  unlisten(listener: QueryListener): void {
-    // Checks for termination but does not raise error, allowing unlisten after
-    // termination to be a no-op.
-    if (this.clientTerminated) {
-      return;
-    }
-    this.asyncQueue.enqueueAndForget(() => {
-      return this.eventMgr.unlisten(listener);
-    });
+    return () => {
+      wrappedObserver.mute();
+      this.asyncQueue.enqueueAndForget(() => this.eventMgr.unlisten(listener));
+    };
   }
 
   async getDocumentFromLocalCache(
@@ -486,24 +479,18 @@ export class FirestoreClient {
     return this.databaseInfo.databaseId;
   }
 
-  addSnapshotsInSyncListener(observer: Observer<void>): void {
+  addSnapshotsInSyncListener(observer: Partial<Observer<void>>): () => void {
     this.verifyNotTerminated();
-    this.asyncQueue.enqueueAndForget(() => {
-      this.eventMgr.addSnapshotsInSyncListener(observer);
-      return Promise.resolve();
-    });
-  }
-
-  removeSnapshotsInSyncListener(observer: Observer<void>): void {
-    // Checks for shutdown but does not raise error, allowing remove after
-    // shutdown to be a no-op.
-    if (this.clientTerminated) {
-      return;
-    }
-    this.asyncQueue.enqueueAndForget(() => {
-      this.eventMgr.removeSnapshotsInSyncListener(observer);
-      return Promise.resolve();
-    });
+    const wrappedObserver = new AsyncObserver(observer);
+    this.asyncQueue.enqueueAndForget(async () =>
+      this.eventMgr.addSnapshotsInSyncListener(wrappedObserver)
+    );
+    return () => {
+      wrappedObserver.mute();
+      this.asyncQueue.enqueueAndForget(async () =>
+        this.eventMgr.removeSnapshotsInSyncListener(wrappedObserver)
+      );
+    };
   }
 
   get clientTerminated(): boolean {

packages/firestore/src/util/async_observer.ts

@@ -30,14 +30,20 @@ export class AsyncObserver<T> implements Observer<T> {
    */
   private muted = false;
 
-  constructor(private observer: Observer<T>) {}
+  constructor(private observer: Partial<Observer<T>>) {}
 
   next(value: T): void {
-    this.scheduleEvent(this.observer.next, value);
+    if (this.observer.next) {
+      this.scheduleEvent(this.observer.next, value);
+    }
   }
 
   error(error: Error): void {
-    this.scheduleEvent(this.observer.error, error);
+    if (this.observer.error) {
+      this.scheduleEvent(this.observer.error, error);
+    } else {
+      console.error('Uncaught Error in snapshot listener:', error);
+    }
   }
 
   mute(): void {