스트림은 메모리에 한 번에 모두 로드하지 않고 바이너리 데이터를 작업하기 위한 중요한 추상화입니다. 파일 읽기 및 쓰기, 네트워크 요청 보내기 및 받기, 대량의 데이터 처리에 일반적으로 사용됩니다.

Bun 은 웹 API 인 ReadableStream 과 WritableStream 을 구현합니다.

NOTE

Bun 은 또한 [`Readable`](https://nodejs.org/api/stream.html#stream_readable_streams), [`Writable`](https://nodejs.org/api/stream.html#stream_writable_streams), [`Duplex`](https://nodejs.org/api/stream.html#stream_duplex_and_transform_streams) 를 포함한 `node:stream` 모듈도 구현합니다. 완전한 문서는 [Node.js 문서](https://nodejs.org/api/stream.html) 를 참조하세요.

간단한 ReadableStream 을 생성하려면:

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

ReadableStream 의 콘텐츠는 for await 구문을 사용하여 청크 단위로 읽을 수 있습니다.

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

// hello
// world

---

직접 ReadableStream

Bun 은 불필요한 데이터 복사 및 큐 관리 로직을 피하는 최적화된 ReadableStream 을 구현합니다.

기존 ReadableStream 을 사용하면 데이터 청크가 인큐 (enqueue) 됩니다. 각 청크는 큐에 저장되어 스트림이 더 많은 데이터를 보낼 준비가 될 때까지 대기합니다.

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

직접 ReadableStream 을 사용하면 데이터 청크가 스트림에 직접 기록됩니다. 큐잉이 발생하지 않으며 청크 데이터를 메모리에 복제할 필요가 없습니다. controller API 는 이를 반영하도록 업데이트되었습니다. .enqueue() 대신 .write 를 호출합니다.

const stream = new ReadableStream({
  type: "direct", 
  pull(controller) {
    controller.write("hello");
    controller.write("world");
  },
});

직접 ReadableStream 을 사용할 때 모든 청크 큐잉은 대상에서 처리됩니다. 스트림의 소비자는 인코딩이나 수정 없이 controller.write() 에 전달된 내용을 정확히 받습니다.

---

비동기 제너레이터 스트림

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 는 직접 ReadableStream 컨트롤러를 반환합니다.

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

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() 하는 것이 유용합니다 (예: WriteableStream 으로). 이를 지원하려면 생성자에 stream: true 를 전달하세요.

const sink = new Bun.ArrayBufferSink();
sink.start({
  stream: true, 
});

sink.write("h");
sink.write("e");
sink.write("l");
sink.flush();
// ArrayBuffer(5) [ 104, 101, 108 ]

sink.write("l");
sink.write("o");
sink.flush();
// ArrayBuffer(5) [ 108, 111 ]

.flush() 메서드는 버퍼링된 데이터를 ArrayBuffer 로 반환하고 (또는 asUint8Array: true 인 경우 Uint8Array) 내부 버퍼를 지웁니다.

내부 버퍼의 크기를 바이트 단위로 수동으로 설정하려면 highWaterMark 에 값을 전달하세요:

const sink = new Bun.ArrayBufferSink();
sink.start({
  highWaterMark: 1024 * 1024, // 1 MB
});

---

참조

/**
 * Fast incremental writer that becomes an `ArrayBuffer` on end().
 */
export class ArrayBufferSink {
  constructor();

  start(options?: {
    asUint8Array?: boolean;
    /**
     * Preallocate an internal buffer of this size
     * This can significantly improve performance when the chunk size is small
     */
    highWaterMark?: number;
    /**
     * On {@link ArrayBufferSink.flush}, return the written data as a `Uint8Array`.
     * Writes will restart from the beginning of the buffer.
     */
    stream?: boolean;
  }): void;

  write(chunk: string | ArrayBufferView | ArrayBuffer | SharedArrayBuffer): number;
  /**
   * Flush the internal buffer
   *
   * If {@link ArrayBufferSink.start} was passed a `stream` option, this will return a `ArrayBuffer`
   * If {@link ArrayBufferSink.start} was passed a `stream` option and `asUint8Array`, this will return a `Uint8Array`
   * Otherwise, this will return the number of bytes written since the last flush
   *
   * This API might change later to separate Uint8ArraySink and ArrayBufferSink
   */
  flush(): number | Uint8Array<ArrayBuffer> | ArrayBuffer;
  end(): ArrayBuffer | Uint8Array<ArrayBuffer>;
}