Understanding RxJS and Observables

Contemporary web applications heavily depend on a multitude of asynchronous events. When a user interacts with a button, the app might initiate a network call. Upon the response from that call, the DOM could be updated. This ongoing cycle of actions and reactions is a daily reality for web developers.

To manage such events, we rely on implementations of the observer pattern. This design pattern enables different parts of an application to respond efficiently to specific events or data streams. This approach underpins event-driven architecture, where functions are executed based on the reception of new data.

Callbacks and promises are the two most prevalent techniques for handling these events. Below is how our earlier example would be implemented with each method:

// callback
document.addEventListener('clicks', () => {
    // promise
    fetch('https://api.mocki.io/v1/b043df5a').then((response) => {
        response.json().then((data) => {
            // update state
        });
    });
});

In this scenario, the document object serves as an infinite stream of clicks events. Likewise, network responses are delivered in fragments, creating a sequence of progress events. Given this combination of asynchronous events and sequential data, the Observable type and the RxJS library become particularly useful.

An Observable is essentially a sequence of values that can be observed. It acts as a conduit for data streams, managing and delivering the data to Observers. There is an ongoing initiative to introduce Observable as a standard type in the ECMAScript specification.

The RxJS library offers the Observable primitive alongside operators that facilitate the composition of asynchronous event sequences.

Observable provides a straightforward, unified API capable of representing a diverse set of data sources, from single values and streams to user interactions and both synchronous and asynchronous data. The true strength of Observables, however, lies in their distinct properties and the guarantees they offer.

The Observable type is lazy and provides the following guarantees:

  • After completion, an error, or an unsubscription, no further messages will be delivered
  • A registered teardown will be executed
  • Upon completion, error, or unsubscription, resources are guaranteed to be cleaned up

Observables also exhibit these characteristics:

  • Compositional: They can be combined using higher-order combinators
  • Lazy: They wait for an observer to subscribe before emitting any data
  • Cancellable: Subscriptions allow you to cancel the underlying operation, such as an event listener or a request

Let's revisit our earlier code example and explore how to adapt it to take full advantage of the Observable's features.

First, we will use the Observable constructor to build a stream of clicks events tied to the document:

import { Observable } from 'rxjs';

const producer = observer => {
    // Create an event handler which sends data to the observer
    let handler = event => observer.next(event);

    // Attach the event handler
    document.addEventListener('click', handler, true);

    // Return a cleanup function which will cancel the event stream
    return () => {
        // Detach the event handler from the element
        document.removeEventListener('click', handler, true);
    };
};

const events = new Observable(producer);

The logic for setting up and tearing down a data stream resides within a producer function. The Observable takes this function and manages the surrounding infrastructure, such as invoking producer upon subscription and executing the cleanup logic when the subscription is terminated.

RxJS ships with a wide array of operators, and fromEvent simplifies creating a stream from DOM events. Here’s how to apply it:

const clicks = fromEvent(document, 'click');

After establishing the click event stream, we can start observing by subscribing:

let subscription = clicks.subscribe(() => console.log('click event occurred'));

This concise version solely provides a function to handle the next event. A more complete syntax looks like this:

let subscription = clicks.subscribe({
    next(val) {console.log('click event occurred')},
    error(err) {console.log('received an error: ' + err)},
    complete() {console.log('stream completed')},
});

Let's recap the lazy and cancellable aspects of Observables mentioned earlier.

They are lazy since the producer function won't run until a subscription is made. This differs from Promises, which initiate the constructor's function right away without waiting for then or other methods to be called.

They are cancellable because invoking unsubscribe on the object returned by subscribe triggers the observable's cleanup function. This cleanup can remove an event listener or, for an XHR, invoke the abort method, effectively cancelling the request.

subscription.unsubscribe();

For an extended period, Promises lacked a cancellation API, but today you can achieve similar control with AbortController.

Composability, the final key characteristic, is what we'll examine now. Consider two Observables: one dispensing click events and another making a network request via the fetch API:

const clicks = fromEvent(document, 'click');
const request = fromFetch('https://api.mocki.io/v1/b043df5a');

We can connect these with the pipe method, which chains them using operators:

clicks.pipe(operator).subscribe((data) => console.log(data));

Here, the operator is a custom function bridging the source observable and the observer:

function operator(source) {
    return new Observable(observer => {
        source.subscribe(() => {
            request.subscribe((response) => {
                response.json().then((data) => {
                    observer.next(data);
                });
            });
        });
    });
}

However, it's more typical to leverage the standard operators included in the RxJS library. The same functionality can be achieved with:

clicks.pipe(
    mergeMap(() => request.pipe(
        switchMap((response) => fromPromise(response.json())))
    )
).subscribe((data) => console.log(data));

A deeper dive into how operators function is available here.

Finally, here's a brief overview of other concepts you're likely to encounter when using RxJS:

  • Observer: A set of callbacks that dictates how to react to the values delivered by an Observable.
  • Subscription: An object representing an Observable's execution, primarily used for cancellation.
  • Subject: Functions as an EventEmitter, and is the unique way to multicast a value or event to multiple Observers.
  • Schedulers: These are centralized dispatchers that manage concurrency, allowing control over when computations occur, such as with setTimeout or requestAnimationFrame.