RxJS Tutorial for Beginners

This article provides a tutorial on RxJS with sample codes and examples.The tutorial provided here will be helpful to developers while using RxJS in building angular or node app. Here, we will mainly discuss the different APIs of RxJS which are useful in day to day development of single page application such as creating observable, subscribing to observable and about different Observable operators.

RxJS is one of the implementation in javascript of ReactiveX API.These API can be used with many others languages such as Java, C#, Python, PHP, ruby etc. It is a library for composing asynchronous and event-based programs by using observable sequences for reactive programming.Reactive programming is a programming paradigm that works with asynchronous data streams.It treats everything as a stream ordered in time and this stream can be generated by Http Requests, UI Events or array-like objects etc. To get started with RxJS, first you need to set up your environment. In my case, I have an angular 5 project generated from CLI and all the examples I will be writing in app.component.ts. You can follow this manual on rectivex.io for complete installation guide.

Observable Using From

Observable represents the idea of an invokable collection of future values or events such as a HTTP call or a button click in the UI.Let's create a simple Observable first using from() which also accepts a list.

import {Observable} from "rxjs";
	 let names = ['Ram', 'Tom', 'Hary'];
	 let  source = Observable.from(names);

This creates an Observable and now we want some observer to subscribe to this Observable. Observer is a collection of callbacks that knows how to listen to values delivered by the Observable.To create an Observer, we require 3 methods - one that will be invoked whenever there is a change in the value in Observable - next(), another when an error occurs - error()and third method will be invoked when the datasource is exhausted - complete(). Following is the implementation of the Observer class.

import {Observer} from "rxjs/Observer";

export class MyObserver implements Observer{
  next(val){
    console.log(val)
  }
  error(e){
    console.log(e)
  }
  complete(){
    console.log("complete");
  }
}

Now to subscribe to this Observable, we can write this - this.source.subscribe(new MyObserver()); and the each name will be printed in the console.The same observer can be defined in following way and we can expect the same result.

 this.source.subscribe(
      value => console.log(value),
      error => console.log(error),
      () => console.log("complete")
    );

Observable Using Create

create() method takes Observer as a parameter that defines the value that will be sent to subscriber from observable.

source = Observable.create(observer => {
    observer.next('Ram')
    observer.next('Tom')
    observer.next('Hary')
	observer.complete()
  });

This also produces the same result.This is a typical example of event based programming. As soon any event happens in the observable, the same will be published to the subscriber and the subscriber that is listening to that Observable will perform the action based on the event.

Observable from Click Events

This is a very obvious use case where we want to raise a click event from the DOM.Let's create an Observable from click event using fromEvent() and print the event in the console.

 let source = Observable.fromEvent(document, 'click');
    source.subscribe(data =>{
      console.log(data);
    })

Observable from Promise Object

The Promise object represents the eventual completion (or failure) of an asynchronous operation, and its resulting value.Below is the syntax to create a promise object.

let promise = new Promise(function(resolve, reject) {
  // function passed to new Promise is called the executor
});
new Promise(executor);

When the promise is created, this executor function runs automatically and once the executor finishes the job, it calls one of the functions that it gets as arguments i.e. resolve or reject and these functions are pre-defined by the JavaScript engine.

We can register consuming functions(subscribers) such as then on promise object for execution based on the state of the promise object - fullfilled or rejected.

promise.then(
  function(result) { 
  console.log("success");
  },
  function(error) {
  console.log("failed");
  });

This way of creating Obervable is important while dealing with promise object during HTTP call.

 let promise = new Promise((resolve, reject) => {
      setTimeout(() => {
        resolve('output')
      }, 1000)
    });
    let source = Observable.fromPromise(promise);
    source.subscribe(data => {
      console.log(data);
    })

When the subscriber subscribes to this observable, output will be printed with a delay of 1 sec.There are many other ways of creating Observable. For the complete list you can follow this.

Observable Use Case in Angular

Below is an example of invoking a HTTP service that returns an Observable.

 createCustomer(customer): Observable{
    return this.http.post(customerBaseUrl, customer);
  }
 

Now, below is an example of Observer for above Observable.

this.apiService.createCustomer(saveCustomerInfoPayload).subscribe(data => {
      if(data.status === 200) {
        this.dialogRef.close();
        this.showError('Customer info saved successfully', 'Notification');
      }else {
        this.showError(data.message, 'Error');
      }
    },
        error => {
        console.log(error)
    });

RxJS Operators

map() allows to perform some operations on the value before emitting to the observer. For example, we can multiply each value with any digit and then emit it to the subscriber.This use case is helpful when we want to modify the API response before processing.

let numbers = [3,9,7];
    let source = Observable.from(numbers).map(value => {
      return 2*value;
    })
    source.subscribe(value => {
      console.log(value);
    })

do operator is used to perform action of the values of the Observable.

 source = Observable.create(observer => {
    observer.next('Ram')
    observer.next('Tom')
    observer.next('Hary')
    observer.complete()
  });
   this.source.do(value => console.log(value))
      .map(value => value.toUpperCase())
      .do(value => console.log(value))
      .subscribe(
      value => console.log(value),
      error => console.log(error),
      () => console.log("complete")
    );

filter() filters out the data that meets given criteria. For example, the following Observable will only emit string that starts with 'H'.

 names = ['Ram', 'Tom', 'Hary', 'Hem'];
 Observable.from(this.names).filter(name => name.startsWith('H'))
      .subscribe(name => {
        console.log(name)
      })

first() operator only emits the first element from the Observable and similarly we have last() operator emits the last element.

Observable.from(this.names).first()
      .subscribe(name => {
        console.log(name)
      })

retry() operator retrun the Observable as many times we want.For example, when any events fail, we can use retry to re-run it.Following code wil run for 3 times.

 source = Observable.create(observer => {
    observer.next('Ram')
    observer.next('Tom')
    observer.next('Hary')
    throw 'exception'
  });
  
   this.source.catch(err => console.log('error ocurred!' + err))
      .retry(2)
      .subscribe(data => console.log(data))

Conclusion

This RxJS tutorial deals with the basics of RxJS to get started with projects using angular JS and node JS.If you have anything that you want to add or share then please share it below in the comment section