Simplify Listen API

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,8 @@ import {
   Unsubscribe
 } from '../../../src/api/observer';
 import { getFirestoreClient } from './components';
+import { AsyncObserver } from '../../../src/util/async_observer';
+import { newQueryForPath } from '../../../src/core/query';
 
 export function getDoc<T>(
   reference: firestore.DocumentReference<T>
@@ -387,11 +386,10 @@ export function onSnapshot<T>(
     };
 
     asyncObserver = getFirestoreClient(firestore).then(firestoreClient =>
-      addDocSnapshotListener(
-        firestoreClient,
-        ref._key,
+      firestoreClient.listen(
+        newQueryForPath(ref._key.path),
         internalOptions,
-        observer
+        new AsyncObserver(observer)
       )
     );
   } else {
@@ -413,11 +411,10 @@ export function onSnapshot<T>(
     validateHasExplicitOrderByForLimitToLast(query._query);
 
     asyncObserver = getFirestoreClient(firestore).then(firestoreClient =>
-      addQuerySnapshotListener(
-        firestoreClient,
+      firestoreClient.listen(
         query._query,
         internalOptions,
-        observer
+        new AsyncObserver(observer)
       )
     );
   }
@@ -458,7 +455,7 @@ export function onSnapshotsInSync(
   const asyncObserver = getFirestoreClient(
     firestoreImpl
   ).then(firestoreClient =>
-    addSnapshotsInSyncListener(firestoreClient, observer)
+    firestoreClient.addSnapshotsInSyncListener(new AsyncObserver(observer))
   );
 
   // TODO(firestorexp): Add test that verifies that we don't raise a snapshot if

packages/firestore/src/api/database.ts

@@ -22,11 +22,10 @@ import * as api from '../protos/firestore_proto_api';
 import { FirebaseApp } from '@firebase/app-types';
 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 {
@@ -485,16 +484,17 @@ export class Firestore implements firestore.FirebaseFirestore, FirebaseService {
     this.ensureClientConfigured();
 
     if (isPartialObserver(arg)) {
-      return addSnapshotsInSyncListener(
-        this._firestoreClient!,
-        arg as PartialObserver<void>
+      return this._firestoreClient!.addSnapshotsInSyncListener(
+        new AsyncObserver(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(
+        new AsyncObserver(observer)
+      );
     }
   }
 
@@ -686,29 +686,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.
  */
@@ -1248,11 +1225,10 @@ 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
+      new AsyncObserver(observer)
     );
   }
 
@@ -1312,40 +1288,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,14 +1298,13 @@ 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
     },
-    {
+    new AsyncObserver({
       next: (snap: ViewSnapshot) => {
         // Remove query first before passing event to user to avoid
         // user actions affecting the now stale query.
@@ -1404,7 +1345,7 @@ export function getDocViaSnapshotListener(
         }
       },
       error: e => result.reject(e)
-    }
+    })
   );
   return result.promise;
 }
@@ -2220,11 +2161,10 @@ export class Query<T = firestore.DocumentData> implements firestore.Query<T> {
 
     validateHasExplicitOrderByForLimitToLast(this._query);
     const firestoreClient = this.firestore.ensureClientConfigured();
-    return addQuerySnapshotListener(
-      firestoreClient,
+    return firestoreClient.listen(
       this._query,
       options,
-      observer
+      new AsyncObserver(observer)
     );
   }
 
@@ -2254,14 +2194,13 @@ export function getDocsViaSnapshotListener(
   options?: firestore.GetOptions
 ): Promise<ViewSnapshot> {
   const result = new Deferred<ViewSnapshot>();
-  const unlisten = addQuerySnapshotListener(
-    firestore,
+  const unlisten = firestore.listen(
     query,
     {
       includeMetadataChanges: true,
       waitForSyncWhenOnline: true
     },
-    {
+    new AsyncObserver({
       next: snapshot => {
         // Remove query first before passing event to user to avoid
         // user actions affecting the now stale query.
@@ -2282,40 +2221,11 @@ export function getDocsViaSnapshotListener(
         }
       },
       error: e => result.reject(e)
-    }
+    })
   );
   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

@@ -43,10 +43,12 @@ import { Query } from './query';
 import { Transaction } from './transaction';
 import { ViewSnapshot } from './view_snapshot';
 import {
-  OnlineComponentProvider,
   MemoryOfflineComponentProvider,
-  OfflineComponentProvider
+  OfflineComponentProvider,
+  OnlineComponentProvider
 } from './component_provider';
+import { Unsubscribe } from '../api/observer';
+import { AsyncObserver } from '../util/async_observer';
 
 const LOG_TAG = 'FirestoreClient';
 const MAX_CONCURRENT_LIMBO_RESOLUTIONS = 100;
@@ -391,24 +393,16 @@ export class FirestoreClient {
 
   listen(
     query: Query,
-    observer: Observer<ViewSnapshot>,
-    options: ListenOptions
-  ): QueryListener {
+    options: ListenOptions,
+    observer: AsyncObserver<ViewSnapshot>
+  ): Unsubscribe {
     this.verifyNotTerminated();
     const listener = new QueryListener(query, observer, 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 () => {
+      observer.mute();
+      this.asyncQueue.enqueueAndForget(() => this.eventMgr.unlisten(listener));
+    };
   }
 
   async getDocumentFromLocalCache(
@@ -486,24 +480,17 @@ export class FirestoreClient {
     return this.databaseInfo.databaseId;
   }
 
-  addSnapshotsInSyncListener(observer: Observer<void>): void {
+  addSnapshotsInSyncListener(observer: AsyncObserver<void>): Unsubscribe {
     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();
-    });
+    this.asyncQueue.enqueueAndForget(async () =>
+      this.eventMgr.addSnapshotsInSyncListener(observer)
+    );
+    return () => {
+      observer.mute();
+      this.asyncQueue.enqueueAndForget(async () =>
+        this.eventMgr.removeSnapshotsInSyncListener(observer)
+      );
+    };
   }
 
   get clientTerminated(): boolean {

packages/firestore/src/util/async_observer.ts

@@ -17,6 +17,7 @@
 
 import { Observer } from '../core/event_manager';
 import { EventHandler } from './misc';
+import { PartialObserver } from '../api/observer';
 
 /*
  * A wrapper implementation of Observer<T> that will dispatch events
@@ -30,14 +31,20 @@ export class AsyncObserver<T> implements Observer<T> {
    */
   private muted = false;
 
-  constructor(private observer: Observer<T>) {}
+  constructor(private observer: PartialObserver<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 {