Function createMultiplexedStream

packages/start-server-core/src/frame-protocol.ts:107–258

(
  jsonStream: ReadableStream<string>,
  rawStreams: Map<number, ReadableStream<Uint8Array>>,
  lateStreamSource?: ReadableStream<LateStreamRegistration>,
)

 * @param lateStreamSource Optional stream of late registrations for streams discovered later
 */
export function createMultiplexedStream(
  jsonStream: ReadableStream<string>,
  rawStreams: Map<number, ReadableStream<Uint8Array>>,
  lateStreamSource?: ReadableStream<LateStreamRegistration>,
): ReadableStream<Uint8Array> {
  // Shared state for the multiplexed stream
  let controller: ReadableStreamDefaultController<Uint8Array>
  let cancelled = false
  const readers: Array<ReadableStreamDefaultReader<any>> = []

  // Helper to enqueue a frame, ignoring errors if stream is closed/cancelled
  const enqueue = (frame: Uint8Array): boolean => {
    if (cancelled) return false
    try {
      controller.enqueue(frame)
      return true
    } catch {
      return false
    }
  }

  // Helper to error the output stream (for fatal errors like JSON stream failure)
  const errorOutput = (error: unknown): void => {
    if (cancelled) return
    cancelled = true
    try {
      controller.error(error)
    } catch {
      // Already errored
    }
    // Cancel all readers to stop other pumps
    for (const reader of readers) {
      reader.cancel().catch(() => {})
    }
  }

  // Pumps a raw stream, sending CHUNK frames and END/ERROR on completion
  async function pumpRawStream(
    streamId: number,
    stream: ReadableStream<Uint8Array>,
  ): Promise<void> {
    const reader = stream.getReader()
    readers.push(reader)
    try {
      while (!cancelled) {
        const { done, value } = await reader.read()
        if (done) {
          enqueue(encodeEndFrame(streamId))
          return
        }
        if (!enqueue(encodeChunkFrame(streamId, value))) return
      }
    } catch (error) {
      // Raw stream error - send ERROR frame, don't fail entire response
      enqueue(encodeErrorFrame(streamId, error))
    } finally {
      reader.releaseLock()
    }