强类型 Web Streams 转换管线

功能说明

该示例使用 Web Streams 标准把字节流按行切分。网络分块可能把一行拆到两个 chunk 中，也可能一个 chunk 包含多行，TransformStream 会缓存未完成的部分，只在遇到换行时输出完整行。它还限制单行最大长度，防止恶意输入无限累积缓冲区。collectStream 则把任意可读流收集成数组。

关键代码

createLineDecoder 使用 TextDecoder 的 stream 模式逐块解码，避免多字节字符被切断。transform 把新文本追加到 buffer，循环查找换行并逐行入队，同时去掉行尾回车以兼容 CRLF。缓冲区超过一兆时调用 controller.error 终止管道并清空。flush 处理最后一段没有换行符的内容。collectStream 通过 getReader 循环读取，在 finally 中释放锁。

使用注意

该实现按 UTF-8 解码，其他编码需要显式构造 TextDecoder。单行上限按字符串长度而非字节数计算，对多字节字符不完全等价。管道通过 pipeThrough 自动传递背压，但下游处理慢时上游会被暂停。controller.error 之后不应再入队。collectStream 会把全部结果放入内存，大文件应逐行消费而不是收集。

验证方式

把一段包含 CRLF 与 LF 的文本切成七字节小块输入，断言得到 alpha、beta、gamma 三行且顺序正确。空流应返回空数组。测试没有结尾换行的输入也能输出最后一行。构造超长单行断言管道报 RangeError，并确认 collectStream 正确释放 reader。

代码示例

export function createLineDecoder(): TransformStream<Uint8Array, string> {
  const decoder = new TextDecoder();
  let buffer = "";
  return new TransformStream<Uint8Array, string>({
    transform(chunk, controller) {
      buffer += decoder.decode(chunk, { stream: true });
      let newline = buffer.indexOf("\n");
      while (newline !== -1) {
        controller.enqueue(buffer.slice(0, newline).replace(/\r$/, ""));
        buffer = buffer.slice(newline + 1);
        newline = buffer.indexOf("\n");
      }
      if (buffer.length > 1_000_000) {
        controller.error(new RangeError("line exceeds the 1 MB limit"));
        buffer = "";
      }
    },
    flush(controller) {
      buffer += decoder.decode();
      if (buffer !== "") controller.enqueue(buffer.replace(/\r$/, ""));
      buffer = "";
    },
  });
}

export async function collectStream<Value>(stream: ReadableStream<Value>): Promise<Value[]> {
  if (!(stream instanceof ReadableStream)) throw new TypeError("stream must be a ReadableStream");
  const values: Value[] = [];
  const reader = stream.getReader();
  try {
    while (true) {
      const { done, value } = await reader.read();
      if (done) return values;
      values.push(value);
    }
  } finally {
    reader.releaseLock();
  }
}

async function* toChunks(text: string): AsyncGenerator<Uint8Array> {
  const bytes = new TextEncoder().encode(text);
  for (let offset = 0; offset < bytes.length; offset += 7) {
    yield bytes.slice(offset, offset + 7);
  }
}

const source = new ReadableStream<Uint8Array>({
  async start(controller) {
    for await (const chunk of toChunks("alpha\r\nbeta\ngamma")) controller.enqueue(chunk);
    controller.close();
  },
});
const lines = await collectStream(source.pipeThrough(createLineDecoder()));
console.log(lines);

const empty = new ReadableStream<Uint8Array>({
  start(controller) {
    controller.close();
  },
});
console.log(await collectStream(empty.pipeThrough(createLineDecoder())));