feat: add ability to close channels (#824)

dev/src/index.ts

@@ -1007,6 +1007,15 @@ export class Firestore {
       });
   }
 
+  /**
+   * Terminates the Firestore client and closes all open streams.
+   *
+   * @return A Promise that resolves when the client is terminated.
+   */
+  terminate(): Promise<void> {
+    return this._clientPool.terminate();
+  }
+
   /**
    * Initializes the client if it is not already initialized. All methods in the
    * SDK can be used after this method completes.

dev/src/pool.ts

@@ -35,6 +35,12 @@ export class ClientPool<T> {
    */
   private activeClients: Map<T, number> = new Map();
 
+  /**
+   * Whether the Firestore instance has been terminated. Once terminated, the
+   * ClientPool can longer schedule new operations.
+   */
+  private terminated = false;
+
   /**
    * @param concurrentOperationLimit The number of operations that each client
    * can handle.
@@ -148,6 +154,9 @@ export class ClientPool<T> {
    * @private
    */
   run<V>(requestTag: string, op: (client: T) => Promise<V>): Promise<V> {
+    if (this.terminated) {
+      throw new Error('The client has already been terminated');
+    }
     const client = this.acquire(requestTag);
 
     return op(client)
@@ -161,6 +170,14 @@ export class ClientPool<T> {
       });
   }
 
+  async terminate(): Promise<void> {
+    this.terminated = true;
+    for (const [client, _requestCount] of this.activeClients) {
+      this.activeClients.delete(client);
+      await this.clientDestructor(client);
+    }
+  }
+
   /**
    * Deletes clients that are no longer executing operations. Keeps up to one
    * idle client to reduce future initialization costs.

dev/system-test/firestore.ts

@@ -127,6 +127,24 @@ describe('Firestore class', () => {
         expect(docs[1].data()).to.deep.equal({f: 'a'});
       });
   });
+
+  it('cannot make calls after the client has been terminated', () => {
+    const ref1 = randomCol.doc('doc1');
+    ref1.onSnapshot(snapshot => {
+      return Promise.reject('onSnapshot() should be called');
+    });
+    return firestore
+      .terminate()
+      .then(() => {
+        return ref1.set({foo: 100});
+      })
+      .then(() => {
+        Promise.reject('set() should have failed');
+      })
+      .catch(err => {
+        expect(err.message).to.equal('The client has already been terminated');
+      });
+  });
 });
 
 describe('CollectionReference class', () => {

dev/test/pool.ts

@@ -191,4 +191,21 @@ describe('Client pool', () => {
     );
     return expect(op).to.eventually.be.rejectedWith('Generated error');
   });
+
+  it('rejects subsequent operations after being terminated', () => {
+    const clientPool = new ClientPool<{}>(1, () => {
+      return {};
+    });
+
+    return clientPool
+      .terminate()
+      .then(() => {
+        clientPool.run(REQUEST_TAG, () =>
+          Promise.reject('Call to run() should have failed')
+        );
+      })
+      .catch((err: Error) => {
+        expect(err.message).to.equal('The client has already been terminated');
+      });
+  });
 });

types/firestore.d.ts

@@ -154,6 +154,13 @@ declare namespace FirebaseFirestore {
     getAll(...documentRefsOrReadOptions: Array<DocumentReference|ReadOptions>):
         Promise<DocumentSnapshot[]>;
 
+    /**
+     * Terminates the Firestore client and closes all open streams.
+     *
+     * @return A Promise that resolves when the client is terminated.
+     */
+    terminate(): Promise<void>;
+
     /**
      * Fetches the root collections that are associated with this Firestore
      * database.