bridgeStoreToAsyncIterable

function bridgeStoreToAsyncIterable<T>(
    store,
    signal,
    shouldYield?,
): AsyncIterable<T>;

Adapts a ReactiveStreamStore into an AsyncIterable, so a push-based reactive store can be driven by pull-based code that consumes a stream by for await-ing it — for example TanStack Query's experimental_streamedQuery.

The bridge only observes the store; it does not open or tear down the connection. Just like every other consumer in this ecosystem — a store does nothing until you connect() it — the caller owns the store's lifecycle: connect() the store yourself (typically binding the same signal via `withSignal()`), and reset() it when you're done if you intend to reuse it. The bridge subscribes, yields the store's current and subsequent values, and unsubscribes when iteration ends.

This is the store-backed counterpart to createAsyncIterableFromDataPublisher. That helper turns a raw DataPublisher directly into an AsyncIterable and queues every message so none are dropped; use it when you have a publisher and no store. bridgeStoreToAsyncIterable instead sits on top of a ReactiveStreamStore, so it reflects the store's unified idle/loading/loaded/error lifecycle and its stale-while-revalidate behaviour — and, because a store only ever holds the latest snapshot, it is latest-wins rather than fully buffered. Note this is also distinct from an RPC subscription's own AsyncIterable (await rpcSubscriptions.someNotifications().subscribe(...)), which vends messages straight off the transport without a store in between.

On iteration it seeds from the store's current snapshot, then yields its lifecycle:

• loaded → yields the value (the one already present when iteration begins, then each subsequent update), unless an optional shouldYield predicate rejects it. Latest-wins: if several notifications land between pulls, only the most recent unconsumed value is yielded (a subscription consumer wants the freshest state, not a backlog).
• error → throws, so the consuming for await rejects. Substitutes a SOLANA_ERROR__SUBSCRIBABLE__STREAM_CLOSED_WITHOUT_ERROR sentinel when the store reports an error with a nullish payload. An error takes precedence over a buffered value: if a loaded value is still pending when an error arrives, that value is dropped and the error propagates (once errored, stop yielding).
• signal aborts → ends the iterable cleanly (no error). A subscription never completes on its own, so signal is how the iterable terminates: aborting it unblocks a parked for await and ends the loop. Bind the same signal to the store's connection (store.withSignal(signal).connect()) so the abort tears the underlying stream down too.

However iteration ends — value exhaustion, error, or abort — the bridge unsubscribes from the store. It does not reset() the store; that is the caller's decision.

Type Parameters

Parameters

Returns

AsyncIterable<T>

An AsyncIterable<T> that yields each store value until the store errors or signal aborts.

Throws

Rethrows the store's error payload when the store transitions to status: 'error', or a SolanaError with code SOLANA_ERROR__SUBSCRIBABLE__STREAM_CLOSED_WITHOUT_ERROR when that payload is nullish.

Example

const store = rpcSubscriptions.slotNotifications().reactiveStore();
const controller = new AbortController();
// The caller owns the connection — bind it to the same signal so an abort tears it down.
store.withSignal(controller.signal).connect();
try {
    for await (const notification of bridgeStoreToAsyncIterable(store, controller.signal)) {
        console.log('Latest slot:', notification.slot);
    }
} catch (e) {
    console.error('The subscription errored', e);
} finally {
    store.reset();
}
// Elsewhere: controller.abort() ends the loop cleanly.

See

• ReactiveStreamStore
• createAsyncIterableFromDataPublisher