Home Tech blog Observables – What is it? And how to use it in your code

Observables – What is it? And how to use it in your code

Łukasz Flak

26.10.2023

Introduction

Observables are an essential part of the Angular framework that provides a powerful way to handle asynchronous data streams. In this article, we will explore what observables are, why they are important, and how you can leverage them in your Angular code. We will also provide code examples to illustrate their usage and demonstrate their benefits.

Understanding Observables

Observables are a representation of a stream of data that can change over time. They are a core part of the Reactive Extensions for JavaScript (RxJS) library, which is widely used in Angular applications. An observable can emit multiple values over time and can be observed by multiple subscribers. It follows the observer design pattern, where an observable is the producer of values, and observers (subscribers) react to those values.

Creating Observables

To create an observable, you can use the Observable class provided by RxJS. Here’s an example of creating a simple observable that emits a sequence of numbers:

import { Observable } from 'rxjs';

const numberObservable = new Observable<number>((observer) => {
  let count = 0;

  const intervalId = setInterval(() => {
    observer.next(count++);
  }, 1000);

  return () => {
    clearInterval(intervalId);
  };
});

In this example, we create an observable that emits a number every second. The Observable constructor takes a function that defines how the observable will produce values. In this case, the function receives an observer object and uses the next method to emit values.

Subscribing to Observables

Once an observable is created, you can subscribe to it to start receiving its emitted values. Here’s how you can subscribe to the numberObservable we created earlier:

numberObservable.subscribe((value) => {
  console.log(value);
});

In this example, we subscribe to the numberObservable and provide a callback function to handle the emitted values. Whenever a new value is emitted, the callback function is called with that value.

Operators and Transformation

Observables provide a wide range of operators that allow you to transform, filter, combine, and manipulate the emitted values. Let’s look at an example using the map operator to transform the emitted numbers into their squares:

import { Observable } from 'rxjs';
import { map } from 'rxjs/operators';

const squaredObservable = numberObservable.pipe(
  map((value) => value * value)
);

squaredObservable.subscribe((value) => {
  console.log(value);
});

In this example, we have used the pipe method to chain the map operator onto the numberObservable. The map operator transforms each emitted value by multiplying it with itself. As a result, we receive the squared values.

Filtering Observables

In addition to transformation, observables allow you to filter the emitted values based on certain conditions. The filter operator is commonly used for this purpose. Let’s consider an example where we filter out even numbers from the numberObservable:

import { Observable } from 'rxjs';
import { filter } from 'rxjs/operators';

const oddNumberObservable = numberObservable.pipe(
  filter((value) => value % 2 !== 0)
);

oddNumberObservable.subscribe((value) => {
  console.log(value);
});

In this example, we have used the filter operator to only allow odd numbers to pass through the observable. The callback function inside the filter operator checks if the value is not divisible by 2 (i.e., an odd number) and only emits those values.

Error Handling

Observables also provide mechanisms to handle errors emitted during the stream. The catchError operator is commonly used to catch errors and handle them neatly. Let’s consider an example where an error is intentionally thrown after a certain condition is met:

import { Observable, throwError } from 'rxjs';
import { catchError } from 'rxjs/operators';

const errorObservable = new Observable<number>((observer) => {
  let count = 0;

  const intervalId = setInterval(() => {
    if (count === 3) {
      observer.error('Something went wrong!');
    } else {
      observer.next(count++);
    }
  }, 1000);

  return () => {
    clearInterval(intervalId);
  };
});

errorObservable.pipe(
  catchError((error) => {
    console.log('Error:', error);
    return throwError('Error occurred');
  })
).subscribe({
  next: (value) => {
    console.log(value);
  },
  error: (error) => {
    console.log('Handled Error:', error);
  },
});

In this example, we have intentionally thrown an error when the count reaches 3. The catchError operator catches the error, logs it, and emits a new error using throwError. The subscriber provides separate handlers ‘next’ and ‘error’ to handle the emitted values and errors respectively.

Unsubscribing and Cleanup

Observables allow for easy cleanup by providing a way to unsubscribe from the stream. When you subscribe to an observable, the subscribe method returns a subscription object that can be used to unsubscribe. Here’s an example:

const subscription = numberObservable.subscribe((value) => {
  console.log(value);
});

// Unsubscribe after 5 seconds
setTimeout(() => {
  subscription.unsubscribe();
}, 5000);

In this example, we store the subscription object in a variable and later call its unsubscribe method after 5 seconds. This ensures that we stop receiving values from the observable and clean up any resources associated with it.

Testing Observables

The proper testing approach for testing Observables is to use the fakeAsync function provided by the @angular/core/testing package. This utility function allows you to write synchronous-looking tests for asynchronous code, including observables.

To demonstrate how to use fakeAsync for testing observables, let’s consider a simple example where we have a service that returns an observable. We want to test whether the observable emits the expected values.

First, let’s assume we have a service called DataService that provides an observable:

import { Injectable } from '@angular/core';
import { Observable, of } from 'rxjs';

@Injectable()
export class DataService {
  getData(): Observable<number> {
    return of(1, 2, 3, 4, 5);
  }
}

Now, let’s write a test using fakeAsync to verify that the observable emits the expected values:

import { fakeAsync, tick } from '@angular/core/testing';
import { DataService } from './data.service';

describe('DataService', () => {
  let service: DataService;

  beforeEach(() => {
    service = new DataService();
  });

  it('should emit values from the observable', fakeAsync(() => {
    let result: number[] = [];

    service.getData().subscribe((value: number) => {
      result.push(value);
    });

    tick(); // Simulate the passage of time

    expect(result).toEqual([1, 2, 3, 4, 5]);
  }));
});

In the test above, we have used the fakeAsync function to wrap our test code. Within the fakeAsync block, we have subscribed to the observable returned by getData() and have stored the emitted values in the result array. Then, we have used the tick function to flush any pending microtasks and advance the virtual clock. After the tick function, we can perform assertions on the result array to ensure that the observable emitted the expected values. By using fakeAsync, we can write clean and synchronous-looking tests for asynchronous observables. It provides a convenient way to control time and handle asynchronous operations in a deterministic manner.

Conclusion

Observables are a powerful tool for handling asynchronous data streams in Angular applications. They provide a flexible and reactive approach to managing and transforming data. By understanding the concepts behind observables and leveraging the RxJS library, you can create more responsive and efficient code. Incorporating observables in your Angular projects will enhance your ability to handle asynchronous operations and build robust applications.

Remember to import the necessary RxJS operators and explore the vast collection of operators available to tailor the behavior of your observables. With observables, you can take full advantage of reactive programming principles and unlock the true potential of asynchronous data handling in Angular. Happy coding!