Understanding RxJS Observables and why you need them

What is RxJS?

RxJS is a framework for reactive programming that makes use of Observables, making it really easy to write asynchronous code. According to the official documentation, this project is a kind of reactive extension to JavaScript with better performance, better modularity, better debuggable call stacks, while staying mostly backwards compatible, with some breaking changes that reduce the API surface. It is the official library used by Angular to handle reactivity, converting pull operations for call-backs into Observables.

Prerequisites

To be able to follow through in this article’s demonstration you should have:

• Node version 11.0 installed on your machine.
• Node Package Manager version 6.7 (usually ships with Node installation).
• Angular CLI version 7.0
• The latest version of Angular (version 7)

// run the command in a terminal
ng version

Confirm that you are using version 7, and update to 7 if you are not.

• Download this tutorial’s starter project here to follow through the demonstrations
• Unzip the project and initialize the node modules in your terminal with this command

npm install

Other things that will be nice to have are:

• Working knowledge of the Angular framework at a beginner level

Understanding Observables: pull vs push

To understand Observables, you have to first understand the pull and push context. In JavaScript, there are two systems of communication called push and pull.

A pull system is basically a function. A function is usually first defined (a process called production) and then somewhere along the line called (this process is called consumption)to return the data or value in the function. For functions, the producer (which is the definition) does not have any idea of when the data is going to be consumed, so the function call literally pulls the return value or data from the producer.

A push system, on the other hand, control rests on the producer, the consumer does not know exactly when the data will get passed to it. A common example is promises in JavaScript, promises (producers) push already resolved value to call-backs (consumers). Another example is RxJS Observables, Observables produces multiple values called a stream (unlike promises that return one value) and pushes them to observers which serve as consumers.

What is a Stream?

A stream is basically a sequence of data values over time, this can range from a simple increment of numbers printed in 6 seconds (0,1,2,3,4,5) or coordinates printed over time, and even the data value of inputs in a form or chat texts passed through web sockets or API responses. These all represent data values that will be collected over time, hence the name stream.

What are Observables?

Streams are important to understand because they are facilitated by RxJS Observables. An Observable is basically a function that can return a stream of values to an observer over time, this can either be synchronously or asynchronously. The data values returned can go from zero to an infinite range of values.

Observers and subscriptions

For Observables to work there needs to be observers and subscriptions. Observables are data source wrappers and then the observer executes some instructions when there is a new value or a change in data values. The Observable is connected to the observer who does the execution through subscription, with a subscribe method the observer connects to the observable to execute a code block.

Observable lifecycle

With some help from observers and subscriptions the Observable instance passes through these four stages throughout its lifetime:

• Creation
• Subscription
• Execution
• Destruction

Creating Observables

If you followed this post from the start, you must have opened the Angular starter project in VS Code. To create an Observable, you have to first import Observable from RxJS in the .ts file of the component you want to create it in. The creation syntax looks something like this:

import { Observable } from "rxjs";

var observable = Observable.create((observer:any) => {
    observer.next('Hello World!')
})

Open your app.component.ts file and copy the code block below into it:

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

Subscribing to Observables

To tell RxJS to execute the code block on the Observable, or in a simpler term, to call the Observable to begin execution you have to use the subscribe method like this:

export class AppComponent implements OnInit{
  title = 'ngcanvas';
  ngOnInit(): void {
    var observable = Observable.create((observer:any) => {
      observer.next('Hello World!')
  })
  observable.subscribe(function logMessage(message:any) {
    console.log(message);
  })
}

This subscribe method will cause “hello world” to be logged in the console.

Executing Observables

The observer is in charge of executing instructions in the Observable, so each observer that subscribes can deliver three values to the Observable:

1. Next value: With the next value, observer sends a value that can be a number, a string or an object. There can be more than one next notifications set on a particular Observable
2. Error value: With the error value, the observer sends a JavaScript exception. If an error is found in the Observable, nothing else can be delivered to the Observable
3. Complete value: With the complete value, the observer sends no value. This usually signals that the subscriptions for that particular Observable is complete. If the complete value is sent, nothing else can be delivered to the Observable.

This can be illustrated with the code block below:

export class AppComponent implements OnInit{
  title = 'ngcanvas';
  ngOnInit(): void {
    var observable = Observable.create((observer:any) => {
      observer.next('I am number 1')
      observer.next('I am number 2')
      observer.error('I am number 3')
      observer.complete('I am number 4')
      observer.next('I am number 5')
  })
  observable.subscribe(function logMessage(message:any) {
    console.log(message);
  })
}
}

If you run the application at this point in the dev server with

ng serve

When you open up the console in the developer tools your log will look like this:

You will notice that either the error value or complete value automatically stops execution and so the number 5 never shows up in the console. This is a simple synchronous exercise. To make it asynchronous, let us wrap timers around some of the values.

export class AppComponent implements OnInit{
  title = 'ngcanvas';
  ngOnInit(): void {
    var observable = Observable.create((observer:any) => {
      observer.next('I am number 1')
      observer.next('I am number 2')
      setInterval(() => {
        observer.next('Random Async log message')
    }, 2000)
    observer.next('I am number 3')
    observer.next('I am number 4')
      setInterval(() => {
        observer.error('This is the end')
    }, 6001)
    observer.next('I am number 5')
  })
  observable.subscribe(function logMessage(message:any) {
    console.log(message);
  })
}
}

This will appear like this in your browser console:

Notice that the display of value was done here asynchronously, with the help of the setInterval module.

Destroying an Observable

To destroy an Observable is to essentially remove it from the DOM by unsubscribing to it. Normally for asynchronous logic, RxJS takes care of unsubscribing and immediately after an error or a complete notification your observable gets unsubscribed. For the knowledge, you can manually trigger unsubscribe with something like this:

return function unsubscribe() {
    clearInterval(observable);
  };

Why Observables are so vital

• Emitting multiple values asynchronously is very easily handled with Observables
• Error handlers can also easily be done inside Observables rather than a construct like promises
• Observables are considered lazy, so in case of no subscription there will be no emission of data values
• Observables can be resolved multiple times as opposed to functions or even promises

Conclusion

We have been given a thorough introduction to Observables, observers and subscriptions in RxJS. We have also been shown the lifecycle process of Observables with practical illustrations. More RxJS posts can be found on the blog, happy hacking!

Plug: LogRocket, a DVR for web apps

LogRocket is a frontend application monitoring solution that lets you replay problems as if they happened in your own browser. Instead of guessing why errors happen, or asking users for screenshots and log dumps, LogRocket lets you replay the session to quickly understand what went wrong. It works perfectly with any app, regardless of framework, and has plugins to log additional context from Redux, Vuex, and @ngrx/store.

In addition to logging Redux actions and state, LogRocket records console logs, JavaScript errors, stacktraces, network requests/responses with headers + bodies, browser metadata, and custom logs. It also instruments the DOM to record the HTML and CSS on the page, recreating pixel-perfect videos of even the most complex single-page apps.

Try it for free.