Reactive Search with RxJS and React Shadcn

Why a search box is a stream

Any screen that lists data eventually grows a search field. That field is a sequence of strings arriving over time, and most of those strings should never become a request. What matters is the timing of the pauses, and the text that is left when a pause is long enough.

That is a good fit for a reactive stream. I have written this component in Angular and in React, and RxJS is the clearest way I have found to describe the rules in one place.

What the stream has to guarantee

A basic search has three rules:

• Send a request when the trimmed text changes, and only after the user has stopped typing for debounceTime.
• If a request is still in flight when a newer query is ready, drop the old one. A slow response for “pi” must not overwrite the results for “pikachu”.
• If a request fails, clear the results and keep the stream alive. The next keystroke still has to work.

An empty box is ignored. The previous results stay on screen until the next non-empty query.

The stream

The source is a Subject of raw input. Everything else is a pure description of what to do with those strings. The React state setters are the only side effects, and they sit in tap and catchError so the pipeline itself stays easy to read top to bottom.

import {type Dispatch, type SetStateAction} from 'react'
import {
  type Observable,
  Subject,
  catchError,
  debounceTime,
  distinctUntilChanged,
  filter,
  map,
  merge,
  of,
  switchMap,
  takeUntil,
  tap,
} from 'rxjs'

export const DEBOUNCE_TIME = 500

export function searchObserver$<T>(
  textSubject$: Subject<string>,
  fetcher$: (text: string) => Observable<T[]>,
  {
    setIsLoading,
    setOptions,
  }: {
    setIsLoading: Dispatch<SetStateAction<boolean>>
    setOptions: Dispatch<SetStateAction<T[]>>
  },
  debounceTimeout = DEBOUNCE_TIME,
) {
  return textSubject$.pipe(
    map((query = '') => query?.trim()),
    // A blank box is not a search.
    filter<string>(Boolean),
    debounceTime(debounceTimeout),
    // Typing "pik" and backspacing to the previous word should not refetch.
    distinctUntilChanged(),
    tap(() => setIsLoading(true)),
    // Drop the previous request when a new debounced query arrives.
    // takeUntil also aborts it on the next keystroke, before that
    // debounce window closes.
    switchMap((text) => fetcher$(text).pipe(takeUntil(textSubject$))),
    tap((options) => {
      setOptions(options)
      setIsLoading(false)
    }),
    catchError((_error, caught) => {
      setOptions([])
      setIsLoading(false)
      // `caught` is this same stream. Resubscribe so one failure
      // does not complete the search for good.
      return merge(of([] as T[]), caught)
    }),
  )
}

Reading the pipe in order:

1. map trims. A default of '' covers a missing value.
2. filter(Boolean) drops ''. Whitespace-only input dies here too, because it was trimmed first.
3. debounceTime waits. Each new character resets the timer, so a burst of typing produces one value: the text present after the quiet period.
4. distinctUntilChanged compares that debounced text with the previous one. If the user edits the field and restores the same query before the timer fires, no request goes out.
5. tap turns loading on only once a query has survived those gates.
6. switchMap subscribes to the fetcher for the latest query and unsubscribes from the previous fetcher when a new debounced query arrives. takeUntil(textSubject$) is stricter than that. The subject emits on every keystroke, including characters that have not passed the debounce yet, so an in-flight request is aborted as soon as the user keeps typing.
7. The second tap writes the options and turns loading off. It runs for a response that actually arrived.
8. catchError runs when the fetcher errors. It clears the options, turns loading off, and resubscribes to the source. Without that resubscribe, the first error would complete the observable and later typing would do nothing. merge(of([]), caught) emits one empty array for any subscriber that is listening, then continues with the restarted stream.

switchMap alone would cancel on the next debounced value. The takeUntil is what cancels on the next raw keystroke. Both are here on purpose: the user should not pay for a request whose query is already stale.

Reuse it from a React hook

The component should not know about subjects. The hook owns the subject, starts the subscription when the component mounts, and stops it on unmount. Calling the function it returns is just next.

import {type Dispatch, type SetStateAction, useEffect, useRef} from 'react'
import {type Observable, Subject} from 'rxjs'

import {searchObserver$} from './searchObserver'

export function useSearch<T>(
  fetcher$: (text: string) => Observable<T[]>,
  componentSetters: {
    setIsLoading: Dispatch<SetStateAction<boolean>>
    setOptions: Dispatch<SetStateAction<T[]>>
  },
  debounceTime?: number,
) {
  const textSubject$ = useRef(new Subject<string>())

  useEffect(() => {
    const watcher = searchObserver$<T>(
      textSubject$.current,
      fetcher$,
      componentSetters,
      debounceTime,
    ).subscribe()

    return () => watcher.unsubscribe()
    // The stream is built once. State setters from useState are stable.
    // The fetcher should be stable too, or this closure keeps the first one.
    // eslint-disable-next-line react-hooks/exhaustive-deps
  }, [])

  return (text: string) => textSubject$.current.next(text)
}

The subject lives in a ref so each render reuses the same hot source. A new Subject per render would orphan the subscription. The effect returns the unsubscribe, which is what stops the stream when the component goes away.

The empty dependency list is a real constraint. The subscription closes over the fetcher from the first render. useState setters are stable, so those are safe. If the fetcher changes identity every render, pass a stable function, or this hook will keep calling the original one.

The request has to be cancellable

Unsubscribing only helps if the request underneath actually stops. fromFetch is the fetch API adapted to an observable: the request starts on subscribe, and unsubscribing aborts it. A cancelled search therefore cannot arrive later and overwrite a newer result. A plain promise that ignores abort would still resolve, and switchMap would only have thrown the value away after the network cost was paid.

import {fromFetch} from 'rxjs/fetch'
import {map, type Observable} from 'rxjs'

import {type Option} from '../components/autocomplete'

const URL = `https://digimoncard.io/api-public/search.php`

export function searchDigimonCards$(search: string): Observable<Option[]> {
  return fromFetch<Record<string, string>[]>(`${URL}?n=${search}`, {
    selector: (response) => response.json(),
  }).pipe(map((cards) => cards.map(({id, name}) => ({label: name, value: id}))))
}

The selector reads the JSON body. The following map is an ordinary RxJS operator that shapes each card into the { label, value } option the autocomplete expects.

Demo

The input below is that hook, wired to a shadcn autocomplete. Type, pause, and change the query mid-request.