# Streams (/runtime/streams)

<!-- agent-signals: reading_time_min: 6 · est_tokens: 2252 · updated: 2026-09-23 -->
Related: [Watch Mode](/runtime/watch-mode.md), [Debugging](/runtime/debugger.md), [REPL](/runtime/repl.md), [bunfig.toml](/runtime/bunfig.md), [File Types](/runtime/file-types.md), [Module Resolution](/runtime/module-resolution.md)

> Use Bun's streams API to work with binary data without loading it all into memory at once

Streams are an important abstraction for working with binary data without loading it all into memory at once. They are commonly used for reading and writing files, sending and receiving network requests, and processing large amounts of data.

Bun implements the Web APIs [`ReadableStream`](https://developer.mozilla.org/en-US/docs/Web/API/ReadableStream) and [`WritableStream`](https://developer.mozilla.org/en-US/docs/Web/API/WritableStream).

<Note>Bun also implements the `node:stream` module, including
[`Readable`](https://nodejs.org/api/stream.html#stream_readable_streams),
[`Writable`](https://nodejs.org/api/stream.html#stream_writable_streams), and
[`Duplex`](https://nodejs.org/api/stream.html#stream_duplex_and_transform_streams). For complete documentation, refer
to the [Node.js docs](https://nodejs.org/api/stream.html).</Note>

To create a `ReadableStream`:

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

You can read the contents of a `ReadableStream` chunk-by-chunk with `for await` syntax.

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

// hello
// world
```

***

## Direct `ReadableStream`

Bun implements an optimized version of `ReadableStream` that avoids unnecessary queue management.

With a traditional `ReadableStream`, you *enqueue* chunks of data. The stream adds each chunk to a queue, where it sits until the stream is ready to send more data.

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

With a direct `ReadableStream`, you write chunks of data directly to the stream. No queueing happens. The `controller` API reflects this: you call `.write()` instead of `.enqueue()`.

```ts
const stream = new ReadableStream({
  type: "direct", // [!code ++]
  pull(controller) {
    controller.write("hello");
    controller.write("world");
    controller.close();
  },
});
```

When using a direct `ReadableStream`, the destination handles all chunk queueing. The destination receives the bytes you pass to `controller.write()`. When the stream is read from JavaScript, Bun buffers the writes and delivers them as `Uint8Array` chunks (strings are UTF-8 encoded).

### Ending the stream

Call `controller.close()` (or `controller.end()`) once everything is written. Every destination flushes what it has buffered and finishes at that point. `controller.close(error)` fails the stream instead: nothing more is flushed and the consumer rejects with `error`.

A destination that takes the whole body (`Bun.serve`, `fetch`, `Bun.write`, `Bun.spawn` stdin, `.text()`, and so on) calls `pull()` once. If it returns a promise, the stream stays open while that promise is pending and ends when it resolves, as if `pull()` had called `controller.close()` last; if it rejects, the stream errors with that reason. A `pull()` that returns without a promise and without closing keeps the stream open until something calls `controller.close()`, so you can hold on to the controller and write from events.

When the stream is read through a reader (`getReader()`, `for await`, `pipeTo()`, `tee()`), `pull()` is a demand signal instead: it is called again for a later read, never while a previous call's promise is still pending, until `controller.close()`.

### Handling backpressure

`controller.write()` returns the number of bytes written, or a &#x2A;*pending `Promise<number>`** when the destination's internal buffer is full (for example, a slow HTTP client). The chunk is accepted either way. Once the destination has gone away (for example the HTTP client disconnected), `write()` returns `0`. The promise resolves once the destination has drained, so `await`ing the result is enough:

```ts
const stream = new ReadableStream({
  type: "direct",
  async pull(controller) {
    for (const chunk of chunks) {
      await controller.write(chunk);
    }
    controller.close();
  },
});
```

`await controller.flush(true)` is equivalent, and you can use it after a write returns a `Promise`.

When the stream is read from JavaScript (`getReader()`, `pipeTo()`, `for await`), the buffer is full once `highWaterMark` bytes (default 64 KiB) are waiting for the reader. Pass `{ highWaterMark }` as the second `ReadableStream` constructor argument to change it. `reader.cancel(reason)` calls the source's `cancel(reason)`.

For default (non-`direct`) `ReadableStream`s and async-generator response bodies, Bun applies this backpressure automatically: it pauses the producer while the destination is backed up.

***

## Async generator streams

Bun also supports async generator functions as a source for `Response` and `Request`. Use async generators to create a `ReadableStream` that fetches data from an asynchronous source.

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

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

You can also use `[Symbol.asyncIterator]` directly.

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

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

The body contains the yielded values only. As with `for await`, the generator's return value is not part of the body.

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

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

For more control over the stream, `yield` returns the direct `ReadableStream` controller.

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

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

***

## `Bun.ArrayBufferSink`

The `Bun.ArrayBufferSink` class is a fast incremental writer for constructing an `ArrayBuffer` of unknown size.

```ts
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 ]
```

To instead retrieve the data as a `Uint8Array`, pass the `asUint8Array` option to the `start` method.

```ts
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 ]
```

The `.write()` method supports strings, typed arrays, `ArrayBuffer`, and `SharedArrayBuffer`.

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

sink.end();
```

Once you call `.end()`, you can't write any more data to the `ArrayBufferSink`. However, when buffering a stream you may want to keep writing data and periodically `.flush()` the contents (say, into a `WritableStream`). To support this, pass `stream: true` to the `start` method.

```ts
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 ]
```

The `.flush()` method returns the buffered data as an `ArrayBuffer` (or `Uint8Array` if `asUint8Array: true`) and clears the internal buffer.

To manually set the size of the internal buffer in bytes, pass a value for `highWaterMark`:

```ts
const sink = new Bun.ArrayBufferSink();
sink.start({
  highWaterMark: 1024 * 1024, // 1 MB  // [!code ++]
});
```

***

## Reference

```ts expandable title="See Typescript Definitions"
/**
 * 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>;
}
```