Streams

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 and WritableStream.

Bun also implements the node:stream module, including Readable, Writable, and Duplex. For complete documentation, refer to the Node.js docs.

To create a ReadableStream:

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

The contents of a ReadableStream can be read chunk-by-chunk with for await syntax.

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

// hello
// world

Direct `ReadableStream`

Bun implements an optimized version of ReadableStream that avoids unnecessary data copying and queue management. With a traditional ReadableStream, chunks of data are enqueued. Each chunk is copied into a queue, where it sits until the stream is ready to send more data.

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

With a direct ReadableStream, chunks of data are written directly to the stream. No queueing happens, and there’s no need to clone the chunk data into memory. The controller API reflects this: you call .write() instead of .enqueue().

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

When using a direct ReadableStream, the destination handles all chunk queueing. The consumer of the stream receives exactly what is passed to controller.write(), without any encoding or modification.

Handling backpressure

controller.write() returns the number of bytes written. When the destination’s internal buffer is full (for example, a slow HTTP client), it returns a negative number instead. The chunk is still accepted — the negative return is a signal to pause and wait for the destination to drain. To wait for the drain, await controller.flush(true):

const stream = new ReadableStream({
  type: "direct",
  async pull(controller) {
    for (const chunk of chunks) {
      const n = controller.write(chunk);
      if (typeof n === "number" && n < 0) {
        // destination is backed up; wait for it to drain before writing more
        await controller.flush(true);
      }
    }
    controller.close();
  },
});

For default (non-direct) ReadableStreams and async-generator response bodies, Bun applies this backpressure automatically — the producer is paused 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.

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

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

You can also use [Symbol.asyncIterator] directly.

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

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

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

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.

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.

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 ]

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

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

sink.end();

Once .end() is called, no more data can be written 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.

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

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:

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

Reference

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

Get Started

Core Runtime

File & Module System

HTTP server

Networking

Data & Storage

Concurrency

Process & System

Interop & Tooling

Utilities

Standards & Compatibility

Contributing

Direct `ReadableStream`

Handling backpressure

Async generator streams

`Bun.ArrayBufferSink`

Reference

​Direct ReadableStream

​Handling backpressure

​Async generator streams

​Bun.ArrayBufferSink

​Reference

Direct `ReadableStream`

Handling backpressure

Async generator streams

`Bun.ArrayBufferSink`

Reference