Getting Started with RxJS

A Javascript or a Node based application invariably deals with lots of asynchronous data. Handling them without a proper library, can easily add a lot of complexities.

Access Anything as a Data Stream

As precisely mentioned in the excerpt above, RxJS library has been designed with the simplicity of an Array in mind. The library allows us to wrap both the synchronous data as well as the complex sources of asynchronous data the same way; so that consumer can access them as data streams from collections.

In fact, the power and huge popularity of this library across applications and frameworks is centered around the two core aspects :

1. Observables : Allows to access anything as simple collections

2. Operators : Provide declarative syntax to address our unique requirements

1. Observables

• Be it a synchronous data coming from a local array or a local variable
• Be it random events generated from button clicks, or
• Or be it some asynchronous data from a http or a series of data from a websocket end point

An observable allows us to wrap any data or source of data, and serve it as a data stream when a consumer subscribes the observable.

Access anything as simple collections

As we will see under Create Operators, we already have many operators like - of, from, range, ajax etc, which can create an Observable wrapping around our data or data point. Even we can easily create a new Observable from scratch for our custom data.

Let's look at some observable snippets, sourcing data from various types of data sources.

import { Observable } from 'rxjs';

//Will emit 10, 20 immediately and 50 after 2 sec
const observable = new Observable((subscriber) => {
  subscriber.next(10);
  subscriber.next(20);
  setTimeout(() => {
    subscriber.next(50);
    subscriber.complete();
  }, 2000);
});

//Will get the custom data on subscription
observable.subscribe((v) => console.log('Got value :' + v ) );

import {ajax} from 'rxjs/ajax';

let echoService = ajax("/api/echo?message=hi");

//Will get the data from the given endpoint on subscription
echoService.subscribe(v => {console.log("Result from echoService :"+ v);})

import {from} from 'rxjs';

const arrayData = from([20,30,50,80]);

//Will get the data from the given array on subscription
arrayData.subscribe(v => {console.log(v);})

import {fromEvent} from 'rxjs';

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

//Will keep collecting any click events as it occours
clicks.subscribe(v => console.log('click...!'));

As we can see, with the observable in hand accessing data from an url, from click events, from a simple array or some custom data source, it's very similar!

• We need not have to bother about how does it access it's data or capture the error internally.
  • All variations like - callack, promise-then-catch, async-await wraped and exposed as observables.
• Simply, subscribe to the observable, it will fetch us the data or a probable error.
  • Now, we can simply focus on our incoming data or the probable error.

A lot more simplified task for us, is not it ?

Now let's move on to the other more interesting and intimidating part of the RxJs library.

2. Operators:

The number of operators RxJs provides is quite intimidating to look at, at it's first sight. But, once you understand their pourpose you would fall in love with them for sure!

It's because each one of them provide you with declarative syntax to address your unique requirements in your application development. These operators can help us avoid lot of custom implementations and save us lot of development and maintenance time.

The functional group summary categories them on a high level. We have discussed them in more details in their respective group and demo pages. Let's take up few use cases to understand their importance.

Declarative Syntax to Address our Unique Requirements

Let's say we want to run a set of 5 tasks - [T1, T2,...., T5]

• concatMap : Run all tasks in Sequence, one after the other

 of(T1,T2,T3,T4,T5).pipe(
   concatMap(task => executeTask(task)),
 ).subscribe()

• mergeMap : Run all in Parallel, for faster execution

 of(T1,T2,T3,T4,T5).pipe(
   mergeMap(task => executeTask(task)),
 ).subscribe()

• exhaustMap : Run the earliest request and discard any concurrent duplicate requests

 of(T1,T2,T3,T4,T5).pipe(
   exhaustMap(task => executeTask(task)),
 ).subscribe()

• switchMap : Run the latest request and cancel any older, obsolate ones

 of(T1,T2,T3,T4,T5).pipe(
   switchMap (task => executeTask(task)),
 ).subscribe()

• forkJoin : Join multiple tasks and combine their end results

 forkJoin(executeTask(T1),
          executeTask(T2),
          executeTask(T3),
 ).subscribe();

 //Output : [T1-Output, T2-Output, T3-Output,]

These are just few examples showing the kind of choice these RxJS operators provide. We can implement multiple variations our custom requirements using a wide range of declarative syntaxes.

Custom implementations, all with minimal effort and roboust, reliable operators!