Understanding RxJS | Third Rock Techkno

Table of Contents

What is RxJS?

RxJS stands for Reactive Extensions for JavaScript. It is used to manage data state in an application and contains various techniques and methods which are useful for a smooth flow of information in the application.

RxJS comes in-built with Angular and is most useful for managing http calls and handling request subscription. It is also internally used in Angular for some of its core logic like Event Emitter. RxJS can also be implemented in other libraries and frameworks such as React, Vue, etc.

The important concepts in RxJS includes,

Observable
Observer
Subscription
Operators
Subject
Schedulers

1. Observables

Observables are part of an app which defines how the data is flowed and managed in the app. It is a technique that passes data from various states of an application. You need to create an observable to initialise it, but it cannot be rendered until it is subscribed.

// Create simple observable that emits three values
const myObservable = of(1, 2, 3);

Note: of() is an RxJS creation operator which creates a new observable, among many others.

The angular HTTP requests are of type observable, they won’t render unless subscribed from wherever they are called.

2. Observer

An observer is used to observe the observable. It contains three types of callbacks that an observable can use. It is simply a set of callbacks.

next - This is required. It can be called multiple times when any value is changed so as to let the subscriber know to subscribe. Every time next is called, a new value is subscribed.
error - Optional. Emits when any error occurs
complete - Optional. Emits when any the execution is completed

// Create observer object for above observable
const myObserver = {
  next: (x: number) => console.log(x),
  error: (err: Error) => console.error(err),
  complete: () => console.log('Completed'),
};

If the error or the complete callback is ignored, the normal execution won’t stop, rather only the callback which isn’t provided won't run. For example, if error callback is ignored, and if any error occurs in the observer, that error won’t be displayed in the observer context, as we did not specify its region.

3. Subscriptions

As explained above, an observable only starts to publish value when someone subscribes to it. Subscription can be understood as a monthly newsletter, whoever has taken the monthly subscription will receive it, as well as terminate it when no longer subscribed. A subscription is like an executable for observable.

.subscribe() method is used for subscription to an observable.

It is not necessary to use an observer for subscribing, as the subscription can already use the next, error and complete types in it, to do the job for us.

// Subscribing to the above observable as...
const subscriber =  myObservable.subscribe(myObserver);
subscriber.unsubscribe();

// 1
// 2
// 3
// Completed

The subscribe method will run every time a .next() is called.

Be careful to always unsubscribe to the observable when not needed, to avoid memory leak, else it could affect the performance and might as well mess data values.

.unsubscribe() method is used to stop the observable subscription. It is available in the subscription class like described in the above example.

In Angular, you can unsubscribe in the ngOnDestroy() lifecycle hook of the component, or use the async pipe which unsubscribes itself when the component is destroyed.

4. Operators

The observable are the core concept of RxJS, but the operators are the one which defines RxJS. RxJS is mostly used with the variable operators it provides out of the box.

Most of the operators are Array-like functions like map, filter, etc. They can be separated in two types, pipeable and creation operator

Pipe operator
These are the types which are used with the .pipe method in an observable. For instance…

theObservable.pipe(map(), filter());

Creation operator
These are the types which are used to initialise the observable

import { of, map, take } from 'rxjs';

of(1, 2, 3, 4, 5)
  .pipe(map(x) => x * x)
  .subscribe((v) => console.log({ value: v }))

// value: 1
// value: 4
// value: 9
// value: 16
// value: 25

of(1, 2, 3, 4, 5)
  .pipe(take(2))
  .subscribe((v) => console.log({ value: v }))

// value: 1
// value: 4

In the first example, of is the creation operator which is used to assign value to an observable, and map is the pipeable operator, which multiplies the values by itself, before output is shown in subscribe

In the second example, another operator take is used, which displays the number of values we want. Here only the first two values will be printed. first operator can be used if we want only the first value.

5. Subject

As per the definition on rxjs, a subject is a special type of Observable that allows values to be multicasted to many Observers. While plain Observables are unicast, subjects are multicast.

import { Subject } from 'rxjs';

const subject = new Subject<number>();

subject.subscribe({
  next: (v) => console.log(`observerA: ${v}`),
});
subject.subscribe({
  next: (v) => console.log(`observerB: ${v}`),
});

subject.next(1);
subject.next(2);

// Logs:
// observerA: 1
// observerB: 1
// observerA: 2
// observerB: 2

The .next tells the subscriber to subscribe, a value is inserted using the .next method.

Various types of subject includes,

BehaviourSubject: Initialised with a default value,
ReplaySubject: Can emit old values to new subscribers,
AsyncSubject: Emit the last value when an observer completed

6. Scheduler

A scheduler controls the execution of a subscription when it starts and when its notifications are delivered.

Resources

https://rxjs.dev/guide/overview
https://angular.io/guide/observables