feat(ClientRequest): await response listeners before emitting the "end" response event

src/interceptors/ClientRequest/NodeClientRequest.ts

@@ -367,32 +367,47 @@ export class NodeClientRequest extends ClientRequest {
         callback?.()
 
         this.logger.info('emitting the custom "response" event...')
-        this.emitter.emit('response', {
+
+        const responseListenersPromise = emitAsync(this.emitter, 'response', {
           response: responseClone,
           isMockedResponse: true,
           request: capturedRequest,
           requestId,
         })
 
-        this.logger.info('request (mock) is completed')
+        responseListenersPromise.then(() => {
+          this.logger.info('request (mock) is completed')
+        })
+
+        // Defer the end of the response until all the response
+        // event listeners are done (those can be async).
+        this.deferResponseEndUntil(responseListenersPromise, this.response)
 
         return this
       }
 
       this.logger.info('no mocked response received!')
 
-      this.once('response-internal', (message: IncomingMessage) => {
-        this.logger.info(message.statusCode, message.statusMessage)
-        this.logger.info('original response headers:', message.headers)
+      this.once(
+        'response-internal',
+        (message: IncomingMessage, originalMessage: IncomingMessage) => {
+          this.logger.info(message.statusCode, message.statusMessage)
+          this.logger.info('original response headers:', message.headers)
 
-        this.logger.info('emitting the custom "response" event...')
-        this.emitter.emit('response', {
-          response: createResponse(message),
-          isMockedResponse: false,
-          request: capturedRequest,
-          requestId,
-        })
-      })
+          this.logger.info('emitting the custom "response" event...')
+
+          const responseListenersPromise = emitAsync(this.emitter, 'response', {
+            response: createResponse(message),
+            isMockedResponse: false,
+            request: capturedRequest,
+            requestId,
+          })
+
+          // Defer the end of the response until all the response
+          // event listeners are done (those can be async).
+          this.deferResponseEndUntil(responseListenersPromise, originalMessage)
+        }
+      )
 
       return this.passthrough(chunk, encoding, callback)
     })
@@ -419,7 +434,7 @@ export class NodeClientRequest extends ClientRequest {
         const firstClone = cloneIncomingMessage(response)
         const secondClone = cloneIncomingMessage(response)
 
-        this.emit('response-internal', secondClone)
+        this.emit('response-internal', secondClone, firstClone)
 
         this.logger.info(
           'response successfully cloned, emitting "response" event...'
@@ -643,4 +658,23 @@ export class NodeClientRequest extends ClientRequest {
     // @ts-ignore "agent" is a private property.
     this.agent?.destroy?.()
   }
+
+  private deferResponseEndUntil(
+    promise: Promise<unknown>,
+    response: IncomingMessage
+  ): void {
+    response.emit = new Proxy(response.emit, {
+      apply(target, thisArg, args) {
+        const [event] = args
+        const callEmit = () => Reflect.apply(target, thisArg, args)
+
+        if (event === 'end') {
+          promise.then(() => callEmit())
+          return
+        }
+
+        return callEmit()
+      },
+    })
+  }
 }

test/modules/http/response/http-await-response-event.test.ts

@@ -0,0 +1,61 @@
+import { vi, it, expect, beforeAll, afterAll, afterEach } from 'vitest'
+import http from 'node:http'
+import { HttpServer } from '@open-draft/test-server/http'
+import { ClientRequestInterceptor } from '../../../../src/interceptors/ClientRequest'
+import { sleep, waitForClientRequest } from '../../../helpers'
+
+const httpServer = new HttpServer((app) => {
+  app.get('/resource', (req, res) => {
+    res.send('original response')
+  })
+})
+
+const interceptor = new ClientRequestInterceptor()
+
+beforeAll(async () => {
+  interceptor.apply()
+  await httpServer.listen()
+})
+
+afterEach(() => {
+  interceptor.removeAllListeners()
+})
+
+afterAll(async () => {
+  interceptor.dispose()
+  await httpServer.close()
+})
+
+it('awaits asynchronous response event listener for a mocked response', async () => {
+  interceptor.on('request', ({ request }) => {
+    request.respondWith(new Response('hello world'))
+  })
+
+  const responseDone = vi.fn()
+  interceptor.on('response', async ({ response }) => {
+    await sleep(200)
+    const text = await response.text()
+    responseDone(text)
+  })
+
+  const request = http.get('http://localhost/')
+  const { text } = await waitForClientRequest(request)
+
+  expect(await text()).toBe('hello world')
+  expect(responseDone).toHaveBeenCalledWith('hello world')
+})
+
+it('awaits asynchronous response event listener for the original response', async () => {
+  const responseDone = vi.fn()
+  interceptor.on('response', async ({ response }) => {
+    await sleep(200)
+    const text = await response.text()
+    responseDone(text)
+  })
+
+  const request = http.get(httpServer.http.url('/resource'))
+  const { text } = await waitForClientRequest(request)
+
+  expect(await text()).toBe('original response')
+  expect(responseDone).toHaveBeenCalledWith('original response')
+})