feat: send mocked response body as ReadableStream to the worker (#1288)

src/setupWorker/glossary.ts

@@ -5,7 +5,7 @@ import {
   LifeCycleEventsMap,
   SharedOptions,
 } from '../sharedOptions'
-import { ServiceWorkerMessage } from '../utils/createBroadcastChannel'
+import { ServiceWorkerMessage } from './start/utils/createMessageChannel'
 import { DefaultBodyType, RequestHandler } from '../handlers/RequestHandler'
 import type { HttpRequestEventMap, Interceptor } from '@mswjs/interceptors'
 import { Path } from '../utils/matching/matchRequestUrl'
@@ -76,11 +76,19 @@ export type ServiceWorkerOutgoingEventTypes =
  * Map of the events that can be sent to the Service Worker
  * only as a part of a single `fetch` event handler.
  */
-export type ServiceWorkerFetchEventTypes =
-  | 'MOCK_SUCCESS'
-  | 'MOCK_NOT_FOUND'
-  | 'NETWORK_ERROR'
-  | 'INTERNAL_ERROR'
+export interface ServiceWorkerFetchEventMap {
+  MOCK_RESPONSE(payload: SerializedResponse): void
+  MOCK_RESPONSE_START(payload: SerializedResponse): void
+
+  MOCK_NOT_FOUND(): void
+  NETWORK_ERROR(payload: { name: string; message: string }): void
+  INTERNAL_ERROR(payload: { status: number; body: string }): void
+}
+
+export interface ServiceWorkerBroadcastChannelMessageMap {
+  MOCK_RESPONSE_CHUNK(payload: Uint8Array): void
+  MOCK_RESPONSE_END(): void
+}
 
 export type WorkerLifecycleEventsMap = LifeCycleEventsMap<Response>
 

src/setupWorker/setupWorker.ts

@@ -9,7 +9,7 @@ import {
 import { createStartHandler } from './start/createStartHandler'
 import { createStop } from './stop/createStop'
 import * as requestHandlerUtils from '../utils/internal/requestHandlerUtils'
-import { ServiceWorkerMessage } from '../utils/createBroadcastChannel'
+import { ServiceWorkerMessage } from './start/utils/createMessageChannel'
 import { jsonParse } from '../utils/internal/jsonParse'
 import { RequestHandler } from '../handlers/RequestHandler'
 import { RestHandler } from '../handlers/RestHandler'

src/utils/worker/createFallbackRequestListener.ts → src/setupWorker/start/createFallbackRequestListener.ts

@@ -10,10 +10,10 @@ import {
   SerializedResponse,
   SetupWorkerInternalContext,
   StartOptions,
-} from '../../setupWorker/glossary'
+} from '../glossary'
 import type { RequiredDeep } from '../../typeUtils'
-import { handleRequest } from '../handleRequest'
-import { parseIsomorphicRequest } from '../request/parseIsomorphicRequest'
+import { handleRequest } from '../../utils/handleRequest'
+import { parseIsomorphicRequest } from '../../utils/request/parseIsomorphicRequest'
 
 export function createFallbackRequestListener(
   context: SetupWorkerInternalContext,

src/setupWorker/start/createFallbackStart.ts

@@ -1,4 +1,4 @@
-import { createFallbackRequestListener } from '../../utils/worker/createFallbackRequestListener'
+import { createFallbackRequestListener } from './createFallbackRequestListener'
 import { SetupWorkerInternalContext, StartHandler } from '../glossary'
 import { printStartMessage } from './utils/printStartMessage'
 

src/utils/worker/createRequestListener.ts → src/setupWorker/start/createRequestListener.ts

@@ -3,17 +3,20 @@ import {
   SerializedResponse,
   SetupWorkerInternalContext,
   ServiceWorkerIncomingEventsMap,
-} from '../../setupWorker/glossary'
+  ServiceWorkerBroadcastChannelMessageMap,
+} from '../glossary'
 import {
   ServiceWorkerMessage,
-  createBroadcastChannel,
-} from '../createBroadcastChannel'
-import { NetworkError } from '../NetworkError'
-import { parseWorkerRequest } from '../request/parseWorkerRequest'
-import { handleRequest } from '../handleRequest'
+  createMessageChannel,
+} from './utils/createMessageChannel'
+import { NetworkError } from '../../utils/NetworkError'
+import { parseWorkerRequest } from '../../utils/request/parseWorkerRequest'
+import { handleRequest } from '../../utils/handleRequest'
 import { RequestHandler } from '../../handlers/RequestHandler'
 import { RequiredDeep } from '../../typeUtils'
 import { MockedResponse } from '../../response'
+import { streamResponse } from './utils/streamResponse'
+import { StrictBroadcastChannel } from '../../utils/internal/StrictBroadcastChannel'
 
 export const createRequestListener = (
   context: SetupWorkerInternalContext,
@@ -26,10 +29,15 @@ export const createRequestListener = (
       ServiceWorkerIncomingEventsMap['REQUEST']
     >,
   ) => {
-    const channel = createBroadcastChannel(event)
+    const messageChannel = createMessageChannel(event)
 
     try {
       const request = parseWorkerRequest(message.payload)
+      const operationChannel =
+        new StrictBroadcastChannel<ServiceWorkerBroadcastChannelMessageMap>(
+          `msw-response-stream-${request.id}`,
+        )
+
       await handleRequest<SerializedResponse>(
         request,
         context.requestHandlers,
@@ -38,15 +46,22 @@ export const createRequestListener = (
         {
           transformResponse,
           onPassthroughResponse() {
-            return channel.send({
+            return messageChannel.send({
               type: 'MOCK_NOT_FOUND',
             })
           },
           onMockedResponse(response) {
-            channel.send({
-              type: 'MOCK_SUCCESS',
-              payload: response,
-            })
+            // Signal the mocked responses without bodies immediately.
+            // There is nothing to stream, so no need to initiate streaming.
+            if (response.body == null) {
+              return messageChannel.send({
+                type: 'MOCK_RESPONSE',
+                payload: response,
+              })
+            }
+
+            // If the mocked response has a body, stream it to the worker.
+            streamResponse(operationChannel, messageChannel, response)
           },
           onMockedResponseSent(
             response,
@@ -69,7 +84,7 @@ export const createRequestListener = (
       if (error instanceof NetworkError) {
         // Treat emulated network error differently,
         // as it is an intended exception in a request handler.
-        return channel.send({
+        return messageChannel.send({
           type: 'NETWORK_ERROR',
           payload: {
             name: error.name,
@@ -81,7 +96,7 @@ export const createRequestListener = (
       if (error instanceof Error) {
         // Treat all the other exceptions in a request handler
         // as unintended, alerting that there is a problem needs fixing.
-        channel.send({
+        messageChannel.send({
           type: 'INTERNAL_ERROR',
           payload: {
             status: 500,

src/utils/worker/createResponseListener.ts → src/setupWorker/start/createResponseListener.ts

@@ -2,7 +2,7 @@ import {
   ServiceWorkerIncomingEventsMap,
   SetupWorkerInternalContext,
 } from '../../setupWorker/glossary'
-import { ServiceWorkerMessage } from '../createBroadcastChannel'
+import { ServiceWorkerMessage } from './utils/createMessageChannel'
 
 export function createResponseListener(context: SetupWorkerInternalContext) {
   return (

src/setupWorker/start/createStartHandler.ts

@@ -2,10 +2,10 @@ import { until } from '@open-draft/until'
 import { getWorkerInstance } from './utils/getWorkerInstance'
 import { enableMocking } from './utils/enableMocking'
 import { SetupWorkerInternalContext, StartHandler } from '../glossary'
-import { createRequestListener } from '../../utils/worker/createRequestListener'
+import { createRequestListener } from './createRequestListener'
 import { requestIntegrityCheck } from '../../utils/internal/requestIntegrityCheck'
 import { deferNetworkRequestsUntil } from '../../utils/deferNetworkRequestsUntil'
-import { createResponseListener } from '../../utils/worker/createResponseListener'
+import { createResponseListener } from './createResponseListener'
 import { validateWorkerScope } from './utils/validateWorkerScope'
 import { devUtils } from '../../utils/internal/devUtils'
 
@@ -25,6 +25,7 @@ export const createStartHandler = (
         createRequestListener(context, options),
       )
 
+      // Handle responses signaled by the worker.
       context.workerChannel.on('RESPONSE', createResponseListener(context))
 
       const instance = await getWorkerInstance(

src/setupWorker/start/utils/createMessageChannel.ts

@@ -0,0 +1,46 @@
+import {
+  ServiceWorkerFetchEventMap,
+  ServiceWorkerIncomingEventsMap,
+} from '../../glossary'
+
+export interface ServiceWorkerMessage<
+  EventType extends keyof ServiceWorkerIncomingEventsMap,
+  EventPayload,
+> {
+  type: EventType
+  payload: EventPayload
+}
+
+export interface WorkerMessageChannel {
+  send<Event extends keyof ServiceWorkerFetchEventMap>(
+    message: Parameters<ServiceWorkerFetchEventMap[Event]>[0] extends undefined
+      ? { type: Event }
+      : {
+          type: Event
+          payload: Parameters<ServiceWorkerFetchEventMap[Event]>[0]
+        },
+  ): void
+}
+
+/**
+ * Creates a communication channel between the client
+ * and the Service Worker associated with the given event.
+ */
+export function createMessageChannel(
+  event: MessageEvent,
+): WorkerMessageChannel {
+  const port = event.ports[0]
+
+  return {
+    /**
+     * Send a text message to the connected Service Worker.
+     */
+    send(message) {
+      if (!port) {
+        return
+      }
+
+      port.postMessage(message)
+    },
+  }
+}

src/setupWorker/start/utils/streamResponse.ts

@@ -0,0 +1,56 @@
+import { invariant } from 'outvariant'
+import { StrictBroadcastChannel } from '../../../utils/internal/StrictBroadcastChannel'
+import {
+  SerializedResponse,
+  ServiceWorkerBroadcastChannelMessageMap,
+} from '../../glossary'
+import { WorkerMessageChannel } from './createMessageChannel'
+
+export async function streamResponse(
+  operationChannel: StrictBroadcastChannel<ServiceWorkerBroadcastChannelMessageMap>,
+  messageChannel: WorkerMessageChannel,
+  mockedResponse: SerializedResponse,
+): Promise<void> {
+  const response = new Response(mockedResponse.body, mockedResponse)
+
+  /**
+   * Delete the ReadableStream response body
+   * so it doesn't get sent via the message channel.
+   * @note Otherwise, an error: cannot clone a ReadableStream if
+   * it hasn't been transformed yet.
+   */
+  delete mockedResponse.body
+
+  // Signal the mock response stream start event on the global
+  // message channel because the worker expects an event in response
+  // to the sent "REQUEST" global event.
+  messageChannel.send({
+    type: 'MOCK_RESPONSE_START',
+    payload: mockedResponse,
+  })
+
+  invariant(response.body, 'Failed to stream mocked response with no body')
+
+  // Read the mocked response body as stream
+  // and pipe it to the worker.
+  const reader = response.body.getReader()
+
+  while (true) {
+    const { done, value } = await reader.read()
+
+    if (!done) {
+      operationChannel.postMessage({
+        type: 'MOCK_RESPONSE_CHUNK',
+        payload: value,
+      })
+      continue
+    }
+
+    operationChannel.postMessage({
+      type: 'MOCK_RESPONSE_END',
+    })
+    operationChannel.close()
+    reader.releaseLock()
+    break
+  }
+}

src/utils/internal/StrictBroadcastChannel.ts

@@ -0,0 +1,27 @@
+const ParentClass =
+  typeof BroadcastChannel == 'undefined'
+    ? class UnsupportedEnvironment {
+        constructor() {
+          throw new Error(
+            'Cannot construct BroadcastChannel in a non-browser environment',
+          )
+        }
+      }
+    : BroadcastChannel
+
+export class StrictBroadcastChannel<
+  MessageMap extends Record<string, any>,
+> extends (ParentClass as unknown as { new (name: string): BroadcastChannel }) {
+  public postMessage<MessageType extends keyof MessageMap>(
+    message: Parameters<MessageMap[MessageType]>[0] extends undefined
+      ? {
+          type: MessageType
+        }
+      : {
+          type: MessageType
+          payload: Parameters<MessageMap[MessageType]>[0]
+        },
+  ): void {
+    return super.postMessage(message)
+  }
+}

test/msw-api/context/delay.node.test.ts

@@ -53,7 +53,6 @@ test('uses realistic server response time when no duration is provided', async (
 
   // Realistic server response time in Node.js is set to 5ms.
   expect(responseTime).toBeGreaterThan(5)
-  expect(responseTime).toBeLessThan(100)
   expect(await res.text()).toBe('john')
 })