带字节上限的流读取

功能说明

该示例把可读流读取为字符串，同时强制一个字节上限。每收到一块数据就累加长度，一旦超过上限立即销毁源流并抛错，避免攻击者用超大响应或无限流耗尽内存。它适合代理下载、Webhook 正文读取、上传内容预览等不可信输入场景。

关键代码

readStreamWithLimit 先确认参数是带 Symbol.asyncIterator 的流，并校验 maxBytes 为正整数。for await 逐块读取，非 Buffer 的块先转成 Buffer，累加长度后再判断是否超限。超限时先调用 destroy 停止底层资源，再抛出带上限值的错误。finally 保证正常结束或异常时都销毁流，最后用 Buffer.concat 合并并按指定编码解码。

使用注意

上限是在读取完整块之后检查的，因此瞬时内存可能略微超过 maxBytes。该函数会等待流结束，不适合无限流或需要边读边处理的场景。destroy 是 Node 可读流的方法，Web ReadableStream 需要改用 getReader 与 cancel。解码使用指定编码，二进制内容应直接返回 Buffer 而不是字符串。上限值应结合业务与压缩后的实际大小设置。

验证方式

用两块 hello 与 world 的流断言在 64 字节上限下返回 hello world。用十字节的流配合 4 字节上限断言抛出包含上限值的错误，并确认流已被销毁。测试非法 maxBytes、非流对象与空流，空流应返回空字符串。再验证 latin1 编码参数能够正确解码。

代码示例

import { Buffer } from "node:buffer";
import { Readable } from "node:stream";

export async function readStreamWithLimit(stream, maxBytes, options = {}) {
  if (stream === null || typeof stream !== "object" || typeof stream[Symbol.asyncIterator] !== "function") {
    throw new TypeError("stream must be an async iterable readable stream");
  }
  if (!Number.isInteger(maxBytes) || maxBytes < 1) {
    throw new RangeError("maxBytes must be a positive integer");
  }
  const { encoding = "utf8" } = options;
  const chunks = [];
  let total = 0;
  try {
    for await (const chunk of stream) {
      const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
      total += buffer.length;
      if (total > maxBytes) {
        if (typeof stream.destroy === "function") stream.destroy();
        throw new Error(`stream exceeded the ${maxBytes} byte limit`);
      }
      chunks.push(buffer);
    }
  } finally {
    if (typeof stream.destroy === "function" && !stream.destroyed) stream.destroy();
  }
  return Buffer.concat(chunks).toString(encoding);
}

const body = Readable.from(["hello ", "world"]);
console.log(await readStreamWithLimit(body, 64));
try {
  await readStreamWithLimit(Readable.from(["0123456789"]), 4);
} catch (error) {
  console.log(error.message);
}