Reactive programming with RxJS

Introduction to RxJS

RxJS (Reactive Extensions for JavaScript) is a library for reactive programming using observables that makes it easier to compose asynchronous or callback-based code. In this post, I’ll be focusing on Observer and Observables, which are used in Angular extensively.

Let’s start off with some basic concepts.

Observer pattern

The observer pattern is a software design pattern in which an object (the Observable) maintains a list of its dependents (the Observers), and notifies them automatically of state changes.

Pushing and Pulling

Pull and Push are two different protocols that describe how a data Producer and a data Consumer communicate.

In Pull systems, the Consumer determines when it receives data from the data Producer. The Producer is unaware of when the data will be delivered to the Consumer.

In Push systems, the Producer determines when to send data to the Consumer. The Consumer is unaware of when it will receive that data.

Observable

Observables are Producers that push values lazily.

const observable = new Observable(subscriber => {
  subscriber.next(1);
  subscriber.next(2);
  setTimeout(() => {
    subscriber.next(3);
    subscriber.complete();
  }, 1000);
});

Observer

An Observer consumes the values delivered by an Observable.

Observers are just objects with three callbacks, one for each type of notification that an Observable may deliver.

const observer = {
  next: x => console.log('Observer got a next value: ' + x),
  error: err => console.error('Observer got an error: ' + err),
  complete: () => console.log('Observer got a complete notification'),
};

To receive these values, we need to subscribe to the Observable.

observable.subscribe(observer);

Operators

1. Creator operators

A creator operator creates a new Observable. For example, of(1, 2, 3) is a creator operator which creates an observable. This observable will emit 1, 2, and 3 successively.

2. Pipeable operators

A Pipeable Operator is a function that takes an Observable as its input and returns another Observable.

For example, map(x => x * 2) is a Pipeable Operator.

map(x => x * 2)(observable).subscribe((v) => console.log(`value: ${v}`));

Having multiple pipeable operators can become unreadable. The .pipe method applies all the operators in the same way, but with a cleaner syntax.

observer.pipe(
  operator1(),
  operator2(),
  operator3(),
)

Subject

A Subject implements both the Observer and Observable interface. It is a special type of Observable that allows values to be multicasted to many Observers.

A Subject is an Observable. You can subscribe to a Subject by providing an Observer, and the Observer will receive values.

A Subject is an Observer. It is an object with the methods next(v), error(e), and complete(). To feed a new value to the Subject, just call next(theValue), and it will be multicasted to the Observers registered to listen to the Subject.

Usage in Angular

To find out more about the usage of Observables in Angular, I recommend the following article in Angular’s documentation.