Review

Author: schmidt-sebastian

packages/firestore/src/core/event_manager.ts

@@ -20,20 +20,15 @@ import { EventHandler } from '../util/misc';
 import { ObjectMap } from '../util/obj_map';
 import { Query } from './query';
 import { SyncEngine, SyncEngineListener } from './sync_engine';
-import { OnlineState, TargetId } from './types';
+import { OnlineState } from './types';
 import { ChangeType, DocumentViewChange, ViewSnapshot } from './view_snapshot';
-import { logError } from '../util/log';
-import { Code, FirestoreError } from '../util/error';
-
-const LOG_TAG = 'EventManager';
 
 /**
  * Holds the listeners and the last received ViewSnapshot for a query being
  * tracked by EventManager.
  */
 class QueryListenersInfo {
   viewSnap: ViewSnapshot | undefined = undefined;
-  targetId: TargetId = 0;
   listeners: QueryListener[] = [];
 }
 
@@ -74,23 +69,7 @@ export class EventManager implements SyncEngineListener {
     }
 
     if (firstListen) {
-      try {
-        const { targetId, snapshot } = await this.syncEngine.listen(query);
-        queryInfo.targetId = targetId;
-        queryInfo.viewSnap = snapshot;
-      } catch (e) {
-        if (e.name === 'IndexedDbTransactionError') {
-          logError(LOG_TAG, `Failed to register query: ${e}`);
-          listener.onError(
-            new FirestoreError(
-              Code.UNAVAILABLE,
-              `Failed to register query: ${e}`
-            )
-          );
-          return;
-        }
-        throw e;
-      }
+      queryInfo.viewSnap = await this.syncEngine.listen(query);
     }
 
     this.queries.set(query, queryInfo!);

packages/firestore/src/core/firestore_client.ts

@@ -27,7 +27,7 @@ import { Datastore } from '../remote/datastore';
 import { RemoteStore } from '../remote/remote_store';
 import { AsyncQueue } from '../util/async_queue';
 import { Code, FirestoreError } from '../util/error';
-import { logDebug } from '../util/log';
+import { logDebug, logError } from '../util/log';
 import { Deferred } from '../util/promise';
 import {
   EventManager,
@@ -398,8 +398,22 @@ export class FirestoreClient {
   ): QueryListener {
     this.verifyNotTerminated();
     const listener = new QueryListener(query, observer, options);
-    this.asyncQueue.enqueueAndForget(() => {
-      return this.eventMgr.listen(listener);
+    this.asyncQueue.enqueueAndForget(async () => {
+      try {
+        await this.eventMgr.listen(listener);
+      } catch (e) {
+        logError(LOG_TAG, `Failed to register query: ${e}`);
+        if (e.name === 'IndexedDbTransactionError') {
+          listener.onError(
+            new FirestoreError(
+              Code.UNAVAILABLE,
+              `Failed to register query: ${e}`
+            )
+          );
+        } else {
+          throw e;
+        }
+      }
     });
     return listener;
   }

packages/firestore/src/core/sync_engine.ts

@@ -102,12 +102,6 @@ class QueryView {
   ) {}
 }
 
-/** Result type returned during initial query registration. */
-export interface QueryRegistration {
-  targetId: TargetId;
-  snapshot?: ViewSnapshot;
-}
-
 /** Tracks a limbo resolution. */
 class LimboResolution {
   constructor(public key: DocumentKey) {}
@@ -220,9 +214,9 @@ export class SyncEngine implements RemoteSyncer, SharedClientStateSyncer {
   /**
    * Initiates the new listen, resolves promise when listen enqueued to the
    * server. All the subsequent view snapshots or errors are sent to the
-   * subscribed handlers. Returns the targetId of the query.
+   * subscribed handlers. Returns the initial snapshot.
    */
-  async listen(query: Query): Promise<QueryRegistration> {
+  async listen(query: Query): Promise<ViewSnapshot> {
     this.assertSubscribed('listen()');
 
     let targetId;
@@ -256,7 +250,7 @@ export class SyncEngine implements RemoteSyncer, SharedClientStateSyncer {
       }
     }
 
-    return { targetId, snapshot: viewSnapshot };
+    return viewSnapshot;
   }
 
   /**

packages/firestore/test/unit/specs/spec_test_runner.ts

@@ -70,7 +70,7 @@ import {
 } from '../../../src/remote/watch_change';
 import { debugAssert, fail } from '../../../src/util/assert';
 import { AsyncQueue, TimerId } from '../../../src/util/async_queue';
-import { FirestoreError } from '../../../src/util/error';
+import { Code, FirestoreError } from '../../../src/util/error';
 import { primitiveComparator } from '../../../src/util/misc';
 import { forEach, objectSize } from '../../../src/util/obj';
 import { ObjectMap } from '../../../src/util/obj_map';
@@ -608,28 +608,36 @@ abstract class TestRunner {
     const queryListener = new QueryListener(query, aggregator, options);
     this.queryListeners.set(query, queryListener);
 
-    await this.queue.enqueue(() => {
-      return this.eventManager.listen(queryListener);
+    const targetAdded = await this.queue.enqueue(async () => {
+      try {
+        await this.eventManager.listen(queryListener);
+        return true;
+      } catch (e) {
+        expect(this.persistence.injectFailures).to.be.true;
+        queryListener.onError(
+          new FirestoreError(Code.UNAVAILABLE, e.message)
+        );
+        return false;
+      }
     });
 
-    if (this.persistence.injectFailures) {
-      // The Watch stream won't open if we are injecting failures.
-      return;
-    }
-
-    // Skip the backoff that may have been triggered by a previous call to
-    // `watchStreamCloses()`.
-    if (
-      this.queue.containsDelayedOperation(TimerId.ListenStreamConnectionBackoff)
-    ) {
-      await this.queue.runDelayedOperationsEarly(
-        TimerId.ListenStreamConnectionBackoff
-      );
-    }
+    if (targetAdded) {
+      // Skip the backoff that may have been triggered by a previous call to
+      // `watchStreamCloses()`.
+      if (
+        this.queue.containsDelayedOperation(
+          TimerId.ListenStreamConnectionBackoff
+        )
+      ) {
+        await this.queue.runDelayedOperationsEarly(
+          TimerId.ListenStreamConnectionBackoff
+        );
+      }
 
-    if (this.isPrimaryClient && this.networkEnabled) {
-      // Open should always have happened after a listen
-      await this.connection.waitForWatchOpen();
+      if (this.isPrimaryClient && this.networkEnabled) {
+        // Open should always have happened after a listen
+        await this.connection.waitForWatchOpen();
+      }
     }
   }