Support logical termination for grpc connection

Author: cherylEnkidu

packages/firestore/src/platform/node/grpc_connection.ts

@@ -153,11 +153,11 @@ export class GrpcConnection implements Connection {
     path: string,
     request: Req,
     authToken: Token | null,
-    appCheckToken: Token | null
+    appCheckToken: Token | null,
+    expected_response_count_?: number
   ): Promise<Resp[]> {
     const results: Resp[] = [];
     const responseDeferred = new Deferred<Resp[]>();
-
     logDebug(
       LOG_TAG,
       `RPC '${rpcName}' invoked (streaming) with request:`,
@@ -172,13 +172,24 @@ export class GrpcConnection implements Connection {
     );
     const jsonRequest = { ...request, database: this.databasePath };
     const stream = stub[rpcName](jsonRequest, metadata);
+    let callback_fired_: Boolean = false;
     stream.on('data', (response: Resp) => {
       logDebug(LOG_TAG, `RPC ${rpcName} received result:`, response);
       results.push(response);
+      if (
+        typeof expected_response_count_ !== 'undefined' &&
+        results.length === expected_response_count_
+      ) {
+        callback_fired_ = true;
+        responseDeferred.resolve(results);
+      }
     });
     stream.on('end', () => {
       logDebug(LOG_TAG, `RPC '${rpcName}' completed.`);
-      responseDeferred.resolve(results);
+      if (!callback_fired_) {
+        callback_fired_ = true;
+        responseDeferred.resolve(results);
+      }
     });
     stream.on('error', (grpcError: grpc.ServiceError) => {
       logDebug(LOG_TAG, `RPC '${rpcName}' failed with error:`, grpcError);

packages/firestore/src/remote/connection.ts

@@ -68,7 +68,8 @@ export interface Connection {
     path: string,
     request: Req,
     authToken: Token | null,
-    appCheckToken: Token | null
+    appCheckToken: Token | null,
+    expected_response_count_?: number
   ): Promise<Resp[]>;
 
   /**

packages/firestore/src/remote/rest_connection.ts

@@ -104,7 +104,8 @@ export abstract class RestConnection implements Connection {
     path: string,
     request: Req,
     authToken: Token | null,
-    appCheckToken: Token | null
+    appCheckToken: Token | null,
+    expected_response_count_?: number
   ): Promise<Resp[]> {
     // The REST API automatically aggregates all of the streamed results, so we
     // can just use the normal invoke() method.