Add useSubscription for rendering based on a subscription

.changeset/icy-sites-clap.md

@@ -0,0 +1,23 @@
+---
+'@solana/react': minor
+---
+
+Add `useSubscription` — a React hook for subscription-based live data. Pass a `ReactiveStreamSource<T>` (satisfied by `PendingRpcSubscriptionsRequest`) and the hook opens the subscription on mount, re-opens whenever the source identity changes, and tears it down on unmount.
+
+```tsx
+function AccountBalance({ address }: { address: Address }) {
+    const client = useClient<ClientWithRpcSubscriptions<AccountNotificationsApi>>();
+    const source = useMemo(() => client.rpcSubscriptions.accountNotifications(address), [client, address]);
+    const { data, slot, error, reconnect } = useSubscription(source);
+    if (error) return <button onClick={reconnect}>Reconnect</button>;
+    return <p>{data ? `${data.lamports} lamports at slot ${slot}` : 'Connecting…'}</p>;
+}
+```
+
+The result reports `status` as one of `loading | loaded | error | disabled`. Notifications shaped as `SolanaRpcResponse<U>` (account/program/signature) are unwrapped automatically: `data` is the inner value and `slot` is lifted from `context.slot`. Raw notifications (slot/logs/root) pass through with `slot: undefined`. Pass `null` for the source to gate the subscription off — useful while inputs aren't yet known. The result then reports `status: 'disabled'`. After a notification arrives, an error transitions to `status: 'error'` while preserving the stale `data`; `reconnect()` returns to `loading` (preserving stale `data` and `error` for stale-while-revalidate) before settling on `loaded` or a fresh `error`.
+
+Optional `getAbortSignal: () => AbortSignal` is a factory invoked on every connection (initial subscribe + every `reconnect()`). Each connection gets a fresh signal that the underlying store composes with its per-connection controller via `AbortSignal.any`. The natural use is per-connection timeouts: `getAbortSignal: () => AbortSignal.timeout(30_000)` gives every connection its own 30-second clock that resets on reconnect. The factory is held in a ref synced to the latest render, so inline closures are fine — no `useCallback` needed. `reconnect()` also accepts an optional `{ abortSignal }` override to replace the factory for one specific attempt (presence-based: omit to use the factory, `{ abortSignal: signal }` to override, `{ abortSignal: undefined }` to opt out).
+
+The hook mirrors `useRequest`'s structure exactly: construct the lazy store via `useMemo`, fire `store.connect()` in a `useEffect`, tear down via `store.reset()` in cleanup. Same StrictMode-safe lifecycle, same vocabulary, same per-call signal API. SSR-safe — on the server the connect effect doesn't run, so the store stays `idle` and the hook reports `status: 'loading'`; first client render hydrates from the same paint and commits the connect.
+
+`SubscriptionResult<T>` and `UseSubscriptionOptions` are exported alongside the hook so plugin hooks built on top can declare their return shape against them. The unwrap is driven by `UnwrapRpcResponse<T>` and `splitSolanaRpcResponse()` from `@solana/rpc-types`, where they live alongside `SolanaRpcResponse`.

packages/react/README.md

@@ -199,6 +199,51 @@ refresh({ abortSignal: undefined }); // no abort signal for this attempt
 refresh(); // omit the key to use the factory (default)
 ```
 
+### `useSubscription(source, options?)`
+
+Subscribe to a stream-store source and surface the latest notification as reactive state. Returns `{ data, error, reconnect, slot, status }` where `status` is one of `'loading' | 'loaded' | 'error' | 'disabled'`. Use it for any RPC subscription (`accountNotifications`, `slotNotifications`, `logsNotifications`, etc.) or any plugin-authored stream that satisfies `ReactiveStreamSource<T>`.
+
+`source` is any `ReactiveStreamSource<T>` — the `{ reactiveStore() }` duck-type satisfied by `PendingRpcSubscriptionsRequest`. Pass `null` to disable. Memoize the source with `useMemo` keyed on whatever inputs it depends on; stable identity is how the hook knows when to tear down and re-open.
+
+Notifications shaped as `SolanaRpcResponse<U>` (account/program/signature notifications) are unwrapped automatically: `data` is the inner value `U` and `slot` is lifted from `context.slot`. Raw notifications (slot/logs/root) pass through with `slot: undefined`.
+
+```tsx
+import { useClient, useSubscription } from '@solana/react';
+import type { Address, AccountNotificationsApi, ClientWithRpcSubscriptions } from '@solana/kit';
+
+function AccountBalance({ address }: { address: Address }) {
+    const client = useClient<ClientWithRpcSubscriptions<AccountNotificationsApi>>();
+    const source = useMemo(() => client.rpcSubscriptions.accountNotifications(address), [client, address]);
+    const { data, slot, error, reconnect } = useSubscription(source);
+    if (error) return <button onClick={reconnect}>Reconnect</button>;
+    return <p>{data ? `${data.lamports} lamports at slot ${slot}` : 'Connecting…'}</p>;
+}
+```
+
+`reconnect()` re-opens the connection. After a `loaded` outcome that transitions to `error`, calling `reconnect()` returns `status` to `'loading'` while preserving the stale `data` and `error` (stale-while-revalidate) → `'loaded'` (or `'error'` again). The hook tears the connection down on unmount via the store's `reset()`; StrictMode's mount → unmount → mount cycle re-opens cleanly.
+
+#### Per-connection cancellation
+
+Pass `getAbortSignal` to attach a cancellation signal to each individual connection — initial subscribe plus every `reconnect()`. The natural use is per-connection timeouts:
+
+```tsx
+const { data, error, reconnect } = useSubscription(source, {
+    // Each connection gets a fresh 30-second clock. `reconnect()` resets it.
+    getAbortSignal: () => AbortSignal.timeout(30_000),
+});
+```
+
+The factory is held in a ref synced to the latest render, so inline closures are fine — no `useCallback` needed. To kill the subscription entirely (e.g. on a route change), set the memoized source to `null` (the result reports `disabled`), or let the component unmount.
+
+`reconnect()` accepts an optional `{ abortSignal }` override that replaces the configured factory for just that attempt — useful when one specific reconnect needs different cancellation semantics:
+
+```tsx
+const userInitiatedCtrl = new AbortController();
+reconnect({ abortSignal: userInitiatedCtrl.signal }); // override: use this signal, ignore the factory
+reconnect({ abortSignal: undefined }); // no abort signal for this attempt
+reconnect(); // omit the key to use the factory (default)
+```
+
 ## Hooks
 
 ### `useSignIn(uiWalletAccount, chain)`

packages/react/src/__tests__/staticStores-test.ts

@@ -1,4 +1,4 @@
-import { disabledActionStore } from '../staticStores';
+import { disabledActionStore, disabledStreamStore } from '../staticStores';
 
 describe('disabledActionStore', () => {
     it('reports a frozen `idle` state with no data or error', () => {
@@ -48,8 +48,65 @@
         unsubscribe();
     });
 
     it('`dispatchAsync()` rejects with an AbortError so accidental awaits surface, not silently hang', async () => {
         const store = disabledActionStore<string>();
         await expect(store.dispatchAsync()).rejects.toMatchObject({ name: 'AbortError' });
     });
 });
+
+describe('disabledStreamStore', () => {
+    it('reports a frozen `idle` state with no data or error', () => {
+        const store = disabledStreamStore<string>();
+        const state = store.getUnifiedState();
+        expect(state).toEqual({ data: undefined, error: undefined, status: 'idle' });
+        expect(Object.isFrozen(state)).toBe(true);
+    });
+
+    it('returns the same getUnifiedState() reference across calls', () => {
+        const store = disabledStreamStore<string>();
+        expect(store.getUnifiedState()).toBe(store.getUnifiedState());
+    });
+
+    it('`connect()` is a no-op — state does not change', () => {
+        const store = disabledStreamStore<string>();
+        const before = store.getUnifiedState();
+        store.connect();
+        store.connect();
+        expect(store.getUnifiedState()).toBe(before);
+    });
+
+    it('`reset()` is a no-op — state does not change', () => {
+        const store = disabledStreamStore<string>();
+        const before = store.getUnifiedState();
+        store.reset();
+        expect(store.getUnifiedState()).toBe(before);
+    });
+
+    it('`retry()` is a no-op — state does not change', () => {
+        const store = disabledStreamStore<string>();
+        const before = store.getUnifiedState();
+        store.retry();
+        expect(store.getUnifiedState()).toBe(before);
+    });
+
+    it('`withSignal(signal).connect()` is a no-op — state does not change, signal is not observed', () => {
+        const store = disabledStreamStore<string>();
+        const ctrl = new AbortController();
+        const before = store.getUnifiedState();
+        store.withSignal(ctrl.signal).connect();
+        ctrl.abort(new Error('would-be-cancellation'));
+        expect(store.getUnifiedState()).toBe(before);
+    });
+
+    it('`subscribe()` never notifies (connect + reset produce no state change to observe)', () => {
+        const store = disabledStreamStore<string>();
+        const listener = jest.fn();
+        const unsubscribe = store.subscribe(listener);
+        store.connect();
+        store.reset();
+        store.retry();
+        store.withSignal(new AbortController().signal).connect();
+        expect(listener).not.toHaveBeenCalled();
+        unsubscribe();
+    });
+});