流 · 文件

流是處理二進位制資料而無需一次性全部載入到記憶體中的重要抽象。它們通常用於讀寫檔案、傳送和接收網路請求以及處理大量資料。

Bun 實現了 Web API 中的 ReadableStream 和 WritableStream。

建立一個 ReadableStream：

TypeScript

const stream = new ReadableStream({
  start(controller) {
    controller.enqueue("hello");
    controller.enqueue("world");
    controller.close();
  },
});

你可以使用 for await 語法逐塊讀取 ReadableStream 的內容。

TypeScript

for await (const chunk of stream) {
  console.log(chunk);
}

// 你好
// 世界

直接 `ReadableStream`

Bun 實現了最佳化版的 ReadableStream，避免了不必要的佇列管理。

使用傳統的 ReadableStream 時，你需要將資料塊入隊。流會將每個資料塊新增到佇列中，直到流準備好傳送更多資料時，這些資料塊才會留在佇列中。

TypeScript

const stream = new ReadableStream({
  start(controller) {
    controller.enqueue("hello");
    controller.enqueue("world");
    controller.close();
  },
});

使用直接 ReadableStream 時，你可以直接將資料塊寫入流中，不會發生排隊。controller API 也體現了這一點：你呼叫 .write() 而不是 .enqueue()。

TypeScript

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：

TypeScript

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。

TypeScript

const response = new Response(
  (async function* () {
    yield "hello";
    yield "world";
  })(),
);

await response.text(); // "helloworld"

你也可以直接使用 [Symbol.asyncIterator]。

TypeScript

const response = new Response({
  [Symbol.asyncIterator]: async function* () {
    yield "hello";
    yield "world";
  },
});

await response.text(); // "helloworld"

訊息體只包含生成器 yield 的值。與 for await 一樣，生成器的傳回值不屬於訊息體。

TypeScript

const response = new Response(
  (async function* () {
    yield "hello";
    return "ignored";
  })(),
);

await response.text(); // "hello"

如需更精細地控制流，yield 會傳回直接 ReadableStream 的 controller。

TypeScript

const response = new Response({
  [Symbol.asyncIterator]: async function* () {
    const controller = yield "hello";
    await controller.end();
  },
});

await response.text(); // "hello"

`Bun.ArrayBufferSink`

Bun.ArrayBufferSink 類是一個用於構造未知大小 ArrayBuffer 的高速增量寫入器。

TypeScript

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 選項。

TypeScript

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。

TypeScript

sink.write("h");
sink.write(new Uint8Array([101, 108]));
sink.write(Buffer.from("lo").buffer);

sink.end();

呼叫 .end() 後，就不能再向 ArrayBufferSink 寫入資料。不過，在緩衝流時，你可能希望繼續寫入資料，並定期對內容呼叫 .flush()（例如，寫入 WritableStream）。為此，請向 start 方法傳入 stream: true。

TypeScript

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：

TypeScript

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>;
}

直接 ReadableStream

結束流

處理背壓

非同步生成器流

Bun.ArrayBufferSink

參考

直接 `ReadableStream`

`Bun.ArrayBufferSink`