fix(wip): proxy events between interceptors

Author: kettanaito

pnpm-lock.yaml

@@ -1,33 +1,35 @@
 import http from 'node:http'
 import https from 'node:https'
-import { HttpRequestEventMap } from '#/src/events/http'
-import { Interceptor } from '#/src/Interceptor'
 import { runInRequestContext } from '#/src/request-context'
 import { applyPatch } from '#/src/utils/apply-patch'
 import { HttpRequestInterceptor } from '#/src/interceptors/http'
-import { propagateHttpEvents } from '#/src/utils/interceptor-utils'
+import { Interceptor } from '#/src/Interceptor'
+import { HttpRequestEventMap } from '#/src/events/http'
+import { proxyEventListeners } from '#/src/utils/interceptor-utils'
 
 export class ClientRequestInterceptor extends Interceptor<HttpRequestEventMap> {
-  static symbol = Symbol('client-request-interceptor')
+  static symbol = Symbol.for('client-request-interceptor')
+
+  #httpInterceptor: HttpRequestInterceptor
 
   constructor() {
     super(ClientRequestInterceptor.symbol)
+
+    this.#httpInterceptor = new HttpRequestInterceptor()
+    this.subscriptions.push(
+      proxyEventListeners({
+        from: this.emitter,
+        to: this.#httpInterceptor['emitter'],
+        filter: (event) => {
+          return event.initiator instanceof http.ClientRequest
+        },
+      })
+    )
   }
 
   protected setup(): void {
-    const httpInterceptor = new HttpRequestInterceptor()
-
-    httpInterceptor.apply()
-    this.subscriptions.push(() => httpInterceptor.dispose())
-
-    const { controller } = propagateHttpEvents(
-      httpInterceptor['emitter'],
-      this.emitter,
-      (event) => {
-        return event.initiator instanceof http.ClientRequest
-      }
-    )
-    this.subscriptions.push(() => controller.abort())
+    this.#httpInterceptor.apply()
+    this.subscriptions.push(() => this.#httpInterceptor.dispose())
 
     this.subscriptions.push(
       applyPatch(http, 'ClientRequest', (ClientRequest) => {

src/interceptors/ClientRequest/index.ts

@@ -1,46 +1,48 @@
+import { Emitter } from 'rettime'
 import { requestContext } from '#/src/request-context'
 import { hasConfigurableGlobal } from '#/src/utils/hasConfigurableGlobal'
-import { applyPatch } from '#/src/utils/apply-patch'
 import { Interceptor } from '#/src/Interceptor'
-import { HttpRequestEventMap } from '../../events/http'
 import { HttpRequestInterceptor } from '#/src/interceptors/http'
+import { applyPatch } from '#/src/utils/apply-patch'
 import { FetchRequest } from '#/src/utils/fetchUtils'
-import { propagateHttpEvents } from '#/src/utils/interceptor-utils'
+import { HttpRequestEventMap } from '#/src/events/http'
+import { proxyEventListeners } from '#/src/utils/interceptor-utils'
 
 export class XMLHttpRequestInterceptor extends Interceptor<HttpRequestEventMap> {
   static symbol = Symbol.for('xhr-interceptor')
 
+  #httpInterceptor: HttpRequestInterceptor
+
   constructor() {
     super(XMLHttpRequestInterceptor.symbol)
+
+    this.#httpInterceptor = new HttpRequestInterceptor()
+    this.subscriptions.push(
+      proxyEventListeners({
+        from: this.emitter,
+        to: this.#httpInterceptor['emitter'],
+        filter: (event) => {
+          if (event.initiator instanceof XMLHttpRequest) {
+            event.request = this.#transformRequest(
+              event.request,
+              event.initiator
+            )
+            return true
+          }
+
+          return false
+        },
+      })
+    )
   }
 
   protected checkEnvironment() {
     return hasConfigurableGlobal('XMLHttpRequest')
   }
 
   protected setup(): void {
-    const httpInterceptor = new HttpRequestInterceptor()
-
-    httpInterceptor.apply()
-    this.subscriptions.push(() => httpInterceptor.dispose())
-
-    this.emitter.hooks.on('beforeEmit', (event) => {
-      event.modify = true
-    })
-
-    const { controller } = propagateHttpEvents(
-      httpInterceptor['emitter'],
-      this.emitter,
-      (event) => {
-        if (event.initiator instanceof XMLHttpRequest) {
-          event.request = this.#transformRequest(event.request, event.initiator)
-          return true
-        }
-
-        return false
-      }
-    )
-    this.subscriptions.push(() => controller.abort())
+    this.#httpInterceptor.apply()
+    this.subscriptions.push(() => this.#httpInterceptor.dispose())
 
     this.logger.info('patching global "XMLHttpRequest"...')
 
@@ -70,11 +72,19 @@ export class XMLHttpRequestInterceptor extends Interceptor<HttpRequestEventMap>
   }
 
   #transformRequest(request: Request, initiator: XMLHttpRequest): Request {
+    const expectedCredentials = initiator.withCredentials
+      ? 'include'
+      : 'same-origin'
+
+    if (request.credentials === expectedCredentials) {
+      return request
+    }
+
     return new FetchRequest(request.url, {
       ...request,
       method: request.method,
       headers: request.headers,
-      credentials: initiator.withCredentials ? 'include' : 'same-origin',
+      credentials: expectedCredentials,
       body: request.body,
     })
   }

src/interceptors/XMLHttpRequest/node.ts

@@ -1,37 +1,39 @@
-import { Interceptor } from '#/src/Interceptor'
-import type { HttpRequestEventMap } from '#/src/events/http'
 import { hasConfigurableGlobal } from '#/src/utils/hasConfigurableGlobal'
 import { canParseUrl } from '#/src/utils/canParseUrl'
 import { requestContext } from '#/src/request-context'
 import { applyPatch } from '#/src/utils/apply-patch'
 import { HttpRequestInterceptor } from '#/src/interceptors/http'
-import { propagateHttpEvents } from '#/src/utils/interceptor-utils'
+import { Interceptor } from '#/src/Interceptor'
+import { HttpRequestEventMap } from '#/src/events/http'
+import { proxyEventListeners } from '#/src/utils/interceptor-utils'
 
 export class FetchInterceptor extends Interceptor<HttpRequestEventMap> {
   static symbol = Symbol.for('fetch-interceptor')
 
+  #httpInterceptor: HttpRequestInterceptor
+
   constructor() {
     super(FetchInterceptor.symbol)
+
+    this.#httpInterceptor = new HttpRequestInterceptor()
+    this.subscriptions.push(
+      proxyEventListeners({
+        from: this.emitter,
+        to: this.#httpInterceptor['emitter'],
+        filter: (event) => {
+          return event.initiator instanceof XMLHttpRequest
+        },
+      })
+    )
   }
 
   protected checkEnvironment() {
     return hasConfigurableGlobal('fetch')
   }
 
   protected setup(): void {
-    const httpInterceptor = new HttpRequestInterceptor()
-
-    httpInterceptor.apply()
-    this.subscriptions.push(() => httpInterceptor.dispose())
-
-    const { controller } = propagateHttpEvents(
-      httpInterceptor['emitter'],
-      this.emitter,
-      (event) => {
-        return event.initiator instanceof Request
-      }
-    )
-    this.subscriptions.push(() => controller.abort())
+    this.#httpInterceptor.apply()
+    this.subscriptions.push(() => this.#httpInterceptor.dispose())
 
     this.subscriptions.push(
       applyPatch(globalThis, 'fetch', (realFetch) => {

src/interceptors/fetch/node.ts

@@ -296,7 +296,10 @@ export class HttpRequestInterceptor extends Interceptor<HttpRequestEventMap> {
       callback?.()
     }
 
-    responseSocket._destroy = () => {
+    responseSocket._destroy = (
+      _error: Error | null,
+      callback: (error: Error | null) => void
+    ) => {
       /**
        * Destroy the socket if the response stream errored.
        * @see https://github.com/mswjs/interceptors/issues/738
@@ -306,6 +309,7 @@ export class HttpRequestInterceptor extends Interceptor<HttpRequestEventMap> {
        * @see https://github.com/nodejs/node/blob/f3adc11e37b8bfaaa026ea85c1cf22e3a0e29ae9/lib/_http_client.js#L586
        */
       socket.destroy()
+      callback(null)
     }
 
     serverResponse.assignSocket(responseSocket)

src/interceptors/http/index.ts

@@ -114,9 +114,7 @@ export async function handleRequest(
       controller: options.controller,
     }
     const requestEvent = new HttpRequestEvent(requestEventData)
-    const requestListenersPromise = options.emitter.emitAsPromise(
-      requestEvent
-    )
+    const requestListenersPromise = options.emitter.emitAsPromise(requestEvent)
 
     await Promise.race([
       // Short-circuit the request handling promise if the request gets aborted.
@@ -152,7 +150,6 @@ export async function handleRequest(
     // If the developer has added "unhandledException" listeners,
     // allow them to handle the error. They can translate it to a
     // mocked response, network error, or forward it as-is.
-    console.log('[DEBUG handleRequest] unhandledException listenerCount:', options.emitter.listenerCount('unhandledException'))
     if (options.emitter.listenerCount('unhandledException') > 0) {
       // Create a new request controller just for the unhandled exception case.
       // This is needed because the original controller might have been already

src/utils/handleRequest.ts

@@ -1,65 +1,45 @@
-import { DefaultEventMap, Emitter, EventMap } from 'rettime'
+import { Emitter } from 'rettime'
 
-export function propagateHttpEvents<SourceEventMap extends DefaultEventMap>(
-  source: Emitter<SourceEventMap>,
-  destination: Emitter<any>,
-  predicate: (event: EventMap.Events<SourceEventMap>) => boolean
-) {
+export function proxyEventListeners<T extends Emitter<any>>(options: {
+  from: T
+  to: T
+  filter: (event: Emitter.Events<T>) => boolean
+}) {
   const controller = new AbortController()
 
-  const propagateEvent = async (event: EventMap.Events<SourceEventMap>) => {
-    if (predicate(event)) {
-      await destination.emitAsPromise(event)
+  const propagateEvent = async (event: Emitter.Events<T>) => {
+    if (options.filter(event)) {
+      await options.from.emitAsPromise(event)
     }
   }
 
-  source.on(
-    '*',
-    async (event) => {
-      if (event.type !== 'response') {
-        await propagateEvent(event)
-      }
-    },
-    { signal: controller.signal }
-  )
-
-  /**
-   * @note Lazily add a "response" listener to the HTTP interceptor if this
-   * interceptor receives a response listener. HTTP interceptor creates a
-   * response parser only if a "response" listener is present.
-   *
-   * Cannot use hooks for this because `removeAllListeners()` in rettime
-   * also removes hooks listeners, breaking lazy registration across tests.
-   */
-  destination.hooks.on(
+  options.from.hooks.on(
     'newListener',
     (type) => {
-      if (
-        type === 'response' &&
-        !source.listeners('response').includes(propagateEvent)
-      ) {
-        source.on('response', propagateEvent, { signal: controller.signal })
+      if (!options.to.listeners(type).includes(propagateEvent)) {
+        options.to.on(type, propagateEvent, { signal: controller.signal })
       }
     },
-    { signal: controller.signal }
+    {
+      persist: true,
+      signal: controller.signal,
+    }
   )
 
-  destination.hooks.on(
+  options.from.hooks.on(
     'removeListener',
     (type) => {
-      if (
-        type === 'response' &&
-        source.listeners('response').includes(propagateEvent)
-      ) {
-        source.removeListener('response', propagateEvent)
+      if (options.from.listenerCount(type) === 1) {
+        options.to.removeListener(type, propagateEvent)
       }
     },
     {
+      persist: true,
       signal: controller.signal,
     }
   )
 
-  return {
-    controller,
+  return () => {
+    controller.abort()
   }
 }