js/stream

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.

constructor(onClose: 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';
class EmptyReader extends Reader { async read() { return null; } }
const reader = new EmptyReader(() => console.log('closed'));
await reader.close();

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';
class EmptyReader extends Reader { async read() { return null; } }
const reader = new EmptyReader();
console.log(reader.closed);
await reader.close();
console.log(reader.closed);

abstract read(): Promise<T | null>

Produce the next value from the stream.

Subclasses must return null to signal EOF. Returning undefined is a value, not EOF, for generic readers. Implementations may throw for source errors; the async iterator forwards those errors to the caller.

import { Reader } from 'fino:stream';
class EmptyReader extends Reader { async read() { return null; } }
console.log(await new EmptyReader().read());

async close(): 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.

import { Reader } from 'fino:stream';
class EmptyReader extends Reader { async read() { return null; } }
const reader = new EmptyReader();
await reader.close();
await reader.close();

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

Advance the async iterator.

next() calls read(). When read() returns null, the reader is closed and the iterator result is { done: true }. Source errors or close callback errors reject the returned promise.

import { Reader } from 'fino:stream';
class EmptyReader extends Reader { async read() { return null; } }
const result = await new EmptyReader().next();
console.log(result.done);

protected abstract doRead(maxBytes: number, options?: BytesReadOptions): Promise<Uint8Array | null>

Read bytes from the underlying source.

Implementations must return at most maxBytes bytes, may return fewer, and must return null on EOF. Empty chunks are allowed but can cause structural helpers to loop, so backends should avoid returning them when possible.

import { BytesReader } from 'fino:stream';
class EmptyBytes extends BytesReader {
  async doRead(_maxBytes) { return null; }
}
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.

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

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 doRead() hook is called. null means EOF.

import { BytesReader } from 'fino:stream';
class EmptyBytes extends BytesReader {
  async doRead(_maxBytes) { return null; }
}
console.log(await new EmptyBytes().read());

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

Read at most maxBytes bytes.

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

async readInto( buffer: Uint8Array, options: Omit<BytesReadOptions, 'maxBytes'> = { } ): Promise<number | null>

Read bytes directly into caller-provided storage.

Returns the number of bytes copied, or null on EOF. The method never copies more than buffer.byteLength and relies on readAtMost() for consumption accounting.

async 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 doRead(maxBytes) {
    if (!this.data.byteLength) return null;
    const out = this.data.subarray(0, maxBytes);
    this.data = this.data.subarray(out.byteLength);
    return out;
  }
}
const reader = new MemoryReader();
console.log((await reader.readExactly(2))?.byteLength);

async 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 doRead(_maxBytes) {
    if (this.done) return null;
    this.done = true;
    return new Uint8Array([97]);
  }
}
console.log(await new OneByte().readByte());

async 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 doRead(maxBytes) {
    if (!this.data.byteLength) return null;
    const out = this.data.subarray(0, maxBytes);
    this.data = this.data.subarray(out.byteLength);
    return out;
  }
}
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.

protected abstract doPull(): Promise<Uint8Array | null>

Pull one raw chunk from the underlying resource.

Subclasses may return any positive chunk size and must return null on EOF. Empty chunks are ignored by the buffering layer and should be rare to avoid busy loops.

import { BufferedBytesReader } from 'fino:stream';
class EmptyBuffered extends BufferedBytesReader {
  async doPull() { return null; }
}
console.log(await new EmptyBuffered().read());

protected async doRead(maxBytes: number, _options?: BytesReadOptions): Promise<Uint8Array | null>

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 doPull() { if (this.done) return null; this.done = true; return new Uint8Array([1, 2]); }
}
console.log((await new OneChunk().read())?.byteLength);

async 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 doPull() {
    if (this.done) return null;
    this.done = true;
    return new Uint8Array([1, 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 doPull() { return this.chunks.shift() ?? null; }
}
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 doPull() {
    if (this.done) return null;
    this.done = true;
    return new Uint8Array([1, 2]);
  }
}
const reader = new OneChunk();
await reader.peek(2);
console.log(reader.takeBuffered(1)[0]);

override async 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 doPull() { if (this.done) return null; this.done = true; return new Uint8Array([1]); }
}
const reader = new OneChunk();
console.log(await reader.readExactly(2));
console.log(reader.buffered);

override async 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 doPull() { return this.chunks.shift() ?? null; }
}
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 doRead() { return null; } }
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 doPull() { return null; } }
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 doPull() { return null; } }
const reader = new EmptyBuffered();
await reader.peek(1);
console.log(reader.eof);

constructor(onClose: () => void | Promise<void> = () => {})

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.

Subclasses should reject writes after this flag is true. The flag is set before the close callback is awaited.

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

abstract write(value: T): Promise<void>

Write one value to the sink.

Subclasses define ordering, backpressure, and failure behavior. Generic Writer does not enforce the closed state; byte writers do.

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 before returning or throw. The base 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);

async flush(): Promise<void>

Flush internally buffered data.

The base implementation is a no-op for unbuffered writers. Buffered subclasses override this to write pending bytes and may throw on sink failures.

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

async close(): Promise<void>

Close the writer and run its close callback once.

Multiple calls are safe. The base class does not flush; subclasses with buffers should override close to flush first.

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

protected abstract 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>

Write one byte buffer.

ArrayBuffer and ArrayBufferView inputs are wrapped in a Uint8Array preserving view byte offsets. The method throws Writer is closed after close and forwards errors from doWrite().

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

async 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);

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. The slice is backed by the writer's internal buffer and should not be retained after the promise resolves.

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

protected async 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.

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

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

Write multiple byte buffers through the coalescing buffer.

Small vectors are accumulated synchronously 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])));

async flush(): Promise<void>

Drain the coalesce buffer.

Calling flush() with no pending bytes is a no-op. Errors from doFlush() reject the returned promise and the pending count has already been reset.

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

async close(): Promise<void>

Flush pending bytes, then close the writer.

The method is idempotent. If flushing throws, the close callback still runs through the finally block and the flush error is rethrown.

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

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();