Skip to main content

Learning RxJS - Reactive Programming in JavaScript

· 6 min read

RxJS (Reactive Extensions for JavaScript) has been a game-changer in how I think about handling asynchronous data. If you've worked with Angular, you've definitely encountered RxJS. But understanding it deeply? That's a different story. Here are my notes from diving into reactive programming.

What is RxJS?​

RxJS is a library for reactive programming using Observables. It gives you an efficient, flexible, and composable way to handle asynchronous and event-based data flows in your applications. Think of it as a powerful toolkit for managing data streams over time.

Core Concepts​

1. Observables​

Observables are the heart of RxJS. They represent streams of data that can emit multiple values over time.

Key characteristics:

  • Streams of Data: Observables can emit multiple values over time
  • Lazy Execution: They won't start emitting data until there's at least one subscriber
  • Three Types of Emissions:
    • next: Regular data values
    • error: Error notification
    • complete: Signal that the Observable has finished

Here's a basic example:

const observable = new Observable<number>(subscriber => {
  subscriber.next(1);
  subscriber.next(2);
  subscriber.next(3);
  
  // Uncomment to see async behavior
  // setTimeout(() => {
  //   subscriber.next(4);
  //   subscriber.complete();
  // }, 1000);
});

observable.subscribe({  
  next(x) { console.log('got value ' + x); },
  error(err) { console.error('something wrong occurred: ' + err); },
  complete() { console.log('done'); }
});

2. Observers​

Observers are objects with callback methods to handle different types of notifications sent by Observables: next, error, and complete. They're the handlers that process the data emitted by Observables.

3. Subscription​

Subscribing to an Observable starts its execution. A subscription represents the execution of an Observable.

Important: Always unsubscribe from Observables that have a long life or are not self-terminating to prevent memory leaks!

const subscription = observable.subscribe(observer);

// Later, when you're done
subscription.unsubscribe();

Operators - The Power Tools​

Operators are functions that enable manipulating the items emitted by Observables. They're what make RxJS so powerful.

Transformation Operators​

map: Transforms each value emitted by the source Observable

of(1, 2, 3)
  .pipe(map(x => x * x))
  .subscribe(value => console.log(value));
// Output: 1, 4, 9

mergeMap / flatMap: Projects each source value to an Observable and merges the emissions

Filtering Operators​

filter: Emits only values that meet a specific condition

of(1, 2, 3, 4, 5)
  .pipe(filter(x => x % 2 === 0))
  .subscribe(value => console.log(value));
// Output: 2, 4

take: Takes the first n values and then completes

Combination Operators​

merge: Turns multiple Observables into a single Observable

const first = of(1, 2, 3);
const second = of(4, 5, 6);
merge(first, second)
  .subscribe(value => console.log(value));
// Output: 1, 2, 3, 4, 5, 6

concat: Concatenates multiple Observables together sequentially

Error Handling​

catchError: Handles errors on the Observable

throwError('Error occurred')
  .pipe(catchError(err => of(err)))
  .subscribe(value => console.log(value));
// Output: Error occurred

Utility Operators​

tap: Transparently performs actions or side-effects, like logging

of(1, 2, 3)
  .pipe(
    tap(value => console.log('Before: ', value)),
    map(value => value * 10),
    tap(value => console.log('After: ', value))
  )
  .subscribe(value => console.log('Output: ', value));

Higher-Order Observables​

This is where things get interesting. Higher-order Observables handle observables of observables - nested subscriptions that are common in real-world applications.

Key Operators​

mergeMap (flatMap): Handles inner Observables concurrently

userIds$.pipe(
  mergeMap(userId => from(fetchUserData(userId)))
).subscribe(userDetails => {
  console.log(`Fetched User Details: ${userDetails.userName}`);
});

concatMap: Handles inner Observables one after another, maintaining order

userIds$.pipe(
  concatMap(userId => from(fetchUserData(userId)))
).subscribe(userDetails => {
  console.log(`Fetched User Details: ${userDetails.userName}`);
});

switchMap: On each emission, unsubscribes from the previous inner Observable and subscribes to the new one. Perfect for search boxes!

exhaustMap: Ignores new inner Observables while the current one is still executing. Great for preventing duplicate submissions.

Subjects and Their Variants​

Subjects are special types of Observables that allow values to be multicasted to many Observers.

Subject​

Basic multicasting - allows values to be broadcasted to multiple subscribers

const subject = new Subject<number>();
subject.subscribe(x => console.log(`Observer A: ${x}`));
subject.next(1);
subject.next(2);

BehaviorSubject​

Requires an initial value and emits its current value to new subscribers. Ideal for representing values over time, like UI state.

const behaviorSubject = new BehaviorSubject(0);
behaviorSubject.subscribe(x => console.log(`Observer A: ${x}`));
behaviorSubject.next(1);
behaviorSubject.next(2);
behaviorSubject.subscribe(x => console.log(`Observer B: ${x}`)); 
// Observer B will receive the latest value (2)

ReplaySubject​

Emits to new subscribers all the items previously emitted by the source Observable, up to a buffer size.

const replaySubject = new ReplaySubject(3); // Buffer size of 3
replaySubject.next(1);
replaySubject.next(2);
replaySubject.next(3);
replaySubject.next(4);
replaySubject.subscribe(x => console.log(`Observer A: ${x}`)); 
// Will receive last 3 values: 2, 3, 4

AsyncSubject​

Emits only the last value upon completion. Useful for scenarios where only the final result matters.

const asyncSubject = new AsyncSubject();
asyncSubject.subscribe(x => console.log(`Observer A: ${x}`));
asyncSubject.next(1);
asyncSubject.next(2);
asyncSubject.complete();
asyncSubject.subscribe(x => console.log(`Observer B: ${x}`)); 
// Both observers will receive only the last value: 2

What I Learned​

Working with RxJS taught me several important lessons:

  1. Think in Streams: Instead of thinking about individual events, think about streams of events over time. This mental shift is key to mastering reactive programming.

  2. Composition Over Complexity: Complex async logic can be broken down into simple, composable operators. Don't try to do everything in one place.

  3. Memory Management Matters: Always unsubscribe! Memory leaks are real, especially in long-running applications.

  4. Choose the Right Operator: There are many operators that seem similar (mergeMap, switchMap, concatMap, exhaustMap), but they behave very differently. Understanding when to use each is crucial.

  5. Start Simple: Begin with basic operators like map, filter, and tap. Build up to higher-order observables as you get comfortable.

  6. Marble Diagrams Help: RxJS documentation uses marble diagrams to visualize how operators work. They're incredibly helpful for understanding timing and emission patterns.

Real-World Use Cases​

In my Angular projects, I've used RxJS for:

  • HTTP Requests: Managing API calls with proper error handling and retry logic
  • Form Validation: Real-time validation with debouncing
  • State Management: Maintaining application state with BehaviorSubject
  • WebSocket Connections: Handling real-time data streams
  • Search Features: Using switchMap to cancel previous searches
  • Auto-save: Debouncing user input before saving

Resources​

If you want to dive deeper into RxJS:

RxJS has a steep learning curve, but once it clicks, it fundamentally changes how you approach asynchronous programming. Whether you're working with Angular or just handling async operations in JavaScript, it's a powerful tool worth mastering.