# @solana/subscribable > Helpers for creating subscription-based event emitters Latest version **8.3.0** (published 2026-09-09) · MIT license · 0 weekly downloads ## Install ```sh npm install @solana/subscribable pnpm add @solana/subscribable yarn add @solana/subscribable bun add @solana/subscribable ``` ## Health **Score 70/100 (B)** — status: active. Positive: has types; esm support; no vulnerabilities; has provenance; recently updated; high maintenance score. Warnings: low downloads. ## Facts | | | |---|---| | Version | 8.3.0 | | Published | 2026-09-09 | | First published | 2024-10-04 | | Weekly downloads | 0 | | License | MIT | | TypeScript types | bundled | | Module format | ESM + CommonJS | | Node | >=20.18.0 | | Dependencies | 2 | | Unpacked size | 473.1 KB | | Known vulnerabilities | 0 | | Install scripts | no | | Provenance | attested (GitHub Actions) | | GitHub stars | 695 | | Author | Solana Labs Maintainers | | Maintainers | solana-devs | | Keywords | blockchain, solana, web3 | ## Links - npm: https://www.npmjs.com/package/@solana/subscribable - Repository: https://github.com/anza-xyz/kit - Homepage: https://www.solanakit.com/api#solanasubscribable - Issues: https://github.com/anza-xyz/kit/issues - npm.io page: https://npm.io/package/@solana/subscribable ## Dependencies (2) - [@solana/errors](https://npm.io/package/@solana/errors.md) 8.3.0 - [@solana/promises](https://npm.io/package/@solana/promises.md) 8.3.0 ## Recent versions - 8.3.0 (latest) — 2026-09-09 - 8.4.0-canary-20260925083045 (canary) — 2026-09-25 - 5.1.0-experimental-20251205104522 (experimental) — 2025-12-05 - 4.0.0 (4x) — 2025-10-08 - 3.0.3 (3x) — 2025-09-11 - 2.0.0 (next) — 2024-11-07 - 2.0.0-rc.4 (rc) — 2024-11-05 - 8.4.0-canary-20260924083259 — 2026-09-24 - 8.4.0-canary-20260923083123 — 2026-09-23 - 8.4.0-canary-20260922155338 — 2026-09-22 - 8.4.0-canary-20260922151509 — 2026-09-22 - 8.4.0-canary-20260922084122 — 2026-09-22 - 8.4.0-canary-20260922083419 — 2026-09-22 - 8.4.0-canary-20260921145441 — 2026-09-21 - 8.4.0-canary-20260921085643 — 2026-09-21 - … 1201 more at https://npm.io/package/@solana/subscribable/versions ## README [![npm][npm-image]][npm-url] [![npm-downloads][npm-downloads-image]][npm-url]
[![code-style-prettier][code-style-prettier-image]][code-style-prettier-url] [code-style-prettier-image]: https://img.shields.io/badge/code_style-prettier-ff69b4.svg?style=flat-square [code-style-prettier-url]: https://github.com/prettier/prettier [npm-downloads-image]: https://img.shields.io/npm/dm/@solana/subscribable?style=flat [npm-image]: https://img.shields.io/npm/v/@solana/subscribable?style=flat [npm-url]: https://www.npmjs.com/package/@solana/subscribable # @solana/subscribable This package contains utilities for creating subscription-based event targets. These differ from the `EventTarget` interface in that the method you use to add a listener returns an unsubscribe function. It is primarily intended for internal use – particularly for those building `RpcSubscriptionChannels` and associated infrastructure. ## Types ### `DataPublisher` This type represents an object with an `on` function that you can call to subscribe to certain data over a named channel. ```ts let dataPublisher: DataPublisher<{ error: SolanaError }>; dataPublisher.on('data', handleData); // ERROR. `data` is not a known channel name. dataPublisher.on('error', e => { console.error(e); }); // OK. ``` ### `ReactiveStreamStore` This type represents a reactive store that holds the latest value published to a data channel. It exposes a `{ getState, retry, subscribe }` contract compatible with `useSyncExternalStore`, Svelte stores, and other reactive primitives. `getState()` returns a discriminated snapshot of the store's lifecycle: ```ts type ReactiveState = | { data: T | undefined; error: unknown; status: 'loading' } | { data: T; error: undefined; status: 'loaded' } | { data: T | undefined; error: unknown; status: 'error' } | { data: undefined; error: undefined; status: 'idle' }; ``` > Also exported as `ReactiveStore` for backwards compatibility. That alias is deprecated and will be removed in a future major release. The store starts in `status: 'idle'`. Call `connect()` to open the underlying stream; the store will transition through `loading` → `loaded` (or `error`). Every subsequent `connect()` transitions back through `loading`, preserving the last known `data` and `error` (stale-while-revalidate). A subsequent `loaded` clears the error; a subsequent `error` replaces it. Call `reset()` to tear down the connection and return to `idle` without permanently killing the store. ```tsx const store: ReactiveStreamStore = /* ... */; // React — snapshot identity is stable between updates, so it can be passed directly. const state = useSyncExternalStore(store.subscribe, store.getState); useEffect(() => { store.connect(); return () => store.reset(); }, [store]); // Stale-while-revalidate: keep showing the last value while a reconnect is in flight. return ( <> {state.data !== undefined && } {state.status === 'loading' && state.data === undefined && } {state.status === 'error' && } ); // Vue const snapshot = shallowRef(store.getState()); store.subscribe(() => { snapshot.value = store.getState(); }); store.connect(); ``` ### `ReactiveActionStore` A framework-agnostic state machine for wrapping an async action (a function you dispatch on demand — like a form submission, a mutation, or an on-click fetch). It exposes a `{ dispatch, getState, subscribe, reset }` contract that bridges trivially into `useSyncExternalStore`, Svelte stores, Vue's `shallowRef`, and similar reactive primitives. The snapshot is a discriminated union: ```ts type ReactiveActionState = | { status: 'idle'; data: undefined; error: undefined } | { status: 'running'; data: TResult | undefined; error: unknown } | { status: 'success'; data: TResult; error: undefined } | { status: 'error'; data: TResult | undefined; error: unknown }; ``` `data` is the last successful result and `error` is the last failure; both survive across transitions so UIs can render stale content (data **or** error) while a retry is in flight. `success` clears `error`; only `reset()` clears `data`. Unlike `ReactiveStreamStore` (which models a stream of values with a separate error channel), `ReactiveActionStore` models a one-shot-per-dispatch lifecycle where errors are part of the snapshot. ### `TypedEventEmitter` This type allows you to type `addEventListener` and `removeEventListener` so that the call signature of the listener matches the event type given. ```ts const emitter: TypedEventEmitter<{ message: MessageEvent }> = new WebSocket('wss://api.devnet.solana.com'); emitter.addEventListener('data', handleData); // ERROR. `data` is not a known event type. emitter.addEventListener('message', message => { console.log(message.origin); // OK. `message` is a `MessageEvent` so it has an `origin` property. }); ``` ### `TypedEventTarget` This type is a superset of `TypedEventEmitter` that allows you to constrain calls to `dispatchEvent`. ```ts const target: TypedEventTarget<{ candyVended: CustomEvent<{ flavour: string }> }> = new EventTarget(); target.dispatchEvent(new CustomEvent('candyVended', { detail: { flavour: 'raspberry' } })); // OK. target.dispatchEvent(new CustomEvent('candyVended', { detail: { flavor: 'raspberry' } })); // ERROR. Misspelling in detail. ``` ## Functions ### `createReactiveActionStore(fn)` Wraps an async function in a `ReactiveActionStore`. Each `dispatch` creates a fresh `AbortController` and aborts the previous one, so a rapid succession of dispatches only produces one final state transition — the outcome of the most recent call. The wrapped function receives the `AbortSignal` as its first argument, followed by the arguments passed to `dispatch`. ```tsx const store = createReactiveActionStore(async (signal: AbortSignal, accountId: Address) => { const response = await fetch(`/api/accounts/${accountId}`, { signal }); return response.json(); }); // React — stale-while-revalidate: keep showing the card during retries. const { data, error, status } = useSyncExternalStore(store.subscribe, store.getState); return ( <> {data !== undefined && } {status === 'running' && } {status === 'error' && store.dispatch(someAccountId)} />} {status === 'idle' && data === undefined && } ); ``` Things to note: - Starts at `{ status: 'idle' }`. `getState()` always returns a defined snapshot. - `dispatch` is a stable reference — safe to pass into memoized callbacks without re-renders. - Two ways to trigger the action: - `dispatch(...)` — fire-and-forget. Returns `undefined` synchronously and never throws; safe to call from UI event handlers without a `.catch`. Failures surface on state as `{ status: 'error' }`. - `dispatchAsync(...)` — returns a promise that resolves to the wrapped function's result. Rejects on failure and with an `AbortError` when superseded or `reset()`. Use from imperative code that needs the resolved value; pair with [`isAbortError`](../promises#isaborterrorerr) from `@solana/promises` to filter abort rejections. - Attach a caller-provided `AbortSignal` to a `dispatch` or `dispatchAsync` call via `store.withSignal(signal)`: ```ts // Per-attempt timeout — fresh signal per call: store.withSignal(AbortSignal.timeout(5_000)).dispatch(someAccountId); // Shared kill switch — bind the wrapper once, reuse everywhere: const killCtrl = new AbortController(); const killable = store.withSignal(killCtrl.signal); killable.dispatch(someAccountId); killable.dispatch(someAccountId); killCtrl.abort(); // cancels in-flight and short-circuits future calls ``` The wrapped signal is composed with the store's internal per-dispatch controller via `AbortSignal.any`, so aborting either cancels the in-flight call and surfaces the abort reason on state. The wrapper exposes only `dispatch` / `dispatchAsync` — `getState` / `subscribe` / `reset` stay on the parent store. - Calling either dispatch while one is in flight aborts the previous call; its outcome is dropped from state regardless of which variant started it. - `data` survives across transitions: a fresh `running` or `error` snapshot carries the last successful result so call sites can keep rendering stale content while a retry is in flight. Only `reset()` clears it. - `reset()` aborts the in-flight dispatch and restores the idle snapshot, clearing both `data` and `error`. - Subscribers are notified only when the snapshot's `status`, `data`, or `error` actually changes, so redundant transitions (`dispatch` while already `running` with the same `data`, `reset` while already `idle`) are silent. - `fn` is captured at construction, so the store holds a closure over whatever `fn` referenced at that moment. In React, create the store once (`useState(() => createReactiveActionStore(...))` or `useRef`) and read the latest closure through a ref if you need it to change between renders — don't call `createReactiveActionStore` directly in a render body. - The store holds strong references to its subscribers. Non-framework consumers that subscribe without unsubscribing will keep their listeners (and anything the listeners close over) alive for the lifetime of the store. ### `createAsyncIterableFromDataPublisher({ abortSignal, dataChannelName, dataPublisher, errorChannelName })` Returns an `AsyncIterable` given a data publisher. The iterable will produce iterators that vend messages published to `dataChannelName` and will throw the first time a message is published to `errorChannelName`. Triggering the abort signal will cause all iterators spawned from this iterator to return once they have published all queued messages. ```ts const iterable = createAsyncIterableFromDataPublisher({ abortSignal: AbortSignal.timeout(10_000), dataChannelName: 'message', dataPublisher, errorChannelName: 'error', }); try { for await (const message of iterable) { console.log('Got message', message); } } catch (e) { console.error('An error was published to the error channel', e); } finally { console.log("It's been 10 seconds; that's enough for now."); } ``` Things to note: - If a message is published over a channel before the `AsyncIterator` attached to it has polled for the next result, the message will be queued in memory. - Messages only begin to be queued after the first time an iterator begins to poll. Channel messages published before that time will be dropped. - If there are messages in the queue and an error occurs, all queued messages will be vended to the iterator before the error is thrown. - If there are messages in the queue and the abort signal fires, all queued messages will be vended to the iterator after which it will return. - Any new iterators created after the first error is encountered will reject with that error when polled. ### `createReactiveStoreFromDataPublisherFactory({ createDataPublisher, dataChannelName, errorChannelName })` Returns a `ReactiveStreamStore` that wires itself to a fresh `DataPublisher` on every `connect()`. Accepts an async factory so the store can tear down a broken stream and open a new one without losing subscribers or the last known value. The factory receives the per-connection `AbortSignal` so the underlying transport can stop when the connection window closes. ```ts const store = createReactiveStoreFromDataPublisherFactory({ createDataPublisher: signal => getDataPublisherFromEventEmitter(new WebSocket(url, { signal })), dataChannelName: 'message', errorChannelName: 'error', }); const unsubscribe = store.subscribe(() => { const snapshot = store.getState(); if (snapshot.status === 'error') console.error('Connection failed:', snapshot.error); else if (snapshot.status === 'loaded') console.log('Latest:', snapshot.data); }); // Fresh 30-second clock per connection attempt: store.withSignal(AbortSignal.timeout(30_000)).connect(); ``` Things to note: - The returned store starts in `status: 'idle'`. Call `connect()` to open the first stream. - `createDataPublisher` is invoked on every `connect()`. The store transitions through `loading`, preserving the last known `data` and `error` (stale-while-revalidate). - If `createDataPublisher` rejects, the store transitions to `status: 'error'` with the rejection as the error. Call `connect()` to try again. - `reset()` aborts the current connection and returns the store to `idle`, clearing `data` and `error`. A follow-up `connect()` opens a fresh stream. - Attach a caller-provided cancellation source via `store.withSignal(signal).connect()` — the signal is composed with the per-connection controller via `AbortSignal.any`. Aborting the caller's signal transitions the store to `error` with that abort reason. ### `bridgeStoreToAsyncIterable(store, signal, shouldYield?)` Adapts a `ReactiveStreamStore` into an `AsyncIterable`, so its _push_-based `subscribe` contract can be driven by _pull_-based consumers like TanStack Query's `experimental_streamedQuery` — anything that consumes a stream by `for await`-ing it. The bridge only _observes_ the store. Just like every other consumer in this package — a store does nothing until you `connect()` it — the caller owns the store's lifecycle: `connect()` it yourself (typically bound to the same `signal`), and `reset()` it when you're done. The bridge subscribes, yields values, and unsubscribes when iteration ends. ```ts 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. ``` This is the store-backed sibling of `createAsyncIterableFromDataPublisher`. Reach for that helper when you have a raw `DataPublisher` and want every message queued and delivered; reach for `bridgeStoreToAsyncIterable` when you already have a `ReactiveStreamStore` and want to consume its unified lifecycle as an iterable. Note this is also distinct from an RPC subscription's own iterable (`await rpcSubscriptions.someNotifications().subscribe({ abortSignal })`), which vends messages straight off the transport without a store in between. Because it reads through a store, its semantics follow the store's snapshot model: - **Seeds from the current snapshot.** On iteration it reads the store's current state, so a value (or error) already present when iteration begins is delivered — you don't have to create the iterable before connecting. - **Latest-wins.** A store only ever holds the most recent value, so if several notifications land between pulls only the freshest survives — a subscription consumer wants the current state, not a backlog. (This is the key behavioural difference from `createAsyncIterableFromDataPublisher`, which queues every message.) - **`error` throws.** A store `error` rejects the consuming `for await`. If the store errors with a nullish payload, a `SolanaError` with code `SOLANA_ERROR__SUBSCRIBABLE__STREAM_CLOSED_WITHOUT_ERROR` is substituted so the failure still surfaces. An error takes precedence over a buffered value. - **Abort ends cleanly.** Aborting `signal` completes the iterable (`done`) rather than throwing — an abort is teardown, not failure. Because a subscription never completes on its own, this is how the iterable terminates; bind the same signal to the store's connection so the abort tears the underlying stream down too. The optional `shouldYield` predicate gates each `loaded` value before it is yielded; return `false` to drop it (useful for e.g. deduping by slot). ### `demultiplexDataPublisher(publisher, sourceChannelName, messageTransformer)` Given a channel that carries messages for multiple subscribers on a single channel name, this function returns a new `DataPublisher` that splits them into multiple channel names. Imagine a channel that carries multiple notifications whose destination is contained within the message itself. ```ts const demuxedDataPublisher = demultiplexDataPublisher(channel, 'message', message => { const destinationChannelName = `notification-for:${message.subscriberId}`; return [destinationChannelName, message]; }); ``` Now you can subscribe to _only_ the messages you are interested in, without having to subscribe to the entire `'message'` channel and filter out the messages that are not for you. ```ts demuxedDataPublisher.on( 'notification-for:123', message => { console.log('Got a message for subscriber 123', message); }, { signal: AbortSignal.timeout(5_000) }, ); ``` ### `getDataPublisherFromEventEmitter(emitter)` Returns an object with an `on` function that you can call to subscribe to certain data over a named channel. The `on` function returns an unsubscribe function. ```ts const socketDataPublisher = getDataPublisherFromEventEmitter(new WebSocket('wss://api.devnet.solana.com')); const unsubscribe = socketDataPublisher.on('message', message => { if (JSON.parse(message.data).id === 42) { console.log('Got response 42'); unsubscribe(); } }); ``` --- _Source: https://npm.io/package/@solana/subscribable · Machine-readable twin of the npm.io package page. Health data is recomputed on every publish._