js/stream

readonly reader: BytesReader

readonly writer: BytesWriter

constructor(capacity: number = 65536)

protected abstract doPullInto( buffer: Uint8Array, options?: BytesReadOptions, ): Promise<ReadResult<number>>

Pull one raw chunk from the underlying resource.

Subclasses may return any positive chunk size in a value result and must return a done result on EOF. Empty chunks are ignored by the buffering layer and should be rare to avoid busy loops. options.signal carries the caller's cancellation request to implementations that wait on an external resource.

import { BufferedBytesReader } from 'fino:stream';
class EmptyBuffered extends BufferedBytesReader {
  async doPullInto() { return { done: true, value: undefined }; }
}
console.log(await new EmptyBuffered().read());

protected async doReadInto( buffer: Uint8Array, options?: BytesReadOptions, ): Promise<ReadResult<number>>

Serve a bounded read from the internal buffer.

The method pulls upstream only while the buffer is empty, skips empty pulls, and returns at most maxBytes. It returns null after upstream EOF and does not over-read from the buffered chunk list.

import { BufferedBytesReader } from 'fino:stream';
class OneChunk extends BufferedBytesReader {
  done = false;
  async doPullInto(buffer) { if (this.done) return { done: true, value: undefined }; this.done = true; buffer.set([1, 2]); return { done: false, value: 2 }; }
}
const result = await new OneChunk().read();
console.log(result.done ? 'eof' : result.value.byteLength);

peek(n: number): Promise<Uint8Array>

Return up to n buffered bytes without consuming them.

The method pulls upstream until at least n bytes are buffered or EOF is reached. It returns a copy of the first min(n, buffered) bytes, so callers can mutate the returned array without changing the internal buffer.

import { BufferedBytesReader } from 'fino:stream';
class OneChunk extends BufferedBytesReader {
  done = false;
  async doPullInto(buffer) {
    if (this.done) return { done: true, value: undefined };
    this.done = true;
    buffer.set([1, 2]);
    return { done: false, value: 2 };
  }
}
const reader = new OneChunk();
console.log((await reader.peek(1))[0]);

scanBuffered(delim: Uint8Array): number

Scan currently buffered bytes for a delimiter without pulling upstream.

Returns the offset one past the first delimiter match, or -1 if the delimiter is absent from the current buffer. Empty delimiters return -1. The method can match delimiters that cross chunk boundaries.

import { BufferedBytesReader } from 'fino:stream';
class Chunked extends BufferedBytesReader {
  chunks = [new Uint8Array([65]), new Uint8Array([10])];
  async doPullInto(buffer) { const chunk = this.chunks.shift(); if (!chunk) return { done: true, value: undefined }; buffer.set(chunk); return { done: false, value: chunk.length }; }
}
const reader = new Chunked();
await reader.peek(2);
console.log(reader.scanBuffered(new Uint8Array([10])));

takeBuffered(n: number): Uint8Array

Remove and return exactly n bytes from the current buffer.

This method never pulls upstream. It throws if fewer than n bytes are buffered. The returned bytes are copied so callers own the buffer.

import { BufferedBytesReader } from 'fino:stream';
class OneChunk extends BufferedBytesReader {
  done = false;
  async doPullInto(buffer) {
    if (this.done) return { done: true, value: undefined };
    this.done = true;
    buffer.set([1, 2]);
    return { done: false, value: 2 };
  }
}
const reader = new OneChunk();
await reader.peek(2);
console.log(reader.takeBuffered(1)[0]);

override readExactly(n: number, options?: BytesReadOptions): Promise<Uint8Array | null>

Read exactly n bytes without consuming on failure.

The method first accumulates bytes with peek(). If EOF arrives before n bytes are available, it returns null and leaves buffered bytes untouched. For n === 0, it returns an empty array.

import { BufferedBytesReader } from 'fino:stream';
class OneChunk extends BufferedBytesReader {
  done = false;
  async doPullInto(buffer) { if (this.done) return { done: true, value: undefined }; this.done = true; buffer[0] = 1; return { done: false, value: 1 }; }
}
const reader = new OneChunk();
console.log(await reader.readExactly(2));
console.log(reader.buffered);

override readUntil( delim: Uint8Array, max: number = 1 << 20, options?: BytesReadOptions, ): Promise<Uint8Array | null>

Read through a delimiter without consuming on failure.

The method scans buffered bytes first, pulls one chunk at a time as needed, and consumes only when a delimiter is found. EOF before a match returns null with buffered bytes preserved. Empty delimiters and scans beyond max throw.

import { BufferedBytesReader } from 'fino:stream';
class Lines extends BufferedBytesReader {
  chunks = [new TextEncoder().encode('a\\n')];
  async doPullInto(buffer) { const chunk = this.chunks.shift(); if (!chunk) return { done: true, value: undefined }; buffer.set(chunk); return { done: false, value: chunk.length }; }
}
const reader = new Lines();
console.log(new TextDecoder().decode(await reader.readUntil(new Uint8Array([10]))));

static over(source: BytesReader): BufferedBytesReader

Wrap an existing byte reader in a buffered reader.

The wrapper pulls from source.read() and closes the source when the buffered reader closes. This is useful for tests and for adding non- consuming peek() and readUntil() behavior to an unbuffered source.

import { BufferedBytesReader, BytesReader } from 'fino:stream';
class EmptyBytes extends BytesReader { async doReadInto() { return { done: true, value: undefined }; } }
const buffered = BufferedBytesReader.over(new EmptyBytes());
console.log(buffered.buffered);

get buffered(): number

Number of bytes currently buffered.

These bytes can be consumed by takeBuffered() without awaiting upstream I/O. The value does not include bytes that may still be available from the underlying resource.

import { BufferedBytesReader } from 'fino:stream';
class EmptyBuffered extends BufferedBytesReader { async doPullInto() { return { done: true, value: undefined }; } }
console.log(new EmptyBuffered().buffered);

override get bufferedBytes(): number

get eof(): boolean

Whether upstream EOF has been reached and all buffered bytes are drained.

The value is false before the first EOF-producing pull, even if no bytes are currently buffered. Use peek() or read() to discover EOF.

import { BufferedBytesReader } from 'fino:stream';
class EmptyBuffered extends BufferedBytesReader { async doPullInto() { return { done: true, value: undefined }; } }
const reader = new EmptyBuffered();
await reader.peek(1);
console.log(reader.eof);

constructor(onClose: () => void | Promise<void> = () => {}, bufferSize: number = 65536)

Create a buffered byte writer.

bufferSize defaults to 64 KiB. A smaller buffer flushes more often; a larger buffer can reduce syscall frequency at the cost of memory. onClose is invoked after pending bytes are flushed.

import { BufferedBytesWriter } from 'fino:stream';
class Sink extends BufferedBytesWriter { async doFlush(_buf) {} }
const writer = new Sink(() => {}, 1024);
await writer.close();

protected abstract doFlush(buf: Uint8Array): Promise<void>

Flush a coalesced byte slice to the underlying resource.

Implementations must emit all bytes in buf or throw. Calls are serialized per writer, so an implementation that suspends on backpressure keeps the descriptor to itself until it returns.

import { BufferedBytesWriter } from 'fino:stream';
class Sink extends BufferedBytesWriter {
  async doFlush(buf) { console.log(buf.byteLength); }
}
await new Sink().write(new Uint8Array([1]));

protected doWrite(buf: Uint8Array): Promise<void>

Coalesce or immediately flush one byte buffer.

Buffers at least as large as the coalesce buffer bypass accumulation after pending bytes are flushed. Smaller buffers are copied into the internal buffer, flushing first if needed.

The base Writer invokes this hook while holding the per-instance FIFO, so callers cannot contend for buffer room or overlap flushes.

import { BufferedBytesWriter } from 'fino:stream';
class Sink extends BufferedBytesWriter { async doFlush(_buf) {} }
await new Sink().write(new Uint8Array([1, 2]));

writev(vecs: Uint8Array[], count: number = vecs.length): Promise<void>

Write multiple byte buffers through the coalescing buffer.

The complete vector batch occupies one writer operation. Small vectors are accumulated and flushed only when the buffer fills. A vector at least as large as the coalesce buffer flushes pending bytes first and then bypasses accumulation.

import { BufferedBytesWriter } from 'fino:stream';
class Sink extends BufferedBytesWriter { async doFlush(_buf) {} }
await new Sink().writev([new Uint8Array([1]), new Uint8Array([2])]);

protected _directAccumulate(buf: Uint8Array): boolean

Synchronously copy buf into the coalesce buffer.

Returns true if the bytes were accumulated; false if the buffer doesn't have enough room (caller must await flush() first, then retry). Only safe to call when buf.byteLength < this.#buf.byteLength.

import { BufferedBytesWriter } from 'fino:stream';
class Sink extends BufferedBytesWriter {
  async doFlush(_buf) {}
  tryAccumulate(buf) { return this._directAccumulate(buf); }
}
console.log(new Sink().tryAccumulate(new Uint8Array([1])));

override async flush(): Promise<void>

Flush-hook implementation that drains the coalesce buffer.

The public base flush() method invokes this after earlier writes. With no pending bytes it is a no-op. Errors from doFlush() reject the caller and the pending count has already been reset.

The pending bytes are taken as a copy: doFlush() can suspend on backpressure, and the coalesce buffer it was handed would otherwise be refilled from offset zero by the next write while those bytes were still on their way out.

import { BufferedBytesWriter } from 'fino:stream';
class Sink extends BufferedBytesWriter { async doFlush(_buf) {} }
const writer = new Sink();
await writer.write(new Uint8Array([1]));
await writer.flush();

protected _takePending(): Uint8Array | null

Return the buffered bytes (a copy) and reset the pending count. Used by subclasses that need to perform a synchronous flush (e.g. on process exit) without going through the async flush path.

import { BufferedBytesWriter } from 'fino:stream';
class Sink extends BufferedBytesWriter {
  async doFlush(_buf) {}
  take() { return this._takePending(); }
}
const writer = new Sink();
await writer.write(new Uint8Array([1]));
console.log(writer.take()?.byteLength);

static over(target: BytesWriter, bufferSize: number = 65536): BufferedBytesWriter

Wrap an existing byte writer with coalescing behavior.

The wrapper flushes by calling target.write(buf) and closes the target when the wrapper closes. bufferSize defaults to 64 KiB.

import { BufferedBytesWriter, BytesWriter } from 'fino:stream';
class Sink extends BytesWriter { async doWrite(_buf) {} }
const buffered = BufferedBytesWriter.over(new Sink(), 4096);
await buffered.close();

constructor(stateOrClose: BytesReadableState | ReaderCloseCallback = () => {})

protected async doReadInto( buffer: Uint8Array, options?: BytesReadOptions, ): Promise<ReadResult<number>>

Read bytes from the underlying source.

Implementations fill at most buffer.byteLength bytes, may fill fewer, and return the number written in a value result or a done result on EOF. Backends should avoid returning zero when possible because structural helpers may need to retry.

import { BytesReader } from 'fino:stream';
class EmptyBytes extends BytesReader {
  async doReadInto(_buffer) { return { done: true, value: undefined }; }
}
console.log(await new EmptyBytes().read());

protected onConsume(_bytes: number): void

Called after bytes are delivered to the public reader caller.

The default implementation is a no-op. Protocol adapters can override this to report backpressure progress, for example by extending QUIC stream and connection flow-control credit. The hook is not called for bytes pulled internally and then stashed after an incomplete structural read.

read(options?: number | BytesReadOptions): Promise<ReadResult<Uint8Array>>

Read one byte chunk.

The default request size is 64 KiB, but subclasses may return fewer bytes. Any bytes stashed by an earlier partial structural read are returned before the underlying doReadInto() hook is called. The returned allocation is owned by the caller and remains stable across later reads. A result with done: true means EOF.

import { BytesReader } from 'fino:stream';
class EmptyBytes extends BytesReader {
  async doReadInto(_buffer) { return { done: true, value: undefined }; }
}
console.log(await new EmptyBytes().read());

readAtMost( maxBytes: number, options: Omit<BytesReadOptions, 'maxBytes'> = {}, ): Promise<ReadResult<Uint8Array>>

Read at most maxBytes bytes.

This is a named convenience around read({ maxBytes }) for protocols that need explicit bounded consumption.

readInto( buffer: ArrayBufferView, options: Omit<BytesReadOptions, 'maxBytes'> = {}, ): Promise<ReadResult<number>>

Read bytes directly into caller-provided view storage.

The reader borrows the exact view, including its offset and length, until the returned promise settles. The caller must not mutate or reuse that region while the read is pending; the reader does not retain it afterward. Returns the number of bytes written in a value result, or a done result on EOF, and never writes more than buffer.byteLength.

The view's byte offset and length are preserved, allowing slices of pooled or arena-allocated buffers to be filled without an intermediate copy.

readExactly(n: number, options?: BytesReadOptions): Promise<Uint8Array | null>

Read exactly n bytes.

Returns an empty array for n === 0. If EOF arrives before n bytes are available, returns null and stashes bytes already read so the next read operation can replay them. Buffered readers override this with atomic, non-consuming failure semantics.

import { BytesReader } from 'fino:stream';
class MemoryReader extends BytesReader {
  data = new Uint8Array([1, 2]);
  async doReadInto(buffer) {
    if (!this.data.byteLength) return { done: true, value: undefined };
    const n = Math.min(buffer.byteLength, this.data.byteLength);
    buffer.set(this.data.subarray(0, n));
    this.data = this.data.subarray(n);
    return { done: false, value: n };
  }
}
const reader = new MemoryReader();
console.log((await reader.readExactly(2))?.byteLength);

readByte(options?: BytesReadOptions): Promise<number | null>

Read a single byte.

The returned number is in the range 0 through 255. null indicates EOF. Stashed bytes from a previous partial structural read are consumed before the underlying source is queried.

import { BytesReader } from 'fino:stream';
class OneByte extends BytesReader {
  done = false;
  async doReadInto(buffer) {
    if (this.done) return { done: true, value: undefined };
    this.done = true;
    buffer[0] = 97;
    return { done: false, value: 1 };
  }
}
console.log(await new OneByte().readByte());

readUntil( delim: Uint8Array, max: number = 1 << 20, options?: BytesReadOptions, ): Promise<Uint8Array | null>

Read through the first delimiter occurrence.

The returned bytes include delim. An empty delimiter throws. EOF before the delimiter returns null; bytes scanned before EOF are stashed for the next read. Scanning more than max bytes without a delimiter throws.

import { BytesReader } from 'fino:stream';
class MemoryReader extends BytesReader {
  data = new TextEncoder().encode('ok\\nrest');
  async doReadInto(buffer) {
    if (!this.data.byteLength) return { done: true, value: undefined };
    const n = Math.min(buffer.byteLength, this.data.byteLength);
    buffer.set(this.data.subarray(0, n));
    this.data = this.data.subarray(n);
    return { done: false, value: n };
  }
}
const line = await new MemoryReader().readUntil(new Uint8Array([10]));
console.log(new TextDecoder().decode(line));

get bufferedBytes(): number

Bytes already held by this reader before another source pull is required.

For the base unbuffered reader this only includes bytes stashed after a partial structural read. Buffered readers include their chunk-list buffer.

readonly reader: BytesReader

readonly writer: BytesWriter

constructor()

constructor(stateOrClose: BytesWritableState | WriterCloseCallback = () => {})

protected doWrite(buf: Uint8Array): Promise<void>

Emit all bytes in buf to the underlying resource.

Implementations must handle partial writes, backpressure, and sink errors internally. The base write() method has already converted input to a Uint8Array and checked the closed state.

import { BytesWriter } from 'fino:stream';
class MemoryWriter extends BytesWriter {
  async doWrite(buf) { console.log(buf.byteLength); }
}
await new MemoryWriter().write(new Uint8Array([1, 2]));

async write(data: ArrayBuffer | ArrayBufferView): Promise<void>

Normalize one byte source admitted by the base writer.

ArrayBuffer and ArrayBufferView inputs are wrapped in a Uint8Array preserving view byte offsets, then forwarded to doWrite() as one ordered writer operation.

import { BytesWriter } from 'fino:stream';
class MemoryWriter extends BytesWriter {
  bytes = 0;
  async doWrite(buf) { this.bytes += buf.byteLength; }
}
const writer = new MemoryWriter();
await writer.write(new ArrayBuffer(4));
console.log(writer.bytes);

writev(vecs: Uint8Array[], count: number = vecs.length): Promise<void>

Write multiple buffers in order.

The default implementation writes up to count vectors sequentially and skips missing or empty entries. Subclasses may override for vectorized system calls. Errors from any individual write abort the sequence.

import { BytesWriter } from 'fino:stream';
class MemoryWriter extends BytesWriter {
  bytes = 0;
  async doWrite(buf) { this.bytes += buf.byteLength; }
}
const writer = new MemoryWriter();
await writer.writev([new Uint8Array([1]), new Uint8Array([2])]);
console.log(writer.bytes);

readonly writer: Writer<T>

Producer half of the channel.

write(value) waits for the reader to accept the value. close() signals EOF after admitted values. close(error) drains them and then surfaces a terminal error.

readonly reader: Reader<T>

Consumer half of the channel.

A standard Reader<T>: iterate it with for await, drive it with reader.read(), or feed another Writer with writer.pipe(reader).

constructor()

Create a connected zero-capacity writer/reader pair.

The two halves share rendezvous state but no value buffer. Reads and writes pair in FIFO order.

constructor((message = 'Channel reader closed before delivery completed'))

constructor(stateOrClose: ReadableState<T, Options> | ReaderCloseCallback = () => {})

Create a reader with an optional close callback.

The callback defaults to a no-op and is invoked at most once, the first time close() is awaited or when iteration reaches EOF. Callback errors reject the close operation.

import { Reader } from 'fino:stream';
const reader = new Reader(() => console.log('closed'));
await reader.close();

static from<T>(source: AsyncIterable<T>): Reader<T>

Wrap an async iterable in the standard pull-based reader facade.

get closed(): boolean

Whether the reader has been closed.

The flag flips before the close callback is awaited. It remains false while the reader is open, even if EOF has not yet been checked.

import { Reader } from 'fino:stream';
const reader = Reader.from((async function* () {})());
console.log(reader.closed);
await reader.close();
console.log(reader.closed);

read(options?: Options): Promise<ReadResult<T>>

Pull the next value from the state as an explicit iterator result.

import { Reader } from 'fino:stream';
const reader = Reader.from((async function* () {})());
console.log(await reader.read());

close(error?: Error): Promise<void>

Close the reader and run its close callback once.

Multiple calls are safe; only the first one invokes onClose. The method does not call read() and does not require EOF. Callback failures reject the returned promise. On channel-backed readers, an omitted error is clean early cancellation; providing an error propagates that error upstream.

import { Reader } from 'fino:stream';
const reader = Reader.from((async function* () {})());
await reader.close();
await reader.close();

async next(): Promise<IteratorResult<T>>

Advance the async iterator.

next() calls read(). When read() returns a done result, the reader is closed. Source errors or close callback errors reject the returned promise.

import { Reader } from 'fino:stream';
const reader = Reader.from((async function* () {})());
const result = await reader.next();
console.log(result.done);

readonly reader: BytesReader

readonly writer: BytesWriter

constructor(segmentCapacity: number = 65536)

readonly writer: Writer<T>

Writer that accepts values into the unbounded FIFO.

readonly reader: Reader<T>

Reader that removes accepted values in FIFO order.

constructor()

Create an empty unbounded channel.

constructor(stateOrClose: WritableState<T> | WriterCloseCallback = () => {})

Create a writer with an optional close callback.

The callback defaults to a no-op and is invoked at most once. Callback failures reject close().

import { Writer } from 'fino:stream';
class NullWriter extends Writer { async write(_value) {} }
const writer = new NullWriter(() => console.log('closed'));
await writer.close();

get closed(): boolean

Whether the writer has been closed.

The flag is set as soon as close() begins, before admitted work drains and before the close callback is awaited. The base class rejects later writes.

import { Writer } from 'fino:stream';
class NullWriter extends Writer { async write(_value) {} }
const writer = new NullWriter();
await writer.close();
console.log(writer.closed);

write(value: T): Promise<void>

Write one value to the state.

import { Writer } from 'fino:stream';
class ArrayWriter extends Writer {
  values = [];
  async write(value) { this.values.push(value); }
}
await new ArrayWriter().write('x');

writeSync?(value: T): void

Write one value synchronously when the writer supports synchronous acceptance.

This method is an optional capability for hot paths that must not yield between producing bytes and updating native state. Implementations should either accept the value completely or throw. The base automatically prevents it from overtaking pending asynchronous work. Writer does not provide a fallback because calling async write() from a sync-only path would hide an ordering bug.

import { Writer } from 'fino:stream';
function writeNow(writer, value) {
  if (writer.writeSync === undefined) throw new Error('sync writes unavailable');
  writer.writeSync(value);
}

async pipe(source: AsyncIterable<T>): Promise<void>

Consume an async iterable and write each value in order.

The method awaits each write() before reading the next source value, preserving backpressure. It does not close the writer or the source.

import { Writer } from 'fino:stream';
class ArrayWriter extends Writer {
  values = [];
  async write(value) { this.values.push(value); }
}
const writer = new ArrayWriter();
await writer.pipe(['a', 'b']);
console.log(writer.values.length);

flush(): Promise<void>

Flush internally buffered data.

Subclass hook for flushing internally buffered data.

import { Writer } from 'fino:stream';
class NullWriter extends Writer { async write(_value) {} }
await new NullWriter().flush();

close(error?: Error): Promise<void>

Close the writer and run its close callback once.

Multiple calls return the same cleanup promise. New operations are rejected immediately; operations admitted earlier drain in FIFO order before the subclass close hook and cleanup callback run. Providing an error closes the stream after that drain and makes the reader reject at the terminal point.

import { Writer } from 'fino:stream';
class NullWriter extends Writer { async write(_value) {} }
const writer = new NullWriter();
await writer.close();
await writer.close();

closeSync?(): void

Close the writer synchronously when the writer supports synchronous close.

This optional capability is for callers that need close state to be visible immediately. Implementations should mark the writer closed before returning and perform only synchronous cleanup. Callers that can yield should continue to use close().

import { Writer } from 'fino:stream';
function closeNow(writer) {
  if (writer.closeSync === undefined) throw new Error('sync close unavailable');
  writer.closeSync();
}

maxBytes?: number

Upper bound on the number of bytes a single read may return.

The reader is free to return fewer bytes, and never returns more. Omitting it lets each method choose its own default: 64 KiB for read, one byte for readByte, and the exact requested count for readExactly.

signal?: AbortSignal | null

Abort signal that cancels the read.

The signal is checked before each underlying source pull, so an already-aborted signal rejects before any bytes are touched and an abort that lands between chunks stops a multi-chunk structural read. The rejection carries the signal's reason. Passing null is equivalent to omitting it.