ANGULAR: Angular Observables and RxJS

Observables are a way of dealing with asynchronous data — data that does not arrive instantly but takes time to come back from a server, a user event, or another source. The key principle is to not pause your application while waiting for the work to complete.

JavaScript has evolved several approaches to handle asynchronous operations:

Approach	Description	Issue
Callbacks	Functions passed as arguments, called when data arrives	Leads to "callback hell" — deeply nested, hard-to-read code
Promises	A cleaner abstraction with built-in error handling (`.catch`) and composability	Can only resolve once; not well-suited for streams of events
Observables	A larger version of Promises — handles streams, multiple values, and complex async pipelines	Requires the RxJS library

The Problem with Synchronous Data Fetching

Without observables, fetching data from a server can block the entire application:

// Blocking - application waits until data arrives
let userData = getUserDataFromServer(); // can take 1-2 seconds
console.log(userData); // nothing else runs until this line finishes

This freezes the UI and creates a poor user experience.

The solution is to make the call observable and subscribe to the result. The application continues running, and the callback fires when the data is ready.

let userDataObs = getUserDataObservable();

userDataObs.subscribe(userData => {
  // This runs when the data arrives — not immediately
  console.log(userData);
});

// Application continues here without waiting

The subscribe method accepts an observer object with three optional callbacks:

let userDataObs = getUserDataObservable();

userDataObs.subscribe({
  next: userData => {
    // Called when new data arrives
    console.log(userData);
  },
  error: error => {
    // Called if an error occurs
    console.error('Error fetching user data:', error);
  },
  complete: () => {
    // Called when the observable completes (not always used)
    console.log('Data stream complete');
  }
});

Callback	When It Fires
`next`	Each time a new value is emitted
`error`	If an error occurs in the stream
`complete`	When the observable finishes emitting values

pipe

The pipe method transforms the data before it reaches the subscribe callback. It is the correct place to process, filter, or transform data from an observable — especially useful when multiple components subscribe to the same stream.

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

let numberObs = from([1, 2, 3, 4]);

numberObs
  .pipe(
    map(num => num + 1)  // adds 1 to every value
  )
  .subscribe(data => console.log(data)); // outputs: 2, 3, 4, 5

Common RxJS Operators

Operator	Description
`map`	Transform each emitted value
`filter`	Only pass values that match a condition
`tap`	Side-effect without transforming (useful for logging)
`catchError`	Handle errors in the stream
`switchMap`	Map to a new observable, cancelling the previous
`debounceTime`	Wait before emitting (useful for search input)

Creating Observables with BehaviorSubject

BehaviorSubject is a special type of observable that holds a current value and emits it to any new subscriber immediately. It is commonly used for shared state in services.

import { BehaviorSubject } from 'rxjs';

// Create a BehaviorSubject with an initial value
let cartCount = new BehaviorSubject<number>(0);

// Subscribe to changes
cartCount.subscribe(count => {
  console.log('Cart count:', count);
});

// Emit a new value
cartCount.next(1);  // logs: Cart count: 1
cartCount.next(2);  // logs: Cart count: 2

BehaviorSubject is ideal for state like:

Shopping cart item count
Current logged-in user
App-wide loading state

Observables in an Angular Service

A typical Angular service using BehaviorSubject for shared state:

import { Injectable } from '@angular/core';
import { BehaviorSubject, Observable } from 'rxjs';
import { IProduct } from './product.model';

@Injectable({
  providedIn: 'root'
})
export class CartService {
  private cartItems$ = new BehaviorSubject<IProduct[]>([]);

  getCartItems(): Observable<IProduct[]> {
    return this.cartItems$.asObservable();
  }

  addItem(product: IProduct): void {
    const current = this.cartItems$.getValue();
    this.cartItems$.next([...current, product]);
  }
}

Consuming the observable in a component:

export class CartComponent implements OnInit, OnDestroy {
  cartItems: IProduct[] = [];
  private subscription: Subscription;

  constructor(private cartService: CartService) {}

  ngOnInit(): void {
    this.subscription = this.cartService.getCartItems()
      .subscribe(items => {
        this.cartItems = items;
      });
  }

  ngOnDestroy(): void {
    this.subscription.unsubscribe(); // prevent memory leaks
  }
}

Observable Data Flow

Data Source (API/Event)
          |
          v
    Observable created
          |
          | pipe()
          v
    Transform / Filter
    (map, filter, etc.)
          |
          | subscribe()
          v
    Component receives data
    and updates the UI

Summary

Observables are the foundation of asynchronous programming in Angular. They replace blocking code with reactive data streams that let the application remain responsive at all times.

Key concepts to remember:

subscribe — attach a callback that fires when data arrives
pipe — transform data before it reaches the subscriber
BehaviorSubject — create your own observable for shared state
Always unsubscribe in ngOnDestroy to prevent memory leaks

Observables and RxJS are what make Angular applications reactive — master them early.