Function readStreamToBufferWithLimit

apps/sim/lib/core/utils/stream-limits.ts:93–145

(
  stream: ReadableStream<Uint8Array> | null | undefined,
  options: ReadStreamWithLimitOptions
)

}

export async function readStreamToBufferWithLimit(
  stream: ReadableStream<Uint8Array> | null | undefined,
  options: ReadStreamWithLimitOptions
): Promise<Buffer> {
  if (!stream) return Buffer.alloc(0)

  const reader = stream.getReader()
  const chunks: Buffer[] = []
  let totalBytes = 0
  const abortFromSignal = () => {
    void reader.cancel(options.signal?.reason).catch(() => {})
  }

  try {
    if (options.signal?.aborted) {
      await reader.cancel(options.signal.reason).catch(() => {})
      throw toError(options.signal.reason ?? new Error('Aborted'))
    }
    options.signal?.addEventListener('abort', abortFromSignal, { once: true })

    while (true) {
      if (options.signal?.aborted) {
        await reader.cancel(options.signal.reason).catch(() => {})
        throw toError(options.signal.reason ?? new Error('Aborted'))
      }

      const { done, value } = await reader.read()
      if (options.signal?.aborted) {
        throw toError(options.signal.reason ?? new Error('Aborted'))
      }
      if (done) break
      if (!value) continue

      totalBytes += value.byteLength
      if (totalBytes > options.maxBytes) {
        await reader.cancel().catch(() => {})
        throw new PayloadSizeLimitError({
          label: options.label,
          maxBytes: options.maxBytes,
          observedBytes: totalBytes,
        })
      }

      await options.onChunk?.(value, totalBytes)
      chunks.push(Buffer.from(value))
    }
  } finally {
    options.signal?.removeEventListener('abort', abortFromSignal)
    reader.releaseLock()
  }

  return Buffer.concat(chunks, totalBytes)
}

export async function readNodeStreamToBufferWithLimit(
  stream: NodeJS.ReadableStream | null | undefined,