createReactiveStoreWithInitialValueAndSlotTracking

function createReactiveStoreWithInitialValueAndSlotTracking<
    TInitialValue,
    TStreamValue,
    TItem,
>(
    config,
): ReactiveStreamStore<
    Readonly<{
        context: Readonly<{
            slot: Slot;
        }>;
        value: TItem;
    }>
>;

Creates a ReactiveStreamStore that combines an initial one-shot fetch with an ongoing stream to keep its state up to date.

The store uses slot-based comparison to ensure that only the most recent value is kept, regardless of whether it came from the initial value source or a stream notification. This prevents stale data from overwriting newer data when the two sources arrive out of order.

The two sources are consumed via their `reactiveStore()` methods rather than by calling send() / subscribe() directly, so any object satisfying the ReactiveActionSource / ReactiveStreamSource duck-types works — including PendingRpcRequest / PendingRpcSubscriptionsRequest and plugin-authored wrappers.

Things to note:

• The returned store starts in status: 'idle'. Call `connect()` to dispatch the initial-value source and open the stream.
• The store transitions through loading until the first value or notification arrives, then to loaded with a SolanaRpcResponse containing the value and the slot context at which it was observed.
• On error from either source, the store transitions to status: 'error' preserving the last known value. Only the first error per connection window is captured.
• A subsequent connect() aborts the current connection, transitions back to status: 'loading' (preserving the last known data and error for stale-while-revalidate), and re-builds both inner stores with a fresh inner abort signal.
• `reset()` aborts the current connection and returns the store to idle, clearing data and error.
• Attach a caller-provided cancellation source via `withSignal()` — store.withSignal(signal).connect() composes the signal with the per-connection controller. Aborting the caller's signal transitions the store to error with that abort reason.

Type Parameters

Parameters

Returns

ReactiveStreamStore<Readonly<{ context: Readonly<{ slot: Slot; }>; value: TItem; }>>

Example

import {
    address,
    createReactiveStoreWithInitialValueAndSlotTracking,
    createSolanaRpc,
    createSolanaRpcSubscriptions,
} from '@solana/kit';
 
const rpc = createSolanaRpc('http://127.0.0.1:8899');
const rpcSubscriptions = createSolanaRpcSubscriptions('ws://127.0.0.1:8900');
const myAddress = address('FnHyam9w4NZoWR6mKN1CuGBritdsEWZQa4Z4oawLZGxa');
 
const balanceStore = createReactiveStoreWithInitialValueAndSlotTracking({
    initialValueSource: rpc.getBalance(myAddress, { commitment: 'confirmed' }),
    initialValueMapper: lamports => lamports,
    streamSource: rpcSubscriptions.accountNotifications(myAddress),
    streamValueMapper: ({ lamports }) => lamports,
});
 
const unsubscribe = balanceStore.subscribe(() => {
    const state = balanceStore.getUnifiedState();
    if (state.status === 'error') {
        console.error('Error:', state.error);
        balanceStore.connect();
    } else if (state.status === 'loaded') {
        console.log(`Balance at slot ${state.data.context.slot}:`, state.data.value);
    }
});
balanceStore.withSignal(AbortSignal.timeout(60_000)).connect();

See

ReactiveStreamStore