# `streaming`

<!-- begin:auto-docs -->
_(no submodules)_
<!-- end:auto-docs -->

## 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

```ts
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

```ts
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.