Stream Handling

Copytype WebMod = typeof import('./stream-ops.web')
 
let _m: WebMod
if (process.env.__NEXT_USE_NODE_STREAMS) {
  _m = require('./stream-ops.node') as typeof import('./stream-ops.node')
} else {
  _m = require('./stream-ops.web') as typeof import('./stream-ops.web')
}

Copyfunction webToReadable(
  stream: ReadableStream<Uint8Array> | import('node:stream').Readable
): import('node:stream').Readable {
  if (process.env.NEXT_RUNTIME === 'edge') {
    throw new Error('webToReadable cannot be used in the edge runtime')
  } else {
    let Readable: typeof import('node:stream').Readable
    if (process.env.TURBOPACK) {
      Readable = (require('node:stream') as typeof import('node:stream'))
        .Readable
    } else if (
      process.env.__NEXT_BUNDLER === 'Webpack' ||
      process.env.__NEXT_BUNDLER === 'Rspack'
    ) {
      Readable = (
        __non_webpack_require__('node:stream') as typeof import('node:stream')
      ).Readable
    } else {
      Readable = (require('node:stream') as typeof import('node:stream'))
        .Readable
    }
    if (stream instanceof Readable) {
      return stream
    }
    return Readable.fromWeb(stream as import('stream/web').ReadableStream)
  }
}

Copyexport function createInlinedDataReadableStream(
  flightStream: ReadableStream<Uint8Array>,
  nonce: string | undefined,
  formState: unknown | null
): ReadableStream<Uint8Array> {
  const startScriptTag = nonce
    ? `<script nonce="${htmlEscapeAttributeString(nonce)}">`
    : '<script>'
 
  const flightReader = flightStream.getReader()
  const decoder = new TextDecoder('utf-8', { fatal: true })
 
  const readable = new ReadableStream({
    type: 'bytes',
    start(controller) {
      try {
        writeInitialInstructions(controller, startScriptTag, formState)
      } catch (error) {
        // during encoding or enqueueing forward the error downstream
        controller.error(error)
      }
    },
    async pull(controller) {
      try {
        const { done, value } = await flightReader.read()
 
        if (value) {
          try {
            const decodedString = decoder.decode(value, { stream: !done })
 
            // The chunk cannot be decoded as valid UTF-8 string as it might
            // have arbitrary binary data.
            writeFlightDataInstruction(
              controller,
              startScriptTag,
              decodedString
            )
          } catch {
            // The chunk cannot be decoded as valid UTF-8 string.
            writeFlightDataInstruction(controller, startScriptTag, value)
          }
        }
 
        if (done) {
          controller.close()
        }
      } catch (error) {
        // There was a problem in the upstream reader or during decoding or enqueuing
        // forward the error downstream
        controller.error(error)
      }
    },
  })
 
  return readable
}

Copyfunction createFlightDataInjectionTransform(
  dataStream: Readable,
  delayDataUntilFirstHtmlChunk: boolean
): Transform {
  let htmlStreamFinished = false
  let pull: Promise<void> | null = null
  let donePulling = false
 
  function startOrContinuePulling(target: Transform) {
    if (!pull) {
      pull = startPulling(target)
    }
    return pull
  }
 
  async function startPulling(target: Transform) {
    if (delayDataUntilFirstHtmlChunk) {
      // Buffer the inlined data stream until we've left the current Task so
      // it's inserted after flushing the shell.
      await atLeastOneTask()
    }
 
    try {
      const iterator = dataStream[Symbol.asyncIterator]()
      while (true) {
        const { done, value } = await iterator.next()
        if (done) {
          donePulling = true
          return
        }
 
        // Prioritize HTML over RSC data: the SSR render produces HTML from
        // the same RSC stream, so yield a task to let HTML flush first.
        if (!delayDataUntilFirstHtmlChunk && !htmlStreamFinished) {
          await atLeastOneTask()
        }
        target.push(value)
      }
    } catch (err) {
      target.destroy(err as Error)
    }
  }
 
  const nodeTransform = new Transform({
    transform(chunk, _encoding, callback) {
      this.push(chunk)
      if (delayDataUntilFirstHtmlChunk) {
        startOrContinuePulling(this)
      }
      callback()
    },
    flush(callback) {
      htmlStreamFinished = true
      if (donePulling) {
        callback()
        return
      }
      startOrContinuePulling(this).then(
        () => callback(),
        (err) => callback(err as Error)
      )
    },
  })
 
  if (!delayDataUntilFirstHtmlChunk) {
    startOrContinuePulling(nodeTransform)
  }
 
  return nodeTransform
}
…
function createHeadInsertionTransform(
  insert: () => Promise<string>
): Transform {
  let inserted = false
  let hasBytes = false
 
  return new Transform({
    async transform(chunk, _encoding, callback) {
      hasBytes = true

Copyexport function indexOfUint8Array(a: Uint8Array, b: Uint8Array) {
  if (b.length === 0) return 0
  if (a.length === 0 || b.length > a.length) return -1
 
  // Use Node's native implementation when available.
  if (typeof Buffer !== 'undefined') {
    const haystack = Buffer.isBuffer(a)
      ? a
      : Buffer.from(a.buffer, a.byteOffset, a.byteLength)
    return haystack.indexOf(b)
  }
 
  // start iterating through `a`
  for (let i = 0; i <= a.length - b.length; i++) {
    let completeMatch = true
    // from index `i`, iterate through `b` and check for mismatch
    for (let j = 0; j < b.length; j++) {
      // if the values do not match, then this isn't a complete match, exit `b` iteration early and iterate to next index of `a`.
      if (a[i + j] !== b[j]) {
        completeMatch = false
        break
      }
    }
 
    if (completeMatch) {
      return i
    }
  }
 
  return -1
}

Copy * Check if two Uint8Arrays are strictly equivalent.
 */
export function isEquivalentUint8Arrays(a: Uint8Array, b: Uint8Array) {
  if (a.length !== b.length) return false
 
  for (let i = 0; i < a.length; i++) {
    if (a[i] !== b[i]) return false
  }
 
  return true
}
 
/**
 * Remove Uint8Array `b` from Uint8Array `a`.
 *
 * If `b` is not in `a`, `a` is returned unchanged.
 *
 * Otherwise, the function returns a new Uint8Array instance with size `a.length - b.length`
 */
export function removeFromUint8Array(a: Uint8Array, b: Uint8Array) {
  const tagIndex = indexOfUint8Array(a, b)
  if (tagIndex === 0) return a.subarray(b.length)
  if (tagIndex > -1) {
    const removed = new Uint8Array(a.length - b.length)
    removed.set(a.subarray(0, tagIndex))
    removed.set(a.subarray(tagIndex + b.length), tagIndex)
    return removed
  } else {
    return a
  }
}

Copyasync function streamToChunks(
  stream: ReadableStream<Uint8Array>
): Promise<Array<Uint8Array>> {
  const reader = stream.getReader()
  const chunks: Array<Uint8Array> = []
 
  while (true) {
    const { done, value } = await reader.read()
    if (done) {
      break
    }
 
    chunks.push(value)
  }
 
  return chunks
}
 
function concatUint8Arrays(chunks: Array<Uint8Array>): Uint8Array {
  const totalLength = chunks.reduce((sum, chunk) => sum + chunk.length, 0)
  const result = new Uint8Array(totalLength)
  let offset = 0
  for (const chunk of chunks) {
    result.set(chunk, offset)
    offset += chunk.length
  }
  return result
}
 
export async function webstreamToUint8Array(
  stream: ReadableStream<Uint8Array>
): Promise<Uint8Array> {
  return concatUint8Arrays(await streamToChunks(stream))
}

Copyfunction concatUint8Arrays(chunks: Array<Uint8Array>): Uint8Array {
  const totalLength = chunks.reduce((sum, chunk) => sum + chunk.length, 0)
  const result = new Uint8Array(totalLength)
  let offset = 0
  for (const chunk of chunks) {
    result.set(chunk, offset)
    offset += chunk.length
  }
  return result
}

Copyexport function createBufferedTransformStream(
  options: BufferedTransformOptions = {}
): TransformStream<Uint8Array, Uint8Array> {
  const { maxBufferByteLength = Infinity } = options
 
  let bufferedChunks: Array<Uint8Array> = []
  let bufferByteLength: number = 0
  let pending: PromiseWithResolvers<void> | undefined
 
  const flush = (controller: TransformStreamDefaultController) => {
    try {
      if (bufferedChunks.length === 0) {
        return
      }
 
      const chunk = new Uint8Array(bufferByteLength)
      let copiedBytes = 0
 
      for (let i = 0; i < bufferedChunks.length; i++) {
        const bufferedChunk = bufferedChunks[i]
        chunk.set(bufferedChunk, copiedBytes)
        copiedBytes += bufferedChunk.byteLength
      }
      // We just wrote all the buffered chunks so we need to reset the bufferedChunks array
      // and our bufferByteLength to prepare for the next round of buffered chunks
      bufferedChunks.length = 0
      bufferByteLength = 0
      controller.enqueue(chunk)
    } catch {
      // If an error occurs while enqueuing, it can't be due to this
      // transformer. It's most likely caused by the controller having been
      // errored (for example, if the stream was cancelled).
    }
  }
 
  const scheduleFlush = (controller: TransformStreamDefaultController) => {
    if (pending) {
      return
    }
 
    const detached = createPromiseWithResolvers<void>()
    pending = detached
 
    scheduleImmediate(() => {
      try {
        flush(controller)
      } finally {
        pending = undefined
        detached.resolve()
      }

Copyexport function createNodeBufferedTransformStream(
  maxBufferByteLength: number = Infinity
): Transform {
  let bufferedChunks: Array<Uint8Array> = []
  let bufferByteLength = 0
  let flushScheduled = false
 
  function flushBuffered(stream: Transform): void {
    if (bufferedChunks.length === 0) return
 
    const merged = new Uint8Array(bufferByteLength)
    let copiedBytes = 0
    for (let i = 0; i < bufferedChunks.length; i++) {
      const bufferedChunk = bufferedChunks[i]
      merged.set(bufferedChunk, copiedBytes)
      copiedBytes += bufferedChunk.byteLength
    }
    bufferedChunks.length = 0
    bufferByteLength = 0
    stream.push(merged)
  }
 
  return new Transform({
    transform(chunk, _encoding, callback) {
      bufferedChunks.push(chunk)
      bufferByteLength += chunk.byteLength
 
      if (bufferByteLength >= maxBufferByteLength) {
        flushBuffered(this)
      } else if (!flushScheduled) {
        flushScheduled = true
        queueMicrotask(() => {
          flushScheduled = false
          flushBuffered(this)
        })
      }
      callback()
    },
    flush(callback) {
      flushBuffered(this)
      callback()
    },
  })
}

Copy// Chunks are sent to the browser in batches to reduce overhead, flushing
// synchronously once this many bytes have accumulated.
const MAX_DEBUG_CHANNEL_BATCH_BYTES = 128 * 1024
 
export interface ReactDebugChannelForBrowser {
  readonly readable: AnyStream
}
 
const reactDebugChannelsByHtmlRequestId = new Map<
  string,
  ReactDebugChannelForBrowser
>()

Copyexport function connectReactDebugChannelForHtmlRequest(
  htmlRequestId: string,
  sendToClient: (message: HmrMessageSentToBrowser) => void
) {
  const debugChannel = reactDebugChannelsByHtmlRequestId.get(htmlRequestId)
 
  if (!debugChannel) {
    return
  }
 
  reactDebugChannelsByHtmlRequestId.delete(htmlRequestId)
 
  connectReactDebugChannel(htmlRequestId, debugChannel, sendToClient)
}

CopyonHMR(req, socket: Socket, head, onUpgrade) {
  wsServer.handleUpgrade(req, socket, head, (client) => {
    const clientIssues: EntryIssuesMap = new Map()
    const subscriptions: Map<string, AsyncIterator<any>> = new Map()
 
    const htmlRequestId = req.url
      ? new URL(req.url, 'http://n').searchParams.get('id')
      : null
 
    // Clients with a request ID are inferred App Router clients. If Cache
    // Components is not enabled, we consider those legacy clients. Pages
    // Router clients are also considered legacy clients. TODO: Maybe mark
    // clients as App Router / Pages Router clients explicitly, instead of
    // inferring it from the presence of a request ID.
    if (htmlRequestId) {
      clientsByHtmlRequestId.set(htmlRequestId, client)
      const enableCacheComponents = nextConfig.cacheComponents
      if (enableCacheComponents) {
        onUpgrade(client, { isLegacyClient: false })
        const cacheStatus = cacheStatusesByHtmlRequestId.get(htmlRequestId)
        if (cacheStatus !== undefined) {
          sendToClient(client, {
            type: HMR_MESSAGE_SENT_TO_BROWSER.CACHE_INDICATOR,
            state: cacheStatus,
          })
          cacheStatusesByHtmlRequestId.delete(htmlRequestId)
        }
      } else {
        onUpgrade(client, { isLegacyClient: true })
      }
 
      connectReactDebugChannelForHtmlRequest(
        htmlRequestId,
        sendToClient.bind(null, client)
      )
 

Copyexport function connectReactDebugChannel(
  requestId: string,
  debugChannel: ReactDebugChannelForBrowser,
  sendToClient: (message: HmrMessageSentToBrowser) => void
) {
  let finished = false
 
  const stop = () => {
    if (finished) {
      return
    }
    finished = true
    sendToClient({
      type: HMR_MESSAGE_SENT_TO_BROWSER.REACT_DEBUG_CHUNK,
      requestId,
      chunk: null,
    })
  }
 
  const onError = (err: unknown) => {
    if (!finished) {
      console.error(
        new Error('React debug channel stream error', { cause: err })
      )
    }
    stop()
  }
 
  const sendChunk = (chunk: Uint8Array) => {
    sendToClient({
      type: HMR_MESSAGE_SENT_TO_BROWSER.REACT_DEBUG_CHUNK,
      requestId,
      chunk,
    })
  }
 
  const { readable } = debugChannel
 
  if (readable instanceof ReadableStream) {
    const reader = readable
      .pipeThrough(
        createBufferedTransformStream({
          maxBufferByteLength: MAX_DEBUG_CHANNEL_BATCH_BYTES,
        })
      )
      .getReader()
 

Copyexport function createBufferedTransformStream(
  options: BufferedTransformOptions = {}
): TransformStream<Uint8Array, Uint8Array> {
  const { maxBufferByteLength = Infinity } = options
 
  let bufferedChunks: Array<Uint8Array> = []
  let bufferByteLength: number = 0
  let pending: PromiseWithResolvers<void> | undefined
 
  const flush = (controller: TransformStreamDefaultController) => {
    try {
      if (bufferedChunks.length === 0) {
        return
      }
 
      const chunk = new Uint8Array(bufferByteLength)
      let copiedBytes = 0
 
      for (let i = 0; i < bufferedChunks.length; i++) {
        const bufferedChunk = bufferedChunks[i]
        chunk.set(bufferedChunk, copiedBytes)
        copiedBytes += bufferedChunk.byteLength
      }
      // We just wrote all the buffered chunks so we need to reset the bufferedChunks array
      // and our bufferByteLength to prepare for the next round of buffered chunks
      bufferedChunks.length = 0
      bufferByteLength = 0
      controller.enqueue(chunk)
    } catch {
      // If an error occurs while enqueuing, it can't be due to this
      // transformer. It's most likely caused by the controller having been
      // errored (for example, if the stream was cancelled).
    }
  }
 
  const scheduleFlush = (controller: TransformStreamDefaultController) => {
    if (pending) {
      return
    }
 
    const detached = createPromiseWithResolvers<void>()
    pending = detached
 
    scheduleImmediate(() => {
      try {
        flush(controller)
      } finally {
        pending = undefined
        detached.resolve()
      }

Copyexport class ReplayableNodeStream {
  private _chunks: Array<Uint8Array> | null
  private _done: boolean
  private _error: Error | null
  private _subscribers: Set<ReplayableStreamSubscriber>
 
  constructor(stream: Readable) {
    this._chunks = []
    this._done = false
    this._error = null
    this._subscribers = new Set()
 
    stream.on('data', (chunk: Buffer | Uint8Array) => {
      const buf = chunk instanceof Uint8Array ? chunk : new Uint8Array(chunk)
      if (this._chunks !== null) {
        this._chunks.push(buf)
      }
      for (const sub of this._subscribers) {
        sub.onChunk(buf)
      }
    })
 
    stream.on('end', () => {
      this._done = true
      for (const sub of this._subscribers) {
        sub.onEnd()
      }
      this._subscribers.clear()
    })
 
    stream.on('error', (err: Error) => {
      this._error = err
      for (const sub of this._subscribers) {
        sub.onError(err)
      }
      this._subscribers.clear()
    })
  }

Copyexport function createNodeStreamWithLateRelease(
  partialChunks: Array<Uint8Array>,
  allChunks: Array<Uint8Array>,
  releaseSignal: AbortSignal
): Readable {
  if (process.env.NEXT_RUNTIME === 'edge') {
    throw new InvariantError(
      'createNodeStreamWithLateRelease cannot be used in the edge runtime'
    )
  } else {
    const { Readable } = require('node:stream') as typeof import('node:stream')
 
    let nextIndex = 0
 
    const readable = new Readable({
      read() {
        while (nextIndex < partialChunks.length) {
          this.push(partialChunks[nextIndex])
          nextIndex++
        }
      },
    })
 
    releaseSignal.addEventListener(
      'abort',
      () => {
        // Flush any remaining chunks from the original set
        while (nextIndex < partialChunks.length) {
          readable.push(partialChunks[nextIndex])
          nextIndex++
        }
        // Flush all chunks since we're now aborted and can't schedule
        // any new work but these chunks might unblock debugInfo
        while (nextIndex < allChunks.length) {
          readable.push(allChunks[nextIndex])
          nextIndex++
        }
 
        setImmediate(() => {
          readable.push(null)
        })
      },
      { once: true }
    )
 
    return readable
  }
}

Copyexport async function pipeToNodeResponse(
  readable: ReadableStream<Uint8Array>,
  res: ServerResponse,
  waitUntilForEnd?: Promise<unknown>,
  clientComponentLoadTracker?: ClientComponentLoadTracker
) {
  try {
    const { errored, destroyed } = res
    if (errored || destroyed) return
 
    const controller = createAbortController(res)
 
    const writer = createWriterFromResponse(
      res,
      waitUntilForEnd,
      clientComponentLoadTracker
    )
 
    await readable.pipeTo(writer, { signal: controller.signal })
  } catch (err: any) {
    if (isAbortError(err)) return
 
    throw new Error('failed to pipe response', { cause: err })
  }
}

Copyexport function isAbortError(e: any): e is Error & { name: 'AbortError' } {
  return e?.name === 'AbortError' || e?.name === ResponseAbortedName
}

Copyconst DEFAULT_BODY_CLONE_SIZE_LIMIT = 10 * 1024 * 1024 // 10MB

Copyfunction replaceRequestBody<T extends IncomingMessage>(
  base: T,
  stream: Readable
): T {
  for (const key in stream) {
    let v = stream[key as keyof Readable] as any
    if (typeof v === 'function') {
      v = v.bind(base)
    }
    base[key as keyof T] = v
  }
 
  return base
}