RxJS: Mastering Reactive Programming in JavaScript ⚡

Understanding RxJS 🤔

Modern JavaScript applications are deeply event-driven — user clicks, HTTP responses, WebSocket messages, timers — all of these are streams of data arriving over time. Managing them with plain callbacks or even Promises can quickly become messy, hard to compose, and even harder to cancel.

This is exactly the problem RxJS (Reactive Extensions for JavaScript) was built to solve. Instead of thinking about data as a single value you fetch once, RxJS encourages you to think about data as a stream — a sequence of values that arrives over time — and gives you a powerful toolkit to transform, filter, combine, and control those streams declaratively.

Whether you're building an Angular application (where RxJS is a first-class citizen), a React app that needs sophisticated async flows, or a Node.js backend, RxJS gives you a composable, consistent model for async and event-based programming.

RxJS Explained 📖

At its heart, RxJS implements the Observer pattern combined with functional programming concepts. The core idea is simple:

A producer emits values over time. A consumer subscribes to receive those values.

RxJS formalises this with a few key concepts:

To install RxJS:

npm install rxjs

Core Building Blocks 🧱

1. Observable

An Observable is the foundation. It represents a stream of zero or more values delivered synchronously or asynchronously.

import { Observable } from 'rxjs';
 
const myObservable$ = new Observable((subscriber) => {
  subscriber.next('Hello');
  subscriber.next('World');
  setTimeout(() => {
    subscriber.next('...after 1 second');
    subscriber.complete();
  }, 1000);
});
 
// Nothing happens until we subscribe!
myObservable$.subscribe({
  next: (value) => console.log('Received:', value),
  error: (err) => console.error('Error:', err),
  complete: () => console.log('Stream complete ✅'),
});
 
// Output:
// Received: Hello
// Received: World
// Received: ...after 1 second
// Stream complete ✅

💡 Key insight: Observables are lazy. They don't execute until you call .subscribe().

2. Subject

A Subject is both an Observable and an Observer. It can multicast — meaning multiple subscribers receive the same emitted values.

import { Subject } from 'rxjs';
 
const subject$ = new Subject();
 
// Two different subscribers
subject$.subscribe((val) => console.log('Subscriber A:', val));
subject$.subscribe((val) => console.log('Subscriber B:', val));
 
subject$.next(1);
subject$.next(2);
 
// Output:
// Subscriber A: 1
// Subscriber B: 1
// Subscriber A: 2
// Subscriber B: 2

RxJS also provides specialised Subject variants:

• BehaviorSubject — stores the latest value and replays it to new subscribers
• ReplaySubject — replays a configurable number of past values to new subscribers
• AsyncSubject — only emits the last value when the sequence completes

import { BehaviorSubject } from 'rxjs';
 
const currentUser$ = new BehaviorSubject({ name: 'Guest' });
 
// New subscriber immediately gets the current value
currentUser$.subscribe((user) => console.log('User:', user.name));
// Output: User: Guest
 
currentUser$.next({ name: 'Alice' });
// Output: User: Alice

3. Operators

Operators are pure functions that take an Observable as input and return a new Observable. They never modify the original stream — they create a new one. You chain them using the .pipe() method.

import { of } from 'rxjs';
import { map, filter } from 'rxjs/operators';
 
of(1, 2, 3, 4, 5, 6)
  .pipe(
    filter((n) => n % 2 === 0),   // Only even numbers
    map((n) => n * 10)             // Multiply by 10
  )
  .subscribe((val) => console.log(val));
 
// Output: 20, 40, 60

4. Creation Operators

RxJS provides many utility functions to create Observables from common sources:

import { of, from, interval, fromEvent, timer } from 'rxjs';
 
// Emit a fixed sequence of values
of('a', 'b', 'c').subscribe(console.log);
 
// Convert a Promise or array into an Observable
from(fetch('/api/data')).subscribe(console.log);
 
// Emit incrementing number every 1 second
interval(1000).subscribe((n) => console.log(`Tick: ${n}`));
 
// Emit after 3 seconds, then every 1 second
timer(3000, 1000).subscribe(console.log);
 
// Turn a DOM event into a stream
fromEvent(document, 'click').subscribe((e) => console.log('Clicked!', e));

5. Subscription & Unsubscription

Every .subscribe() call returns a Subscription object. Always unsubscribe when you're done to prevent memory leaks — especially in components.

import { interval } from 'rxjs';
 
const sub = interval(500).subscribe((n) => console.log(n));
 
// Unsubscribe after 3 seconds
setTimeout(() => {
  sub.unsubscribe();
  console.log('Unsubscribed!');
}, 3000);

Common RxJS Operators Explained 🛠️

This is where the real power of RxJS shines. These higher-order mapping operators are among the most important — and most misunderstood — in the library.

map 🔄

The simplest transformation operator. Applies a function to each emitted value and emits the result — identical in concept to Array.prototype.map.

import { of } from 'rxjs';
import { map } from 'rxjs/operators';
 
of(1, 2, 3)
  .pipe(map((n) => n * n))
  .subscribe(console.log);
// Output: 1, 4, 9

Use it when: You want to transform each value in a stream without any async work.

switchMap 🔀

switchMap maps each value to an inner Observable, but cancels the previous inner Observable whenever a new value arrives. It only ever subscribes to the most recent inner stream.

import { fromEvent, interval } from 'rxjs';
import { switchMap, take } from 'rxjs/operators';
 
// Every time the user clicks, start a fresh 3-second countdown
// If they click again before it finishes, the old one is cancelled
fromEvent(document, 'click')
  .pipe(
    switchMap(() => interval(1000).pipe(take(3)))
  )
  .subscribe((n) => console.log(`Tick: ${n}`));

Real-world use case: Autocomplete / typeahead search — cancel the previous HTTP request when the user keeps typing.

import { fromEvent } from 'rxjs';
import { debounceTime, distinctUntilChanged, switchMap, map } from 'rxjs/operators';
import { ajax } from 'rxjs/ajax';
 
const searchInput = document.getElementById('search');
 
fromEvent(searchInput, 'input')
  .pipe(
    map((e) => e.target.value),
    debounceTime(300),
    distinctUntilChanged(),
    switchMap((query) => ajax.getJSON(`/api/search?q=${query}`))
  )
  .subscribe((results) => console.log('Results:', results));

⚠️ Use switchMap when only the latest result matters. If the previous request is still in-flight, it gets cancelled.

mergeMap (flatMap) 🌊

mergeMap maps each value to an inner Observable and merges all inner Observables concurrently. Unlike switchMap, it does NOT cancel previous inner streams — it runs all of them in parallel.

import { of, interval } from 'rxjs';
import { mergeMap, take, map } from 'rxjs/operators';
 
of('A', 'B', 'C')
  .pipe(
    mergeMap((letter) =>
      interval(1000).pipe(
        take(3),
        map((n) => `${letter}: ${n}`)
      )
    )
  )
  .subscribe(console.log);
 
// All 3 inner Observables run concurrently — output is interleaved
// A: 0, B: 0, C: 0, A: 1, B: 1, C: 1 ...

Real-world use case: Fire multiple HTTP requests in parallel and handle each response as it arrives.

import { from } from 'rxjs';
import { mergeMap } from 'rxjs/operators';
import { ajax } from 'rxjs/ajax';
 
const userIds = [1, 2, 3, 4, 5];
 
from(userIds)
  .pipe(
    mergeMap((id) => ajax.getJSON(`/api/users/${id}`))
  )
  .subscribe((user) => console.log('Fetched user:', user));

⚠️ Order is NOT guaranteed. Responses arrive as they complete, not in the order they were requested.

concatMap 🚂

concatMap maps each value to an inner Observable, but queues them and subscribes to them one at a time, waiting for each to complete before starting the next. Order is always preserved.

import { of, timer } from 'rxjs';
import { concatMap, map } from 'rxjs/operators';
 
of('First', 'Second', 'Third')
  .pipe(
    concatMap((label) =>
      timer(1000).pipe(map(() => `${label} done`))
    )
  )
  .subscribe(console.log);
 
// Output (one per second, always in order):
// First done
// Second done
// Third done

Real-world use case: Sequential API calls where each depends on the previous completing, or uploading files one at a time.

import { from } from 'rxjs';
import { concatMap } from 'rxjs/operators';
import { ajax } from 'rxjs/ajax';
 
const operations = [
  { url: '/api/step1', body: { action: 'init' } },
  { url: '/api/step2', body: { action: 'process' } },
  { url: '/api/step3', body: { action: 'finalize' } },
];
 
from(operations)
  .pipe(
    concatMap((op) => ajax.post(op.url, op.body))
  )
  .subscribe((response) => console.log('Step complete:', response));

exhaustMap 🚫

exhaustMap maps the first value to an inner Observable and ignores all new values until that inner Observable completes. It's the opposite strategy to switchMap.

import { fromEvent, timer } from 'rxjs';
import { exhaustMap, map } from 'rxjs/operators';
 
const loginButton = document.getElementById('login-btn');
 
fromEvent(loginButton, 'click')
  .pipe(
    exhaustMap(() =>
      // Simulate a login request taking 2 seconds
      timer(2000).pipe(map(() => 'Login successful!'))
    )
  )
  .subscribe(console.log);
 
// No matter how many times the user clicks, only the FIRST click
// triggers a request. Subsequent clicks are ignored until it completes.

Real-world use case: Preventing double-submission of a form or duplicate login requests on button mashing.

💡 The strategies compared:

• switchMap → cancel previous, use latest
• mergeMap → run all concurrently
• concatMap → queue, run sequentially
• exhaustMap → ignore new while busy

combineLatestAll 🔗

combineLatestAll collects a set of inner Observables and combines their latest values into a single array whenever any of them emits. It waits for all inner Observables to emit at least once before emitting.

import { interval, of } from 'rxjs';
import { map, combineLatestAll, take } from 'rxjs/operators';
 
// Create an Observable that emits 3 inner Observables
of(1000, 2000, 3000)
  .pipe(
    map((ms) => interval(ms).pipe(take(5), map((n) => `Every ${ms}ms: ${n}`))),
    combineLatestAll()
  )
  .subscribe(console.log);
 
// Emits an array with the latest value from each inner Observable
// e.g. ['Every 1000ms: 3', 'Every 2000ms: 1', 'Every 3000ms: 0']

Real-world use case: Combining the latest state from multiple independent data streams (e.g., user preferences + product list + cart state) to drive a UI render.

import { combineLatest } from 'rxjs';
 
// Combine three separate streams into one
combineLatest([user$, products$, cart$])
  .subscribe(([user, products, cart]) => {
    renderDashboard(user, products, cart);
  });

concatAll 📋

concatAll subscribes to an Observable of Observables (a "higher-order Observable") and concatenates the inner streams in order, waiting for each to finish before subscribing to the next.

import { interval, of } from 'rxjs';
import { map, take, concatAll } from 'rxjs/operators';
 
of(1, 2, 3)
  .pipe(
    map((n) => interval(1000).pipe(take(2), map((i) => `Stream ${n}, tick ${i}`))),
    concatAll()
  )
  .subscribe(console.log);
 
// Output (strictly sequential):
// Stream 1, tick 0
// Stream 1, tick 1
// Stream 2, tick 0
// Stream 2, tick 1
// Stream 3, tick 0
// Stream 3, tick 1

💡 concatAll() is equivalent to concatMap((x) => x) — it just flattens a pre-existing higher-order Observable.

concatWith 🔗➡️

concatWith appends one or more Observables to the current one, subscribing to each only after the previous completes. It's a clean way to sequence multiple streams.

import { of, timer } from 'rxjs';
import { concatWith, map } from 'rxjs/operators';
 
of('Starting...').pipe(
  concatWith(
    timer(1000).pipe(map(() => 'Step 1 complete')),
    timer(1000).pipe(map(() => 'Step 2 complete')),
    of('All done! 🎉')
  )
).subscribe(console.log);
 
// Output:
// Starting...
// (1 second later) Step 1 complete
// (1 second later) Step 2 complete
// All done! 🎉

Real-world use case: Chaining an initialisation stream with a data-loading stream and then a "ready" notification.

Conclusion 🎯

RxJS is one of the most powerful tools in the JavaScript ecosystem for managing asynchronous and event-driven code. While the learning curve can feel steep at first, mastering just a handful of operators opens up elegant solutions to problems that would otherwise require complex, fragile imperative code.

Here's a quick decision guide for the operators we covered:

The best way to master RxJS is to use it on real problems and to visualise streams. Don't try to learn all 100+ operators at once — start with map, filter, switchMap, and combineLatest, and you'll cover 80% of everyday use cases.

📚 Further Learning

Keep exploring with these excellent resources:

• 📖 Official RxJS Documentation — the definitive reference for every operator and concept
• 🎓 Learn RxJS — clear operator examples with real-world use cases
• 🎨 RxMarbles — interactive marble diagrams to visualise how operators behave
• 🔬 RxViz — animated visualisation of RxJS Observables in the browser
• 🎥 RxJS Crash Course (YouTube) — a beginner-friendly video series
• 🅰️ RxJS & Angular — A Deep Dive — how Angular uses RxJS under the hood
• ⭐ RxJS on GitHub — browse source code and contribute

💬 Have questions or a favourite RxJS pattern? Drop a comment below — reactive programming is a journey best taken together! 🚀