Reactive programming in RxJS

Alaeddine DouagiWeb Developer

A list that arrives over time

An array is a pull collection. You decide when to read items[0], and the value is already there. A promise is a push of a single future value: you register a reaction, and the value is delivered once. An Observable is the general case of that push. It can deliver zero values, one value, or many, on whatever schedule the source uses, and then it either fails or finishes.

That is the whole of the idea behind RxJS. The library, currently at 7.8, is the JavaScript implementation of ReactiveX. It gives you a type for those streams and a set of operators that return new streams. The current import style, since 7.2, takes creation functions and operators from 'rxjs' itself. fromFetch still comes from 'rxjs/fetch'.

import {filter, map, range} from 'rxjs'

range(1, 5)
  .pipe(
    filter((n) => n % 2 === 1),
    map((n) => n * 10),
  )
  .subscribe((n) => console.log(n))
// 10, 30, 50

range(1, 5) does not run when you create it. It runs when someone subscribes. pipe does not change range(1, 5). It returns a new observable that filters and multiplies whatever the source emits. Subscribing to the result is what subscribes upstream.

The contract

Creating an observable means handing RxJS a function that runs per subscriber. That function receives a subscriber and may return a teardown.

import {Observable} from 'rxjs'

const letters$ = new Observable((subscriber) => {
  subscriber.next('H')
  subscriber.next('i')
  subscriber.complete()

  return () => {
    // Release whatever this execution acquired.
  }
})

const subscription = letters$.subscribe({
  next: (value) => console.log(value),
  error: (error) => console.error(error),
  complete: () => console.log('done'),
})

subscription.unsubscribe()

Three notifications exist:

next delivers a value. It can happen any number of times, including zero.
error delivers a failure and ends the stream.
complete ends the stream successfully.

After error or complete, further notifications are ignored. A stream does not emit an error and then a completion, and it does not resume afterward. The observable guide is worth reading with that rule in mind, because most confusing bugs are a stream that ended when you expected it to stay open.

Execution is lazy. A cold observable runs its setup once per subscriber. new Observable, range, fromFetch, and fromEvent are all cold. Two subscribers to fromEvent attach two listeners, and two subscribers to fromFetch send two requests. A Subject is hot: pushing a value notifies current subscribers only, and a value that fired before you subscribed is gone. Add share() when one listener or one request should serve several subscribers.

import {fromEvent, map} from 'rxjs'

const positions$ = fromEvent(document, 'click').pipe(
  map((event) => ({
    x: event.clientX,
    y: event.clientY,
  })),
)

const subscription = positions$.subscribe((position) => {
  console.log(position.x, position.y)
})

Unsubscribe when the listener should detach. The teardown returned from the subscribe function is how a custom observable drops timers, sockets, and abort controllers. Operators such as take, takeUntil, and first unsubscribe for you when their condition is met.

Operators are decisions about the values

Once values are a stream, the useful work is deciding which values survive, how they change shape, and how they line up with other streams.

map and filter behave like the array methods, applied to each arrival. debounceTime(ms) emits the latest value only after ms milliseconds of silence, and it drops the values that were waiting. distinctUntilChanged drops a value that is equal to the previous one. scan is reduce for a stream that may never complete: it emits each intermediate accumulation.

import {fromEvent, map, scan} from 'rxjs'

fromEvent(document.querySelector('#count'), 'click')
  .pipe(
    map(() => 1),
    scan((total, tick) => total + tick, 0),
  )
  .subscribe((total) => console.log(total))

Combination is a separate family. merge interleaves two streams as their values arrive. concat subscribes to the second only after the first completes, which means a source that never completes will starve everything after it. combineLatest waits until every source has emitted once, then emits the latest tuple whenever any source emits. forkJoin waits until every source completes and emits a single tuple of final values. It is the closest thing in RxJS to Promise.all, and it is the wrong tool for a click stream, because a click stream does not complete.

The four ways to flatten

The step that usually decides whether RxJS feels clear or haunted is a higher-order observable: a stream whose values are themselves streams. A search box produces one of these. Each query should become a request, and a request is already an observable or a promise.

Flattening answers one question. A new inner stream has arrived, and an older one may still be running. What happens to the older one?

concatMap queues later inner streams until the active one completes. Reach for it when order must be preserved, such as sequential saves.
mergeMap subscribes to every inner stream immediately. Reach for it when the work is independent and results may arrive out of order.
switchMap unsubscribes from the previous inner stream. Reach for it when only the latest result matters.
exhaustMap ignores new inner streams until the active one finishes. Reach for it when a second click during a submit must do nothing.

The higher-order observables guide states the same four behaviors. Applied to a typeahead, they are not interchangeable.

mergeMap would let every keystroke’s request run. Responses return in network order, so the results for an earlier, slower query can replace the results for the query the user is looking at now.

concatMap would run those requests one after another and deliver them in keystroke order. The user would wait through queries they have already moved past.

exhaustMap would keep the first request and ignore typing until it finished. That is what you want on a submit button, where a double click should not create a second order. It is a poor search box.

switchMap unsubscribes from the previous inner observable and subscribes to the new one. Paired with debounceTime, it is the typeahead shape:

import {
  debounceTime,
  distinctUntilChanged,
  filter,
  from,
  fromEvent,
  map,
  switchMap,
} from 'rxjs'

const input = document.querySelector('#search')

fromEvent(input, 'input')
  .pipe(
    map((event) => event.target.value.trim()),
    filter((query) => query.length > 0),
    debounceTime(300),
    distinctUntilChanged(),
    switchMap((query) =>
      from(
        fetch(`/api/search?q=${encodeURIComponent(query)}`).then((response) => {
          if (!response.ok) throw new Error(`Search failed: ${response.status}`)
          return response.json()
        }),
      ),
    ),
  )
  .subscribe((results) => console.log(results))

There is a limit hiding in that snippet. from(fetch(...)) will ignore the result of an unsubscribed inner observable, but the browser request keeps going, because a bare fetch was given no AbortSignal. Cancellation is real only when unsubscription aborts the underlying work. fromFetch from 'rxjs/fetch' does that. I used it for a React autocomplete in Reactive Search with RxJS and React Shadcn, where the same switchMap pipeline also has to survive an error and stay subscribed for the next query.

mergeMap accepts a concurrent limit, mergeMap(project, 4), when you want parallelism with a ceiling. Leaving it unlimited in front of a network call is how a stream tips over a backend.

Stopping is part of the design

A subscription is a live execution. In a page component, the execution should end when the component does.

import {fromEvent, interval, switchMap, takeUntil} from 'rxjs'
import {fromFetch} from 'rxjs/fetch'

const stop$ = fromEvent(document.querySelector('#stop'), 'click')

interval(1000)
  .pipe(
    switchMap(() =>
      fromFetch('/api/health', {
        selector: (response) => response.json(),
      }),
    ),
    takeUntil(stop$),
  )
  .subscribe((status) => console.log(status))

takeUntil completes the output when stop$ emits. That unsubscribes from interval and from the inner fromFetch, and fromFetch aborts the request when its subscription ends. A bare fetch(...).then(...) inside switchMap would only drop the result. The browser request would keep going. In React, the same role is played by the cleanup function of useEffect: call subscription.unsubscribe() there. In Angular, takeUntilDestroyed() does it from an injection context.

Errors travel on the same channel as values. A callback-style API can forget to call the error branch. An observable cannot emit a failure quietly: either an operator such as catchError handles it, or the subscriber’s error path runs and the stream is done. catchError can return a replacement observable, which is how a search stream clears its results and resubscribes instead of dying on the first 500.

Where a stream earns its cost

RxJS is a cost. It adds a vocabulary and a subscription lifecycle, and a single request fired from a click handler is clearer as async/await.

A stream earns that cost when the problem is about time and cancellation together: typeahead, dragging, websockets, polling that must stop, or several event sources that have to be combined. Model the source, write the decisions as operators, pick the flattening behavior from the table above, and decide what unsubscribes it. The operators are the program.