RxJS - Observables

An observable is a function that creates an observer and attaches it to the source where values are expected from, for example, cpcks, mouse events from a dom element or an Http request, etc.

Observer is an object with callback functions, that will get called when there is interaction to the Observable, i.e., the source has interacted for an example button cpck, Http request, etc.

We are going to discuss following topics in this chapter −

Create Observable

Subscribe Observable

Execute Observable

Create Observable

The observable can be created using observable constructor and also using observable create method and by passing subscribe function as an argument to it as shown below −

testrx.js

import { Observable } from  rxjs ;

var observable = new Observable(
   function subscribe(subscriber) {
      subscriber.next("My First Observable")
   }
);

We have created an observable and added a message “My First Observable” using subscriber.next method available inside Observable.

We can also create Observable using, Observable.create() method as shown below −

testrx.js

import { Observable } from  rxjs ;
var observer = Observable.create(
   function subscribe(subscriber) {
      subscriber.next("My First Observable")
   }
);

Subscribe Observable

You can subscribe to an observable as follows −

testrx.js

import { Observable } from  rxjs ;

var observer = new Observable(
   function subscribe(subscriber) {
      subscriber.next("My First Observable")
   }
);
observer.subscribe(x => console.log(x));

When the observer is subscribed, it will start the execution of the Observable.

This is what we see in the browser console −

Execute Observable

An observable gets executed when it is subscribed. An observer is an object with three methods that are notified,

next() − This method will send values pke a number, string, object etc.

complete() − This method will not send any value and indicates the observable as completed.

error() − This method will send the error if any.

Let us create the observable with all three notifications and execute the same.

testrx.js

import { Observable } from  rxjs ;
var observer = new Observable(
   function subscribe(subscriber) {
      try {
         subscriber.next("My First Observable");
         subscriber.next("Testing Observable");
         subscriber.complete();
      } catch(e){
         subscriber.error(e);
      }
   }
);
observer.subscribe(x => console.log(x), (e)=>console.log(e), 
   ()=>console.log("Observable is complete"));

In the above code, we have added, next, complete and error method.

try{
   subscriber.next("My First Observable");
   subscriber.next("Testing Observable");
   subscriber.complete();
} catch(e){
   subscriber.error(e);
}

To execute next, complete and error, we have to call the subscribe method as shown below −

observer.subscribe(x => console.log(x), (e)=>console.log(e), 
   ()=>console.log("Observable is complete"));

The error method will be invoked only if there is an error.

This is the output seen in the browser −