Merging Persistent Stream refactor

packages/firestore/src/remote/datastore.ts

@@ -1,4 +1,3 @@
-import { WatchStreamListener, WriteStreamListener } from './persistent_stream';
 /**
  * Copyright 2017 Google Inc.
  *
@@ -24,7 +23,7 @@ import { Mutation, MutationResult } from '../model/mutation';
 import { assert } from '../util/assert';
 import { Code, FirestoreError } from '../util/error';
 import { AsyncQueue } from '../util/async_queue';
-
+import { WatchStreamListener, WriteStreamListener } from './persistent_stream';
 import { Connection } from './connection';
 import {
   PersistentListenStream,

packages/firestore/src/remote/remote_store.ts

@@ -76,9 +76,6 @@ const MAX_PENDING_WRITES = 10;
  * - pulling pending mutations from LocalStore and sending them to Datastore.
  * - retrying mutations that failed because of network problems.
  * - acking mutations to the SyncEngine once they are accepted or rejected.
- *
- * RemoteStore always starts out offline. A call to `enableNetwork()`
- * initializes the network connection.
  */
 export class RemoteStore implements TargetMetadataProvider {
   /**
@@ -111,19 +108,16 @@ export class RemoteStore implements TargetMetadataProvider {
    */
   private listenTargets: { [targetId: number]: QueryData } = {};
 
-  private networkEnabled = false;
-
   private watchStream: PersistentListenStream;
   private writeStream: PersistentWriteStream;
   private watchChangeAggregator: WatchChangeAggregator = null;
 
   /**
    * Set to true by enableNetwork() and false by disableNetwork() and indicates
-   * the user-preferred network state. A network connection is only established
-   * if `networkAllowed` is true, the client is primary and there are
-   * outstanding mutations or active listens.
+   * the user-preferred network state.
    */
-  private networkAllowed = true;
+  private networkEnabled = false;
+
   private isPrimary = false;
 
   private onlineStateTracker: OnlineStateTracker;
@@ -165,60 +159,25 @@ export class RemoteStore implements TargetMetadataProvider {
    * Starts up the remote store, creating streams, restoring state from
    * LocalStore, etc.
    */
-<<<<<<< HEAD
   start(): Promise<void> {
-    // Start is a no-op for RemoteStore.
-    return Promise.resolve();
-  }
-
-  private isNetworkEnabled(): boolean {
-    assert(
-      (this.watchStream == null) === (this.writeStream == null),
-      'WatchStream and WriteStream should both be null or non-null'
-    );
-    return this.watchStream != null;
-=======
-  async start(): Promise<void> {
-    await this.enableNetwork();
->>>>>>> master
+    return this.enableNetwork();
   }
 
   /** Re-enables the network. Idempotent. */
   async enableNetwork(): Promise<void> {
-<<<<<<< HEAD
-    this.networkAllowed = true;
+    this.networkEnabled = true;
 
-    if (this.isPrimary) {
-      if (this.isNetworkEnabled()) {
-        return;
-      }
-
-      // Create new streams (but note they're not started yet).
-      this.watchStream = this.datastore.newPersistentWatchStream();
-      this.writeStream = this.datastore.newPersistentWriteStream();
-=======
-    if (!this.networkEnabled) {
-      this.networkEnabled = true;
+    if (this.canUseNetwork()) {
       this.writeStream.lastStreamToken = await this.localStore.getLastStreamToken();
->>>>>>> master
-
-      // Load any saved stream token from persistent storage
-      return this.localStore.getLastStreamToken().then(token => {
-        this.writeStream.lastStreamToken = token;
 
-<<<<<<< HEAD
-        if (this.shouldStartWatchStream()) {
-          this.startWatchStream();
-        } else {
-          this.onlineStateTracker.set(OnlineState.Unknown);
-        }
+      if (this.shouldStartWatchStream()) {
+        this.startWatchStream();
+      } else {
+        this.onlineStateTracker.set(OnlineState.Unknown);
+      }
 
-        return this.fillWritePipeline(); // This may start the writeStream.
-      });
-=======
       // This will start the write stream if necessary.
       await this.fillWritePipeline();
->>>>>>> master
     }
   }
 
@@ -227,41 +186,31 @@ export class RemoteStore implements TargetMetadataProvider {
    * enableNetwork().
    */
   async disableNetwork(): Promise<void> {
-<<<<<<< HEAD
-    this.networkAllowed = false;
-
+    this.networkEnabled = false;
     this.disableNetworkInternal();
-=======
-    await this.disableNetworkInternal();
 
->>>>>>> master
     // Set the OnlineState to Offline so get()s return from cache, etc.
     this.onlineStateTracker.set(OnlineState.Offline);
   }
 
-  private async disableNetworkInternal(): Promise<void> {
-    if (this.networkEnabled) {
-      this.networkEnabled = false;
-
-      this.writeStream.stop();
-      this.watchStream.stop();
-
-      if (this.writePipeline.length > 0) {
-        log.debug(
-          LOG_TAG,
-          `Stopping write stream with ${
-            this.writePipeline.length
-          } pending writes`
-        );
-        this.writePipeline = [];
-      }
+  private disableNetworkInternal(): void {
+    this.writeStream.stop();
+    this.watchStream.stop();
 
-      this.cleanUpWatchStreamState();
+    if (this.writePipeline.length > 0) {
+      log.debug(
+        LOG_TAG,
+        `Stopping write stream with ${this.writePipeline.length} pending writes`
+      );
+      this.writePipeline = [];
     }
+
+    this.cleanUpWatchStreamState();
   }
 
   shutdown(): Promise<void> {
     log.debug(LOG_TAG, 'RemoteStore shutting down.');
+    this.networkEnabled = false;
     this.disableNetworkInternal();
 
     // Set the OnlineState to Unknown (rather than Offline) to avoid potentially
@@ -355,9 +304,7 @@ export class RemoteStore implements TargetMetadataProvider {
   }
 
   private canUseNetwork(): boolean {
-    // TODO(mikelehen): This could take into account isPrimary when we merge
-    // with multitab.
-    return this.networkEnabled;
+    return this.isPrimary && this.networkEnabled;
   }
 
   private cleanUpWatchStreamState(): void {
@@ -555,7 +502,7 @@ export class RemoteStore implements TargetMetadataProvider {
    */
   private canAddToWritePipeline(): boolean {
     return (
-      this.networkEnabled && this.writePipeline.length < MAX_PENDING_WRITES
+      this.canUseNetwork() && this.writePipeline.length < MAX_PENDING_WRITES
     );
   }
 
@@ -739,10 +686,11 @@ export class RemoteStore implements TargetMetadataProvider {
   async handleUserChange(user: User): Promise<void> {
     log.debug(LOG_TAG, 'RemoteStore changing users: uid=', user.uid);
 
-    if (this.networkEnabled) {
+    if (this.canUseNetwork()) {
       // Tear down and re-create our network streams. This will ensure we get a fresh auth token
       // for the new user and re-fill the write pipeline with new mutations from the LocalStore
       // (since mutations are per-user).
+      this.networkEnabled = false;
       this.disableNetworkInternal();
       this.onlineStateTracker.set(OnlineState.Unknown);
       await this.enableNetwork();
@@ -755,9 +703,9 @@ export class RemoteStore implements TargetMetadataProvider {
   async applyPrimaryState(isPrimary: boolean): Promise<void> {
     this.isPrimary = isPrimary;
 
-    if (isPrimary && this.networkAllowed) {
+    if (isPrimary && this.networkEnabled) {
       await this.enableNetwork();
-    } else if (!isPrimary && this.isNetworkEnabled()) {
+    } else if (!isPrimary) {
       this.disableNetworkInternal();
       this.onlineStateTracker.set(OnlineState.Unknown);
     }

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

@@ -494,17 +494,11 @@ abstract class TestRunner {
   }
 
   async shutdown(): Promise<void> {
-<<<<<<< HEAD
-    if (this.started) {
-      await this.doShutdown();
-    }
-=======
     await this.queue.enqueue(async () => {
-      await this.remoteStore.shutdown();
-      await this.persistence.shutdown(/* deleteData= */ true);
-      await this.destroyPersistence();
+      if (this.started) {
+        await this.doShutdown();
+      }
     });
->>>>>>> master
   }
 
   /** Runs a single SpecStep on this runner. */

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

@@ -1245,4 +1245,30 @@ describeSpec('Writes:', [], () => {
       );
     }
   );
+
+  specTest(
+    'Mutation are not sent twice after primary failover',
+    ['multi-client'],
+    () => {
+      const query = Query.atPath(path('collection'));
+      const docA = doc('collection/a', 0, { k: 'a' });
+      const docB = doc('collection/b', 0, { k: 'b' });
+
+      return client(0)
+        .expectPrimaryState(true)
+        .userSets('collection/a', { k: 'a' })
+        .userSets('collection/b', { k: 'b' })
+        .client(1)
+        .stealPrimaryLease()
+        .writeAcks('collection/a', 1000, { expectUserCallback: false })
+        .client(0)
+        .expectUserCallbacks({
+          acknowledged: ['collection/a']
+        })
+        .stealPrimaryLease()
+        .writeAcks('collection/b', 2000)
+        .userListens(query)
+        .expectEvents(query, { added: [docA, docB], fromCache: true });
+    }
+  );
 });