Code complete, need to fix Event

anager tests by adding callback interface

Author: schmidt-sebastian

packages/firestore/exp/dependencies.json

@@ -23,11 +23,14 @@ import {
   OfflineComponentProvider,
   OnlineComponentProvider
 } from '../../../src/core/component_provider';
-import { 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 { RemoteStore } from '../../../src/remote/remote_store';
+import {
+  RemoteStore,
+  remoteStoreHandleCredentialChange
+} from '../../../src/remote/remote_store';
 import { Persistence } from '../../../src/local/persistence';
 import { EventManager } from '../../../src/core/event_manager';
 export const LOG_TAG = 'ComponentProvider';
@@ -65,7 +68,7 @@ export async function setOfflineComponentProvider(
     firestore._queue.enqueueAndForget(() =>
       // TODO(firestorexp): Make sure handleUserChange is a no-op if user
       // didn't change
-      offlineComponentProvider.localStore.handleUserChange(user)
+      handleUserChange(offlineComponentProvider.localStore, user)
     )
   );
   // When a user calls clearPersistence() in one client, all other clients
@@ -95,7 +98,7 @@ export async function setOnlineComponentProvider(
   // precedence over the offline component provider.
   firestore._setCredentialChangeListener(user =>
     firestore._queue.enqueueAndForget(() =>
-      onlineComponentProvider.remoteStore.handleCredentialChange(user)
+      remoteStoreHandleCredentialChange(onlineComponentProvider.remoteStore, user)
     )
   );
   onlineDeferred.resolve(onlineComponentProvider);

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

@@ -55,13 +55,13 @@ import {
   setOfflineComponentProvider,
   setOnlineComponentProvider
 } from './components';
-
 import { DEFAULT_HOST, DEFAULT_SSL } from '../../../lite/src/api/components';
 import { DatabaseInfo } from '../../../src/core/database_info';
 import { AutoId } from '../../../src/util/misc';
 import { User } from '../../../src/auth/user';
 import { CredentialChangeListener } from '../../../src/api/credentials';
 import { logDebug } from '../../../src/util/log';
+import { fillWritePipeline } from '../../../src/remote/remote_store';
 
 const LOG_TAG = 'Firestore';
 
@@ -198,6 +198,8 @@ export function enableIndexedDbPersistence(
   // `getOnlineComponentProvider()`
   const settings = firestoreImpl._getSettings();
 
+  const onlineComponentProvider = new OnlineComponentProvider();
+
   // TODO(firestoreexp): Add forceOwningTab
   return setOfflineComponentProvider(
     firestoreImpl,
@@ -209,7 +211,12 @@ export function enableIndexedDbPersistence(
       forceOwningTab: false
     },
     new IndexedDbOfflineComponentProvider()
-  );
+  )
+    .then(() =>
+      setOnlineComponentProvider(firestoreImpl, new OnlineComponentProvider())
+    )
+    // Enqueue writes from a previous session
+    .then(() => fillWritePipeline(onlineComponentProvider.remoteStore));
 }
 
 export function enableMultiTabIndexedDbPersistence(
@@ -238,9 +245,12 @@ export function enableMultiTabIndexedDbPersistence(
       forceOwningTab: false
     },
     offlineComponentProvider
-  ).then(() =>
-    setOnlineComponentProvider(firestoreImpl, onlineComponentProvider)
-  );
+  )
+    .then(() =>
+      setOnlineComponentProvider(firestoreImpl, onlineComponentProvider)
+    )
+    // Enqueue writes from a previous session
+    .then(() => fillWritePipeline(onlineComponentProvider.remoteStore));
 }
 
 export function clearIndexedDbPersistence(

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

@@ -33,11 +33,15 @@ import {
   applyPrimaryState,
   applyTargetState,
   getActiveClients,
-  handleCredentialChange,
+  syncEngineHandleCredentialChange,
   newSyncEngine,
   SyncEngine
 } from './sync_engine';
-import { RemoteStore } from '../remote/remote_store';
+import {
+  newRemoteStore,
+  RemoteStore,
+  remoteStoreApplyPrimaryState, remoteStoreShutdown
+} from '../remote/remote_store';
 import { EventManager } from './event_manager';
 import { AsyncQueue } from '../util/async_queue';
 import { DatabaseId, DatabaseInfo } from './database_info';
@@ -341,13 +345,12 @@ export class OnlineComponentProvider {
         OnlineStateSource.SharedClientState
       );
 
-    this.remoteStore.remoteSyncer.handleCredentialChange = handleCredentialChange.bind(
+    this.remoteStore.syncEngine.handleCredentialChange = syncEngineHandleCredentialChange.bind(
       null,
       this.syncEngine
     );
 
-    await this.remoteStore.start();
-    await this.remoteStore.applyPrimaryState(this.syncEngine.isPrimaryClient);
+    await remoteStoreApplyPrimaryState(this.remoteStore, this.syncEngine.isPrimaryClient);
   }
 
   createEventManager(cfg: ComponentConfiguration): EventManager {
@@ -361,7 +364,7 @@ export class OnlineComponentProvider {
   }
 
   createRemoteStore(cfg: ComponentConfiguration): RemoteStore {
-    return new RemoteStore(
+    return newRemoteStore(
       this.localStore,
       this.datastore,
       cfg.asyncQueue,
@@ -388,6 +391,6 @@ export class OnlineComponentProvider {
   }
 
   terminate(): Promise<void> {
-    return this.remoteStore.shutdown();
+    return remoteStoreShutdown(this.remoteStore);
   }
 }

packages/firestore/src/core/component_provider.ts

@@ -28,7 +28,13 @@ import { GarbageCollectionScheduler, Persistence } from '../local/persistence';
 import { Document, NoDocument } from '../model/document';
 import { DocumentKey } from '../model/document_key';
 import { Mutation } from '../model/mutation';
-import { RemoteStore } from '../remote/remote_store';
+import {
+  fillWritePipeline,
+  remoteStoreHandleCredentialChange,
+  RemoteStore,
+  enableNetwork,
+  disableNetwork, remoteStoreShutdown
+} from '../remote/remote_store';
 import { AsyncQueue, wrapInUserErrorIfRecoverable } from '../util/async_queue';
 import { Code, FirestoreError } from '../util/error';
 import { logDebug } from '../util/log';
@@ -205,7 +211,7 @@ export class FirestoreClient {
         ).then(this.initializationDone.resolve, this.initializationDone.reject);
       } else {
         this.asyncQueue.enqueueRetryable(() =>
-          this.remoteStore.handleCredentialChange(user)
+          remoteStoreHandleCredentialChange(this.remoteStore, user)
         );
       }
     });
@@ -224,7 +230,7 @@ export class FirestoreClient {
     this.verifyNotTerminated();
     return this.asyncQueue.enqueue(() => {
       this.persistence.setNetworkEnabled(true);
-      return this.remoteStore.enableNetwork();
+      return enableNetwork(this.remoteStore);
     });
   }
 
@@ -289,6 +295,8 @@ export class FirestoreClient {
         await this.terminate();
       });
 
+      await fillWritePipeline(this.remoteStore);
+
       persistenceResult.resolve();
     } catch (error) {
       // Regardless of whether or not the retry succeeds, from an user
@@ -368,7 +376,7 @@ export class FirestoreClient {
     this.verifyNotTerminated();
     return this.asyncQueue.enqueue(() => {
       this.persistence.setNetworkEnabled(false);
-      return this.remoteStore.disableNetwork();
+      return disableNetwork(this.remoteStore);
     });
   }
 
@@ -382,7 +390,7 @@ export class FirestoreClient {
           this.gcScheduler.stop();
         }
 
-        await this.remoteStore.shutdown();
+        await remoteStoreShutdown(this.remoteStore);
         await this.sharedClientState.shutdown();
         await this.persistence.shutdown();
 
@@ -573,7 +581,7 @@ export function enqueueNetworkEnabled(
 ): Promise<void> {
   return asyncQueue.enqueue(() => {
     persistence.setNetworkEnabled(enabled);
-    return enabled ? remoteStore.enableNetwork() : remoteStore.disableNetwork();
+    return enabled ? enableNetwork(remoteStore) : disableNetwork(remoteStore);
   });
 }
 

packages/firestore/src/core/firestore_client.ts

@@ -48,7 +48,14 @@ import { DocumentKey } from '../model/document_key';
 import { Mutation } from '../model/mutation';
 import { BATCHID_UNKNOWN, MutationBatchResult } from '../model/mutation_batch';
 import { RemoteEvent, TargetChange } from '../remote/remote_event';
-import { RemoteStore } from '../remote/remote_store';
+import {
+  canUseNetwork,
+  fillWritePipeline,
+  RemoteStore,
+  remoteStoreApplyPrimaryState,
+  remoteStoreListen,
+  remoteStoreUnlisten
+} from '../remote/remote_store';
 import { debugAssert, debugCast, fail, hardAssert } from '../util/assert';
 import { Code, FirestoreError } from '../util/error';
 import { logDebug } from '../util/log';
@@ -338,7 +345,7 @@ export async function syncEngineListen(
       status === 'current'
     );
     if (syncEngineImpl.isPrimaryClient) {
-      syncEngineImpl.remoteStore.listen(targetData);
+      remoteStoreListen(syncEngineImpl.remoteStore, targetData);
     }
   }
 
@@ -439,7 +446,7 @@ export async function syncEngineUnlisten(
       )
         .then(() => {
           syncEngineImpl.sharedClientState.clearQueryState(queryView.targetId);
-          syncEngineImpl.remoteStore.unlisten(queryView.targetId);
+          remoteStoreUnlisten(syncEngineImpl.remoteStore, queryView.targetId);
           removeAndCleanupTarget(syncEngineImpl, queryView.targetId);
         })
         .catch(ignoreIfPrimaryLeaseLoss);
@@ -477,7 +484,7 @@ export async function syncEngineWrite(
     syncEngineImpl.sharedClientState.addPendingMutation(result.batchId);
     addMutationCallback(syncEngineImpl, result.batchId, userCallback);
     await emitNewSnapsAndNotifyLocalStore(syncEngineImpl, result.changes);
-    await syncEngineImpl.remoteStore.fillWritePipeline();
+    await fillWritePipeline(syncEngineImpl.remoteStore);
   } catch (e) {
     // If we can't persist the mutation, we reject the user callback and
     // don't send the mutation. The user can then retry the write.
@@ -725,7 +732,7 @@ export async function registerPendingWritesCallback(
   callback: Deferred<void>
 ): Promise<void> {
   const syncEngineImpl = debugCast(syncEngine, SyncEngineImpl);
-  if (!syncEngineImpl.remoteStore.canUseNetwork()) {
+  if (!canUseNetwork(syncEngineImpl.remoteStore)) {
     logDebug(
       LOG_TAG,
       'The network is disabled. The task returned by ' +
@@ -888,7 +895,7 @@ function removeLimboTarget(
     return;
   }
 
-  syncEngineImpl.remoteStore.unlisten(limboTargetId);
+  remoteStoreUnlisten(syncEngineImpl.remoteStore, limboTargetId);
   syncEngineImpl.activeLimboTargetsByKey = syncEngineImpl.activeLimboTargetsByKey.remove(
     key
   );
@@ -960,7 +967,8 @@ function pumpEnqueuedLimboResolutions(syncEngineImpl: SyncEngineImpl): void {
       key,
       limboTargetId
     );
-    syncEngineImpl.remoteStore.listen(
+    remoteStoreListen(
+      syncEngineImpl.remoteStore,
       new TargetData(
         queryToTarget(newQueryForPath(key.path)),
         limboTargetId,
@@ -1064,7 +1072,7 @@ async function applyDocChanges(
   return viewChange.snapshot;
 }
 
-export async function handleCredentialChange(
+export async function syncEngineHandleCredentialChange(
   syncEngine: SyncEngine,
   user: User
 ): Promise<void> {
@@ -1181,7 +1189,7 @@ export async function applyBatchState(
     // If we are the primary client, we need to send this write to the
     // backend. Secondary clients will ignore these writes since their remote
     // connection is disabled.
-    await syncEngineImpl.remoteStore.fillWritePipeline();
+    await fillWritePipeline(syncEngineImpl.remoteStore);
   } else if (batchState === 'acknowledged' || batchState === 'rejected') {
     // NOTE: Both these methods are no-ops for batches that originated from
     // other clients.
@@ -1217,9 +1225,9 @@ export async function applyPrimaryState(
       /*transitionToPrimary=*/ true
     );
     syncEngineImpl._isPrimaryClient = true;
-    await syncEngineImpl.remoteStore.applyPrimaryState(true);
+    await remoteStoreApplyPrimaryState(syncEngineImpl.remoteStore, true);
     for (const targetData of activeQueries) {
-      syncEngineImpl.remoteStore.listen(targetData);
+      remoteStoreListen(syncEngineImpl.remoteStore, targetData);
     }
   } else if (isPrimary === false && syncEngineImpl._isPrimaryClient !== false) {
     const activeTargets: TargetId[] = [];
@@ -1238,7 +1246,7 @@ export async function applyPrimaryState(
           );
         });
       }
-      syncEngineImpl.remoteStore.unlisten(targetId);
+      remoteStoreUnlisten(syncEngineImpl.remoteStore, targetId);
     });
     await p;
 
@@ -1249,15 +1257,15 @@ export async function applyPrimaryState(
     );
     resetLimboDocuments(syncEngineImpl);
     syncEngineImpl._isPrimaryClient = false;
-    await syncEngineImpl.remoteStore.applyPrimaryState(false);
+    await remoteStoreApplyPrimaryState(syncEngineImpl.remoteStore, false);
   }
 }
 
 // PORTING NOTE: Multi-Tab only.
 function resetLimboDocuments(syncEngine: SyncEngine): void {
   const syncEngineImpl = debugCast(syncEngine, SyncEngineImpl);
   syncEngineImpl.activeLimboResolutionsByTarget.forEach((_, targetId) => {
-    syncEngineImpl.remoteStore.unlisten(targetId);
+    remoteStoreUnlisten(syncEngineImpl.remoteStore, targetId);
   });
   syncEngineImpl.limboDocumentRefs.removeAllReferences();
   syncEngineImpl.activeLimboResolutionsByTarget = new Map<
@@ -1447,7 +1455,7 @@ export async function applyActiveTargetsChange(
       targetData.targetId,
       /*current=*/ false
     );
-    syncEngineImpl.remoteStore.listen(targetData);
+    remoteStoreListen(syncEngineImpl.remoteStore, targetData);
   }
 
   for (const targetId of removed) {
@@ -1464,7 +1472,7 @@ export async function applyActiveTargetsChange(
       /* keepPersistedTargetData */ false
     )
       .then(() => {
-        syncEngineImpl.remoteStore.unlisten(targetId);
+        remoteStoreUnlisten(syncEngineImpl.remoteStore, targetId);
         removeAndCleanupTarget(syncEngineImpl, targetId);
       })
       .catch(ignoreIfPrimaryLeaseLoss);
@@ -1473,15 +1481,15 @@ export async function applyActiveTargetsChange(
 
 function ensureWatchCallbacks(syncEngine: SyncEngine): SyncEngineImpl {
   const syncEngineImpl = debugCast(syncEngine, SyncEngineImpl);
-  syncEngineImpl.remoteStore.remoteSyncer.applyRemoteEvent = applyRemoteEvent.bind(
+  syncEngineImpl.remoteStore.syncEngine.applyRemoteEvent = applyRemoteEvent.bind(
     null,
     syncEngineImpl
   );
-  syncEngineImpl.remoteStore.remoteSyncer.getRemoteKeysForTarget = getRemoteKeysForTarget.bind(
+  syncEngineImpl.remoteStore.syncEngine.getRemoteKeysForTarget = getRemoteKeysForTarget.bind(
     null,
     syncEngineImpl
   );
-  syncEngineImpl.remoteStore.remoteSyncer.rejectListen = rejectListen.bind(
+  syncEngineImpl.remoteStore.syncEngine.rejectListen = rejectListen.bind(
     null,
     syncEngineImpl
   );
@@ -1490,11 +1498,11 @@ function ensureWatchCallbacks(syncEngine: SyncEngine): SyncEngineImpl {
 
 function ensureWriteCallbacks(syncEngine: SyncEngine): SyncEngineImpl {
   const syncEngineImpl = debugCast(syncEngine, SyncEngineImpl);
-  syncEngineImpl.remoteStore.remoteSyncer.applySuccessfulWrite = applySuccessfulWrite.bind(
+  syncEngineImpl.remoteStore.syncEngine.applySuccessfulWrite = applySuccessfulWrite.bind(
     null,
     syncEngineImpl
   );
-  syncEngineImpl.remoteStore.remoteSyncer.rejectFailedWrite = rejectFailedWrite.bind(
+  syncEngineImpl.remoteStore.syncEngine.rejectFailedWrite = rejectFailedWrite.bind(
     null,
     syncEngineImpl
   );