fix: request body reads, response piping

Author: kettanaito

src/interceptors/http/index.ts

@@ -1,7 +1,7 @@
 import net from 'node:net'
 import { Readable, Writable } from 'node:stream'
 import { invariant } from 'outvariant'
-import { Interceptor, INTERNAL_REQUEST_ID_HEADER_NAME } from '../../Interceptor'
+import { Interceptor } from '../../Interceptor'
 import { type HttpRequestEventMap } from '../../glossary'
 import { SocketInterceptor } from '../net'
 import { FetchResponse } from '../../utils/fetchUtils'
@@ -81,32 +81,44 @@ export class HttpRequestInterceptor extends Interceptor<HttpRequestEventMap> {
                 controller,
                 emitter: this.emitter,
                 onResponse: async (response) => {
+                  if (this.emitter.listenerCount('response') > 0) {
+                    const responseClone = response.clone()
+                    process.nextTick(() => {
+                      emitAsync(this.emitter, 'response', {
+                        requestId,
+                        request,
+                        response: responseClone,
+                        isMockedResponse: true,
+                      })
+                    })
+                  }
+
                   await respondWith({
                     socket,
                     connectionOptions: options,
                     request,
                     response,
                   })
-                  await emitAsync(this.emitter, 'response', {
-                    requestId,
-                    request,
-                    response,
-                    isMockedResponse: true,
-                  })
                 },
                 async onRequestError(response) {
+                  if (this.emitter.listenerCount('response') > 0) {
+                    const responseClone = response.clone()
+                    process.nextTick(() => {
+                      emitAsync(this.emitter, 'response', {
+                        requestId,
+                        request,
+                        response: responseClone,
+                        isMockedResponse: true,
+                      })
+                    })
+                  }
+
                   await respondWith({
                     socket,
                     connectionOptions: options,
                     request,
                     response,
                   })
-                  await emitAsync(this.emitter, 'response', {
-                    requestId,
-                    request,
-                    response,
-                    isMockedResponse: true,
-                  })
                 },
                 onError(error) {
                   if (error instanceof Error) {
@@ -116,22 +128,6 @@ export class HttpRequestInterceptor extends Interceptor<HttpRequestEventMap> {
               })
 
               if (!isRequestHandled) {
-                // If the user didn't register any response listeners, no need to pay the
-                // price of routing the entire response message through the parser.
-                if (this.emitter.listenerCount('response') > 0) {
-                  createHttpResponseParserStream({
-                    socket,
-                    onResponse: async (response) => {
-                      await emitAsync(this.emitter, 'response', {
-                        requestId,
-                        request,
-                        response,
-                        isMockedResponse: false,
-                      })
-                    },
-                  })
-                }
-
                 const passthroughSocket = socket.passthrough()
 
                 /**
@@ -147,14 +143,39 @@ export class HttpRequestInterceptor extends Interceptor<HttpRequestEventMap> {
                 passthroughSocket.parser = socket.parser
                 // @ts-expect-error Node.js internals.
                 passthroughSocket.parser.socket = passthroughSocket
+
+                // If the user didn't register any response listeners, no need to pay the
+                // price of routing the entire response message through the parser.
+                if (this.emitter.listenerCount('response') > 0) {
+                  const responseStream = createHttpResponseParserStream({
+                    onResponse: async (response) => {
+                      await emitAsync(this.emitter, 'response', {
+                        requestId,
+                        request,
+                        response,
+                        isMockedResponse: false,
+                      })
+                    },
+                  })
+
+                  passthroughSocket
+                    .on('data', (chunk) => responseStream.write(chunk))
+                    .on('close', () => responseStream.end())
+                }
               }
             },
           })
 
           // Write the message header to the parser manually because it's already been written
           // on the socket so it won't get piped.
           requestParser.write(toBuffer(chunk, encoding))
-          socket.pipe(requestParser)
+
+          socket
+            .on('write', (chunk, encoding, callback) => {
+              requestParser.write(chunk, encoding, callback)
+            })
+            .on('finish', () => requestParser.end())
+            .on('error', () => requestParser.end())
         })
       })
     })
@@ -211,13 +232,7 @@ function createHttpRequestParserStream(options: {
          * @note Provide the `read()` method so a `Readable` could be
          * used as the actual request body (the stream calls "read()").
          */
-        read() {
-          // If the user attempts to read the request body,
-          // flush the write buffer to trigger the callbacks.
-          // This way, if the request stream ends in the write callback,
-          // it will indeed end correctly.
-          // flushWriteBuffer()
-        },
+        read() {},
       })
 
       const request = new Request(url, {
@@ -254,25 +269,18 @@ function createHttpRequestParserStream(options: {
     },
   })
 
-  const parserStream = new Writable({
+  return new Writable({
     write(chunk, encoding, callback) {
       parser.execute(toBuffer(chunk, encoding))
       callback()
     },
   })
-
-  parserStream
-    .once('finish', () => parser.free())
-    .once('close', () => parser.free())
-
-  return parserStream
 }
 
 function createHttpResponseParserStream(options: {
-  socket: MockSocket
   onResponse: (response: Response) => void
 }) {
-  const { socket, onResponse } = options
+  const { onResponse } = options
   const responseRawHeadersBuffer: Array<string> = []
   let responseBodyStream: Readable | undefined
 
@@ -323,11 +331,12 @@ function createHttpResponseParserStream(options: {
     },
   })
 
-  socket
-    .on('push', (chunk, encoding) => {
+  return new Writable({
+    write(chunk, encoding, callback) {
       parser.execute(toBuffer(chunk, encoding))
-    })
-    .once('close', () => parser.free())
+      callback()
+    },
+  })
 }
 
 /**

test/modules/http/compliance/events.test.ts

@@ -9,6 +9,9 @@ const httpServer = new HttpServer((app) => {
   app.get('/', (req, res) => {
     res.send('original-response')
   })
+  app.post('/', (req, res) => {
+    res.send()
+  })
 })
 
 const interceptor = new HttpRequestInterceptor()
@@ -18,6 +21,10 @@ beforeAll(async () => {
   await httpServer.listen()
 })
 
+afterEach(() => {
+  interceptor.removeAllListeners()
+})
+
 afterAll(async () => {
   interceptor.dispose()
   await httpServer.close()
@@ -48,10 +55,10 @@ it('emits the "request" event for an outgoing request without body', async () =>
   expect(request.body).toBe(null)
 })
 
-it('emits the "request" event for an outgoing request with a body', async () => {
+it('emits the "request" event for a bypassed request with a body', async () => {
   const requestListener =
     vi.fn<(...args: HttpRequestEventMap['request']) => void>()
-  interceptor.once('request', requestListener)
+  interceptor.on('request', requestListener)
 
   const request = http.request(httpServer.http.url('/'), {
     method: 'POST',
@@ -62,21 +69,22 @@ it('emits the "request" event for an outgoing request with a body', async () =>
   })
   request.write('post-payload')
   request.end()
+
   await waitForClientRequest(request)
 
   expect(requestListener).toHaveBeenCalledTimes(1)
 
-  const { request: requestFromListener } = requestListener.mock.calls[0][0]
-  expect(requestFromListener).toBeInstanceOf(Request)
-  expect(requestFromListener.method).toBe('POST')
-  expect(requestFromListener.url).toBe(httpServer.http.url('/'))
+  const { request: interceptedRequest } = requestListener.mock.calls[0][0]
+  expect(interceptedRequest).toBeInstanceOf(Request)
+  expect(interceptedRequest.method).toBe('POST')
+  expect(interceptedRequest.url).toBe(httpServer.http.url('/'))
   expect(
-    Object.fromEntries(requestFromListener.headers.entries())
+    Object.fromEntries(interceptedRequest.headers.entries())
   ).toMatchObject({
     'content-type': 'text/plain',
     'x-custom-header': 'yes',
   })
-  expect(await requestFromListener.text()).toBe('post-payload')
+  await expect(interceptedRequest.text()).resolves.toBe('post-payload')
 })
 
 it('emits the "response" event for a mocked response', async () => {
@@ -99,28 +107,28 @@ it('emits the "response" event for a mocked response', async () => {
   const {
     response,
     requestId,
-    request: requestFromListener,
+    request: interceptedRequest,
     isMockedResponse,
   } = responseListener.mock.calls[0][0]
   expect(response).toBeInstanceOf(Response)
   expect(response.status).toBe(200)
-  expect(await response.text()).toBe('hello world')
+  await expect(response.text()).resolves.toBe('hello world')
   expect(isMockedResponse).toBe(true)
 
   expect(requestId).toMatch(REQUEST_ID_REGEXP)
-  expect(requestFromListener).toBeInstanceOf(Request)
-  expect(requestFromListener.method).toBe('GET')
-  expect(requestFromListener.url).toBe('http://localhost/')
+  expect(interceptedRequest).toBeInstanceOf(Request)
+  expect(interceptedRequest.method).toBe('GET')
+  expect(interceptedRequest.url).toBe('http://localhost/')
   expect(
-    Object.fromEntries(requestFromListener.headers.entries())
+    Object.fromEntries(interceptedRequest.headers.entries())
   ).toMatchObject({
     'x-custom-header': 'yes',
   })
-  expect(requestFromListener.body).toBe(null)
+  expect(interceptedRequest.body).toBe(null)
 
   // Must respond with the mocked response.
   expect(res.statusCode).toBe(200)
-  expect(await text()).toBe('hello world')
+  await expect(text()).resolves.toBe('hello world')
 })
 
 it('emits the "response" event for a bypassed response', async () => {
@@ -135,31 +143,35 @@ it('emits the "response" event for a bypassed response', async () => {
   })
   const { res, text } = await waitForClientRequest(request)
 
-  // Must emit the "response" interceptor event.
-  expect(responseListener).toHaveBeenCalledTimes(1)
+  // Must respond with the mocked response.
+  expect.soft(res.statusCode).toBe(200)
+  await expect.soft(text()).resolves.toBe('original-response')
+
+  expect(
+    responseListener,
+    'Must emit the "response" event'
+  ).toHaveBeenCalledTimes(1)
+
   const {
     response,
     requestId,
-    request: requestFromListener,
+    request: interceptedRequest,
     isMockedResponse,
   } = responseListener.mock.calls[0][0]
+
   expect(response).toBeInstanceOf(Response)
   expect(response.status).toBe(200)
-  expect(await response.text()).toBe('original-response')
+  await expect(response.text()).resolves.toBe('original-response')
   expect(isMockedResponse).toBe(false)
 
   expect(requestId).toMatch(REQUEST_ID_REGEXP)
-  expect(requestFromListener).toBeInstanceOf(Request)
-  expect(requestFromListener.method).toBe('GET')
-  expect(requestFromListener.url).toBe(httpServer.http.url('/'))
+  expect(interceptedRequest).toBeInstanceOf(Request)
+  expect(interceptedRequest.method).toBe('GET')
+  expect(interceptedRequest.url).toBe(httpServer.http.url('/'))
   expect(
-    Object.fromEntries(requestFromListener.headers.entries())
+    Object.fromEntries(interceptedRequest.headers.entries())
   ).toMatchObject({
     'x-custom-header': 'yes',
   })
-  expect(requestFromListener.body).toBe(null)
-
-  // Must respond with the mocked response.
-  expect(res.statusCode).toBe(200)
-  expect(await text()).toBe('original-response')
+  expect(interceptedRequest.body).toBe(null)
 })