suaveplan/packages/api-stack/api/api-worker-rpc/src/streaming/streaming.md

streaming

(no submodules)

Purpose

Caller-side primitives for consuming streaming RPC methods. The server publishes a stream-chunk frame per yielded value plus a final stream-end; on the caller side, streamRequest(client, dispatcher, method, params, opts) returns an AsyncIterable whose Symbol.asyncIterator pulls chunks lazily from a bounded internal buffer. Consumer-side for await semantics drive the protocol: each next() call pulls one buffered chunk, and a consumer-side return() (triggered by break out of for await) sends a cancel frame so the remote producer stops generating.

The StreamDispatcher is the small registry that routes inbound stream frames to the right caller; the raw WorkerRpcClient accepts the dispatcher's dispatch method as its onStreamFrame option.

Features

• AsyncIterable<unknown> return type — composes with for await and standard async-iteration helpers.
• Bounded backpressure via BoundedQueue — fail/drop-oldest/drop-newest policies share the same primitive as client's outbound queue.
• AbortSignal cancels the iterator with a name-"AbortError" throw; pre-aborted signals cancel synchronously on first next().
• Stream-end error frames reconstruct the original throw on the caller via envelopeToError.
• Consumer-side return() (e.g. break out of for await) sends a cancel frame so the remote producer can release resources.
• StreamDispatcher is a trivial id-keyed routing table.

Usage: Basic

import { createRawWorkerRpcClient } from "@suaveplan/api-worker-rpc/client";
import { StreamDispatcher, streamRequest } from "@suaveplan/api-worker-rpc/streaming";

const dispatcher = new StreamDispatcher();
const client = createRawWorkerRpcClient(port, {
    onStreamFrame: dispatcher.dispatch,
});

for await (const row of streamRequest(client, dispatcher, "rows", [])) {
    process(row);
}

Usage: Advanced

import { StreamDispatcher, streamRequest } from "@suaveplan/api-worker-rpc/streaming";

const ac = new AbortController();
setTimeout(() => ac.abort(), 5_000);

try {
    for await (const chunk of streamRequest(client, dispatcher, "tail", [logId], {
        signal: ac.signal,
        queue: { highWaterMark: 256, mode: "drop-oldest" },
    })) {
        ui.append(chunk);
    }
} catch (err) {
    if (err instanceof Error && err.name === "AbortError") {
        // intentional cancel
    } else {
        throw err;
    }
}

API Reference

• streamRequest(client, dispatcher, method, params, opts?): AsyncIterable<unknown>.
• interface StreamRequestOptions { signal?: AbortSignal; queue?: BackpressurePolicy }.
• class StreamDispatcher:
  • dispatch: (frame) => void — pass to the client's onStreamFrame.
  • register(id, handler): void
  • unregister(id): void
  • clear(): void

Implementation Notes

• The internal buffer is a BoundedQueue<unknown>; the default policy caps it at 64 chunks. Switch to drop-oldest for log-tail workloads where the latest matters more than completeness.
• next() resolves either from the buffer (synchronously, via microtask) or when the next stream-chunk arrives. The tryFlush helper is the single state-transition point that satisfies a pending pull.
• Iterator return() sends a cancel frame, marks the stream ended, flushes any pending pull as { done: true }, and unregisters from the dispatcher.
• AbortSignal aborts behave identically to cancel except the thrown error has name === "AbortError" to match the standard cancellation contract.