React hooks
useObservable()
A React hook that returns the current/latest value from an observable. Store updates are deferred by default via useDeferredValue: urgent renders keep the previous value while a background render catches up. That makes it safe to suspend on the returned value without replacing already-revealed UI with a Suspense fallback.
The deferral is identity-coherent: unlike a bare useDeferredValue(useObservable(...)), the observable identity and its value are deferred as one snapshot, and when the observable identity changes (e.g. it is memoized on a document id that just changed) the hook falls back to the live value — the initialValue, or the new observable’s last emission when it is already live elsewhere — so the previous identity’s value never renders under the new one.
initialValue is required: it is what renders until the observable emits. Every value is a valid initial value — undefined included, pass it explicitly — and omitting the argument throws during render. Functions act as initializers, exactly like useState: pass () => value to compute the initial value lazily, and an initializer returning the function when the initial value should be a function itself. When there is no meaningful initial value, use useObservablePromise instead.
Mounts, remounts, and <Activity> reveals still render the current snapshot synchronously (no initial-value flash once a value has been emitted). The observable is never subscribed during render — the initialValue paints first and the live subscription starts on commit, keeping subscribe-time side effects out of the render phase. Keep the observable’s identity stable across renders (useMemo, useState, module scope, or React Compiler memoization): like useSyncExternalStore’s subscribe, an observable rebuilt on every render is re-subscribed on every render, and when it synchronously replays a value that differs from the initialValue this forces a render loop. On the server, this hook renders the resolved initialValue — exactly what the client’s first paint will show — and never subscribes the observable.
Prefer this hook for previews, validation, lists, and other non-input reads. Use useSyncObservable for controlled inputs.
Signature
function useObservable<T>(
observable$: Observable<T>,
initialValue: T | (() => T),
options?: UseObservableOptions,
): T
function useObservable<T, InitialValue>(
observable$: Observable<T>,
initialValue: InitialValue | (() => InitialValue),
options?: UseObservableOptions,
): T | InitialValue
interface UseObservableOptions {
disabled?: boolean
}disabled pauses the live subscription (later emissions stop updating the component; the last value is kept). A disabled hook performs no subscriptions at all — disabled: true means zero subscriptions until it is re-enabled — see the guide.
Example
import {useMemo} from 'react'
import {useObservable} from 'react-rx'
import {interval} from 'rxjs'
function MyComponent() {
const observable = useMemo(() => interval(100), [])
const number = useObservable(observable, 0)
return <>The number is {number}</>
}useSyncObservable()
A React hook that returns the current/latest value from an observable synchronously via useSyncExternalStore. This is the v4 useObservable behavior.
Use it when the value feeds a controlled input, or must stay consistent within the same event.
initialValue is required and follows the same rules as useObservable: every value is valid (undefined included), functions act as useState-style initializers, and omitting the argument throws during render. The server always renders the resolved initialValue.
Caveat: store mutations cannot be marked as Transitions. Suspending on a value returned by this hook replaces already-visible content with the nearest Suspense fallback — see the useSyncExternalStore caveats . Compare the two hooks in the Suspense example.
Signature
function useSyncObservable<T>(
observable$: Observable<T>,
initialValue: T | (() => T),
options?: UseObservableOptions,
): T
function useSyncObservable<T, InitialValue>(
observable$: Observable<T>,
initialValue: InitialValue | (() => InitialValue),
options?: UseObservableOptions,
): T | InitialValueThe overload without initialValue is deprecated. v7 removes it and requires the argument.
useSyncObservable(observable$, undefined) is a drop-in replacement with the same type and, in
v6, the same behavior. See the v6 to v7 migration
guide.
Example
import type {ChangeEvent} from 'react'
import {useMemo} from 'react'
import {useObservableSubject, useSyncObservable} from 'react-rx'
import {map} from 'rxjs'
function SearchField() {
const [changes$, handleChange] = useObservableSubject<ChangeEvent<HTMLInputElement>>()
const text$ = useMemo(() => changes$.pipe(map((event) => event.currentTarget.value)), [changes$])
// Controlled input values must update synchronously.
const text = useSyncObservable(text$, '')
return <input value={text} onChange={handleChange} />
}useObservablePromise()
A React hook that turns an observable into a use()-compatible promise for Suspense and Activity pre-rendering.
Signature
function useObservablePromise<T>(
observable: Observable<T>,
options?: UseObservablePromiseOptions,
): ObservablePromise<T>
interface UseObservablePromiseOptions {
disabled?: boolean
ttl?: number
}
type ObservablePromise<T> = Promise<T> &
({status: 'pending'} | {status: 'fulfilled'; value: T} | {status: 'rejected'; reason: unknown})The hook does not suspend, and mounting renders never subscribe the source: the fetch starts when the component that called the hook commits, when an already-live consumer re-renders with a new observable (so startTransition / useDeferredValue swaps fetch and commit on their own; see the Transitions and refetching example), or via preloadObservablePromise. Pass the returned promise as a prop to a child component that reads it with React’s use, with a <Suspense> boundary between the hook caller and that child — the caller must be able to commit while the child suspends. Never call use() on the promise in the same component that called the hook (or without a boundary in between): it suspends on its own pending promise before the fetch can start and deadlocks, the same wrong usage as use()-ing a promise created during your own render, and it is not guarded against. Suspends until the first emission; later emissions update without re-suspending. Errors reject the promise (Error Boundary). Hidden <Activity> trees calling the hook stay paused until revealed.
Client components only: on the server the observable is never subscribed, so server rendering emits the Suspense fallback and the fetch starts after hydration. react-rx is not a library for React Server Components or server-only flows — see server rendering in the guide. Also see the guide for startWith caveats, disabled / ttl, Activity patterns, and when to prefer useObservable.
Example
import {Suspense, use, useMemo} from 'react'
import {useObservablePromise} from 'react-rx'
import {fromFetch} from 'rxjs/fetch'
function Profile({url}: {url: string}) {
const data$ = useMemo(() => fromFetch(url, {selector: (r) => r.json()}), [url])
const promise = useObservablePromise(data$)
return (
<Suspense fallback="Loading…">
<Pre promise={promise} />
</Suspense>
)
}
function Pre({promise}: {promise: Promise<unknown>}) {
return <pre>{JSON.stringify(use(promise), null, 2)}</pre>
}preloadObservablePromise()
Warm the useObservablePromise cache before any consumer is live (for example on mouseenter, in a route loader, or ahead of a transition swap so it commits with no pending period). Not a hook — callable anywhere. Returns the same promise instance the hook would return for that observable.
Calling it starts the source subscription immediately, before any component has committed. On the server it is a no-op: it returns an inert, forever-pending promise and subscribes nothing (react-rx never subscribes observables on the server), so preloads in shared/isomorphic code only take effect in the browser. Pending entries are never timed out, so a never-emitting / hung observable keeps both the promise and the subscription alive until it settles. Prefer RxJS timeout (or cancel the source) when a preload can stall.
Signature
function preloadObservablePromise<T>(
observable: Observable<T>,
options?: {ttl?: number},
): ObservablePromise<T>Default ttl is 5000 (longer than the hook default) so a hover-warmed value survives until click/navigation.
createObservableCache()
Creates a keyed cache of shared observables for request-like sources (ones that complete), so every caller of a key gets the same observable. Not a hook — call it once, in module scope or wherever the data layer lives.
Sharing the identity is the point: useObservablePromise caches its promise per observable, so consumers of a key share one request and one settled promise, and a sibling or a remount renders the loaded value instead of suspending again. An observable built per component (useMemo(() => fetchUser(id), [id])) is a new identity for every consumer, and every one of them fetches and suspends on its own.
Pass the cache’s ttl to the hook as well: a remount within that window reuses the promise the hook settled, so it renders no fallback. The two windows nest rather than coincide — the hook holds the cached observable for its own ttl after the last consumer unmounts, and the cache’s ttl only starts once it lets go. So a remount in between renders a fallback but makes no request, and a key is refetched only after both have elapsed: ttl: 60_000 on each means two minutes, not one.
share dedupes the rest: concurrent subscribers join the same request, and its result replays to later ones. A key is released ttl after its last subscriber leaves — and an entry nothing ever subscribed to, such as one a render that never committed created, ttl after it was created. The same ttl is the grace an abandoned request gets: one that loses every subscriber keeps running for ttl, long enough for a remount to rejoin it, and is torn down with its key after that rather than left connected for nobody. Teardown unsubscribes the source; whether it also aborts the work in flight is the input’s business — fromFetch aborts, a plain promise keeps going and has its result dropped. cache.clear() forgets every key at once, for invalidating after a mutation: live subscribers keep theirs, and everything else is torn down now — pending timers included — rather than waiting out its ttl.
Failures are never cached: an error releases the key before it reaches subscribers, so the next caller — an error handler reading the cache included — retries on a fresh entry. A request that completes without emitting counts as a failure, erroring with RxJS EmptyError the same way firstValueFrom and useObservablePromise treat an empty source. Since nothing holds the failure, a key that keeps failing is fetched again by every render that reads it — bound that with RxJS retry inside fetch when an endpoint can stay down. Keys are plain strings, built by the caller from whatever identifies the request.
The cached observable ends at the first value the request produces and then stays open, because subscribers are what keep the key — a completing source would release it the moment the result arrived, and the hook stops subscribing to a terminated source. The hooks handle the open end, but outside them take the result with first() or firstValueFrom, not an operator that waits for completion.
Expect one value per key. A source that keeps emitting, or that fails after its result, is truncated at that first value rather than forwarded to subscribers who have already settled — a late failure would otherwise release the key while they still hold the result, and every one of them would refetch on its next render. Keyed streams that should keep emitting are a different primitive; this one caches requests.
Signature
function createObservableCache<T>(
fetch: (key: string) => ObservableInput<T>,
options: {ttl: number},
): ObservableCache<T>
interface ObservableCache<T> {
(key: string): Observable<T>
clear(): void
}Example
import {Suspense, use} from 'react'
import {createObservableCache, useObservablePromise} from 'react-rx'
import {fromFetch} from 'rxjs/fetch'
const USER_TTL = 60_000
const users = createObservableCache(
(id: string) =>
fromFetch(`/api/users/${id}`, {
// `fromFetch` resolves 4xx/5xx the way `fetch` does, so without this an error
// body is a successful result and gets cached as a user for the whole ttl
selector: (response) => {
if (!response.ok) {
throw new Error(`GET /api/users/${id} failed with ${response.status}`)
}
return response.json()
},
}),
{ttl: USER_TTL},
)
function Avatar({id}: {id: string}) {
const promise = useObservablePromise(users(id), {ttl: USER_TTL})
return (
<Suspense fallback={<AvatarSkeleton />}>
<AvatarImage promise={promise} />
</Suspense>
)
}
function AvatarImage({promise}: {promise: Promise<{imageUrl: string}>}) {
return <img src={use(promise).imageUrl} alt="" />
}useObservableSubject()
Creates an RxJS Subject scoped to the component instance and returns its two halves: an observable of the values pushed into it, plus a stable handler that pushes them.
Only the observable side of the Subject is exposed, so the pipeline cannot accidentally push into the stream and the handler cannot be subscribed. The handler is referentially stable, so it can be passed straight to event props or memoized children without useCallback. Read the observable with useObservable / useSyncObservable when the pipeline produces something to render, or subscribe it in an effect for side-effect-only pipelines. Values emitted while nothing is subscribed are dropped, exactly like a Subject.
useObservableEvent was removed in v7; this hook is the replacement. Call sites migrated to it on v6.1 need no further changes. See the v6 → v7 migration guide.
Signature
function useObservableSubject<T>(): [events$: Observable<T>, handleEvent: (event: T) => void]Example
import {useMemo} from 'react'
import {useObservableSubject, useSyncObservable} from 'react-rx'
import {map} from 'rxjs'
const ShowSliderValue = () => {
const [input$, handleChange] = useObservableSubject<string>()
const value$ = useMemo(() => input$.pipe(map((value) => Number(value))), [input$])
// The derived stream is the state — no setState mirror needed.
const value = useSyncObservable(value$, 1)
return (
<>
<input
type="range"
value={value}
onChange={(event) => handleChange(event.currentTarget.value)}
min={1}
max={10}
/>
<div>Value is: {value}</div>
</>
)
}Side-effect-only pipelines (analytics, persistence, …) subscribe the observable in an effect instead:
import {useEffect} from 'react'
import {useObservableSubject} from 'react-rx'
import {concatMap} from 'rxjs'
function SaveSearchButton({term}: {term: string}) {
const [saves$, handleSave] = useObservableSubject<string>()
useEffect(() => {
const subscription = saves$.pipe(concatMap((t) => saveSearch(t))).subscribe()
return () => subscription.unsubscribe()
}, [saves$])
return <button onClick={() => handleSave(term)}>Save search</button>
}What happened to useObservableEvent()?
It was removed in v7. Use useObservableSubject and read the derived stream with one of the other hooks — see Handling events and the v6 → v7 migration guide.