An Introduction to Observables in RxJS and Angular

Reactive programming has become an essential paradigm in modern web development, and RxJS (Reactive Extensions for JavaScript) is at the heart of it. Angular, one of the leading frameworks for building web applications, leverages RxJS for handling asynchronous events. This tutorial will walk you through the basics of Observables in RxJS and how they are used in Angular.

What is RxJS?

RxJS is a library for composing asynchronous and event-based programs by using observable sequences. It brings the power of reactive programming to JavaScript, enabling you to work with streams of data in a declarative manner.

What are Observables?

At the core of RxJS is the Observable. An Observable is a representation of any set of values over any amount of time. These values can be of any type: numbers, strings, objects, or even other Observables. The key idea is that Observables can emit values asynchronously.

Creating Observables

You can create an Observable in several ways. Here are a few common methods:

Using the Observable constructor:

JavaScript

import { Observable } from 'rxjs';

const observable = new Observable(subscriber => {
  subscriber.next('Hello');
  subscriber.next('World');
  subscriber.complete();
});

2. Using creation operators like of, from, and interval:

JavaScript

import { of, from, interval } from 'rxjs';

const observable1 = of(1, 2, 3);
const observable2 = from([10, 20, 30]);
const observable3 = interval(1000); // emits values every second

Subscribing to Observables

To start receiving values from an Observable, you need to subscribe to it. Subscribing involves providing callback functions to handle the emitted values, errors, and the completion of the Observable.

JavaScript

const subscription = observable.subscribe(
  value => console.log(value),  // Next handler
  error => console.error(error), // Error handler
  () => console.log('Complete')  // Completion handler
);

Unsubscribing from Observables

It is crucial to unsubscribe from Observables when they are no longer needed, especially to avoid memory leaks in Angular applications. This can be done by calling the unsubscribe method on the subscription.

JavaScript

subscription.unsubscribe();

You can also tell your subscription to unsubscribe on its own using the pipe function and operators like take or takeUntil.

JavaScript

// A subject is a special type of Observable
private destroy$ = new Subject<void>();

observable$
  .pipe(takeUntil(this.destroy$))
  .subscribe(data => console.log(data));

ngOnDestroy() {
  this.destroy$.next();
  this.destroy$.complete();
}

And in Angular 16+ we have a new unsubscribe operator takeUntilDestroyed

TypeScript

import { takeUntilDestroyed } from '@angular/core/rxjs-interop';

@Component({
  ...
})
export class Component{
  constructor() {
    interval(1000)
      // stop checking every second if component is destroyed
      .pipe(takeUntilDestroyed()) 
      .subscribe(console.log);
  }
}

Using Observables in Angular

Angular integrates seamlessly with RxJS. Observables are commonly used in Angular for handling asynchronous operations, such as HTTP requests and user inputs.

HTTP Requests

The Angular HttpClient service returns Observables for HTTP operations, making it easy to handle asynchronous data. This is one of the few Observables that you do not need to unsubscribe from as it is handled for you.

TypeScript

import { HttpClient } from '@angular/common/http';
import { Component, OnInit } from '@angular/core';

@Component({
  selector: 'app-data',
  template: '<div *ngIf="data">{{ data }}</div>'
})
export class DataComponent implements OnInit {
  data: any;

  constructor(private http: HttpClient) {}

  ngOnInit() {
    this.http.get('https://api.example.com/data')
      .subscribe(response => this.data = response);
  }
}

Forms and User Input

Angular’s Reactive Forms module also uses Observables to handle form value changes and validation status.

TypeScript

import { Component, OnInit } from '@angular/core';
import { FormBuilder, FormGroup } from '@angular/forms';

@Component({
  selector: 'app-form',
  template: `
    <form [formGroup]="form">
      <input formControlName="name">
    </form>
    <p>{{ formValue }}</p>
  `
})
export class FormComponent implements OnInit {
  form: FormGroup;
  formValue: string;

  constructor(private fb: FormBuilder) {}

  ngOnInit() {
    this.form = this.fb.group({
      name: ['']
    });

    // this will create a memory leak because it does not unsubscribe
    this.form.get('name').valueChanges.subscribe(value => {
      this.formValue = value;
    });
  }
}

Operators

RxJS comes with a rich set of operators that allow you to manipulate Observables. Operators are functions that take an Observable as input and return another Observable. Common operators include map, filter, mergeMap, and switchMap.

Example: Using `map` and `filter`

JavaScript

import { of } from 'rxjs';
import { map, filter } from 'rxjs/operators';

const numbers = of(1, 2, 3, 4, 5);

const squaredEvenNumbers = numbers.pipe(
  filter(n => n % 2 === 0),
  map(n => n * n)
);

squaredEvenNumbers.subscribe(x => console.log(x)); // Output: 4, 16

Example: Using `switchMap` for HTTP Requests

switchMap is particularly useful for handling HTTP requests where you need to switch to a new Observable and cancel the previous one.

JavaScript

import { Component, OnInit } from '@angular/core';
import { FormBuilder, FormGroup } from '@angular/forms';
import { HttpClient } from '@angular/common/http';
import { takeUntilDestroyed } from '@angular/core/rxjs-interop';
import { switchMap } from 'rxjs/operators';

@Component({
  selector: 'app-search',
  template: `
    <form [formGroup]="form">
      <input formControlName="query">
    </form>
    <ul>
      <li *ngFor="let item of results">{{ item }}</li>
    </ul>
  `
})
export class SearchComponent implements OnInit {
  public form: FormGroup;
  public results: any[];

  // used to tell subscriptions in the component when to unsubscribe
  private destroyRef = inject(DestroyRef);
  
  constructor(private fb: FormBuilder, private http: HttpClient) {}

  ngOnInit() {
    this.form = this.fb.group({
      query: ['']
    });

    this.form.get('query').valueChanges.pipe(
      // unsub when component is destroyed to avoid memrory leak
      takeUntilDestroyed(this.destroyRef),
      // listen to changes to the query field and switch to a new observable 
      // created by doing an http request based on the query
      switchMap(query => this.http.get<any[]>(`https://api.example.com/search?q=${query}`))
    ).subscribe(data => this.results = data);
  }
}

Conclusion

Observables are a powerful way to handle asynchronous data in JavaScript, and RxJS provides a comprehensive toolkit for working with them. In Angular, Observables are used extensively, from HTTP requests to form handling and beyond. By understanding the basics of Observables and how to use them in Angular, you can build more responsive and efficient web applications. Happy coding!