Understanding RxJS use cases (part I)

Understanding the right moment to reach for RxJS

Across the web, numerous tutorials and courses do a solid job of explaining RxJS primitives — functions, Observables, Subjects, and operators. Yet, simply knowing that these tools exist is rarely enough to unlock the full potential of reactive programming. The real challenge lies in recognizing which tool fits a given problem. This article explores practical scenarios where specific RxJS utility functions make code cleaner, more maintainable, and easier to read.

The first part focuses on the most commonly used stream-combining functions rather than operators. Since the operator list is extensive, their use cases will be covered in later installments.

Merging streams together

Functions like combineLatest, merge, and zip help merge multiple Observables into one flow of data. Most developers are aware of their mechanics, but spotting when a problem calls for them is often trickier. Let us walk through these stream-combining utilities and identify the situations where each one shines.

merge: when only the occurrence of an event matters

The merge operator produces a stream that emits whenever any of the source Observables emits. Consider this basic illustration:

// create three observable streams
// one of strings, other of numbers, and the other of booleans
const numbers$ = interval(1000).pipe(
  map(i => i + 1), 
  take(3),
); // take only the first three
const strings$ = interval(2000).pipe(
  map(i => 'a' + i),
  take(3),
);
const booleans$ = interval(3000).pipe(
  map(i => i % 2 === 0),
  take(3),
);
// merge the three streams together
const merged$ = merge([numbers$, strings$, booleans$]);
// subscribe to the merged stream
merged$.subscribe(console.log);

Here, a stream of numbers, strings, and booleans is combined into a single flow. Values appear at varying intervals, so the merged stream is genuinely mixed — sometimes a number, sometimes a boolean, sometimes a string. Because the emitted value changes type frequently, it becomes clear that the actual payload is irrelevant; only the fact that something was emitted counts. That may sound limited, but several real-world situations fit perfectly.

Take user inactivity detection. Suppose we want to log a user out or display a warning after a period of no interaction. We must track events such as clicks, scrolls, and right-clicks, then react when nothing has happened for a while. This is how it looks:

const ACTIVE_EVENTS = [
  'click', 'scroll', 'contextmenu', 'dblclick', 'mousemove',
];
// you can add as many events as you want to define "being inactive"

merge(...ACTIVE_EVENTS.map(event => fromEvent(document, event))).pipe(
bufferWhen(() => interval(10_000)),
filter(events => events.length === 0),
).subscribe(() => alert('You have been inactive for ten seconds!'))

In this snippet, multiple Observables created with fromEvent watch distinct browser events that signal user activity. These are merged into a single stream, and we respond when no events arrive for a set duration. Whether the event was a scroll, a double-click, or a regular click is irrelevant. The reasoning boils down to: if only the occurrence matters — not the event type — merge is the right choice.

(The example also uses bufferWhen; a full explanation of that operator is outside this article’s scope. Read about it here, and see a Twitter discussion with alternative explanations here.)

combineLatest: when values from all sources are needed

There are cases where two or more separate events each trigger an update in the same part of the UI. Unlike the previous scenario, the emitted values themselves matter — and more specifically, we need the latest value from every source to compute a final result that the UI reflects.

Imagine a dynamic form where fields come from a backend API endpoint, and the values a user enters into some fields affect validation rules on others. For instance, if the user chooses to supply an address separately, the “Address Line” field becomes mandatory. On one side, the form emits values that may trigger UI changes. On the other side, we load dynamic field definitions from a backend via a custom service using HttpClient, which returns an Observable. Any change from either side — the form or the dynamic configuration — must update the UI, and we need both latest values to calculate the result and perform side effects such as setting validators on FormControl instances.

Here is how combineLatest neatly handles this:

export class ExampleComponent {
    // rest of the component code omitted for the sake of brevity
    dynamicControls$ = this.controlsService.getDynamicControls();
    formValue$ = combineLatest([
        this.form.valueChanges,
        this.dynamicControls$,
    ]).pipe(
        tap(([value]) => {
            if (value.attachmentsRequired) {
                this.controls.attachments
                    .setValidators(Validators.required);
            } else {
                this.controls.attachments.clearValidators();
            }
        }),
        map(([value, controls]) => {
            const controlsValue = { ...value, ...controls };
            return controlsValue;
        }),
    );

    constructor(
    private readonly controlsService: ControlsService,
    ) { }
}

We take both latest values, perform side effects based on conditions from one source, and then merge them into a single output. Using merge here would fail because it emits a value from just one Observable at a time, whereas combineLatest gives us the fresh value along with the latest value from the other Observable.

forkJoin: when only the complete result counts

forkJoin is arguably the most well-known stream-combiner in RxJS. It is often described as the reactive equivalent of Promise.all. When data must be fetched from multiple API endpoints, forkJoin waits until every request completes before doing anything on the UI:

homePageData$ = forkJoin([
    this.userService.getUserInfo(),
    this.dataService.getData(),
    this.otherDataService.getOtherData(),
]).pipe(
    map(([userInfo, data, otherData]) => ({
        userInfo,
        data,
        otherData,
    })),
    catchError(error => of({/*error object*/})),
);

In practice, this function is typically paired with HTTP calls.

pairwise: inspecting values from the past

At times we need to pair the current value of an Observable with a value it emitted previously. Consider a profile edit form that comes prefilled with data. There is a “Save” button, but it should stay disabled until the user actually modifies something.

For simple cases, the form’s dirty flag works. But what if the user edits a field and then changes it back to the original value? The form is still marked dirty even though the data is identical to the initial state. We could compare the current form value against the previous one to confirm a real difference exists. This is precisely where pairwise steps in:

disabled$ = this.form.valueChanges.pipe(
    pairwise(),
    map(([prev, current]) => {
        return this.utilitiesService.isEqual(prev, current);
        // will disable the button if the form has not changed
    }),
);

(The isEqual method performs a deep comparison between the previous and current form values.)

withLatestFrom: incorporate another value without responding to its emissions

There are situations where a calculation on the source Observable must factor in a value coming from a different Observable, yet we deliberately want to avoid triggering that calculation when the secondary Observable emits. This contrasts with combineLatest, which fires whenever any participating Observable produces a new value.

Consider a login flow that should only redirect when a redirect_url query parameter exists. The parameter value can be read from the queryParamMap Observable, but the redirect should not occur merely because that parameter changes. Instead, the redirect should happen solely after the login HTTP request completes successfully:

this.authService.login(credentials).pipe(
    withLatestFrom(
      this.route.queryParamMap.pipe(startWith(new Map())),
    ),
).subscribe(([, params]) => {
    if (params.get('redirectUrl')) {
        const navUrl = params.get('redirectUrl') ?? '/home';
        this.router.navigateByUrl(decodeURIComponent(navUrl));
    }
});

Here, withLatestFrom retrieves the query parameters at the moment the login succeeds. Changes to the query parameters themselves do not trigger the redirect; the redirect is only initiated once the login call has completed without error.

What lies ahead

This piece covered the purposes of the functions and operators used to combine Observables. The following article will shift focus to operators that act on individual streams, examining their practical applications so they can be adopted with greater confidence.