An RxJS primer

RxJS is a library for working with things that happen over time: clicks and keypresses, responses from a server, timers, messages arriving on a WebSocket. It gives all of them the same shape, the observable, and a set of operators to transform and combine them.

Angular is built on it, so if you're writing Angular you're already using it. HttpClient returns observables, and so do the router's events and a form's value changes.

Let's start with what an observable is, then build up to a search box that's surprisingly hard to get right without one.

Observables

An observable is a source of values that arrive over time. You can make one by hand, with a function that's given a subscriber to send values to:

import { Observable } from 'rxjs';

const numbers = new Observable((subscriber) => {
    subscriber.next(1);
    subscriber.next(2);
    subscriber.next(3);
    subscriber.complete();
});

numbers.subscribe({
    next: (value) => console.log(value),
    complete: () => console.log('done'),
});

1
2
3
done

Nothing happens when the observable is made. The function runs when something subscribes, and it runs again for every subscriber.

A subscriber gets three kinds of notification: next for each value, error if something goes wrong, and complete when there won't be any more. After an error or complete, nothing else arrives. Most of the time you only want the values, so you can pass just one function:

numbers.subscribe((value) => console.log(value));

Making observables

You rarely need to write one by hand, because RxJS has functions for the usual sources:

import { of, from, interval, timer, fromEvent } from 'rxjs';

of(1, 2, 3);                      //  these values, then complete
from([1, 2, 3]);                  //  the items of an array
from(fetch('/api/user'));         //  a promise's result
interval(1000);                   //  0, 1, 2... one a second, forever
timer(500);                       //  0 after half a second, then complete
fromEvent(document, 'click');     //  every click, forever

Unsubscribing

Some observables never complete. An interval goes on forever, and so do clicks, so subscribe returns a subscription that you use to stop listening:

const subscription = interval(1000).subscribe((n) => console.log(n));

setTimeout(() => subscription.unsubscribe(), 3500);   //  logs 0, 1 and 2

When you write an observable yourself, its function can return what to do on unsubscribe. This is how interval stops its timer and fromEvent removes its event listener:

const ticks = new Observable((subscriber) => {
    const id = setInterval(() => subscriber.next(Date.now()), 1000);
    return () => clearInterval(id);   //  runs when the subscriber unsubscribes
});

Forgetting to unsubscribe is the most common bug with RxJS. A component goes away but its subscription keeps running, holding on to the component and doing work that nobody sees.

Operators

Operators make a new observable from an existing one. You chain them with pipe, and each one works on the values coming out of the one before:

import { interval } from 'rxjs';
import { filter, map, take } from 'rxjs/operators';

interval(1000).pipe(
    filter((n) => n % 2 === 0),
    map((n) => n * 10),
    take(3),
).subscribe((value) => console.log(value));   //  0, 20, 40

filter and map work like the array functions with the same names, except that the values are spread out in time. take(3) passes on three values and then completes, which also unsubscribes from the interval, so this one tidies up after itself.

A few more come up all the time:

• tap(fn) runs a function for each value without changing it, which is handy for logging.
• startWith(value) sends a value first, before any of the others.
• scan((total, value) => ..., seed) is reduce over time: it sends the running total after each value.
• debounceTime(ms) waits until values have stopped arriving for a while, then sends the last one.
• distinctUntilChanged() drops a value that's the same as the one before it.

Subjects

A subject is an observable that you push values into yourself. It has next, error and complete like a subscriber, and everything subscribed to it gets what's pushed. A BehaviorSubject also remembers its latest value, and gives it to each new subscriber straight away:

import { BehaviorSubject } from 'rxjs';

const selectedId = new BehaviorSubject(null);

selectedId.subscribe((id) => console.log('selected', id));   //  selected null
selectedId.next(42);                                          //  selected 42
console.log(selectedId.getValue());                           //  42

In Angular this is the usual way for a service to hold some state that components watch. The service keeps the subject to itself and hands out asObservable(), so nothing else can change it:

@Injectable({ providedIn: 'root' })
export class SelectionService {
    private selectedId = new BehaviorSubject<number | null>(null);
    readonly selectedId$ = this.selectedId.asObservable();

    select(id: number) {
        this.selectedId.next(id);
    }
}

The $ on the end of a name is the usual way of saying that it's an observable.

Combining streams

Some functions take several observables and make one. merge passes on every value from all of them as it arrives, so here a save happens on a click or on Ctrl+S:

import { merge, fromEvent } from 'rxjs';
import { filter } from 'rxjs/operators';

const saves$ = merge(
    fromEvent(saveButton, 'click'),
    fromEvent(document, 'keydown').pipe(filter((event) => event.ctrlKey && event.key === 's')),
);

combineLatest sends an array of the latest value from each of them, every time any one of them changes:

import { combineLatest } from 'rxjs';
import { map } from 'rxjs/operators';

const total$ = combineLatest([quantity$, price$]).pipe(
    map(([quantity, price]) => quantity * price),
);

It doesn't send anything until every one of them has sent at least one value, which is a common surprise. startWith on each of them gives it something to begin with.

A search box

Here's where RxJS really earns its place. A search box should search as you type, but not on every keypress. It shouldn't search again for the same text. And when answers come back out of order, it should only ever show the results for the latest text. With callbacks and flags that's fiddly to get right. With RxJS it's one pipe:

import { fromEvent } from 'rxjs';
import { ajax } from 'rxjs/ajax';
import { map, debounceTime, distinctUntilChanged, switchMap } from 'rxjs/operators';

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

const results$ = fromEvent(input, 'input').pipe(
    map(() => input.value.trim()),
    debounceTime(300),
    distinctUntilChanged(),
    switchMap((term) => ajax.getJSON('/api/search?q=' + encodeURIComponent(term))),
);

results$.subscribe((results) => showResults(results));

Each keypress becomes the text in the box. debounceTime waits until typing has paused for 300ms, and distinctUntilChanged skips a search for the text we've just searched for (typing a letter and deleting it again, say). Then switchMap turns each term into a request.

switchMap is the important one. For each value it subscribes to a new observable, here a request, and passes on what that sends. When the next value arrives, it unsubscribes from the previous one before starting the next. Unsubscribing from an ajax request aborts it, so a slow answer to an old search can never overwrite the results for a newer one:

It has relatives for when you want something else to happen to the one before. mergeMap lets them all run at once, concatMap queues them up and runs them one at a time, and exhaustMap ignores new values while one is still running, which is right for a submit button that shouldn't submit twice.

Errors

An error ends a stream. That matters in the search box: if one request fails, the error comes out of switchMap, the whole stream ends, and the search box stops working for good. So catch it inside, on the request, where it only ends that one request:

import { of } from 'rxjs';
import { catchError } from 'rxjs/operators';

switchMap((term) => ajax.getJSON('/api/search?q=' + encodeURIComponent(term)).pipe(
    catchError(() => of([])),   //  show no results, and keep the search box working
)),

catchError swaps the error for another observable, here one that sends an empty list. The other tool is retry(n), which subscribes again, up to n times, before letting the error through. For a request, that means trying it again.

In Angular

In Angular you mostly get observables rather than make them. HttpClient's get returns one that makes the request when it's subscribed to, sends the response, and completes. The async pipe subscribes from the template and unsubscribes when the component goes away, which takes care of that most common bug. Here's the search box as a component:

@Component({
    selector: 'app-search',
    template: `
        <input [formControl]="search">
        <ul>
            <li *ngFor="let result of results$ | async">{{ result.title }}</li>
        </ul>
    `,
})
export class SearchComponent {
    search = new FormControl('');

    results$ = this.search.valueChanges.pipe(
        debounceTime(300),
        distinctUntilChanged(),
        switchMap((term) => this.http.get<SearchResult[]>('/api/search', { params: { q: term } }).pipe(
            catchError(() => of([])),
        )),
    );

    constructor(private http: HttpClient) {}
}

The form control's valueChanges takes the place of the input events, and the template never subscribes or unsubscribes by hand.

Things to know

• Observables are lazy. Nothing happens until something subscribes, so an HttpClient call that nothing subscribes to never makes its request.
• Each subscription runs the observable again. Subscribe twice to the same request (two async pipes on it, say) and it's made twice. share() makes the subscribers share one run, and shareReplay(1) also gives late subscribers the last value.
• Unsubscribe from anything that doesn't complete by itself. Let the async pipe do it, use take(1) when you only want one value, or takeUntil with a subject that fires when the component is destroyed.
• An error ends the stream, so catch errors where a failure should only end one piece of work.