I mentioned the core components of RxJS in the introduction of RxJS page. As you know, the components are Observables, Observers, Operators, Subjects and Schedulers. These core components are nothing but the Building Blocks of RxjS. In this tutorial, you will learn how they work together to provide a powerful set of functionalities. Once you have a high-level understanding of them, you will then learn each one of these building blocks in detail in the latter part of this tutorial. So, let us look at them one by one.

I will use the code shown below to take you through RxJS journey, you can also run it on stackblitz.com to see the output.

import { from } from "rxjs";

let nums = [1, 2, 4, 34, 56, 789];

let numsObservable$ = from(nums); //Observable

let observer = {
  next: num => console.log(num),
  error: err => console.error(err),
  complete: () => console.log("Execution completed")
};

numsObservable$.subscribe(observer); //Observer subscribed
  • On line-1, the necessary functions are imported from the rxjs library.
  • Next, on line-3, nums is an array of numbers which is then converted to an Observable.
  • On line-5, I created an rxjs observable, named numObservable$, with the help of from which converts an array to an observable. Note appending $ at the end of Observables is a naming convention.
  • Nothing gets executed or invoked when we just create an observable. To perform any action, you need to subscribe to the Observable.
  • So, on line-13, numsObservable$ is subscribed with an object named observer. Note, this observer has 3 properties, next, error and complete.
  • Once the execution completes, you can see the “Execution completed” message in the console. Please make sure to look into the console in stackblitz.com.

What is an Observable?

An Observable is a stream of events or a stream of data that can arrive over time. Using RxJS functions, you can create an Observable from almost any source (events, web services, socket data and etc). In other languages, an Observable is a collection of future data.

In the above example, I used from operator to create Observable from an array. Similarly, there is also fromEvent operator to create from JavaScript events.

Observables are Lazy or Cold. They don’t activate any producer unless subscribe method is invoked.

Below code show how fromEvent is used to create an Observable from a click event.

import { fromEvent } from "rxjs";

let clicksObservable$ = fromEvent(document, "click");

What is an Observer?

Observable represents the data source that can be Observed and whoever is interested can subscribe to Observables. An observer is the code who registers an interest in the data stream through subscription.

Observer provides support for iterations over an Observable sequence.

Below is the source code of Observer from Rxjs GitHub repo.

import { Observer } from './types';
import { config } from './config';
import { hostReportError } from './util/hostReportError';

export const empty: Observer<any> = {
  closed: true,
  next(value: any): void { /* noop */},
  error(err: any): void {
    if (config.useDeprecatedSynchronousErrorHandling) {
      throw err;
    } else {
      hostReportError(err);
    }
  },
  complete(): void { /*noop*/ }
};

So basically, an observer has 3 methods, next, error and complete. It is important to understand each one of them.

let observer = {
  next: num => console.log(num),
  error: err => console.error(err),
  complete: () => console.log("Execution completed")
};

next – Accepts one param (value). This is the data that is being pushed/produced by the Observable. So every time observable pushes data next (value) will receive the value.

error – Accepts one param as an error. This method gets executed if there is an error thrown from the Observer.

complete – Does not accept any param. Gets executed once the Observable completed its operation/execution.

What is an Operator?

Operators are functions that allow to manipulate the values produced by Observables. They are another reason why everyone like to use RxJS. I will use the code example below to explain to you the concepts of Operators.

Code Example – Run on Stackblitz.com, make sure you open the console.

import { from } from "rxjs";
import { map } from "rxjs/operators";

const nObservable$ = from([1, 2, 3, 5, 56, 567, 89]);
nObservable$
.pipe(
  map(val => val * 2)
)
.subscribe(
  value => console.log(value)
);

The above code doubles the number stored in an array. Important is to note the use of map function, you can relate it to Array.prototype.map(). As you noticed, map operator is put inside pipe method.

Let us look into another example that captures the mouse click and logs the x-coordinate. I will make use of the pluck operator to do this. This example is also included in the above Stackblitz.com code snippet.

import { fromEvent } from "rxjs";
import { pluck } from "rxjs/operators";

const clicksObservable$ = fromEvent(document, "click");
clicksObservable$
  .pipe(pluck("clientX"))
  .subscribe({
    next: clientX => console.log(clientX),
    error: err => console.error(err),
    complete: () => console.log("Execution completed")
  });

There are a bunch of Operators present in rxjs/operators. You will learn the commonly used ones in the latter part of the tutorial.

Subject in RxJS

Subject pushes data to multiple Observers, wherein Observable does it to a single observer at a time. Subject allows you to emit new values, unlike passively waiting for data in case of Observables.

Subjects are used in multicasting values, wherein Observables are used for unicasting.

If you are new to RxJS, take a break to grab a cup of Tea, Coffee or whatever you like and then look into the code example below to understand Subjects at a high level.

If you check the source code of Subject on GitHub, you can see it allows us to invoke .next, .error, .complete() etc on the subject. But this is not possible on the Observables. Let us look at a simple example below, run it on Stackblitz to see the output.

Let us look at the example below, I have only shown you invocation of next(), you can also invoke error(err) and complete() as and when needed.

import { Subject } from "rxjs";

const mySubject$ = new Subject<number>();

mySubject$.subscribe({
  next : (num) => console.log("ObserverA: " + num)
});
mySubject$.subscribe({
  next : (num) => console.log("ObserverB: " + num)
});

mySubject$.next(5);
mySubject$.next(101);

Output

ObserverA: 5
ObserverB: 5
ObserverA: 101
ObserverB: 101

As shown above, you can use next(value) to emit the next value to multiple Observers. When the emitting of the values is completed, you can invoke complete() as below. 

mySubject$.complete();

Schedulers in RxJS

Schedulers are another important part of the rxjs building blocks. A scheduler controls when a subscription starts and when notifications are delivered. There are several schedulers out of the box like queueScheduler, asyncScheduler, asapScheduler and etc.

In a simple project, you may not even need to touch this part, however, it is useful when you need precise control on the context. I am using a simple code taken from the official site to demonstrate to you the working of schedulers.

Run the code on Stackblitz.com to check the output.

import { Observable, asyncScheduler } from 'rxjs';
import { observeOn } from 'rxjs/operators';

const observable = new Observable((subscriber) => {
  subscriber.next(1);
  subscriber.next(2);
  subscriber.next(3);
  subscriber.complete();
}).pipe(
  observeOn(asyncScheduler)
);

console.log('just before subscribe');
observable.subscribe({
  next(x) {
    console.log('got value ' + x)
  },
  error(err) {
    console.error('something wrong occurred: ' + err);
  },
  complete() {
     console.log('done');
  }
});
console.log('just after subscribe');

Output

just before subscribe
just after subscribe
got value 1
got value 2
got value 3
done

Learn RxJS building blocks

By |Last Updated: April 1st, 2024|Categories: RxJS|

Table of Contents