fix: pausing recorder, handling passthrough http

Author: kettanaito

src/interceptors/http/index.ts

@@ -37,70 +37,85 @@ export class HttpRequestInterceptor extends Interceptor<HttpRequestEventMap> {
     })
 
     socketInterceptor.on('connection', ({ options, socket }) => {
-      socket.once('write', (chunk, encoding) => {
-        const firstFrame = chunk.toString()
+      socket.runInternally(() => {
+        socket.once('write', (chunk, encoding) => {
+          const firstFrame = chunk.toString()
 
-        if (!firstFrame.includes('HTTP/')) {
-          return
-        }
-
-        // Get the request method from the first frame because it's faster
-        // and we need this before initiating the HTTP parser.
-        const method = firstFrame.split(' ')[0]
+          if (!firstFrame.includes('HTTP/')) {
+            return
+          }
 
-        invariant(
-          method != null,
-          'Failed to handle HTTP request: expected a valid HTTP method but got %s',
-          method,
-          options
-        )
+          // Get the request method from the first frame because it's faster
+          // and we need this before initiating the HTTP parser.
+          const method = firstFrame.split(' ')[0]
 
-        const requestParser = createHttpRequestParserStream({
-          requestOptions: {
+          invariant(
+            method != null,
+            'Failed to handle HTTP request: expected a valid HTTP method but got %s',
             method,
-            ...options,
-          },
-          onRequest: async (request) => {
-            const requestId = createRequestId()
-            const controller = new RequestController(request)
-
-            const isRequestHandled = await handleRequest({
-              request,
-              requestId,
-              controller,
-              emitter: this.emitter,
-              async onResponse(response) {
-                await respondWith({
-                  socket,
-                  connectionOptions: options,
-                  request,
-                  response,
-                })
-              },
-              async onRequestError(response) {
-                await respondWith({
-                  socket,
-                  connectionOptions: options,
-                  request,
-                  response,
-                })
-              },
-              onError(error) {
-                if (error instanceof Error) {
-                  socket.destroy(error)
-                }
-              },
-            })
-
-            if (!isRequestHandled) {
-              socket.passthrough()
-            }
-          },
+            options
+          )
+
+          const requestParser = createHttpRequestParserStream({
+            requestOptions: {
+              method,
+              ...options,
+            },
+            onRequest: async (request) => {
+              const requestId = createRequestId()
+              const controller = new RequestController(request)
+
+              const isRequestHandled = await handleRequest({
+                request,
+                requestId,
+                controller,
+                emitter: this.emitter,
+                async onResponse(response) {
+                  await respondWith({
+                    socket,
+                    connectionOptions: options,
+                    request,
+                    response,
+                  })
+                },
+                async onRequestError(response) {
+                  await respondWith({
+                    socket,
+                    connectionOptions: options,
+                    request,
+                    response,
+                  })
+                },
+                onError(error) {
+                  if (error instanceof Error) {
+                    socket.destroy(error)
+                  }
+                },
+              })
+
+              if (!isRequestHandled) {
+                const passthroughSocket = socket.passthrough()
+
+                /**
+                 * @note Creating a passthroughsocket does NOT trigger the "socket" event
+                 * from `http.ClientRequest` where the request, parser, and socket get
+                 * associated. Recreate that association on the passthrough socket manually.
+                 * @see https://github.com/nodejs/node/blob/134625d76139b4b3630d5baaf2efccae01ede564/lib/_http_client.js#L890
+                 */
+                // @ts-expect-error Internal Node.js property.
+                passthroughSocket._httpMessage = socket._httpMessage
+                // @ts-expect-error Internal Node.js property.c
+                passthroughSocket.parser = socket.parser
+                // @ts-expect-error Internal Node.js property.
+                passthroughSocket.parser.socket = passthroughSocket
+              }
+            },
+          })
+
+          // Write the header again because at this point it's already been written.
+          requestParser.write(toBuffer(chunk, encoding))
+          socket.pipe(requestParser)
         })
-
-        // Write the header again because at this point it's already been written.
-        requestParser.write(toBuffer(chunk, encoding))
-        socket.pipe(requestParser)
       })
     })
   }

src/interceptors/net/mock-socket.ts

@@ -34,7 +34,7 @@ export class MockSocket extends net.Socket {
       onEntry: (entry) => {
         if (
           entry.type === 'apply' &&
-          entry.metadata.property === 'passthrough'
+          ['runInternally', 'passthrough'].includes(entry.metadata.property)
         ) {
           return false
         }
@@ -54,6 +54,7 @@ export class MockSocket extends net.Socket {
         }
       },
     })
+
     return this[kRecorder].socket
   }
 
@@ -76,31 +77,54 @@ export class MockSocket extends net.Socket {
     const [chunk, encoding, callback] = normalizeSocketWriteArgs(
       args as WriteArgs
     )
-    this.emit('write', chunk, encoding, callback)
+    this.runInternally(() => {
+      this.emit('write', chunk, encoding, callback)
+    })
     return true
   }
 
   public push(chunk: any, encoding?: BufferEncoding): boolean {
-    this.emit('push', chunk, encoding)
+    this.runInternally(() => {
+      this.emit('push', chunk, encoding)
+    })
     return super.push(chunk, encoding)
   }
 
   public end(...args: any) {
     const [chunk, encoding, callback] = normalizeSocketWriteArgs(
       args as WriteArgs
     )
-    this.emit('write', chunk, encoding, callback)
+    this.runInternally(() => {
+      this.emit('write', chunk, encoding, callback)
+    })
     return super.end.apply(this, args)
   }
 
+  /**
+   * Invokes the given callback without its actions being recorded.
+   * Use this for internal logic that must not be replayed on the passthrough socket.
+   */
+  public runInternally(callback: () => void) {
+    try {
+      this[kRecorder].pause()
+      callback()
+    } finally {
+      this[kRecorder].resume()
+    }
+  }
+
   /**
    * Establishes the actual connection behind this socket.
    * Replays all the consumer interaction on the passthrough socket
    * and mirrors all the subsequent mock socket interactions onto the passthrough socket.
    */
-  public passthrough(): void {
+  public passthrough(): net.Socket {
     const socket = this.options.createConnection()
     this[kRecorder].replay(socket)
     this[kPassthroughSocket] = socket
+
+    socket.on('error', () => console.log('ERR!'))
+
+    return socket
   }
 }

src/interceptors/net/socket-recorder.ts

@@ -5,6 +5,8 @@ const kSocketRecorder = Symbol('kSocketRecorder')
 export interface SocketRecorder<T extends net.Socket> {
   socket: T
   replay: (newSocket: net.Socket) => void
+  pause: () => void
+  resume: () => void
 }
 
 export interface SocketRecorderEntry {
@@ -29,6 +31,7 @@ export function createSocketRecorder<T extends net.Socket>(
     ) => void
   }
 ): SocketRecorder<T> {
+  let isRecording = true
   const entries: Array<SocketRecorderEntry> = []
 
   Object.defineProperty(socket, kSocketRecorder, {
@@ -38,6 +41,10 @@ export function createSocketRecorder<T extends net.Socket>(
   })
 
   const addEntry = (entry: SocketRecorderEntry) => {
+    if (!isRecording) {
+      return
+    }
+
     if (options?.onEntry?.(entry) !== false) {
       entries.push(entry)
     }
@@ -61,7 +68,11 @@ export function createSocketRecorder<T extends net.Socket>(
                 type: 'apply',
                 metadata: { property },
                 replay(newSocket) {
-                  fn.apply(newSocket, argArray)
+                  Reflect.apply(
+                    newSocket[property as keyof net.Socket] as Function,
+                    newSocket,
+                    argArray
+                  )
                 },
               })
             }
@@ -110,6 +121,12 @@ export function createSocketRecorder<T extends net.Socket>(
       }
       entries.length = 0
     },
+    pause() {
+      isRecording = false
+    },
+    resume() {
+      isRecording = true
+    },
   }
 }