WHATWG Streams

Elide implements the [WHATWG Streams Standard][streams-spec] — ReadableStream, WritableStream, TransformStream, their default and byte readers / writers / controllers, plus the two built-in queuing strategies (CountQueuingStrategy, ByteLengthQueuingStrategy). All thirteen interfaces are exposed on globalThis per the spec; instances follow the spec's state machine, backpressure model, and Promise-based read/write contracts.

[streams-spec]: https://streams.spec.whatwg.org/

Module surface

declare class ReadableStream<R = any> {
  constructor(
    underlyingSource?: UnderlyingSource<R>,
    strategy?: QueuingStrategy<R>,
  );
  readonly locked: boolean;
  cancel(reason?: any): Promise<undefined>;
  getReader(options?: { mode?: "byob" }): ReadableStreamDefaultReader<R> | ReadableStreamBYOBReader;
  pipeThrough<T>(transform: ReadableWritablePair<T, R>, options?: StreamPipeOptions): ReadableStream<T>;
  pipeTo(destination: WritableStream<R>, options?: StreamPipeOptions): Promise<undefined>;
  tee(): [ReadableStream<R>, ReadableStream<R>];
}

declare class WritableStream<W = any> {
  constructor(
    underlyingSink?: UnderlyingSink<W>,
    strategy?: QueuingStrategy<W>,
  );
  readonly locked: boolean;
  abort(reason?: any): Promise<undefined>;
  close(): Promise<undefined>;
  getWriter(): WritableStreamDefaultWriter<W>;
}

declare class TransformStream<I = any, O = any> {
  constructor(
    transformer?: Transformer<I, O>,
    writableStrategy?: QueuingStrategy<I>,
    readableStrategy?: QueuingStrategy<O>,
  );
  readonly readable: ReadableStream<O>;
  readonly writable: WritableStream<I>;
}

declare class CountQueuingStrategy {
  constructor(init: { highWaterMark: number });
  readonly highWaterMark: number;
  readonly size: () => 1;
}

declare class ByteLengthQueuingStrategy {
  constructor(init: { highWaterMark: number });
  readonly highWaterMark: number;
  readonly size: (chunk: { byteLength: number }) => number;
}

Usage

Construct + consume

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

const reader = rs.getReader();
console.log((await reader.read()).value); // "hello"
console.log((await reader.read()).value); // "world"
console.log((await reader.read()).done);  // true

Async pull

const rs = new ReadableStream({
  async pull(controller) {
    const data = await fetchSomething();
    if (data === null) controller.close();
    else controller.enqueue(data);
  },
});

Pipe

await sourceStream.pipeTo(destinationStream, { signal: abortController.signal });

Transform

const upper = new TransformStream({
  transform(chunk, controller) {
    controller.enqueue(chunk.toUpperCase());
  },
});
sourceStream.pipeThrough(upper).pipeTo(destination);

Backpressure

ReadableStream and WritableStream automatically apply backpressure when their internal queues exceed the configured highWaterMark. The default high-water mark is 1 (per the spec); pass a CountQueuingStrategy or ByteLengthQueuingStrategy instance — or any object exposing highWaterMark / size — to tune the behaviour.

const rs = new ReadableStream(
  { /* underlying source */ },
  new CountQueuingStrategy({ highWaterMark: 16 }),
);

Byte streams

Pass type: "bytes" to ReadableStream to construct a byte-mode stream. Such streams accept ArrayBufferView chunks via controller.enqueue(view) and support both default readers (which auto-allocate Uint8Array chunks when autoAllocateChunkSize is set) and BYOB readers via stream.getReader({ mode: "byob" }).

const rs = new ReadableStream({
  type: "bytes",
  start(controller) {
    controller.enqueue(new Uint8Array([1, 2, 3]));
    controller.close();
  },
});
const reader = rs.getReader({ mode: "byob" });
const buf = new Uint8Array(8);
const { value, done } = await reader.read(buf);

ArrayBuffer transfer and detachment

The Streams Standard requires byte streams to transfer the caller-supplied ArrayBuffer whenever a chunk crosses the controller↔reader boundary. Transfer detaches the original buffer (its byteLength becomes 0, every view raises TypeError on access) and replaces it with a fresh internal copy whose lifetime the controller owns. This holds for three call sites:

Call site	Spec §	Behaviour
`controller.enqueue(view)` (byte controller)	§3.13.5.7 step 4	Raises `TypeError` if `view.buffer` is already detached. The internal copy retains `view.byteOffset` / `view.byteLength` semantics.
`reader.read(view)` (BYOB)	§3.13.5.21 step 14	Detaches `view.buffer` synchronously before the read enters the pending-pull-into list. After `read()` returns, mutating the original `view[i]` is impossible — a fresh `Uint8Array` (sharing element type, offset, length with the original) is delivered as `value`.
`byobRequest.respondWithNewView(view)`	§3.13.5.27	Same detach-and-copy as `pullInto`. The user-source must not reuse the supplied buffer.

Memory cost: each transfer allocates one ArrayBuffer whose backing store equals the caller-supplied byte length. There is no in-place transfer fast-path — the spec requires a distinct buffer so user code cannot observe partial writes through the original view. Callers passing very large buffers should be aware that pipelines flowing through BYOB readers will allocate proportionally; for high-throughput byte plumbing prefer autoAllocateChunkSize (which uses controller-owned buffers from the start) over user-supplied views.

The detach is synchronous with the call:

const buf = new Uint8Array(64);
const readPromise = reader.read(buf);
// buf.buffer is already detached — buf[0] = 1 throws TypeError.
const { value } = await readPromise;
// `value` is a fresh Uint8Array. The original `buf` is unusable.

Re-using a transferred buffer raises TypeError. Library code that wants to recycle buffers across reads must obtain a fresh allocation from the read result and manage it externally.

Threading

All streams operations run on the realm's JS thread. The implementation never schedules work on a worker thread, ForkJoinPool, CompletableFuture executor, or any background Executor; every microtask is dispatched through JSContext.enqueuePromiseJob (which is serialised on the realm's owning thread). User-supplied start / pull / cancel / write / close / transform callbacks are therefore guaranteed to observe the same single-threaded execution model as ordinary script code.

If you bridge a stream to background I/O (e.g. by enqueueing chunks from a callback that runs on a JVM thread you control), you must marshal the chunk back to the JS thread before calling controller.enqueue — graal-js's Context is single-threaded by default, and calling into the stream from a non-owning thread will fail an IllegalStateException from the engine. The recommended pattern is a JS-thread executor that owns the realm; the incoming chunk is enqueued onto that executor and Context.enter() / enqueueMicrotask runs the actual controller.enqueue.

Compatibility

Feature	Status
`ReadableStream` (default + byte)	Full
`WritableStream`	Full
`TransformStream`	Full
`getReader()` / `getReader({ mode: "byob" })`	Full
`getWriter()`	Full
`pipeTo` (with preventClose / preventAbort / preventCancel / signal)	Full
`pipeThrough`	Full
`tee` (default + byte streams)	Full
`CountQueuingStrategy` / `ByteLengthQueuingStrategy`	Full
`controller.signal` (writable)	Full
Async iteration on `ReadableStream` (`[Symbol.asyncIterator]` + `values({preventCancel})`)	Full
`ReadableStream.from(iterable)` (sync + async iterables)	Full
Transferable streams via `port.postMessage(stream, [stream])`	Full (same-realm; cross-isolate gated on MessagePort transfer infrastructure)