RxJS — Complete Reactive Programming & Observables Guide

RxJS is a library for reactive programming using Observables — it makes composing asynchronous and event-based programs easier by treating data streams as sequences that can be transformed, filtered, and combined.

What You’ll Learn

By the end of this tutorial, you’ll create and subscribe to Observables, transform data streams with operators like map/filter/switchMap, use Subjects for multicasting, handle errors gracefully, and apply reactive patterns to HTTP requests and user input.

Why RxJS Matters

Modern applications are inherently asynchronous — user clicks, API responses, timer events, WebSocket messages. Managing all these with callbacks or promises leads to complex, hard-to-debug code. RxJS provides a unified model: everything is a stream. HTTP request? Stream. Mouse click? Stream. Array of data? Stream. DodaTech uses RxJS in Doda Browser’s extension event system, where multiple user actions and API responses must be composed into coherent workflows.

Security note: Understanding Rxjs helps build more secure applications — a core principle at DodaTech, where tools like Durga Antivirus Pro and Doda Browser rely on solid implementation practices.

RxJS Learning Path

    flowchart LR
  A[Async JavaScript] --> B[RxJS]
  B --> C[Observables]
  B --> D[Operators]
  B --> E[Subjects]
  B --> F[Error Handling]
  C --> G[Angular / React]
  style B fill:#3b82f6,stroke:#fff,color:#fff
  

What is Reactive Programming? — The Water Pipe Analogy

Think of data as water flowing through pipes:

• Observable = the water source (tap). Data flows out over time
• Operators = filters and attachments on the pipe. They transform the water (filter, change pressure, combine with other pipes)
• Subscription = your glass at the end of the pipe. You receive the water when it arrives
• Subject = a tap that multiple people can drink from simultaneously

Why this matters: With callbacks, each async operation is handled differently. With RxJS, everything is a stream — clicks, HTTP responses, timers, arrays. One model, infinite applications.

Core Concepts

import { Observable, from, of } from "rxjs";
import { map, filter, debounceTime, switchMap, catchError } from "rxjs/operators";

// Create an Observable
const observable = new Observable(subscriber => {
  subscriber.next(1);
  subscriber.next(2);
  subscriber.next(3);
  setTimeout(() => {
    subscriber.next(4);
    subscriber.complete();
  }, 1000);
});

// Subscribe
observable.subscribe({
  next: x => console.log(x),     // handles each value
  error: err => console.error(err), // handles errors
  complete: () => console.log("Done") // cleanup
});

Line-by-line:

• new Observable(subscriber => { ... }) — creates a data source. The function runs when someone subscribes
• subscriber.next(1) — emits a value. Like water flowing from the tap
• subscriber.complete() — signals that no more values will be emitted
• .subscribe({ next, error, complete }) — starts the stream. Nothing happens until someone subscribes (Observables are lazy)

Key insight: Observables are lazy — they don’t start producing values until you call .subscribe(). Promises execute immediately; Observables wait for subscribers.

Creating Observables

// From an array
from([1, 2, 3, 4, 5]);
// emits: 1, 2, 3, 4, 5

// From individual values
of("a", "b", "c");
// emits: a, b, c

// From a promise
from(fetch("/api/data"));
// emits the response

// From an event
fromEvent(document, "click");
// emits every click event

// Timer
import { interval, timer } from "rxjs";
interval(1000);  // emits 0, 1, 2, ... every second
timer(3000);     // emits 0 after 3 seconds, then completes

Common Operators

Transformation

of(1, 2, 3).pipe(
  map(x => x * 2)
);
// emits: 2, 4, 6

Filtering

from([1, 2, 3, 4, 5]).pipe(
  filter(x => x > 2)
);
// emits: 3, 4, 5

Debounce + SwitchMap (Type-Ahead Search)

This is the classic RxJS pattern — user input, debounced, with cancelable API requests:

input.valueChanges.pipe(
  debounceTime(300),              // wait 300ms after last keystroke
  distinctUntilChanged(),          // don't search if value hasn't changed
  switchMap(term => this.search(term))  // cancel previous request, start new one
).subscribe(results => {
  this.results = results;
});

Line-by-line:

• debounceTime(300) — waits 300ms after the user stops typing. Without this, every keystroke triggers an API call
• distinctUntilChanged() — skips the search if the user types the same value (e.g., paste + blur same value)
• switchMap(term => this.search(term)) — maps the input to an API call. If a new input arrives before the previous API call completes, it cancels the old one automatically. This prevents out-of-order responses.

Why this is powerful: With promises, handling rapid user input requires manual cancellation flags. RxJS handles it with one operator.

Error Handling

http.get("/api/data").pipe(
  catchError(err => {
    console.error("API error:", err);
    return of({ error: err.message });  // return default value
  })
).subscribe(data => console.log(data));

Subjects — Multicasting

Subjects are both an Observable and an Observer — they can emit values AND be subscribed to:

import { Subject, BehaviorSubject, ReplaySubject } from "rxjs";

// Subject — basic multicast
const subject = new Subject();
subject.subscribe(v => console.log("A:", v));
subject.next(1); // A: 1
// Late subscribers miss past values

// BehaviorSubject — requires initial value, replays last on subscribe
const state = new BehaviorSubject("initial");
state.getValue(); // "initial"
state.subscribe(v => console.log(v)); // immediately gets "initial"

// ReplaySubject — replays N previous values to new subscribers
const replay = new ReplaySubject(2);
replay.next(1); replay.next(2); replay.next(3);
replay.subscribe(v => console.log(v)); // gets 2, 3 (last 2)

When to use each:

• Subject — simple event bus (like EventEmitter)
• BehaviorSubject — state management (always has a current value)
• ReplaySubject — replay history to late subscribers

RxJS in Angular

RxJS is foundational in Angular:

• HTTP — HttpClient returns Observables
• Forms — valueChanges is an Observable on form controls
• Router — params, queryParams are Observables
• State management — NgRx is built on RxJS

// Angular example
this.route.params.pipe(
  switchMap(params => this.userService.getUser(params.id))
).subscribe(user => this.user = user);

Common Mistakes

1. Not unsubscribing (memory leaks)

Every subscription consumes memory. Always unsubscribe when the component is destroyed. Use takeUntil, async pipe (Angular), or store the subscription and call .unsubscribe().

2. Nesting subscriptions

// BAD: nested subscriptions
obs1.subscribe(val1 => {
  obs2.subscribe(val2 => {
    console.log(val1, val2);
  });
});

// GOOD: use combineLatest or forkJoin
combineLatest([obs1, obs2]).subscribe(([val1, val2]) => {
  console.log(val1, val2);
});

3. Using switchMap when you need mergeMap

switchMap cancels the previous inner observable. mergeMap runs all in parallel. Use switchMap for search (cancel previous), mergeMap for parallel requests (no cancellation).

4. Forgetting that Observables are cold (lazy)

An Observable does nothing until subscribed. If you create an Observable and don’t .subscribe(), nothing happens — which might surprise you if you’re used to Promises executing immediately.

5. Not handling errors in the stream

An unhandled error in an Observable kills the entire subscription. Always add catchError or retry operators.

Practice Questions

1. What is the difference between an Observable and a Promise?

Answer: Observables are lazy (nothing happens until subscription), can emit multiple values over time, and are cancellable. Promises execute immediately, emit one value, and aren’t cancellable.

2. What does switchMap do and when should you use it?

Answer: switchMap maps each emission to an inner Observable, cancelling the previous one if still active. Use it for type-ahead search or any scenario where only the latest request matters.

3. Why should you unsubscribe from Observables?

Answer: To prevent memory leaks. Every active subscription holds references to callbacks and closures. If a component is destroyed without unsubscribing, the callback can still execute and may try to update destroyed DOM.

4. What is the difference between a Subject and a BehaviorSubject?

Answer: A Subject doesn’t have an initial value and new subscribers don’t receive past emissions. A BehaviorSubject requires an initial value and replays the latest value to new subscribers.

Challenge

Build a real-time search component: capture input changes, debounce 300ms, fetch results from an API, cancel previous requests if new input arrives, handle errors gracefully, and display results. Do it entirely with RxJS operators — no promise .then() calls.

FAQ

Try It Yourself

# 1. Install RxJS
mkdir rxjs-practice && cd rxjs-practice
npm init -y
npm install rxjs

# 2. Create a practice file
cat > index.js << 'EOF'
const { of, from, interval } = require("rxjs");
const { map, filter, take } = require("rxjs/operators");

// Create a stream
of(1, 2, 3, 4, 5).pipe(
  filter(x => x % 2 === 0),
  map(x => x * 10)
).subscribe({
  next: val => console.log("Value:", val),
  complete: () => console.log("Done!")
});
// Output: Value: 20, Value: 40, Done!

// Timer stream
interval(500).pipe(
  take(5)
).subscribe(val => console.log("Tick:", val));
// Output: Tick: 0, Tick: 1, Tick: 2, Tick: 3, Tick: 4
EOF

# 3. Run it
node index.js

What’s Next

Explore more JavaScript patterns:

Related topics to explore:

• Async JavaScript
• Node.js Streams
• Event-Driven Architecture

What’s Next

Congratulations on completing this Rxjs tutorial! Here’s where to go from here:

• Practice daily — Consistency is more important than long study sessions
• Build a project — Apply what you learned by building something real
• Explore related topics — Check out other tutorials in the same category
• Join the community — Discuss with other learners and share your progress

Remember: every expert was once a beginner. Keep coding!