RxJS Observables

RxJS stands for Reactive Extension for JavaScript. As the name implied RxJS is used to implement reactive programming in JavaScript. Reactive programming deals with implementing asynchronous streams. RxJS provides an implementation of Observable type which is used in Angular to implement a producer – subscriber pattern.

An observable is a function that produces a stream of values and emits it to its subscribers over time. When you subscribe to an observable, you are an observer. Observer receive the emitted values.

Observables can emit 3 types of notifications

next – To emit the next value in the list
complete – When the observable is done with emitting all the values
error – When an error occurred in the observable.

We can understand how the observables and observer work with a simple example as follows.

The following code creates an observable which emits values “item-1”, “item=2” upto “item-n” when it is done with emitting all the value it calls the complete() to indicate that all the values have been emitted. If any exception occurs then the error() is called to indicate that the process errored out.

var observable = Observable.create((observer:any) => {

try{

    observer.next('Item-1')

    observer.next('Item-2')
    …
    observer.next('Item-n')

    observer.complete()   

}
          catch(ex) {

    observer.error(ex)

}

})

Now we have the observable ready, we need to subscribe to it to receive the values emitted by the observable, this is done as follows by subscribing to the observable.

observable.subscribe(

    (x:any) => console.log(x),

    (error:any) => console.log(error)

);

Here the observer subscribes to the observable and prints the values received from the observable to the console, if an error is emitted then it displays the error in the console.