Angular rxjs tutorial to create observables

In this tutorial, we are going to learn about what is an observable and how to create your own observables using rxjs.

Angular uses the observables for handling asynchronous tasks like http requests and EventEmitters. In the angular router, observables are used to update the route params data.

Rxjs

RxJS is a JavaScript library for the reactive programming using Observables, which makes easier to work with asynchronous or callback-based code.

Rxjs offers us a different type of functions to create a new observable.

What are Observables?

• Observables are the functions which can return a single value or even multiple values in sequence and they can handle both synchronous and asynchronous tasks.

• Observables can only run when you subscribe to the data.

Creating our own observable

Let’s learn it by creating our own observable using rxjs.

This tutorial assumes that you already created a new angular project using cli.

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 {

  ngOnInit() {
   const observable1 = new Observable(subscriber => {      subscriber.next(1);      subscriber.next(2);      subscriber.next(3);     });   });}

In the above code, we first imported Observable constructor from the rxjs package and we are passing a function as its first argument.

By using subscriber.next method we are passing a three values 1,2,3 in sequence.

Invoking observable

Creating an observable won’t do anything we need to invoke the observable by using subscribe method provided by the Observable constructor.

The subscribe method takes the observer object as its argument.

Observer: is a collection of callbacks that knows how to listen to values delivered by the Observable.

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 {


  ngOnInit() {
    const observable1 = new Observable(subscriber => {
      subscriber.next(1);
      subscriber.next(2);
      subscriber.next(3);
    });

    observable1.subscribe({      next: (val) => console.log(val)    });  }
}

Now, run your project and open your browser you will see three log values are displayed which are 1,2,3 which comes from our observable.

An observable can also have an error and complete method which helps in working with asynchronous tasks like http requests because in such cases data is not available immediately so that our observer reacts to the future data available in observable.

In some cases, we can also get an error and observer reacts to that error also.

error method: It runs when an error occurs

complete method: It runs when a given task is completed.

Let’s update our observable by adding these two methods.

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 {

  ngOnInit() {
    const observable1 = new Observable(subscriber => {
      subscriber.next(1);
      subscriber.next(2);
      subscriber.error(setTimeout(() => {
        return 'this is err msg';
      }, 3000));

      // once we call complete the below code can't execute
      subscriber.complete();
      subscriber.next(3); // this line is unreachable    });

    observable1.subscribe({
      next: (val) => console.log(val),
      error: (errormsg) => console.log(errormsg),      complete: () => console.log('task is completed')    });
  }
}

In the above code, we have added error method by passing a setTimeout method so that we can see an error msg in our console after 3s.

Once we call complete method in Observable the below code not reachable like in JavaScript functions code is not reachable after return keyword.

Builtin observables

Rxjs can give us some built-in observables let’s see interval observable.

import { Component, OnInit } from '@angular/core';
import { Observable, interval } from 'rxjs';

@Component({
  selector: 'app-root',
  templateUrl: './app.component.html',
  styleUrls: ['./app.component.css']
})
export class AppComponent implements OnInit {

  ngOnInit() {
    const observable = interval(1000);
    observable.subscribe({
      next: (val) => console.log(val)
    });
  }
}

Now this interval observable will push new value to the subscriber on every 1s.

Unsubscribe method

By default, subscriptions are enabled by calling subscribe method but our subscriptions are active until we call unsubscribe method which is provided by the subscribe method.

Example: In youtube, if we click on bell icon notifications are enabled and we start receiving any new updates related to that channel but in future, we don’t need subscription we need to unsubscribe it.

import { Component, OnInit } from '@angular/core';
import { Observable, interval } from 'rxjs';

@Component({
  selector: 'app-root',
  templateUrl: './app.component.html',
  styleUrls: ['./app.component.css']
})
export class AppComponent implements OnInit {

  ngOnInit() {
    const observable = interval(1000);

    const subscription = observable.subscribe({
      next: (val) => console.log(val)
    });

    // by calling this method we are cancelling our subscription
    subscription.unsubscribe();  }
}

In the above example, we are unsubscribing the interval observable subscription by calling unsubscribe method.