Fix tests.

Author: taeold

spec/common/providers/https.spec.ts

@@ -803,6 +803,33 @@ describe("onCallHandler", () => {
       expect(resp.body).to.equal([...data, ""].join("\n"));
     });
 
+    it("stops processing when client disconnects", async () => {
+      const mockReq = mockRequest(
+        { message: "test abort" },
+        "application/json",
+        {},
+        { accept: "text/event-stream" }
+      ) as any;
+
+
+      const fn = https.onCallHandler(
+        {
+          cors: { origin: true, methods: "POST" },
+        },
+        async (req, resp) => {
+          resp.write("initial message");
+          mockReq.emit('close');
+          resp.write("should not be sent");
+          return "done";
+        },
+        "gcfv2"
+      );
+
+      const resp = await runHandler(fn, mockReq);
+
+      expect(resp.body).to.equal(`data: {"message":"initial message"}\n`);
+    });
+
     describe("Heartbeats", () => {
       let clock: sinon.SinonFakeTimers;
 
@@ -836,7 +863,7 @@ describe("onCallHandler", () => {
         );
 
         const handlerPromise = runHandler(fn, mockReq as any);
-        clock.tick(11_000);
+        await clock.tickAsync(11_000)
         const resp = await handlerPromise;
         expect(resp.body).to.include(': ping\n: ping\ndata: {"result":"done"}');
       });
@@ -862,7 +889,7 @@ describe("onCallHandler", () => {
         );
 
         const handlerPromise = runHandler(fn, mockReq as any);
-        clock.tick(31_000);
+        await clock.tickAsync(31_000);
         const resp = await handlerPromise;
         expect(resp.body).to.include('data: {"result":"done"}');
       });

spec/fixtures/mockrequest.ts

@@ -1,18 +1,20 @@
+import { EventEmitter } from 'node:stream';
+
 import * as jwt from 'jsonwebtoken';
 import * as jwkToPem from 'jwk-to-pem';
 import * as nock from 'nock';
 import * as mockJWK from '../fixtures/credential/jwk.json';
 import * as mockKey from '../fixtures/credential/key.json';
 
 // MockRequest mocks an https.Request.
-export class MockRequest {
+export class MockRequest extends EventEmitter {
   public method: 'POST' | 'GET' | 'OPTIONS' = 'POST';
 
   constructor(
     readonly body: any,
     readonly headers: { [name: string]: string }
   ) {
-    // This block intentionally left blank.
+    super()
   }
 
   public header(name: string): string {

spec/helper.ts

@@ -52,6 +52,10 @@ export function runHandler(
       private headers: { [name: string]: string } = {};
       private callback: () => void;
 
+      constructor() {
+        request.on("close", () => this.end())
+      }
+
       public status(code: number) {
         this.statusCode = code;
         return this;

src/common/providers/https.ts

@@ -871,29 +871,32 @@ function wrapOnCallHandler<Req = any, Res = any>(
 
       // If there was some result, encode it in the body.
       const responseBody: HttpResponseBody = { result };
+
       if (acceptsStreaming) {
         res.write(encodeSSE(responseBody));
         res.end();
       } else {
         res.status(200).send(responseBody);
       }
     } catch (err) {
-      let httpErr = err;
-      if (!(err instanceof HttpsError)) {
-        // This doesn't count as an 'explicit' error.
-        logger.error("Unhandled error", err);
-        httpErr = new HttpsError("internal", "INTERNAL");
-      }
+      if (!abortController.signal.aborted) {
+        let httpErr = err;
+        if (!(err instanceof HttpsError)) {
+          // This doesn't count as an 'explicit' error.
+          logger.error("Unhandled error", err);
+          httpErr = new HttpsError("internal", "INTERNAL");
+        }
 
-      const { status } = httpErr.httpErrorCode;
-      const body = { error: httpErr.toJSON() };
-      if (req.header("accept") === "text/event-stream") {
-        res.send(encodeSSE(body));
-      } else {
-        res.status(status).send(body);
+        const { status } = httpErr.httpErrorCode;
+        const body = { error: httpErr.toJSON() };
+        if (req.header("accept") === "text/event-stream") {
+          res.send(encodeSSE(body));
+        } else {
+          res.status(status).send(body);
+        }
       }
     } finally {
-      cleanup()
+      cleanup();
     }
   };
 }