Merge d7b5d2a into e8b950f

Author: schmidt-sebastian

packages/firestore/exp/dependencies.json

@@ -23,10 +23,14 @@ import {
   OfflineComponentProvider,
   OnlineComponentProvider
 } from '../../../src/core/component_provider';
-import {handleUserChange, LocalStore} from '../../../src/local/local_store';
+import { handleUserChange, LocalStore } from '../../../src/local/local_store';
 import { Deferred } from '../../../src/util/promise';
 import { logDebug } from '../../../src/util/log';
-import { SyncEngine } from '../../../src/core/sync_engine';
+import {
+  SyncEngine,
+  syncEngineListen,
+  syncEngineUnlisten
+} from '../../../src/core/sync_engine';
 import { RemoteStore } from '../../../src/remote/remote_store';
 import { Persistence } from '../../../src/local/persistence';
 import { EventManager } from '../../../src/core/event_manager';
@@ -153,9 +157,14 @@ export function getRemoteStore(firestore: Firestore): Promise<RemoteStore> {
 }
 
 export function getEventManager(firestore: Firestore): Promise<EventManager> {
-  return getOnlineComponentProvider(firestore).then(
-    components => components.eventManager
-  );
+  return getOnlineComponentProvider(firestore).then(components => {
+    const eventManager = components.eventManager;
+    eventManager.subscribe(
+      syncEngineListen.bind(null, components.syncEngine),
+      syncEngineUnlisten.bind(null, components.syncEngine)
+    );
+    return eventManager;
+  });
 }
 
 export function getPersistence(firestore: Firestore): Promise<Persistence> {

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

@@ -29,9 +29,11 @@ import {
 import {
   applyActiveTargetsChange,
   applyBatchState,
+  applyOnlineStateChange,
   applyPrimaryState,
   applyTargetState,
   getActiveClients,
+  handleCredentialChange,
   newSyncEngine,
   SyncEngine
 } from './sync_engine';
@@ -332,20 +334,26 @@ export class OnlineComponentProvider {
     this.syncEngine = this.createSyncEngine(cfg);
     this.eventManager = this.createEventManager(cfg);
 
+    this.syncEngine.subscribe(this.eventManager);
+
     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);
   }
 
   createEventManager(cfg: ComponentConfiguration): EventManager {
-    return new EventManager(this.syncEngine);
+    return new EventManager();
   }
 
   createDatastore(cfg: ComponentConfiguration): Datastore {
@@ -360,7 +368,8 @@ export class OnlineComponentProvider {
       this.datastore,
       cfg.asyncQueue,
       onlineState =>
-        this.syncEngine.applyOnlineStateChange(
+        applyOnlineStateChange(
+          this.syncEngine,
           onlineState,
           OnlineStateSource.RemoteStore
         ),
@@ -372,7 +381,6 @@ export class OnlineComponentProvider {
     return newSyncEngine(
       this.localStore,
       this.remoteStore,
-      this.datastore,
       this.sharedClientState,
       cfg.initialUser,
       cfg.maxConcurrentLimboResolutions,

packages/firestore/src/core/component_provider.ts

@@ -19,7 +19,7 @@ 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 { SyncEngineListener } from './sync_engine';
 import { OnlineState } from './types';
 import { ChangeType, DocumentViewChange, ViewSnapshot } from './view_snapshot';
 import { wrapInUserErrorIfRecoverable } from '../util/async_queue';
@@ -45,6 +45,10 @@ export interface Observer<T> {
  * EventManager is responsible for mapping queries to query event emitters.
  * It handles "fan-out". -- Identical queries will re-use the same watch on the
  * backend.
+ *
+ * PORTING NOTE: On Web, EventManager requires a call to `subscribe()` to
+ * register SyncEngine's `listen()` and `unlisten()` functionality. This allows
+ * users to tree-shake the Watch logic.
  */
 export class EventManager implements SyncEngineListener {
   private queries = new ObjectMap<Query, QueryListenersInfo>(
@@ -56,11 +60,21 @@ export class EventManager implements SyncEngineListener {
 
   private snapshotsInSyncListeners: Set<Observer<void>> = new Set();
 
-  constructor(private syncEngine: SyncEngine) {
-    this.syncEngine.subscribe(this);
+  /** Callback invoked when a Query is first listen to. */
+  private onListen?: (query: Query) => Promise<ViewSnapshot>;
+  /** Callback invoked once all listeners to a Query are removed. */
+  private onUnlisten?: (query: Query) => Promise<void>;
+
+  subscribe(
+    onListen: (query: Query) => Promise<ViewSnapshot>,
+    onUnlisten: (query: Query) => Promise<void>
+  ): void {
+    this.onListen = onListen;
+    this.onUnlisten = onUnlisten;
   }
 
   async listen(listener: QueryListener): Promise<void> {
+    debugAssert(!!this.onListen, 'onListen not set');
     const query = listener.query;
     let firstListen = false;
 
@@ -72,7 +86,7 @@ export class EventManager implements SyncEngineListener {
 
     if (firstListen) {
       try {
-        queryInfo.viewSnap = await this.syncEngine.listen(query);
+        queryInfo.viewSnap = await this.onListen(query);
       } catch (e) {
         const firestoreError = wrapInUserErrorIfRecoverable(
           e,
@@ -102,6 +116,7 @@ export class EventManager implements SyncEngineListener {
   }
 
   async unlisten(listener: QueryListener): Promise<void> {
+    debugAssert(!!this.onUnlisten, 'onUnlisten not set');
     const query = listener.query;
     let lastListen = false;
 
@@ -116,7 +131,7 @@ export class EventManager implements SyncEngineListener {
 
     if (lastListen) {
       this.queries.delete(query);
-      return this.syncEngine.unlisten(query);
+      return this.onUnlisten(query);
     }
   }
 

packages/firestore/src/core/event_manager.ts

@@ -39,7 +39,13 @@ import {
   Observer,
   QueryListener
 } from './event_manager';
-import { SyncEngine } from './sync_engine';
+import {
+  registerPendingWritesCallback,
+  SyncEngine,
+  syncEngineListen,
+  syncEngineUnlisten,
+  syncEngineWrite
+} from './sync_engine';
 import { View } from './view';
 import { SharedClientState } from '../local/shared_client_state';
 import { AutoId } from '../util/misc';
@@ -277,6 +283,11 @@ export class FirestoreClient {
       this.syncEngine = onlineComponentProvider.syncEngine;
       this.eventMgr = onlineComponentProvider.eventManager;
 
+      this.eventMgr.subscribe(
+        syncEngineListen.bind(null, this.syncEngine),
+        syncEngineUnlisten.bind(null, this.syncEngine)
+      );
+
       // When a user calls clearPersistence() in one client, all other clients
       // need to be terminated to allow the delete to succeed.
       this.persistence.setDatabaseDeletedListener(async () => {
@@ -405,9 +416,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;
   }
 
@@ -480,7 +491,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;
   }
@@ -549,7 +560,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;
 }
 
@@ -570,9 +583,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;
 }