Closing the write and the watch stream after 60s of idleness

packages/firestore/src/api/database.ts

@@ -188,6 +188,7 @@ export class Firestore implements firestore.Firestore, FirebaseService {
   // Operations on the _firestoreClient don't block on _firestoreReady. Those
   // are already set to synchronize on the async queue.
   private _firestoreClient: FirestoreClient | undefined;
+  private _queue = new AsyncQueue();
   public _dataConverter: UserDataConverter;
 
   constructor(databaseIdOrApp: FirestoreDatabase | FirebaseApp) {
@@ -310,7 +311,7 @@ export class Firestore implements firestore.Firestore, FirebaseService {
       PlatformSupport.getPlatform(),
       databaseInfo,
       this._config.credentials,
-      new AsyncQueue()
+      this._queue
     );
     return this._firestoreClient.start(persistence);
   }
@@ -374,7 +375,9 @@ export class Firestore implements firestore.Firestore, FirebaseService {
     },
     // Exposed via INTERNAL for use in tests.
     disableNetwork: () => this._firestoreClient.disableNetwork(),
-    enableNetwork: () => this._firestoreClient.enableNetwork()
+    enableNetwork: () => this._firestoreClient.enableNetwork(),
+    drainAsyncQueue: (executeDelayedTasks: boolean) =>
+      this._queue.drain(executeDelayedTasks)
   };
 
   collection(pathString: string): firestore.CollectionReference {

packages/firestore/src/local/local_store.ts

@@ -582,9 +582,9 @@ export class LocalStore {
 
   /**
    * Gets the mutation batch after the passed in batchId in the mutation queue
-   *  or null if empty.
-   *  @param afterBatchId If provided, the batch to search after.
-   *  @returns The next mutation or null if there wasn't one.
+   * or null if empty.
+   * @param afterBatchId If provided, the batch to search after.
+   * @returns The next mutation or null if there wasn't one.
    */
   nextMutationBatch(afterBatchId?: BatchId): Promise<MutationBatch | null> {
     return this.persistence.runTransaction('Get next mutation batch', txn => {

packages/firestore/src/remote/datastore.ts

@@ -58,30 +58,24 @@ export class Datastore {
     private initialBackoffDelay?: number
   ) {}
 
-  public newPersistentWriteStream(
-    listener: WriteStreamListener
-  ): PersistentWriteStream {
+  public newPersistentWriteStream(): PersistentWriteStream {
     return new PersistentWriteStream(
       this.databaseInfo,
       this.queue,
       this.connection,
       this.credentials,
       this.serializer,
-      listener,
       this.initialBackoffDelay
     );
   }
 
-  public newPersistentWatchStream(
-    listener: WatchStreamListener
-  ): PersistentListenStream {
+  public newPersistentWatchStream(): PersistentListenStream {
     return new PersistentListenStream(
       this.databaseInfo,
       this.queue,
       this.connection,
       this.credentials,
       this.serializer,
-      listener,
       this.initialBackoffDelay
     );
   }