RxJS Interview Questions (Free Preview)

RxJS Fundamentals

What is the difference between cold and hot Observables?

The 30-Second Answer: Cold Observables create a new independent execution for each subscriber (like Netflixâ€”each viewer starts their own playback), while hot Observables share a single execution among all subscribers (like broadcast TVâ€”everyone sees the same live stream). Cold Observables are unicast; hot Observables are multicast.

The 2-Minute Answer (If They Want More): The cold vs. hot distinction is crucial for understanding Observable behavior and avoiding bugs. A cold Observable starts producing values only when subscribed to, and each subscription gets its own independent execution. This is like calling a functionâ€”each call runs separately. Most RxJS creation operators (of, from, ajax, interval) create cold Observables by default.

Cold Observables are perfect for scenarios where each subscriber needs its own data stream: HTTP requests where each subscriber should trigger a separate API call, file reads, or calculated sequences. The downside is that multiple subscriptions can lead to duplicate work (like making the same HTTP request multiple times).

A hot Observable, in contrast, produces values regardless of subscriptions. Subscribers tap into an ongoing stream and only receive values emitted after they subscribeâ€”they miss earlier emissions. Examples include mouse events, WebSocket messages, or subjects. Hot Observables are multicastâ€”all subscribers share the same execution and receive the same values.

You can convert cold to hot using operators like share(), shareReplay(), or publish(). This is essential for performance optimization when you want multiple subscribers to share a single HTTP request or expensive computation. Understanding this distinction helps you predict Observable behavior, especially regarding when side effects execute and whether subscribers receive the same or different values.

Code Example:

import { Observable, interval, share, Subject } from 'rxjs';
import { ajax } from 'rxjs/ajax';

// COLD OBSERVABLE: Each subscriber gets independent execution
const coldObservable$ = new Observable(subscriber => {
  console.log('Cold Observable: New execution started');
  subscriber.next(Math.random()); // Each subscriber gets different random number
  subscriber.complete();
});

coldObservable$.subscribe(value => console.log('Subscriber 1:', value));
coldObservable$.subscribe(value => console.log('Subscriber 2:', value));
// Logs:
// "Cold Observable: New execution started"
// "Subscriber 1: 0.123..." (unique random)
// "Cold Observable: New execution started"
// "Subscriber 2: 0.456..." (different random)

// HOT OBSERVABLE: Shared execution via Subject
const hotObservable$ = new Subject();

hotObservable$.subscribe(value => console.log('Hot Subscriber 1:', value));
hotObservable$.subscribe(value => console.log('Hot Subscriber 2:', value));

hotObservable$.next(Math.random());
// Logs:
// "Hot Subscriber 1: 0.789..." (same random)
// "Hot Subscriber 2: 0.789..." (same random)

// PRACTICAL: Converting cold to hot for HTTP request sharing
const httpRequest$ = ajax.getJSON('/api/data'); // Cold: each subscribe = new request

// Without sharing (BAD - makes 2 HTTP requests)
httpRequest$.subscribe(data => console.log('Component 1:', data));
httpRequest$.subscribe(data => console.log('Component 2:', data));

// With sharing (GOOD - makes 1 HTTP request, shares result)
const sharedRequest$ = httpRequest$.pipe(share());
sharedRequest$.subscribe(data => console.log('Component 1:', data));
sharedRequest$.subscribe(data => console.log('Component 2:', data));

Mermaid Diagram:

flowchart TD
    subgraph Cold Observable
        C[Observable Definition] --> S1[Subscriber 1 subscribes]
        C --> S2[Subscriber 2 subscribes]
        S1 --> E1[Independent Execution 1]
        S2 --> E2[Independent Execution 2]
        E1 --> V1[Values: 1, 2, 3]
        E2 --> V2[Values: 1, 2, 3]
    end

    subgraph Hot Observable
        H[Shared Execution] --> V[Emits: 1, 2, 3, 4, 5]
        HS1[Subscriber 1 joins at start] --> V
        HS2[Subscriber 2 joins late] -.->|Misses 1, 2| V
        V --> R1[Subscriber 1 gets: 1, 2, 3, 4, 5]
        V --> R2[Subscriber 2 gets: 3, 4, 5]
    end

    style E1 fill:#e1f5ff
    style E2 fill:#e1f5ff
    style H fill:#fff3cd
    style R2 fill:#f8d7da

References:

↑ Back to top

What is RxJS and what problems does it solve?

The 30-Second Answer: RxJS (Reactive Extensions for JavaScript) is a library for composing asynchronous and event-based programs using observable sequences. It solves problems like callback hell, race conditions, and complex state management by providing a unified API for handling multiple async values over time.

The 2-Minute Answer (If They Want More): RxJS provides a powerful toolkit for managing asynchronous data streams in JavaScript applications. Before RxJS, developers struggled with callback pyramids, managing multiple event listeners, coordinating parallel async operations, and handling error propagation across async boundaries.

RxJS addresses these challenges by treating everything as a stream of data over time. Whether you're dealing with HTTP requests, user input, WebSocket messages, or timer events, RxJS provides a consistent, composable API. It offers over 100 operators for transforming, filtering, combining, and managing these streams.

The library shines in scenarios like autocomplete search (debouncing user input, canceling previous requests), real-time data dashboards (merging multiple WebSocket streams), complex form validation (combining multiple async validators), and managing application state reactively. By using declarative operators instead of imperative event handlers, your code becomes more predictable, testable, and maintainable.

RxJS is the foundation of reactive programming in Angular and is increasingly used in React, Vue, and vanilla JavaScript applications for managing complex async workflows.

Code Example:

import { fromEvent, debounceTime, map, switchMap, catchError } from 'rxjs';
import { ajax } from 'rxjs/ajax';

// Problem: Autocomplete search with debouncing and request cancellation
const searchInput = document.getElementById('search');

// Solution with RxJS
fromEvent(searchInput, 'input')
  .pipe(
    // Extract the input value
    map(event => (event.target as HTMLInputElement).value),
    // Wait 300ms after user stops typing
    debounceTime(300),
    // Cancel previous request if new input arrives
    switchMap(query =>
      ajax.getJSON(`/api/search?q=${query}`).pipe(
        // Handle errors gracefully
        catchError(error => of([]))
      )
    )
  )
  .subscribe(results => {
    // Update UI with search results
    displayResults(results);
  });

// Without RxJS, you'd need manual debounce timers,
// XMLHttpRequest abort logic, and complex state tracking

Mermaid Diagram:

flowchart TD
    A[User Input Events] --> B[debounceTime]
    B --> C[switchMap to API]
    C --> D{Request Status}
    D -->|Success| E[Display Results]
    D -->|Error| F[Handle Error]
    G[New Input] -.->|Cancels| C

    style A fill:#e1f5ff
    style E fill:#d4edda
    style F fill:#f8d7da

References:

↑ Back to top

What is reactive programming?

The 30-Second Answer: Reactive programming is a declarative programming paradigm focused on data streams and the propagation of change. Instead of imperatively telling the computer what to do step-by-step, you define relationships between data streams and let the system automatically propagate changes through those relationships.

The 2-Minute Answer (If They Want More): Reactive programming represents a fundamental shift in how we think about program flow. Traditional imperative programming follows a "pull" model where you actively request data when needed. Reactive programming uses a "push" model where data sources notify consumers when new values are available.

Think of a spreadsheet: when you change cell A1, any cells with formulas referencing A1 automatically update. That's reactive programming. You've declared the relationships (formulas), and the system handles propagation of changes.

In software, reactive programming treats asynchronous events, user interactions, API responses, and state changes as streams of values over time. You compose these streams using functional operators (map, filter, merge) to create data transformation pipelines. The key principles include: treating time as a first-class citizen, declarative composition over imperative coordination, automatic error propagation, and backpressure handling for managing data flow rates.

This paradigm excels in event-driven applications, real-time systems, and scenarios with multiple interdependent async data sources. It leads to more maintainable code because the flow of data is explicit and side effects are isolated, making programs easier to reason about, test, and debug.

Code Example:

import { interval, fromEvent, combineLatest, map } from 'rxjs';

// Reactive programming example: Auto-saving form with status indicator
const formData$ = fromEvent(document.getElementById('form'), 'input').pipe(
  map(() => getFormData()) // Stream of form states
);

const autoSaveTimer$ = interval(5000); // Stream of timer ticks every 5s

// Declaratively combine streams: save whenever timer fires AND form has data
combineLatest([formData$, autoSaveTimer$]).pipe(
  map(([data, _]) => data), // Extract form data
)
.subscribe(data => {
  saveToServer(data); // Side effect: save to server
  showSaveStatus('Saved at ' + new Date().toLocaleTimeString());
});

// Traditional imperative approach would require:
// - Manual timer management
// - State variables to track form changes
// - Event listener cleanup
// - Coordination logic between timer and form events

Mermaid Diagram:

flowchart LR
    A[Data Source 1] -->|push| C[Operator Chain]
    B[Data Source 2] -->|push| C
    C -->|transform| D[Operator Chain]
    D -->|push| E[Observer/Consumer]

    F[Traditional Pull Model] -.->|request| G[Data Source]
    G -.->|return| F

    style A fill:#e1f5ff
    style B fill:#e1f5ff
    style E fill:#d4edda
    style F fill:#fff3cd
    style G fill:#fff3cd

References:

↑ Back to top

What is an Observable and how does it differ from a Promise?

The 30-Second Answer: An Observable is a lazy, cancellable stream that can emit multiple values over time, while a Promise is an eager, non-cancellable operation that resolves to a single value once. Observables don't execute until subscribed to, can be cancelled via unsubscribe, and are ideal for handling continuous data streams.

The 2-Minute Answer (If They Want More): Observables and Promises both handle asynchronous operations, but they have fundamental differences in behavior and capabilities. A Promise starts executing immediately when created (eager), represents a single future value, and cannot be cancelled once initiated. You use .then() to consume the result, and once settled (resolved or rejected), a Promise is done.

An Observable, however, is lazyâ€”it doesn't execute until someone subscribes to it. This means you can define an Observable, pass it around, and it won't do any work until explicitly subscribed. Each subscription can trigger independent execution, making Observables ideal for data sources that should be re-executed (like HTTP requests in retry scenarios).

Observables can emit zero, one, or multiple values over their lifetime, making them perfect for streams like WebSocket messages, mouse movements, or interval timers. They're also cancellableâ€”calling unsubscribe() stops execution and cleans up resources. Observables have a rich operator ecosystem (map, filter, debounce, switchMap, etc.) for composing complex async workflows declaratively.

The key trade-off: use Promises for simple, one-time async operations (single HTTP request, reading a file). Use Observables when you need cancellation, multiple values over time, or complex stream composition. In Angular, you can convert between them using toPromise() or from(), but it's generally better to stay in Observable-land for consistency.

Code Example:

import { Observable, from } from 'rxjs';

// PROMISE: Eager, single value, non-cancellable
const promise = new Promise((resolve) => {
  console.log('Promise executing immediately!');
  setTimeout(() => resolve('Promise result'), 1000);
});
// Logs "Promise executing immediately!" right away, even without .then()

promise.then(value => console.log(value)); // Single value: "Promise result"

// OBSERVABLE: Lazy, multiple values, cancellable
const observable$ = new Observable((subscriber) => {
  console.log('Observable executing only when subscribed!');

  let count = 0;
  const interval = setInterval(() => {
    subscriber.next(count++); // Can emit multiple values
    if (count === 3) {
      subscriber.complete(); // Signal completion
    }
  }, 1000);

  // Cleanup function runs on unsubscribe
  return () => {
    console.log('Cleaning up!');
    clearInterval(interval);
  };
});
// Nothing logged yet - Observable hasn't executed

const subscription = observable$.subscribe({
  next: (value) => console.log('Observable value:', value),
  complete: () => console.log('Observable completed')
});
// Now logs "Observable executing..." and emits: 0, 1, 2

// Cancellation
setTimeout(() => {
  subscription.unsubscribe(); // Stops execution, runs cleanup
}, 2500);

// Converting between them
const promiseFromObservable = observable$.toPromise(); // Gets last emitted value
const observableFromPromise = from(promise); // Wraps Promise as Observable

Mermaid Diagram:

flowchart TD
    subgraph Promise
        P1[Created] -->|Executes Immediately| P2[Pending]
        P2 -->|Single Value| P3[Resolved/Rejected]
        P3 -->|Done| P4[Cannot Cancel]
    end

    subgraph Observable
        O1[Defined] -.->|Lazy| O2[subscribe called]
        O2 -->|Executes| O3[Emitting Values]
        O3 -->|next, next, next| O4[Multiple Values]
        O4 -->|complete or error| O5[Done]
        O3 -.->|unsubscribe| O6[Cancelled & Cleanup]
    end

    style P3 fill:#d4edda
    style P4 fill:#f8d7da
    style O4 fill:#d4edda
    style O6 fill:#fff3cd

References:

↑ Back to top

What is the Observer pattern in RxJS?

The 30-Second Answer: The Observer pattern in RxJS is a design pattern where an Observable (subject) maintains a list of Observers (subscribers) and notifies them of state changes by calling their next(), error(), or complete() methods. It enables a one-to-many relationship where multiple consumers can react to data emitted by a single source.

The 2-Minute Answer (If They Want More): The Observer pattern is the foundational design pattern underlying RxJS. It establishes a subscription mechanism where Observers register interest in an Observable's data stream. When the Observable emits values, it pushes those values to all registered Observers, creating an inversion of control compared to traditional pull-based approaches.

In RxJS, an Observer is an object with three optional methods: next(value) receives emitted values, error(err) handles errors, and complete() signals stream completion. When you call observable.subscribe(), you're registering an Observer with the Observable. The Observable then "pushes" data to Observers over time.

This pattern solves the problem of tight coupling between data sources and consumers. The Observable doesn't need to know who's listening or how many listeners there areâ€”it just emits values. Observers don't need to poll for dataâ€”they react when notified. This decoupling makes code more modular and testable.

RxJS extends the classic Observer pattern with additional features: lazy execution (Observables don't start until subscribed), cancellation (via unsubscribe), and a rich operator ecosystem for stream composition. Unlike traditional event emitters, RxJS Observables also handle completion and errors as first-class concepts, making error propagation and resource cleanup more predictable.

Code Example:

import { Observable } from 'rxjs';

// Observable: The subject being observed
const dataStream$ = new Observable(subscriber => {
  console.log('Observable: Starting to produce values');

  // Emit values to all observers
  subscriber.next('First value');
  subscriber.next('Second value');

  // Simulate async operation
  setTimeout(() => {
    subscriber.next('Third value');
    subscriber.complete(); // Signal completion
  }, 1000);

  // Error example (uncomment to test)
  // subscriber.error(new Error('Something went wrong'));

  // Cleanup logic
  return () => console.log('Observable: Cleanup on unsubscribe');
});

// Observer: Object with next, error, complete methods
const observer1 = {
  next: (value) => console.log('Observer 1 received:', value),
  error: (err) => console.error('Observer 1 error:', err),
  complete: () => console.log('Observer 1: Stream completed')
};

const observer2 = {
  next: (value) => console.log('Observer 2 received:', value),
  error: (err) => console.error('Observer 2 error:', err),
  complete: () => console.log('Observer 2: Stream completed')
};

// Register observers (subscribe)
const subscription1 = dataStream$.subscribe(observer1);
const subscription2 = dataStream$.subscribe(observer2);

// Shorthand subscribe syntax (more common)
dataStream$.subscribe(
  value => console.log('Observer 3 received:', value), // next
  error => console.error('Observer 3 error:', error),  // error
  () => console.log('Observer 3: Stream completed')    // complete
);

// Unsubscribe when done (important for cleanup)
setTimeout(() => {
  subscription1.unsubscribe();
  subscription2.unsubscribe();
}, 2000);

Mermaid Diagram:

flowchart TB
    O[Observable/Subject] --> |next value| OBS1[Observer 1]
    O --> |next value| OBS2[Observer 2]
    O --> |next value| OBS3[Observer 3]

    OBS1 --> N1[next method]
    OBS1 --> E1[error method]
    OBS1 --> C1[complete method]

    O -.->|error| OBSE[All Observers]
    O -.->|complete| OBSC[All Observers]

    subgraph Observer Interface
        N1
        E1
        C1
    end

    style O fill:#e1f5ff
    style OBS1 fill:#d4edda
    style OBS2 fill:#d4edda
    style OBS3 fill:#d4edda
    style OBSE fill:#f8d7da

References:

↑ Back to top

Subjects

What is the difference between Subject, BehaviorSubject, ReplaySubject, and AsyncSubject?

The 30-Second Answer: Subject emits values only to current subscribers. BehaviorSubject requires an initial value and always replays the latest value to new subscribers. ReplaySubject replays a specified number of past values to new subscribers. AsyncSubject only emits the last value when the stream completes.

The 2-Minute Answer (If They Want More): These four Subject variants handle value emission and replay behavior differently:

Subject is the base type - it multicasts values to all current subscribers but doesn't store any values. New subscribers only receive values emitted after their subscription.

BehaviorSubject stores the most recent value and requires an initial value at creation. Any new subscriber immediately receives the current value upon subscription, making it perfect for representing "current state" like user authentication status or selected theme.

ReplaySubject maintains a buffer of previously emitted values (configurable size) and replays them to new subscribers. You can specify how many values to buffer or a time window. This is useful for scenarios like activity logs where you want new subscribers to catch up on recent events.

AsyncSubject only emits the very last value and only after the stream completes. Until completion, it doesn't emit anything to subscribers. This is useful for operations that have a single result after completion, similar to a Promise.

Code Example:

import { Subject, BehaviorSubject, ReplaySubject, AsyncSubject } from 'rxjs';

// 1. Regular Subject - no initial value, no replay
const subject = new Subject<number>();
subject.subscribe(v => console.log('Subject A:', v));
subject.next(1);
subject.subscribe(v => console.log('Subject B:', v)); // Misses value 1
subject.next(2);
// Output: Subject A: 1, Subject A: 2, Subject B: 2

console.log('---');

// 2. BehaviorSubject - requires initial value, replays latest
const behaviorSubject = new BehaviorSubject<number>(0);
behaviorSubject.subscribe(v => console.log('Behavior A:', v));
behaviorSubject.next(1);
behaviorSubject.next(2);
behaviorSubject.subscribe(v => console.log('Behavior B:', v)); // Gets current value 2
behaviorSubject.next(3);
// Output: Behavior A: 0, Behavior A: 1, Behavior A: 2, Behavior B: 2, Behavior A: 3, Behavior B: 3

console.log('---');

// 3. ReplaySubject - replays last N values
const replaySubject = new ReplaySubject<number>(2); // Buffer size: 2
replaySubject.subscribe(v => console.log('Replay A:', v));
replaySubject.next(1);
replaySubject.next(2);
replaySubject.next(3);
replaySubject.subscribe(v => console.log('Replay B:', v)); // Gets last 2 values: 2, 3
// Output: Replay A: 1, Replay A: 2, Replay A: 3, Replay B: 2, Replay B: 3

console.log('---');

// 4. AsyncSubject - only emits last value on complete
const asyncSubject = new AsyncSubject<number>();
asyncSubject.subscribe(v => console.log('Async A:', v));
asyncSubject.next(1);
asyncSubject.next(2);
asyncSubject.subscribe(v => console.log('Async B:', v));
asyncSubject.next(3);
asyncSubject.complete(); // Only now both subscribers receive value 3
// Output: Async A: 3, Async B: 3

// Practical use cases
class UserService {
  // Current user state - new components get current value immediately
  private currentUser = new BehaviorSubject<User | null>(null);
  currentUser$ = this.currentUser.asObservable();

  // Recent notifications - new subscribers see last 5
  private notifications = new ReplaySubject<Notification>(5);
  notifications$ = this.notifications.asObservable();

  // API result - only emit when request completes
  private apiResult = new AsyncSubject<ApiResponse>();
  apiResult$ = this.apiResult.asObservable();
}

Mermaid Diagram (if helpful for visualization):

flowchart TD
    subgraph Subject
        S1[Subject] -->|Only future values| NewSub1[New Subscriber]
        S1 -->|Value 1, 2, 3| ExistingSub1[Existing Subscriber]
    end

    subgraph BehaviorSubject
        S2[BehaviorSubject<br/>Initial: 0] -->|Latest value: 3| NewSub2[New Subscriber]
        S2 -->|0, 1, 2, 3| ExistingSub2[Existing Subscriber]
    end

    subgraph ReplaySubject
        S3[ReplaySubject<br/>Buffer: 2] -->|Last 2 values: 2, 3| NewSub3[New Subscriber]
        S3 -->|1, 2, 3| ExistingSub3[Existing Subscriber]
    end

    subgraph AsyncSubject
        S4[AsyncSubject<br/>On Complete] -->|Last value: 3| NewSub4[New Subscriber]
        S4 -->|Last value: 3| ExistingSub4[Existing Subscriber]
    end

    style S1 fill:#ffcccc
    style S2 fill:#ccffcc
    style S3 fill:#ccccff
    style S4 fill:#ffffcc

References:

↑ Back to top

What is a Subject in RxJS?

The 30-Second Answer: A Subject is a special type of Observable that acts as both an observer and an observable, allowing you to multicast values to multiple subscribers. Unlike regular Observables that are unicast (each subscriber gets an independent execution), Subjects share a single execution among all subscribers.

The 2-Minute Answer (If They Want More): A Subject in RxJS is essentially a bridge between the observer pattern and the observable pattern. It maintains a list of subscribers and can manually emit values to all of them at once using next(), error(), and complete() methods.

The key difference from regular Observables is that Subjects are "hot" - they emit values regardless of whether there are subscribers, and late subscribers don't receive past values. When you subscribe to a Subject, you're registering a listener that will receive future emissions, not triggering a new execution.

This makes Subjects perfect for event buses, shared state management, and converting callback-based APIs to reactive streams. They're commonly used in Angular services to share data between components or to create custom event emitters.

However, Subjects should be used carefully as they introduce state and side effects into your reactive code. In many cases, operators like share() or shareReplay() can provide similar multicasting behavior without manually managing a Subject.

Code Example:

import { Subject } from 'rxjs';

// Create a Subject
const messageSubject = new Subject<string>();

// First subscriber
messageSubject.subscribe({
  next: (msg) => console.log('Subscriber 1:', msg)
});

// Emit a value - both subscribers will receive it
messageSubject.next('Hello');

// Second subscriber (joins later)
messageSubject.subscribe({
  next: (msg) => console.log('Subscriber 2:', msg)
});

// Both subscribers receive this
messageSubject.next('World');

// Output:
// Subscriber 1: Hello
// Subscriber 1: World
// Subscriber 2: World

// Practical example: Event bus service in Angular
class EventBusService {
  private userLoggedIn = new Subject<User>();

  // Expose as Observable to prevent external next() calls
  userLoggedIn$ = this.userLoggedIn.asObservable();

  notifyUserLogin(user: User) {
    this.userLoggedIn.next(user);
  }
}

Mermaid Diagram (if helpful for visualization):

flowchart TD
    S[Subject]

    S -->|next value| Sub1[Subscriber 1]
    S -->|next value| Sub2[Subscriber 2]
    S -->|next value| Sub3[Subscriber 3]

    E[External Code] -->|calls next| S

    style S fill:#f9f,stroke:#333,stroke-width:4px
    style E fill:#bbf,stroke:#333,stroke-width:2px

References:

↑ Back to top

Creation Operators

What is the difference between of, from, and fromEvent?

The 30-Second Answer: of emits individual arguments as separate values, from converts arrays/iterables/promises into sequential emissions, and fromEvent creates Observables from DOM or Node.js events. Use of for discrete values, from for collections/promises, and fromEvent for event listeners.

The 2-Minute Answer (If They Want More): These three creation operators serve different purposes despite appearing similar at first glance.

of takes a comma-separated list of arguments and emits each one synchronously as a separate value before completing. It's perfect when you have a fixed set of known values you want to emit. Think of it as converting individual items into a stream: of(1, 2, 3) emits three separate values.

from converts various data structures into Observables. It accepts arrays, array-like objects, promises, iterables, and other Observables. When given an array [1, 2, 3], it unpacks it and emits each element individually. When given a promise, it waits for resolution and emits the resolved value. This makes from ideal for converting existing data structures into reactive streams.

fromEvent is specifically designed for event sources like DOM events, Node.js EventEmitter events, or jQuery events. It sets up an event listener and emits every event occurrence. Unlike of and from which complete after emitting their values, fromEvent creates an infinite Observable that emits values indefinitely until unsubscribed.

The key distinction: of is for discrete known values, from is for converting collections or promises, and fromEvent is for ongoing event streams. This affects completion behaviorâ€”of and from complete automatically, while fromEvent requires manual unsubscription.

Code Example:

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

// of: Emits each argument as separate values
const of$ = of(1, 2, 3);
of$.subscribe({
  next: val => console.log('of emitted:', val),
  complete: () => console.log('of completed')
});
// Output: 1, 2, 3, then completes

// of with objects (emits the objects themselves)
const ofObjects$ = of({ id: 1 }, { id: 2 }, { id: 3 });
ofObjects$.subscribe(val => console.log('of object:', val));

// of with array (emits the ENTIRE array as one value)
const ofArray$ = of([1, 2, 3]);
ofArray$.subscribe(val => console.log('of array:', val));
// Output: [1, 2, 3] (single emission)

// from: Converts array to individual emissions
const fromArray$ = from([1, 2, 3]);
fromArray$.subscribe({
  next: val => console.log('from emitted:', val),
  complete: () => console.log('from completed')
});
// Output: 1, 2, 3, then completes

// from: Converts Promise to Observable
const promise = fetch('https://api.example.com/data');
const fromPromise$ = from(promise);
fromPromise$.subscribe({
  next: response => console.log('Promise resolved:', response),
  complete: () => console.log('Promise Observable completed')
});

// from: Converts iterable (like Set, Map)
const set = new Set([1, 2, 3]);
const fromSet$ = from(set);
fromSet$.subscribe(val => console.log('from Set:', val));

// fromEvent: Creates Observable from DOM events
const button = document.getElementById('btn');
const clicks$ = fromEvent(button, 'click');

// Note: fromEvent never completes on its own
clicks$.pipe(take(3)).subscribe({
  next: event => console.log('Click detected:', event.target),
  complete: () => console.log('Stopped listening after 3 clicks')
});

// fromEvent with Node.js EventEmitter
const eventEmitter = new EventEmitter();
const messages$ = fromEvent(eventEmitter, 'message');
messages$.subscribe(msg => console.log('Message:', msg));

Mermaid Diagram:

flowchart LR
    subgraph "of operator"
        A1[of 1, 2, 3] --> B1[Emit: 1]
        B1 --> C1[Emit: 2]
        C1 --> D1[Emit: 3]
        D1 --> E1[Complete]
    end

    subgraph "from operator"
        A2[from Promise/Array] --> B2{Type?}
        B2 -->|Array| C2[Unpack & Emit Each]
        B2 -->|Promise| D2[Wait & Emit Result]
        C2 --> E2[Complete]
        D2 --> E2
    end

    subgraph "fromEvent operator"
        A3[fromEvent target, 'click'] --> B3[Setup Listener]
        B3 --> C3[Emit on Each Event]
        C3 --> C3
        D3[Never Completes] -.-> C3
    end

    style E1 fill:#4CAF50
    style E2 fill:#4CAF50
    style D3 fill:#FF9800

References:

↑ Back to top

Transformation Operators

What is the map operator and how does it work?

The 30-Second Answer: The map operator transforms each value emitted by an Observable by applying a projection function to it, similar to Array.map(). It takes each incoming value, applies your transformation function, and emits the result downstream in a one-to-one mapping.

The 2-Minute Answer (If They Want More): The map operator is one of the most fundamental transformation operators in RxJS. It works exactly like JavaScript's Array.map(), but for Observable streams. For every value that comes through the Observable, map applies a function to transform that value and emits the transformed result.

Map maintains a one-to-one relationship between input and output values. If your source Observable emits 5 values, map will also emit exactly 5 values (the transformed versions). This makes it predictable and easy to reason about.

I use map whenever I need to transform data without changing the structure of the Observable stream itself. Common use cases include extracting properties from objects, converting data types, formatting strings, performing calculations, or mapping API responses to domain models. It's synchronous and doesn't introduce any additional Observables into the stream.

The key distinction is that map transforms values, while operators like mergeMap, switchMap, and concatMap transform values into new Observables and then flatten them. If your transformation function returns a plain value (not an Observable), use map. If it returns an Observable, you need one of the flattening operators.

Code Example:

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

// Basic transformation: doubling numbers
const numbers$ = of(1, 2, 3, 4, 5);
numbers$.pipe(
  map(num => num * 2)
).subscribe(result => console.log(result));
// Output: 2, 4, 6, 8, 10

// Extracting properties from objects
interface User {
  id: number;
  name: string;
  email: string;
}

const users$ = of<User>(
  { id: 1, name: 'Alice', email: 'alice@example.com' },
  { id: 2, name: 'Bob', email: 'bob@example.com' }
);

users$.pipe(
  map(user => user.name) // Extract just the name
).subscribe(name => console.log(name));
// Output: 'Alice', 'Bob'

// Transforming API response to domain model
interface ApiResponse {
  data: { firstName: string; lastName: string; };
  metadata: { timestamp: number; };
}

interface UserModel {
  fullName: string;
  createdAt: Date;
}

const apiResponse$ = of<ApiResponse>({
  data: { firstName: 'John', lastName: 'Doe' },
  metadata: { timestamp: 1640000000000 }
});

apiResponse$.pipe(
  map(response => ({
    fullName: `${response.data.firstName} ${response.data.lastName}`,
    createdAt: new Date(response.metadata.timestamp)
  } as UserModel))
).subscribe(user => console.log(user));
// Output: { fullName: 'John Doe', createdAt: Date }

// Chaining multiple map operations
const input$ = fromEvent<MouseEvent>(document, 'click');
input$.pipe(
  map(event => ({ x: event.clientX, y: event.clientY })), // Extract coordinates
  map(coords => `Position: ${coords.x}, ${coords.y}`),    // Format as string
  map(str => str.toUpperCase())                            // Transform to uppercase
).subscribe(result => console.log(result));
// Output: 'POSITION: 150, 200' (example values)

Mermaid Diagram:

flowchart LR
    A[Source: 1, 2, 3] --> B[map x => x * 2]
    B --> C[Output: 2, 4, 6]

    style A fill:#e1f5ff
    style B fill:#fff4e1
    style C fill:#e8f5e9

References:

↑ Back to top

What is the difference between map, mergeMap, switchMap, and concatMap?

The 30-Second Answer: map transforms values one-to-one, while mergeMap, switchMap, and concatMap are "flattening operators" that transform values into inner Observables and then flatten them. mergeMap runs all inner Observables concurrently, switchMap cancels previous inner Observables when a new one starts, and concatMap queues them to run sequentially.

The 2-Minute Answer (If They Want More): The fundamental difference lies in what these operators do and how they handle Observable emissions. The map operator performs simple value transformation - it takes a value, applies a function, and emits the result. It's synchronous and maintains a one-to-one relationship.

The other three operators (mergeMap, switchMap, concatMap) are "higher-order mapping operators" or "flattening operators." They're used when your transformation function returns an Observable rather than a plain value. These operators not only apply the transformation but also subscribe to the resulting inner Observables and flatten their emissions back into the outer stream.

The critical difference between these flattening operators is their concurrency strategy:

mergeMap (also called flatMap) subscribes to all inner Observables as they arrive and merges their emissions. All inner Observables run concurrently with no cancellation. This is ideal for operations where order doesn't matter and you want maximum parallelism, like making multiple independent API calls.

switchMap cancels the previous inner Observable whenever a new value arrives from the source. Only the most recent inner Observable is active at any time. This is perfect for scenarios like search typeahead where you only care about the latest request and want to cancel outdated ones.

concatMap queues inner Observables and subscribes to them one at a time in sequence. It waits for each inner Observable to complete before starting the next one. This maintains strict ordering and is ideal for operations that must happen sequentially, like saving items to a database in order.

Code Example:

import { of, interval, fromEvent } from 'rxjs';
import { map, mergeMap, switchMap, concatMap, take, delay } from 'rxjs/operators';

// Helper function to simulate API call
const apiCall = (id: number, delayMs: number) =>
  of(`Response for ${id}`).pipe(delay(delayMs));

// ============================================
// 1. map - Simple value transformation
// ============================================
of(1, 2, 3).pipe(
  map(x => x * 2) // Returns a plain value
).subscribe(console.log);
// Output: 2, 4, 6 (immediate)

// âťŚ This would be WRONG - map expects plain values, not Observables
// of(1, 2, 3).pipe(
//   map(x => of(x * 2)) // Returns Observable - creates Observable<Observable<number>>
// ).subscribe(console.log);
// Output: Observable, Observable, Observable (nested Observables!)

// ============================================
// 2. mergeMap - Concurrent execution
// ============================================
console.log('=== mergeMap ===');
of(1, 2, 3).pipe(
  mergeMap(x => apiCall(x, (4 - x) * 1000)) // 3s, 2s, 1s delays
).subscribe(console.log);
// Output order: Response for 3, Response for 2, Response for 1
// All requests run concurrently; fastest completes first

// Real-world: Loading multiple resources in parallel
const userIds$ = of(101, 102, 103);
userIds$.pipe(
  mergeMap(id => fetch(`/api/users/${id}`).then(r => r.json()))
).subscribe(user => console.log('User loaded:', user));
// All API calls fire immediately and resolve as they complete

// ============================================
// 3. switchMap - Cancel previous on new emission
// ============================================
console.log('=== switchMap ===');
of(1, 2, 3).pipe(
  switchMap(x => apiCall(x, (4 - x) * 1000))
).subscribe(console.log);
// Output: Response for 3
// Requests for 1 and 2 are cancelled when newer values arrive

// Real-world: Search typeahead
const searchInput$ = fromEvent<InputEvent>(
  document.getElementById('search'),
  'input'
);

searchInput$.pipe(
  map(event => (event.target as HTMLInputElement).value),
  switchMap(searchTerm =>
    fetch(`/api/search?q=${searchTerm}`).then(r => r.json())
  )
).subscribe(results => displayResults(results));
// Previous search requests are cancelled when user keeps typing

// ============================================
// 4. concatMap - Sequential execution
// ============================================
console.log('=== concatMap ===');
of(1, 2, 3).pipe(
  concatMap(x => apiCall(x, (4 - x) * 1000))
).subscribe(console.log);
// Output order: Response for 1, Response for 2, Response for 3
// Each request waits for the previous one to complete

// Real-world: Sequential database writes
interface SaveOperation {
  id: number;
  data: any;
}

const operations$ = of<SaveOperation>(
  { id: 1, data: 'First' },
  { id: 2, data: 'Second' },
  { id: 3, data: 'Third' }
);

operations$.pipe(
  concatMap(op =>
    fetch('/api/save', {
      method: 'POST',
      body: JSON.stringify(op)
    }).then(r => r.json())
  )
).subscribe(result => console.log('Saved:', result));
// Each save waits for the previous one to complete

// ============================================
// Side-by-side comparison with timing
// ============================================
const source$ = interval(1000).pipe(take(3)); // Emits 0, 1, 2 at 0s, 1s, 2s

// Each inner Observable takes 2 seconds
const innerObs = (x: number) => of(`Result: ${x}`).pipe(delay(2000));

console.log('Starting at:', new Date().toISOString());

// mergeMap: All run concurrently
source$.pipe(mergeMap(innerObs)).subscribe(
  val => console.log('mergeMap:', val, new Date().toISOString())
);
// Output at ~2s: Result: 0
// Output at ~3s: Result: 1
// Output at ~4s: Result: 2

// switchMap: Only last completes
source$.pipe(switchMap(innerObs)).subscribe(
  val => console.log('switchMap:', val, new Date().toISOString())
);
// Output at ~4s: Result: 2 (only the last one, others cancelled)

// concatMap: Sequential queue
source$.pipe(concatMap(innerObs)).subscribe(
  val => console.log('concatMap:', val, new Date().toISOString())
);
// Output at ~2s: Result: 0
// Output at ~4s: Result: 1
// Output at ~6s: Result: 2

function displayResults(results: any) {
  console.log('Search results:', results);
}

Mermaid Diagram:

flowchart TD
    Start[Source emits: 1, 2, 3]

    Start --> Map[map x => x * 2]
    Map --> MapOut[Output: 2, 4, 6]

    Start --> Merge[mergeMap x => apiCall x]
    Merge --> MergeInner1[Inner Observable 1 âŹ±ď¸Ź3s]
    Merge --> MergeInner2[Inner Observable 2 âŹ±ď¸Ź2s]
    Merge --> MergeInner3[Inner Observable 3 âŹ±ď¸Ź1s]
    MergeInner1 --> MergeOut[3, 2, 1 as completed]
    MergeInner2 --> MergeOut
    MergeInner3 --> MergeOut

    Start --> Switch[switchMap x => apiCall x]
    Switch --> SwitchInner1[Inner Observable 1 âťŚ cancelled]
    Switch --> SwitchInner2[Inner Observable 2 âťŚ cancelled]
    Switch --> SwitchInner3[Inner Observable 3 âś… active]
    SwitchInner3 --> SwitchOut[Only: 3]

    Start --> Concat[concatMap x => apiCall x]
    Concat --> ConcatInner1[Inner Observable 1 âŹ±ď¸Ź wait]
    ConcatInner1 --> ConcatInner2[Inner Observable 2 âŹ±ď¸Ź wait]
    ConcatInner2 --> ConcatInner3[Inner Observable 3 âŹ±ď¸Ź wait]
    ConcatInner3 --> ConcatOut[1, 2, 3 in order]

    style Map fill:#e1f5ff
    style Merge fill:#fff4e1
    style Switch fill:#ffe1f5
    style Concat fill:#e8f5e9
    style SwitchInner1 fill:#ffcdd2
    style SwitchInner2 fill:#ffcdd2

References:

↑ Back to top

Combination Operators

What is the difference between merge, concat, and forkJoin?

The 30-Second Answer: merge subscribes to all observables simultaneously and emits values as they arrive, concat subscribes to observables sequentially (waiting for each to complete before moving to the next), and forkJoin waits for all observables to complete then emits an array of their final values. Choose merge for parallel execution, concat for sequential ordering, and forkJoin for parallel execution with a single combined result.

The 2-Minute Answer (If They Want More): These three operators represent fundamentally different strategies for combining multiple observables, each suited to specific use cases.

merge is the "free-for-all" operator - it subscribes to all input observables immediately and forwards emissions as they happen in real-time. If you have three HTTP requests and use merge, all three fire simultaneously, and results appear in whatever order they complete. This is perfect when you want parallel execution and care about individual results as they arrive, like displaying search results from multiple sources or handling multiple user input streams.

concat enforces strict sequential ordering - it subscribes to the first observable and waits for it to complete before moving to the second, then the third, and so on. This is essential when order matters or when later observables depend on earlier ones completing. For example, authenticating a user before fetching their profile, or processing a queue of operations that must happen in sequence.

forkJoin is the "Promise.all" of RxJS - it subscribes to all observables in parallel (like merge) but doesn't emit anything until every single observable completes. Then it emits one array containing the last value from each observable. This is ideal when you need multiple independent async operations to finish before proceeding, like loading all required data for a dashboard or waiting for multiple API calls before rendering a page. Be warned: if any observable never completes or errors, forkJoin will never emit.

Code Example:

import { merge, concat, forkJoin, of, delay } from 'rxjs';

// Three observables that complete at different times
const fast$ = of('Fast').pipe(delay(100));
const medium$ = of('Medium').pipe(delay(300));
const slow$ = of('Slow').pipe(delay(500));

// MERGE: All start immediately, emit as they complete
// Output: "Fast" (100ms), "Medium" (300ms), "Slow" (500ms)
merge(fast$, medium$, slow$).subscribe(val =>
  console.log('merge:', val)
);

// CONCAT: Sequential execution, wait for each to complete
// Output: "Fast" (100ms), "Medium" (400ms total), "Slow" (900ms total)
concat(fast$, medium$, slow$).subscribe(val =>
  console.log('concat:', val)
);

// FORKJOIN: Wait for all to complete, emit array of final values
// Output: ["Fast", "Medium", "Slow"] (500ms - when slowest completes)
forkJoin([fast$, medium$, slow$]).subscribe(val =>
  console.log('forkJoin:', val)
);

// Real-world example: Loading dashboard data
interface DashboardData {
  user: User;
  stats: Stats;
  notifications: Notification[];
}

function loadDashboard(): Observable<DashboardData> {
  return forkJoin({
    user: this.userService.getCurrentUser(),
    stats: this.analyticsService.getStats(),
    notifications: this.notificationService.getRecent()
  }).pipe(
    map(data => ({
      user: data.user,
      stats: data.stats,
      notifications: data.notifications
    }))
  );
}

// Real-world example: Sequential authentication flow
function authenticateAndLoad(): Observable<UserProfile> {
  return concat(
    this.authService.login(credentials),  // Wait for login
    this.userService.fetchProfile(),      // Then fetch profile
    this.settingsService.loadPreferences() // Then load settings
  );
}

// Real-world example: Search multiple sources in parallel
function searchEverywhere(query: string): Observable<SearchResult> {
  return merge(
    this.searchService.searchDocuments(query),
    this.searchService.searchUsers(query),
    this.searchService.searchProjects(query)
  ).pipe(
    // Results appear as each search completes
    tap(result => this.displayResult(result))
  );
}

Mermaid Diagram:

flowchart TD
    subgraph "MERGE - Parallel, Emit As They Arrive"
        M1[Observable 1] -->|emit immediately| MO[Output]
        M2[Observable 2] -->|emit immediately| MO
        M3[Observable 3] -->|emit immediately| MO
    end

    subgraph "CONCAT - Sequential, One After Another"
        C1[Observable 1] -->|complete| C2[Observable 2]
        C2 -->|complete| C3[Observable 3]
        C3 --> CO[Output]
    end

    subgraph "FORKJOIN - Parallel, Wait For All"
        F1[Observable 1] -.->|wait| FW[Wait for ALL]
        F2[Observable 2] -.->|wait| FW
        F3[Observable 3] -.->|wait| FW
        FW -->|all complete| FO[Output Array]
    end

References:

↑ Back to top

Error Handling

What is the difference between retry and retryWhen?

The 30-Second Answer: The retry operator automatically resubscribes to the source Observable a fixed number of times when errors occur, using immediate retries. The retryWhen operator provides fine-grained control over retry logic, allowing you to implement delays, exponential backoff, conditional retries, and custom retry strategies based on the error itself.

The 2-Minute Answer (If They Want More): Both retry and retryWhen are used to handle transient failures by resubscribing to the source Observable, but they differ significantly in control and flexibility.

retry is the simpler operator. You provide a count (or leave it empty for infinite retries), and it automatically resubscribes to the source Observable immediately when an error occurs. This is useful for simple scenarios where you want to retry a fixed number of times without delay, such as quick network requests where immediate retry might succeed.

retryWhen, on the other hand, gives you complete control over the retry mechanism. It takes a callback function that receives an Observable of errors. You can transform this error stream to control when and if retries happen. This allows you to implement sophisticated strategies like exponential backoff (increasing delay between retries), conditional retries (only retry certain error types), maximum retry limits with custom logic, or even user-prompted retries.

The key architectural difference is that retry is declarative and simple, while retryWhen is imperative and powerful. With retryWhen, you can add delays between retries using delay, limit retries using take, implement exponential backoff using scan, or even combine retry logic with other operators. This makes retryWhen essential for production applications where you need resilient, intelligent retry behavior that doesn't overwhelm failing services.

Code Example:

import {
  retry, retryWhen, delay, take, scan, tap, mergeMap,
  throwError, timer, of
} from 'rxjs';
import { ajax } from 'rxjs/ajax';

// 1. Simple retry - immediate retries
const simpleRetry$ = ajax.getJSON('/api/data').pipe(
  retry(3), // Retry immediately up to 3 times
  catchError(error => {
    console.error('Failed after 3 immediate retries');
    return of({ error: true });
  })
);

// 2. Conditional retry - only retry specific errors
const conditionalRetry$ = ajax.getJSON('/api/data').pipe(
  retry({
    count: 3,
    // Only retry on network errors, not 404s
    resetOnSuccess: true
  }),
  catchError(error => of({ error: true }))
);

// 3. retryWhen with fixed delay
const fixedDelayRetry$ = ajax.getJSON('/api/data').pipe(
  retryWhen(errors =>
    errors.pipe(
      tap(error => console.log('Error occurred, retrying in 2s:', error)),
      delay(2000), // Wait 2 seconds before retry
      take(3) // Only retry 3 times
    )
  ),
  catchError(error => of({ error: true }))
);

// 4. Exponential backoff with retryWhen
const exponentialBackoff$ = ajax.getJSON('/api/data').pipe(
  retryWhen(errors =>
    errors.pipe(
      // Track retry count and calculate delay
      scan((retryCount, error) => {
        if (retryCount >= 4) {
          throw error; // Max retries exceeded
        }
        return retryCount + 1;
      }, 0),
      tap(retryCount => {
        const delayMs = Math.pow(2, retryCount) * 1000;
        console.log(`Retry attempt ${retryCount}, waiting ${delayMs}ms`);
      }),
      // Exponential delay: 2s, 4s, 8s, 16s
      mergeMap(retryCount =>
        timer(Math.pow(2, retryCount) * 1000)
      )
    )
  ),
  catchError(error => {
    console.error('Max retries with exponential backoff exceeded');
    return of({ error: true });
  })
);

// 5. Conditional retry based on error type
const smartRetry$ = ajax.getJSON('/api/data').pipe(
  retryWhen(errors =>
    errors.pipe(
      mergeMap((error, index) => {
        // Don't retry client errors (4xx)
        if (error.status >= 400 && error.status < 500) {
          return throwError(() => error);
        }

        // Retry server errors (5xx) with delay
        if (index < 3) {
          console.log(`Retrying server error (attempt ${index + 1})`);
          return timer(1000 * (index + 1));
        }

        // Max retries exceeded
        return throwError(() => error);
      })
    )
  ),
  catchError(error => of({ error: true, status: error.status }))
);

// 6. Advanced: Exponential backoff with jitter
const exponentialBackoffWithJitter$ = ajax.getJSON('/api/data').pipe(
  retryWhen(errors =>
    errors.pipe(
      scan((acc, error) => ({
        count: acc.count + 1,
        error
      }), { count: 0, error: null }),
      tap(({ count }) => {
        if (count > 5) {
          throw new Error('Max retries exceeded');
        }
      }),
      mergeMap(({ count }) => {
        // Exponential backoff: 2^count * 1000ms
        const exponentialDelay = Math.pow(2, count) * 1000;
        // Add jitter: random 0-1000ms to prevent thundering herd
        const jitter = Math.random() * 1000;
        const totalDelay = exponentialDelay + jitter;

        console.log(`Retry ${count}, delay: ${totalDelay.toFixed(0)}ms`);
        return timer(totalDelay);
      })
    )
  ),
  catchError(error => of({ error: true }))
);

// 7. Comparison: retry vs retryWhen for same scenario
// Using retry (simple, immediate)
const withRetry$ = ajax.getJSON('/api/data').pipe(
  retry(3)
);

// Using retryWhen (with control over retry logic)
const withRetryWhen$ = ajax.getJSON('/api/data').pipe(
  retryWhen(errors =>
    errors.pipe(
      take(3) // Same 3 retries, but we could add delay/logic
    )
  )
);

Mermaid Diagram:

flowchart TD
    A[Source Observable] --> B{Error Occurs}

    B -->|retry| C[Immediate Resubscribe]
    C --> D{Retry Count < Max?}
    D -->|Yes| A
    D -->|No| E[Propagate Error]

    B -->|retryWhen| F[Error Stream]
    F --> G[Custom Logic/Transform]
    G --> H{Should Retry?}
    H -->|Yes + Delay| I[Wait for Timer]
    I --> A
    H -->|Yes + Immediate| A
    H -->|No| E

    G --> J[Add Delay]
    G --> K[Check Error Type]
    G --> L[Exponential Backoff]
    G --> M[Limit Attempts]

    style C fill:#ffcccc
    style G fill:#ccffcc

References:

↑ Back to top

Observables and Subscriptions

What is a Subscription and how do you manage it?

The 30-Second Answer: A Subscription represents the execution of an Observable and provides the unsubscribe() method to cancel that execution and free up resources. You manage subscriptions by storing the returned subscription object and calling unsubscribe() when done, or by using operators like takeUntil() for automatic cleanup.

The 2-Minute Answer (If They Want More): When you call subscribe() on an Observable, it returns a Subscription object that represents the ongoing execution. This subscription maintains the connection between the Observable and the observer, and continues to deliver values until the Observable completes, errors, or you manually unsubscribe.

Managing subscriptions is critical to prevent memory leaks in your application. Every active subscription consumes memory and potentially holds references to DOM elements, event listeners, or timers. In long-lived applications like SPAs, forgetting to unsubscribe can lead to significant performance degradation and bugs.

There are several strategies for managing subscriptions: storing them in component properties and unsubscribing in lifecycle hooks, using a Subscription container to group multiple subscriptions, leveraging RxJS operators like takeUntil() to automatically complete streams, or using Angular's async pipe which handles subscriptions automatically.

The most robust pattern is the takeUntil() approach with a destroy subject. You create a subject that emits when the component is destroyed, and pipe all your subscriptions through takeUntil(destroy$). This centralizes cleanup logic and ensures no subscriptions are forgotten, especially as your component grows in complexity.

Code Example:

import { Component, OnDestroy } from '@angular/core';
import { Subscription, Subject, interval, fromEvent } from 'rxjs';
import { takeUntil, take } from 'rxjs/operators';

@Component({
  selector: 'app-example',
  template: `<div>Check console for subscription management</div>`
})
export class ExampleComponent implements OnDestroy {
  // Method 1: Store individual subscriptions
  private subscription1: Subscription;
  private subscription2: Subscription;

  // Method 2: Subscription container
  private subscriptions = new Subscription();

  // Method 3: takeUntil pattern (RECOMMENDED)
  private destroy$ = new Subject<void>();

  constructor() {
    this.demonstrateSubscriptionManagement();
  }

  demonstrateSubscriptionManagement() {
    // Method 1: Manual subscription management
    this.subscription1 = interval(1000).subscribe(
      value => console.log('Interval:', value)
    );

    // Method 2: Add to subscription container
    this.subscriptions.add(
      interval(2000).subscribe(value => console.log('Timer:', value))
    );

    // Another subscription added to container
    this.subscriptions.add(
      fromEvent(document, 'click').subscribe(
        event => console.log('Click:', event)
      )
    );

    // Method 3: takeUntil pattern (automatically unsubscribes)
    interval(500)
      .pipe(takeUntil(this.destroy$))
      .subscribe(value => console.log('Auto-cleanup:', value));

    // Method 4: Self-completing Observables (no cleanup needed)
    interval(1000)
      .pipe(take(5)) // Completes after 5 emissions
      .subscribe(value => console.log('Self-completing:', value));
  }

  ngOnDestroy() {
    // Clean up method 1: Individual unsubscribe
    if (this.subscription1) {
      this.subscription1.unsubscribe();
    }
    if (this.subscription2) {
      this.subscription2.unsubscribe();
    }

    // Clean up method 2: Unsubscribe all at once
    this.subscriptions.unsubscribe();

    // Clean up method 3: Trigger takeUntil completion
    this.destroy$.next();
    this.destroy$.complete();
  }
}

// Advanced: Subscription utility service
class SubscriptionService {
  private subscriptions = new Map<string, Subscription>();

  add(key: string, subscription: Subscription): void {
    if (this.subscriptions.has(key)) {
      this.subscriptions.get(key)?.unsubscribe();
    }
    this.subscriptions.set(key, subscription);
  }

  unsubscribe(key: string): void {
    const sub = this.subscriptions.get(key);
    if (sub) {
      sub.unsubscribe();
      this.subscriptions.delete(key);
    }
  }

  unsubscribeAll(): void {
    this.subscriptions.forEach(sub => sub.unsubscribe());
    this.subscriptions.clear();
  }
}

Mermaid Diagram:

flowchart TD
    A[Observable.subscribe] --> B[Subscription Created]
    B --> C{Active Subscription}
    C -->|Values Emitted| D[observer.next]
    C -->|Observable Completes| E[Auto Cleanup]
    C -->|Observable Errors| F[Auto Cleanup]
    C -->|Manual Unsubscribe| G[subscription.unsubscribe]
    C -->|takeUntil Triggers| H[Auto Unsubscribe]
    D --> C
    E --> I[Resources Freed]
    F --> I
    G --> I
    H --> I
    I --> J[Memory Released]

References:

↑ Back to top

Higher-Order Observables

What is the mergeAll operator?

The 30-Second Answer: mergeAll is a flattening operator that subscribes to all inner Observables concurrently and merges their emissions into a single output stream. Unlike map which returns Observables, mergeAll automatically subscribes to them and flattens the results - it's essentially what mergeMap does after the mapping function.

The 2-Minute Answer (If They Want More): mergeAll is the "lower-level" building block that powers mergeMap. When you have a higher-order Observable (an Observable emitting Observables), mergeAll subscribes to each inner Observable as it arrives and merges all their values into the output stream. It handles all inner Observables concurrently with no limit by default, though you can specify a concurrency limit.

The relationship between mergeMap and mergeAll is simple: mergeMap(fn) is equivalent to map(fn) followed by mergeAll(). Understanding this helps demystify the "Map" operators - they're just map + the corresponding "All" operator. For example, switchMap = map + switchAll, concatMap = map + concatAll, and exhaustMap = map + exhaustAll.

You typically use mergeMap in practice because it combines the mapping and flattening in one step. However, mergeAll is useful when you already have a higher-order Observable from another source, or when you want to explicitly separate the mapping logic from the flattening strategy for clarity.

Code Example:

import { interval, fromEvent, of } from 'rxjs';
import { map, take, mergeAll, mergeMap } from 'rxjs/operators';

// Example 1: Basic mergeAll - flattening a higher-order Observable
const clicks$ = fromEvent(document, 'click');

// Without mergeAll - you get Observable objects
const higherOrder$ = clicks$.pipe(
  map(() => interval(1000).pipe(take(3)))
);

higherOrder$.subscribe(obs => {
  console.log('Got Observable:', obs); // Observable {_subscribe: Ć’}
});

// With mergeAll - you get the actual values
const flattened$ = clicks$.pipe(
  map(() => interval(1000).pipe(take(3))),
  mergeAll() // Subscribe to all inner Observables concurrently
);

flattened$.subscribe(value => {
  console.log('Got value:', value); // 0, 1, 2 (from each click)
});

// Example 2: mergeMap vs map + mergeAll (they're equivalent)
const ids = [1, 2, 3];

// Using mergeMap (common approach)
of(...ids).pipe(
  mergeMap(id => fetch(`/api/user/${id}`).then(r => r.json()))
).subscribe(user => console.log('User (mergeMap):', user));

// Using map + mergeAll (explicit separation)
of(...ids).pipe(
  map(id => fetch(`/api/user/${id}`).then(r => r.json())),
  mergeAll() // Flatten the Observables from map
).subscribe(user => console.log('User (map+mergeAll):', user));

// Example 3: Concurrency control with mergeAll
const urls = [
  '/api/data/1',
  '/api/data/2',
  '/api/data/3',
  '/api/data/4',
  '/api/data/5'
];

// Limit to 2 concurrent HTTP requests at a time
of(...urls).pipe(
  map(url => fetch(url).then(r => r.json())),
  mergeAll(2) // Only 2 concurrent inner subscriptions
).subscribe(data => {
  console.log('Data received:', data);
});

// Example 4: Real-world use case - processing multiple file uploads
interface UploadResult {
  fileId: string;
  url: string;
  status: 'success' | 'error';
}

const uploadFile = (file: File): Promise<UploadResult> => {
  const formData = new FormData();
  formData.append('file', file);

  return fetch('/api/upload', {
    method: 'POST',
    body: formData
  }).then(r => r.json());
};

const fileInput = document.getElementById('file-input') as HTMLInputElement;
const uploadButton = document.getElementById('upload-btn') as HTMLButtonElement;

fromEvent(uploadButton, 'click').pipe(
  map(() => {
    const files = Array.from(fileInput.files || []);
    // Create an array of upload Observables
    return files.map(file => uploadFile(file));
  }),
  // Flatten and execute all uploads concurrently
  mergeAll() // Could use mergeAll(3) to limit to 3 concurrent uploads
).subscribe({
  next: (result: UploadResult) => {
    console.log(`File ${result.fileId} uploaded:`, result.url);
  },
  error: err => console.error('Upload failed:', err),
  complete: () => console.log('All uploads complete')
});

// Example 5: Combining multiple Observable sources
const source1$ = interval(1000).pipe(take(3), map(x => `A${x}`));
const source2$ = interval(1500).pipe(take(3), map(x => `B${x}`));
const source3$ = interval(2000).pipe(take(3), map(x => `C${x}`));

// Merge all sources into one stream
of(source1$, source2$, source3$).pipe(
  mergeAll()
).subscribe(value => {
  console.log('Merged value:', value);
  // Output (interleaved): A0, A1, B0, A2, B1, C0, B2, C1, C2
});

Mermaid Diagram:

flowchart TD
    A[Outer Observable] -->|emits| B[Inner Obs 1: --1--2--3--|]
    A -->|emits| C[Inner Obs 2: --4--5--|]
    A -->|emits| D[Inner Obs 3: --6--7--8--|]

    B --> E[mergeAll]
    C --> E
    D --> E

    E -->|concurrent merge| F[Output: --1--4--2--6--3--5--7--8--|]

    G[Alternative: mergeAll concurrent=1] --> H[Output: --1--2--3--4--5--6--7--8--|]

    style A fill:#e1f5ff
    style B fill:#fff4e1
    style C fill:#fff4e1
    style D fill:#fff4e1
    style E fill:#e1ffe1
    style F fill:#ffe1f5
    style G fill:#e1ffe1
    style H fill:#ffe1f5

References:

↑ Back to top

Best Practices and Performance

What is the takeUntil pattern for managing subscriptions?

The 30-Second Answer: The takeUntil pattern uses a notifier Observable (typically a Subject) that emits when you want to unsubscribe from a stream. By piping .pipe(takeUntil(notifier$)) before subscribing, the subscription automatically completes when the notifier emits, providing a centralized and declarative way to manage subscription cleanup in components or services.

The 2-Minute Answer (If They Want More): The takeUntil pattern is the most elegant and recommended approach for managing subscription lifecycles in Angular and other RxJS applications. Instead of manually tracking and unsubscribing from individual subscriptions, you create a single Subject (commonly named destroy$) that acts as a cleanup signal. All Observables in your component or service pipe through takeUntil(this.destroy$) before subscribing, which means they'll automatically complete when the destroy Subject emits.

This pattern shines in Angular components where you have multiple subscriptions that all need cleanup when the component is destroyed. In the ngOnDestroy lifecycle hook, you simply call this.destroy$.next() and this.destroy$.complete(), which triggers completion of all subscriptions using that notifier. This eliminates the need for subscription arrays, manual tracking, or calling unsubscribe multiple times.

The pattern is declarative and self-documentingâ€”when you see takeUntil(this.destroy$), you immediately understand the subscription's lifecycle. It also composes well with other operators and prevents the common pitfall of forgetting to unsubscribe from one of many subscriptions. The operator works by subscribing to both the source Observable and the notifier; when the notifier emits any value, takeUntil unsubscribes from the source and completes the stream.

Important considerations: always place takeUntil as the last operator in your pipe (except for operators that must run during cleanup like finalize), complete the notifier Subject in addition to emitting from it to prevent memory leaks of the notifier itself, and remember that takeUntil doesn't help with synchronous emissions that occur before the pipe is set up.

Code Example:

import { Component, OnDestroy, Injectable } from '@angular/core';
import { Subject, interval, fromEvent, merge } from 'rxjs';
import { takeUntil, switchMap, tap, finalize } from 'rxjs/operators';

// BASIC PATTERN - Component example
@Component({
  selector: 'app-user-dashboard',
  template: `
    <div>{{ userCount }}</div>
    <button (click)="loadData()">Load</button>
  `
})
class UserDashboardComponent implements OnDestroy {
  userCount = 0;

  // Create the notifier Subject
  private destroy$ = new Subject<void>();

  constructor(
    private userService: UserService,
    private analyticsService: AnalyticsService
  ) {
    // All subscriptions use takeUntil
    this.setupAutoRefresh();
    this.setupAnalytics();
  }

  private setupAutoRefresh() {
    // Auto-refresh every 30 seconds
    interval(30000)
      .pipe(
        switchMap(() => this.userService.getUsers()),
        takeUntil(this.destroy$) // Stops when component destroyed
      )
      .subscribe(users => {
        this.userCount = users.length;
      });
  }

  private setupAnalytics() {
    // Track user interactions
    merge(
      fromEvent(document, 'click'),
      fromEvent(document, 'scroll')
    )
      .pipe(
        takeUntil(this.destroy$) // Both event streams stop together
      )
      .subscribe(event => {
        this.analyticsService.track(event);
      });
  }

  loadData() {
    this.userService.getUsers()
      .pipe(
        takeUntil(this.destroy$) // Even manual calls are protected
      )
      .subscribe(users => {
        this.userCount = users.length;
      });
  }

  ngOnDestroy() {
    // Single cleanup point for all subscriptions
    this.destroy$.next(); // Emit to trigger takeUntil
    this.destroy$.complete(); // Complete the subject itself
  }
}

// ADVANCED PATTERN - Service with multiple destroy scenarios
@Injectable()
class DataCacheService implements OnDestroy {
  private destroy$ = new Subject<void>();
  private resetCache$ = new Subject<void>();

  // Notifier that combines destroy OR cache reset
  private stopSignal$ = merge(this.destroy$, this.resetCache$);

  private cache$ = this.http.get('/api/data')
    .pipe(
      // Stream stops on destroy OR reset
      takeUntil(this.stopSignal$),
      shareReplay({ bufferSize: 1, refCount: true })
    );

  getData() {
    return this.cache$;
  }

  resetCache() {
    this.resetCache$.next(); // Triggers takeUntil in cache$
  }

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

// PATTERN WITH MULTIPLE NOTIFIERS - Different lifecycles
@Component({
  selector: 'app-multi-lifecycle',
  template: '<div>Multi-lifecycle example</div>'
})
class MultiLifecycleComponent implements OnDestroy {
  private destroy$ = new Subject<void>();
  private userLogout$ = new Subject<void>();
  private tabChange$ = new Subject<void>();

  constructor(private authService: AuthService) {
    // Some subscriptions end on component destroy
    interval(1000)
      .pipe(takeUntil(this.destroy$))
      .subscribe(val => console.log('Heartbeat:', val));

    // Some subscriptions end when user logs out
    this.authService.getUserData()
      .pipe(
        takeUntil(merge(this.destroy$, this.userLogout$))
      )
      .subscribe(userData => console.log('User data:', userData));

    // Some subscriptions end when tab changes
    this.loadTabData()
      .pipe(
        takeUntil(merge(this.destroy$, this.tabChange$))
      )
      .subscribe(data => console.log('Tab data:', data));
  }

  onTabChange() {
    this.tabChange$.next(); // Stops current tab subscriptions
  }

  onLogout() {
    this.userLogout$.next(); // Stops user-related subscriptions
  }

  ngOnDestroy() {
    this.destroy$.next(); // Stops all subscriptions
    this.destroy$.complete();
    this.userLogout$.complete();
    this.tabChange$.complete();
  }

  private loadTabData() {
    return interval(5000).pipe(map(i => `Tab data ${i}`));
  }
}

// OPERATOR ORDERING - takeUntil placement matters
class OperatorOrderingExample {
  private destroy$ = new Subject<void>();

  incorrectOrder() {
    interval(1000)
      .pipe(
        takeUntil(this.destroy$), // âťŚ WRONG - finalize won't run
        finalize(() => console.log('Cleanup'))
      )
      .subscribe();
  }

  correctOrder() {
    interval(1000)
      .pipe(
        finalize(() => console.log('Cleanup')), // âś“ Runs on unsubscribe
        takeUntil(this.destroy$) // âś“ Last operator (except finalize)
      )
      .subscribe();
  }

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

// BASE CLASS PATTERN - Reusable across components
abstract class BaseComponent implements OnDestroy {
  protected destroy$ = new Subject<void>();

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

@Component({
  selector: 'app-derived',
  template: '<div>Derived component</div>'
})
class DerivedComponent extends BaseComponent {
  constructor(private dataService: DataService) {
    super();

    // Can use destroy$ from base class
    this.dataService.getData()
      .pipe(takeUntil(this.destroy$))
      .subscribe(data => console.log(data));
  }
}

Mermaid Diagram:

flowchart TD
    A[Component Created] --> B[Create destroy$ Subject]
    B --> C[Setup Observable 1]
    B --> D[Setup Observable 2]
    B --> E[Setup Observable N]

    C --> F[pipe takeUntil destroy$]
    D --> G[pipe takeUntil destroy$]
    E --> H[pipe takeUntil destroy$]

    F --> I[Subscribe]
    G --> I
    H --> I

    I --> J{Component Active?}
    J -->|Yes| K[Observables Emit]
    K --> J
    J -->|No| L[ngOnDestroy Called]
    L --> M[destroy$.next]
    M --> N[All takeUntil Triggered]
    N --> O[All Subscriptions Complete]
    O --> P[destroy$.complete]

    style M fill:#ff6b6b
    style N fill:#ffd43b
    style O fill:#51cf66

References:

↑ Back to top

Item added to your cart