Understanding RxJS Observables

Let’s look into Reactive Programming with RxJS

Photo by Jeffrey Workman on Unsplash

In this blog post I am going to give you a basic understanding of RxJS observables and how to use them in Angular.

First, let me mitigate the confusion by simplifying the terms in the topic of the article.

What’s Reactive Programming?

Reactive programming means programming with asynchronous data streams. A stream is a sequence of data ordered in time. Data streams can be created with anything like variables, changes on a variable, properties, user inputs, data structures, click events, HTTP requests, etc. So with reactive programming, you can observe data streams and when you observe that a value is emitted you can react accordingly.

A little bit more about data streams.

A stream does not emit only the values as stated above. During its timeline a stream can also emit an error when some unexpected event occurred causing the stream to end, or a complete signal as the stream may complete when the user closes the window. So we can catch these streams to execute functions according to them.

What’s RxJS?

RxJS is a JavaScript library that allows us to work with reactive programming. RxJS uses observables to compose asynchronous code and provides functions to work with observables.

The Big Topic: Observables

What’s an observable?

An observable is a function that acts as a wrapper for a data stream. They support to pass messages inside your application. An observable is useless until an observer subscribes to it. An observer is an object which consumes the data emitted by the observable. An observer keeps receiving data values from the observable until the observable is completed, or the observer unsubscribes from the observable. Otherwise observers can receive data values from the observable continuously and asynchronously. So we can perform various operations such as updating the user interface, or passing the JSON response.

There are 4 stages for a life cycle of an observable.

1. Creation
2. Subscription
3. Execution
4. Destruction

Let’s understand these 4 stages with Angular.

You need to import the RxJS library to create an observable.

import { Component } from '@angular/core';
import { Observable } from 'rxjs';
@Component({
   selector: 'app-root',
   templateUrl: './app.component.html',
   styleUrls: ['./app.component.scss']
})
export class AppComponent {
   title = 'observable';
ngOnInit() {
   var observable = Observable.create()
}
}

This is how you can create an observable. But observables are useless until you subscribe to them. The below code shows how you can subscribe to an observable.

import { Component } from '@angular/core';
import { Observable } from 'rxjs';
@Component({
selector: 'app-root',
templateUrl: './app.component.html',
styleUrls: ['./app.component.scss']
})
export class AppComponent {
   title = 'observable';
ngOnInit() {
   var observable = Observable.create((observer: any) => {
      observer.next('Hello world!')
   })
observable.subscribe(function showMessage(msg: any) {
      console.log(msg)
   })
}
}

When you check the console, observer.next() value can be seen there.

Now let’s try executing observables.

import { Component } from '@angular/core';
import { Observable } from 'rxjs';
@Component({
   selector: 'app-root',
   templateUrl: './app.component.html',
   styleUrls: ['./app.component.scss']
})
export class AppComponent {
   title = 'observable';
ngOnInit() {
   var observable = Observable.create((observer: any) => {
      observer.next('A')
      observer.next('B')
      observer.error('Error occured.')
      observer.complete('Data stream is over.')
      observer.next('C')
   })
observable.subscribe(function showMessage(msg: any) {
      console.log(msg)
   })
}
}

If an error or completion occurred it will automatically stop the execution. You can only see ‘A’ and ‘B’ printed in the console.

Let’s try making this asynchronous with the help of the setInterval() method.

import { Component } from '@angular/core';
import { Observable } from 'rxjs';
@Component({
   selector: 'app-root',
   templateUrl: './app.component.html',
   styleUrls: ['./app.component.scss']
})
export class AppComponent {
   title = 'observable';
ngOnInit() {
   var observable = Observable.create((observer: any) => {
      observer.next('A')
      observer.next('B')
setInterval(() => {
         observer.next("This is a random message.")
      }, 2000)
observer.next('C')
      observer.next('D')
setInterval(() => {
         observer.error("Error occured.")
      }, 4001)
observer.next('E')
   })
observable.subscribe(function showMessage(msg: any) {
      console.log(msg)
   })
}
}

Your console will look like below. Here we the display of the values is done asynchronously.

Destroying an observable means removing it from the DOM by unsubscribing to it. Normally RxJS takes care of destroying an observable by unsubscribing to it after an error or a completion occurred.

Importance of Observables

Observables are lazy. It means that if it is not subscribed then it will not emit data. And also, handling asynchronous data emitting that has multiple data values is easier with observables.

So this is a basic introduction to RxJS observables with Angular and hope it helped you to understand what observables are.

Happy Coding!