流
使用 Bun 的流 API 以无需一次性将全部数据加载到内存中即可处理二进制数据
流是处理二进制数据而无需一次性全部加载到内存中的重要抽象。它们通常用于读写文件、发送和接收网络请求以及处理大量数据。
Bun 实现了 Web API 中的 ReadableStream 和 WritableStream。
创建一个 ReadableStream：
const stream = new ReadableStream({
start(controller) {
controller.enqueue("hello");
controller.enqueue("world");
controller.close();
},
});
你可以使用 for await 语法逐块读取 ReadableStream 的内容。
for await (const chunk of stream) {
console.log(chunk);
}
// 你好
// 世界
直接 ReadableStream
Bun 实现了优化版的 ReadableStream，避免了不必要的队列管理。
使用传统的 ReadableStream 时，你需要将数据块入队。流会将每个数据块添加到队列中，直到流准备好发送更多数据时，这些数据块才会留在队列中。
const stream = new ReadableStream({
start(controller) {
controller.enqueue("hello");
controller.enqueue("world");
controller.close();
},
});
使用直接 ReadableStream 时，你可以直接将数据块写入流中，不会发生排队。controller API 也体现了这一点：你调用 .write() 而不是 .enqueue()。
const stream = new ReadableStream({
type: "direct", // [!code ++]
pull(controller) {
controller.write("hello");
controller.write("world");
controller.close();
},
});
使用直接 ReadableStream 时，目标端会处理所有数据块的排队。目标端会接收你传递给 controller.write() 的字节。从 JavaScript 读取流时，Bun 会缓冲写入内容，并以 Uint8Array 数据块的形式传递（字符串会进行 UTF-8 编码）。
结束流
所有内容写入完毕后，调用 controller.close()（或 controller.end()）。每个目标端都会刷新已缓冲的内容，并在此时结束。controller.close(error) 则会使流失败：不会再刷新任何内容，消费者会因 error 而拒绝。
接收整个消息体的目标端（Bun.serve、fetch、Bun.write、Bun.spawn stdin、.text() 等）会调用一次 pull()。如果它返回一个 promise，那么在该 promise 处于待定状态期间，流会保持打开；当 promise 兑现时，流会结束，效果相当于 pull() 最后调用了 controller.close()；如果 promise 被拒绝，流会因该原因而出错。pull() 若不返回 promise 且不关闭流，则流会保持打开，直到有内容调用 controller.close()，因此你可以保留 controller，并从事件中写入内容。
通过 reader（getReader()、for await、pipeTo()、tee()）读取流时，pull() 则是一种需求信号：在之前调用返回的 promise 仍处于待定状态时不会再次调用，而是会在后续读取时再次调用，直到 controller.close()。
处理背压
controller.write() 会返回已写入的字节数；如果目标端的内部缓冲区已满（例如，HTTP 客户端速度较慢），则会返回一个待定的 Promise<number>。无论哪种情况，数据块都会被接收。目标端断开连接后（例如，HTTP 客户端断开连接），write() 会返回 0。当目标端缓冲区排空后，promise 就会兑现，因此只需对结果使用 await：
const stream = new ReadableStream({
type: "direct",
async pull(controller) {
for (const chunk of chunks) {
await controller.write(chunk);
}
controller.close();
},
});
await controller.flush(true) 效果相同，也可以在写入返回 Promise 后使用它。
从 JavaScript 读取流时（getReader()、pipeTo()、for await），当有 highWaterMark 字节（默认为 64 KiB）等待 reader 读取时，缓冲区就会被视为已满。将 { highWaterMark } 作为 ReadableStream 构造函数的第二个参数传入，即可更改此值。reader.cancel(reason) 会调用源的 cancel(reason)。
对于默认（非 direct）的 ReadableStream 和异步生成器响应体，Bun 会自动应用此背压：当目标端积压时，它会暂停生产者。
异步生成器流
Bun 也支持将异步生成器函数作为 Response 和 Request 的数据源。使用异步生成器创建一个从异步源获取数据的 ReadableStream。
const response = new Response(
(async function* () {
yield "hello";
yield "world";
})(),
);
await response.text(); // "helloworld"
你也可以直接使用 [Symbol.asyncIterator]。
const response = new Response({
[Symbol.asyncIterator]: async function* () {
yield "hello";
yield "world";
},
});
await response.text(); // "helloworld"
消息体只包含生成器 yield 的值。与 for await 一样，生成器的返回值不属于消息体。
const response = new Response(
(async function* () {
yield "hello";
return "ignored";
})(),
);
await response.text(); // "hello"
如需更精细地控制流，yield 会返回直接 ReadableStream 的 controller。
const response = new Response({
[Symbol.asyncIterator]: async function* () {
const controller = yield "hello";
await controller.end();
},
});
await response.text(); // "hello"
Bun.ArrayBufferSink
Bun.ArrayBufferSink 类是一个用于构造未知大小 ArrayBuffer 的高速增量写入器。
const sink = new Bun.ArrayBufferSink();
sink.write("h");
sink.write("e");
sink.write("l");
sink.write("l");
sink.write("o");
sink.end();
// ArrayBuffer(5) [ 104, 101, 108, 108, 111 ]
如果需要以 Uint8Array 格式获取数据，可以向 start 方法传入 asUint8Array 选项。
const sink = new Bun.ArrayBufferSink();
sink.start({
asUint8Array: true, // [!code ++]
});
sink.write("h");
sink.write("e");
sink.write("l");
sink.write("l");
sink.write("o");
sink.end();
// Uint8Array(5) [ 104, 101, 108, 108, 111 ]
.write() 方法支持字符串、类型化数组、ArrayBuffer 和 SharedArrayBuffer。
sink.write("h");
sink.write(new Uint8Array([101, 108]));
sink.write(Buffer.from("lo").buffer);
sink.end();
调用 .end() 后，就不能再向 ArrayBufferSink 写入数据。不过，在缓冲流时，你可能希望继续写入数据，并定期对内容调用 .flush()（例如，写入 WritableStream）。为此，请向 start 方法传入 stream: true。
const sink = new Bun.ArrayBufferSink();
sink.start({
stream: true, // [!code ++]
});
sink.write("h");
sink.write("e");
sink.write("l");
sink.flush();
// ArrayBuffer(3) [ 104, 101, 108 ]
sink.write("l");
sink.write("o");
sink.flush();
// ArrayBuffer(2) [ 108, 111 ]
.flush() 方法会将缓冲的数据作为 ArrayBuffer 返回（如果设置了 asUint8Array: true，则返回 Uint8Array），并清空内部缓冲区。
如果需要手动设置内部缓冲区的大小（以字节为单位），请传入 highWaterMark：
const sink = new Bun.ArrayBufferSink();
sink.start({
highWaterMark: 1024 * 1024, // 1 MB // [!code ++]
});
参考
/**
* 快速增量写入器，调用 end() 后生成 `ArrayBuffer`。
*/
export class ArrayBufferSink {
constructor();
start(options?: {
asUint8Array?: boolean;
/**
* 预分配此大小的内部缓冲区
* 当数据块较小时，能显著提升性能
*/
highWaterMark?: number;
/**
* 在 {@link ArrayBufferSink.flush} 时返回写入数据为 `Uint8Array`。
* 写入将从缓冲区开头重新开始。
*/
stream?: boolean;
}): void;
write(chunk: string | ArrayBufferView | ArrayBuffer | SharedArrayBuffer): number;
/**
* 刷新内部缓冲区
*
* 如果 {@link ArrayBufferSink.start} 传入了 `stream` 选项，则返回一个 `ArrayBuffer`
* 若同时传入了 `stream` 和 `asUint8Array`，则返回 `Uint8Array`
* 否则，返回自上次刷新以来写入的字节数
*
* 该 API 以后可能拆分为 Uint8ArraySink 和 ArrayBufferSink
*/
flush(): number | Uint8Array<ArrayBuffer> | ArrayBuffer;
end(): ArrayBuffer | Uint8Array<ArrayBuffer>;
}