Make View processing logic optional

packages/firestore/src/core/component_provider.ts

@@ -29,9 +29,11 @@ import {
 import {
   applyActiveTargetsChange,
   applyBatchState,
+  applyOnlineStateChange,
   applyPrimaryState,
   applyTargetState,
   getActiveClients,
+  handleCredentialChange,
   newSyncEngine,
   SyncEngine
 } from './sync_engine';
@@ -333,12 +335,16 @@ export class OnlineComponentProvider {
     this.eventManager = this.createEventManager(cfg);
 
     this.sharedClientState.onlineStateHandler = onlineState =>
-      this.syncEngine.applyOnlineStateChange(
+      applyOnlineStateChange(
+        this.syncEngine,
         onlineState,
         OnlineStateSource.SharedClientState
       );
 
-    this.remoteStore.syncEngine = this.syncEngine;
+    this.remoteStore.remoteSyncer.handleCredentialChange = handleCredentialChange.bind(
+      null,
+      this.syncEngine
+    );
 
     await this.remoteStore.start();
     await this.remoteStore.applyPrimaryState(this.syncEngine.isPrimaryClient);
@@ -360,7 +366,8 @@ export class OnlineComponentProvider {
       this.datastore,
       cfg.asyncQueue,
       onlineState =>
-        this.syncEngine.applyOnlineStateChange(
+        applyOnlineStateChange(
+          this.syncEngine,
           onlineState,
           OnlineStateSource.RemoteStore
         ),
@@ -372,7 +379,6 @@ export class OnlineComponentProvider {
     return newSyncEngine(
       this.localStore,
       this.remoteStore,
-      this.datastore,
       this.sharedClientState,
       cfg.initialUser,
       cfg.maxConcurrentLimboResolutions,

packages/firestore/src/core/event_manager.ts

@@ -19,7 +19,12 @@ import { debugAssert } from '../util/assert';
 import { EventHandler } from '../util/misc';
 import { ObjectMap } from '../util/obj_map';
 import { canonifyQuery, Query, queryEquals, stringifyQuery } from './query';
-import { SyncEngine, SyncEngineListener } from './sync_engine';
+import {
+  SyncEngine,
+  SyncEngineListener,
+  syncEngineListen,
+  syncEngineUnlisten
+} from './sync_engine';
 import { OnlineState } from './types';
 import { ChangeType, DocumentViewChange, ViewSnapshot } from './view_snapshot';
 import { wrapInUserErrorIfRecoverable } from '../util/async_queue';
@@ -47,79 +52,19 @@ export interface Observer<T> {
  * backend.
  */
 export class EventManager implements SyncEngineListener {
-  private queries = new ObjectMap<Query, QueryListenersInfo>(
+  queries = new ObjectMap<Query, QueryListenersInfo>(
     q => canonifyQuery(q),
     queryEquals
   );
 
-  private onlineState = OnlineState.Unknown;
+  onlineState = OnlineState.Unknown;
 
   private snapshotsInSyncListeners: Set<Observer<void>> = new Set();
 
-  constructor(private syncEngine: SyncEngine) {
+  constructor(readonly syncEngine: SyncEngine) {
     this.syncEngine.subscribe(this);
   }
 
-  async listen(listener: QueryListener): Promise<void> {
-    const query = listener.query;
-    let firstListen = false;
-
-    let queryInfo = this.queries.get(query);
-    if (!queryInfo) {
-      firstListen = true;
-      queryInfo = new QueryListenersInfo();
-    }
-
-    if (firstListen) {
-      try {
-        queryInfo.viewSnap = await this.syncEngine.listen(query);
-      } catch (e) {
-        const firestoreError = wrapInUserErrorIfRecoverable(
-          e,
-          `Initialization of query '${stringifyQuery(listener.query)}' failed`
-        );
-        listener.onError(firestoreError);
-        return;
-      }
-    }
-
-    this.queries.set(query, queryInfo);
-    queryInfo.listeners.push(listener);
-
-    // Run global snapshot listeners if a consistent snapshot has been emitted.
-    const raisedEvent = listener.applyOnlineStateChange(this.onlineState);
-    debugAssert(
-      !raisedEvent,
-      "applyOnlineStateChange() shouldn't raise an event for brand-new listeners."
-    );
-
-    if (queryInfo.viewSnap) {
-      const raisedEvent = listener.onViewSnapshot(queryInfo.viewSnap);
-      if (raisedEvent) {
-        this.raiseSnapshotsInSyncEvent();
-      }
-    }
-  }
-
-  async unlisten(listener: QueryListener): Promise<void> {
-    const query = listener.query;
-    let lastListen = false;
-
-    const queryInfo = this.queries.get(query);
-    if (queryInfo) {
-      const i = queryInfo.listeners.indexOf(listener);
-      if (i >= 0) {
-        queryInfo.listeners.splice(i, 1);
-        lastListen = queryInfo.listeners.length === 0;
-      }
-    }
-
-    if (lastListen) {
-      this.queries.delete(query);
-      return this.syncEngine.unlisten(query);
-    }
-  }
-
   onWatchChange(viewSnaps: ViewSnapshot[]): void {
     let raisedEvent = false;
     for (const viewSnap of viewSnaps) {
@@ -180,7 +125,7 @@ export class EventManager implements SyncEngineListener {
   }
 
   // Call all global snapshot listeners that have been set.
-  private raiseSnapshotsInSyncEvent(): void {
+  raiseSnapshotsInSyncEvent(): void {
     this.snapshotsInSyncListeners.forEach(observer => {
       observer.next();
     });
@@ -357,3 +302,72 @@ export class QueryListener {
     this.queryObserver.next(snap);
   }
 }
+
+export async function eventManagerListen(
+  eventManager: EventManager,
+  listener: QueryListener
+): Promise<void> {
+  const query = listener.query;
+  let firstListen = false;
+
+  let queryInfo = eventManager.queries.get(query);
+  if (!queryInfo) {
+    firstListen = true;
+    queryInfo = new QueryListenersInfo();
+  }
+
+  if (firstListen) {
+    try {
+      queryInfo.viewSnap = await syncEngineListen(
+        eventManager.syncEngine,
+        query
+      );
+    } catch (e) {
+      const firestoreError = wrapInUserErrorIfRecoverable(
+        e,
+        `Initialization of query '${stringifyQuery(listener.query)}' failed`
+      );
+      listener.onError(firestoreError);
+      return;
+    }
+  }
+
+  eventManager.queries.set(query, queryInfo);
+  queryInfo.listeners.push(listener);
+
+  // Run global snapshot listeners if a consistent snapshot has been emitted.
+  const raisedEvent = listener.applyOnlineStateChange(eventManager.onlineState);
+  debugAssert(
+    !raisedEvent,
+    "applyOnlineStateChange() shouldn't raise an event for brand-new listeners."
+  );
+
+  if (queryInfo.viewSnap) {
+    const raisedEvent = listener.onViewSnapshot(queryInfo.viewSnap);
+    if (raisedEvent) {
+      eventManager.raiseSnapshotsInSyncEvent();
+    }
+  }
+}
+
+export async function eventManagerUnlisten(
+  eventManager: EventManager,
+  listener: QueryListener
+): Promise<void> {
+  const query = listener.query;
+  let lastListen = false;
+
+  const queryInfo = eventManager.queries.get(query);
+  if (queryInfo) {
+    const i = queryInfo.listeners.indexOf(listener);
+    if (i >= 0) {
+      queryInfo.listeners.splice(i, 1);
+      lastListen = queryInfo.listeners.length === 0;
+    }
+  }
+
+  if (lastListen) {
+    eventManager.queries.delete(query);
+    return syncEngineUnlisten(eventManager.syncEngine, query);
+  }
+}

packages/firestore/src/core/firestore_client.ts

@@ -33,9 +33,15 @@ import {
   EventManager,
   ListenOptions,
   Observer,
-  QueryListener
+  QueryListener,
+  eventManagerListen,
+  eventManagerUnlisten
 } from './event_manager';
-import { SyncEngine } from './sync_engine';
+import {
+  registerPendingWritesCallback,
+  SyncEngine,
+  syncEngineWrite
+} from './sync_engine';
 import { View } from './view';
 import { SharedClientState } from '../local/shared_client_state';
 import { AutoId } from '../util/misc';
@@ -401,9 +407,9 @@ export class FirestoreClient {
     this.verifyNotTerminated();
 
     const deferred = new Deferred<void>();
-    this.asyncQueue.enqueueAndForget(() => {
-      return this.syncEngine.registerPendingWritesCallback(deferred);
-    });
+    this.asyncQueue.enqueueAndForget(() => 
+       registerPendingWritesCallback(this.syncEngine, deferred)
+    );
     return deferred.promise;
   }
 
@@ -415,10 +421,14 @@ export class FirestoreClient {
     this.verifyNotTerminated();
     const wrappedObserver = new AsyncObserver(observer);
     const listener = new QueryListener(query, wrappedObserver, options);
-    this.asyncQueue.enqueueAndForget(() => this.eventMgr.listen(listener));
+    this.asyncQueue.enqueueAndForget(() =>
+      eventManagerListen(this.eventMgr, listener)
+    );
     return () => {
       wrappedObserver.mute();
-      this.asyncQueue.enqueueAndForget(() => this.eventMgr.unlisten(listener));
+      this.asyncQueue.enqueueAndForget(() =>
+        eventManagerUnlisten(this.eventMgr, listener)
+      );
     };
   }
 
@@ -476,7 +486,7 @@ export class FirestoreClient {
     this.verifyNotTerminated();
     const deferred = new Deferred<void>();
     this.asyncQueue.enqueueAndForget(() =>
-      this.syncEngine.write(mutations, deferred)
+      syncEngineWrite(this.syncEngine, mutations, deferred)
     );
     return deferred.promise;
   }
@@ -545,7 +555,9 @@ export function enqueueWrite(
   mutations: Mutation[]
 ): Promise<void> {
   const deferred = new Deferred<void>();
-  asyncQueue.enqueueAndForget(() => syncEngine.write(mutations, deferred));
+  asyncQueue.enqueueAndForget(() =>
+    syncEngineWrite(syncEngine, mutations, deferred)
+  );
   return deferred.promise;
 }
 
@@ -566,9 +578,9 @@ export function enqueueWaitForPendingWrites(
   syncEngine: SyncEngine
 ): Promise<void> {
   const deferred = new Deferred<void>();
-  asyncQueue.enqueueAndForget(() => {
-    return syncEngine.registerPendingWritesCallback(deferred);
-  });
+  asyncQueue.enqueueAndForget(() =>
+    registerPendingWritesCallback(syncEngine, deferred)
+  );
   return deferred.promise;
 }
 
@@ -581,10 +593,12 @@ export function enqueueListen(
 ): Unsubscribe {
   const wrappedObserver = new AsyncObserver(observer);
   const listener = new QueryListener(query, wrappedObserver, options);
-  asyncQueue.enqueueAndForget(() => eventManger.listen(listener));
+  asyncQueue.enqueueAndForget(() => eventManagerListen(eventManger, listener));
   return () => {
     wrappedObserver.mute();
-    asyncQueue.enqueueAndForget(() => eventManger.unlisten(listener));
+    asyncQueue.enqueueAndForget(() =>
+      eventManagerUnlisten(eventManger, listener)
+    );
   };
 }