WIP

Author: schmidt-sebastian

packages/firestore/exp/dependencies.json

@@ -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,11 @@ export function enableIndexedDbPersistence(
       forceOwningTab: false
     },
     new IndexedDbOfflineComponentProvider()
-  );
+  )
+    .then(() =>
+      setOnlineComponentProvider(firestoreImpl, new OnlineComponentProvider())
+    )
+    .then(() => fillWritePipeline(onlineComponentProvider.remoteStore));
 }
 
 export function enableMultiTabIndexedDbPersistence(
@@ -238,9 +244,11 @@ export function enableMultiTabIndexedDbPersistence(
       forceOwningTab: false
     },
     offlineComponentProvider
-  ).then(() =>
-    setOnlineComponentProvider(firestoreImpl, onlineComponentProvider)
-  );
+  )
+    .then(() =>
+      setOnlineComponentProvider(firestoreImpl, onlineComponentProvider)
+    )
+    .then(() => fillWritePipeline(onlineComponentProvider.remoteStore));
 }
 
 export function clearIndexedDbPersistence(

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

@@ -37,7 +37,7 @@ import {
   newSyncEngine,
   SyncEngine
 } from './sync_engine';
-import { RemoteStore } from '../remote/remote_store';
+import { newRemoteStore, RemoteStore } from '../remote/remote_store';
 import { EventManager } from './event_manager';
 import { AsyncQueue } from '../util/async_queue';
 import { DatabaseId, DatabaseInfo } from './database_info';
@@ -361,7 +361,7 @@ export class OnlineComponentProvider {
   }
 
   createRemoteStore(cfg: ComponentConfiguration): RemoteStore {
-    return new RemoteStore(
+    return newRemoteStore(
       this.localStore,
       this.datastore,
       cfg.asyncQueue,

packages/firestore/src/core/component_provider.ts

@@ -28,7 +28,7 @@ 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, RemoteStore } from '../remote/remote_store';
 import { AsyncQueue, wrapInUserErrorIfRecoverable } from '../util/async_queue';
 import { Code, FirestoreError } from '../util/error';
 import { logDebug } from '../util/log';
@@ -289,6 +289,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

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,
@@ -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
   );