DEV Community

Cover image for Exploring the Depths of Observables and RxJS in Angular Applications

Posted on

Exploring the Depths of Observables and RxJS in Angular Applications

Observables and the RxJS library play a pivotal role in the development of reactive and asynchronous applications using Angular. This comprehensive article delves deeply into the intricacies of RxJS, exploring advanced concepts, sophisticated operators, and practical examples that exemplify their application in intricate scenarios within Angular projects.

Custom Observables: Constructing Tailored Data Streams

Beyond the pre-built Observables furnished by the RxJS library, the capacity to create custom data streams is indispensable. This proves especially valuable when generating sequences of values that are not sourced externally, such as from APIs or events. Here, we construct a custom Observable emitting a sequence of values at regular intervals.

import { Observable } from 'rxjs';

// Creating a bespoke Observable
const customObservable = new Observable<number>((observer) => {
  let count = 0;
  const intervalId = setInterval(() => {;
  }, 1000);

  // Clearing the interval on subscription cancellation
  return () => {

// Subscribing to the custom Observable
const subscription = customObservable.subscribe((value) => {
  console.log(`Emitted value: ${value}`);

// Unsubscribing after 5 seconds
setTimeout(() => {
}, 5000);
Enter fullscreen mode Exit fullscreen mode

In the provided example, a custom Observable named customObservable is established. It emits values at 1-second intervals. Upon subscription cancellation, the interval is cleared to avert memory leaks.

Sharing Observables: Multicasting for Efficiency

There arise instances where sharing an Observable amongst multiple observers mitigates redundant API calls and resource consumption. The share operator facilitates this sharing. Herein, we explore the process of sharing a stream of numbers at regular intervals.

import { interval } from 'rxjs';
import { share } from 'rxjs/operators';

// Generating an Observable emitting numbers at regular intervals
const sharedObservable = interval(1000).pipe(share());

// First observer
sharedObservable.subscribe((value) => {
  console.log(`Observer 1: ${value}`);

// Second observer (3 seconds later)
setTimeout(() => {
  sharedObservable.subscribe((value) => {
    console.log(`Observer 2: ${value}`);
}, 3000);
Enter fullscreen mode Exit fullscreen mode

In the aforementioned example, the share operator guarantees the sharing of the number stream between observers. Thus, the initial observer commences counting from zero, and the subsequent observer continues from the point where the first left off.

Advanced RxJS Operators: Manipulating Data Streams

The RxJS library boasts a diverse array of operators designed to transform, combine, and manipulate data streams. Advanced operators like concatMap, exhaustMap, and bufferTime are scrutinized below.

import { of, interval, fromEvent } from 'rxjs';
import { concatMap, exhaustMap, bufferTime } from 'rxjs/operators';

// ConcatMap: Emitting values sequentially from inner Observables
of(1, 2, 3).pipe(
  concatMap((value) => interval(1000))
).subscribe((value) => {
  console.log(`ConcatMap: ${value}`);

// ExhaustMap: Ignoring new emissions during the activity of an inner Observable
fromEvent(document, 'click').pipe(
  exhaustMap(() => interval(1000))
).subscribe((value) => {
  console.log(`ExhaustMap: ${value}`);

// BufferTime: Grouping button clicks within time intervals
const clicks$ = fromEvent(document, 'click');

).subscribe((clicks) => {
  console.log(`Clicked ${clicks.length} times in the last 2 seconds.`);
Enter fullscreen mode Exit fullscreen mode

The concatMap operator consecutively emits values from inner Observables, while the exhaustMap operator disregards new emissions while an inner Observable is active. The bufferTime operator groups button clicks occurring within a designated time interval.

Advanced Asynchronous Management Strategies: Addressing Complex Scenarios

Advanced strategies for managing asynchronous operations encompass utilizing ReplaySubject to replay historical values and implementing timeouts and retries via the timeout and retry operators.

import { ReplaySubject, interval, throwError } from 'rxjs';
import { timeout, retry, delay, catchError } from 'rxjs/operators';

// ReplaySubject: Replaying past values
const replaySubject = new ReplaySubject(2);;;

replaySubject.subscribe((value) => {
  console.log(`ReplaySubject Subscriber 1: ${value}`);

replaySubject.subscribe((value) => {
  console.log(`ReplaySubject Subscriber 2: ${value}`);

// Timeout and retry: Managing timeouts and retries
const source$ = interval(1000).pipe(
  retry(errors => errors.pipe(delay(1000)))

  value => console.log(`Received: ${value}`),
  err => console.error(`Error: ${err}`)
Enter fullscreen mode Exit fullscreen mode

The initial example demonstrates the application of the ReplaySubject to replay previous values. The second example showcases handling timeouts and retries using the timeout and retry operators.

Advanced Applications in Angular Projects: Implementation of Complex Features

Within the realm of Angular projects, these advanced concepts can be translated into robust features, including advanced error handling within Observables and real-time data streaming utilizing WebSockets.

import { of, fromEvent } from 'rxjs';
import { catchError } from 'rxjs/operators';

// Error Handling in Observables
  map(data => { throw new Error('Oops!'); }),
  catchError(error => of('fallback data'))
).subscribe(result => {
  console.log(`Error Handling: ${result}`);

// Real-Time Data Streaming with WebSockets
import { Observable, webSocket } from 'rxjs';

const socket$: Observable<any> = webSocket('ws://localhost:8080');

  (message) => console.log('WebSocket Message:', message),
  (err) => console.error('WebSocket Error:', err),
  () => console.log('WebSocket Closed')

// Sending a message
socket$.next({ type: 'chat', text: 'Hello, WebSocket!' });
Enter fullscreen mode Exit fullscreen mode

The initial example showcases the use of catchError for error handling within Observables. The subsequent example illustrates the implementation of a real-time data streaming mechanism through WebSockets and Observables.


In this article, we conducted an exploration of Observables and RxJS, encompassing advanced concepts, intricate operators, and practical applications within Angular projects. Armed with a profound comprehension of these facets, developers are equipped to surmount complex challenges and construct high-quality reactive and asynchronous applications within the Angular ecosystem.

Top comments (0)