DEV Community 4h ago

RxJS in Practice: Reactive Programming, Operators, Patterns, and Real-World Examples

RxJS is not just a library of operators. It is a different way of thinking about asynchronous data. If you work with Angular, Node.js, React, or any application that deals with asynchronous operations, you will eventually encounter problems involving:

HTTP requests
User input
WebSockets
Events
Timers
State changes
Multiple asynchronous operations
Cancellation
Race conditions
Retries
Error handling

RxJS provides a powerful model for solving these problems through Reactive Programming. In this article, we will move from the theory to practical examples and finally build a realistic Angular-style application using RxJS.

1. What Is RxJS?

RxJS (Reactive Extensions for JavaScript) is a library for composing asynchronous and event-based programs using Observables. Instead of thinking:

"I have a function that will eventually return a value."

You start thinking:

"I have a stream of values that may arrive over time."

For example:

const observable$ = new Observable(subscriber => {
  subscriber.next(1);
  subscriber.next(2);
  subscriber.next(3);
});

The observable represents a stream:

1 ───> 2 ───> 3 ───> complete

But an Observable doesn't have to emit only three values. It could represent:

User clicks: click ↓ click ↓ click ↓ click ↓ ...
HTTP Request: Response ↓ Complete
WebSocket: Message ↓ Message ↓ Message ↓ Message ↓ ...

This is the fundamental idea behind RxJS.

2. What Is Reactive Programming?

Reactive programming is a programming paradigm based on:

Reacting to changes and events over time.

Traditional imperative programming often looks like:

const value = getValue();
console.log(value);

You ask for a value and receive it.

Reactive programming looks more like:

value$.subscribe(value => {
  console.log(value);
});

You are saying: "Whenever a value arrives, execute this logic."

This becomes extremely powerful when dealing with asynchronous systems.

3. The Core RxJS Concepts

RxJS revolves around several important concepts:

Observable → Subscribe → Receive values → Operators → Transform / Filter / Combine → Observer

The most important concepts are:

Observable
Observer
Subscriber
Subscription
Operators
Subject
Schedulers

Let's understand them.

4. Observable

An Observable represents a stream of values over time.

Example:

import { Observable } from 'rxjs';

const observable$ = new Observable<number>(subscriber => {
  subscriber.next(10);
  subscriber.next(20);
  subscriber.next(30);
  subscriber.complete();
});

Subscribe:

observable$.subscribe(value => {
  console.log(value);
});

Output:

10 20 30

The $ suffix is a common naming convention: users$, products$, orders$, searchResults$. It usually means: "This variable represents an Observable." It is a convention, not a language requirement.

5. Observable Lifecycle

An Observable can emit three kinds of notifications:

next
error
complete

Example:

const observable$ = new Observable<number>(subscriber => {
  subscriber.next(1);
  subscriber.next(2);
  subscriber.next(3);
  subscriber.complete();
});

Conceptually:

next(1) ↓
next(2) ↓
next(3) ↓
complete()

An Observable can also fail:

const observable$ = new Observable<number>(subscriber => {
  subscriber.next(1);
  subscriber.error(new Error('Something went wrong'));
});

Then:

next(1) ↓
error

Once an Observable sends either complete() or error(...), the stream terminates.

6. Observer

An Observer defines what should happen when the Observable emits values, errors, or completes.

observable$.subscribe({
  next: value => {
    console.log('Value:', value);
  },
  error: error => {
    console.error('Error:', error);
  },
  complete: () => {
    console.log('Completed');
  }
});

This is more explicit than:

observable$.subscribe(value => {
  console.log(value);
});

For production code, the object form is often easier to maintain when error or completion handling matters.

7. Subscription

When you subscribe:

const subscription = observable$.subscribe({
  next: value => console.log(value)
});

you receive a Subscription. You can unsubscribe:

subscription.unsubscribe();

This matters for long-lived streams such as:

WebSockets
DOM events
Intervals
Subjects
Application-wide streams

Example:

const subscription = interval(1000).subscribe(value => {
  console.log(value);
});

setTimeout(() => {
  subscription.unsubscribe();
}, 5000);

The interval stops after approximately five seconds.

8. Cold Observables

A cold Observable starts its producer separately for each subscriber.

Example:

const observable$ = new Observable<number>(subscriber => {
  console.log('Producer started');
  subscriber.next(Math.random());
});

Subscribe twice:

observable$.subscribe(value => {
  console.log('Subscriber 1:', value);
});

observable$.subscribe(value => {
  console.log('Subscriber 2:', value);
});

The producer runs twice. Conceptually:

Subscriber 1 ↓ Producer A
Subscriber 2 ↓ Producer B

Each subscriber gets its own execution.

9. Hot Observables

A hot Observable represents a shared source. For example:

WebSocket ↓
├── Subscriber A
├── Subscriber B
└── Subscriber C

The source exists independently of individual subscribers.

This concept becomes especially important when using:

Subject
share()
shareReplay()

10. Operators

Operators are one of the most important parts of RxJS. They allow you to transform and control streams.

For example:

source$.pipe(
  map(value => value * 2),
  filter(value => value > 10)
).subscribe(value => {
  console.log(value);
});

Think of an RxJS pipeline as:

Source → map → filter → subscribe

11. map

map transforms every emitted value.

of(1, 2, 3).pipe(
  map(value => value * 10)
).subscribe(console.log);

Output:

10 20 30

Conceptually:

1 → map → 10
2 → map → 20
3 → map → 30

A very common example is extracting data from an HTTP response:

this.http.get<ApiResponse<User[]>>('/api/users').pipe(
  map(response => response.data)
);

Now the consumer receives only User[] instead of the complete response object.

12. filter

filter allows only values matching a condition.

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

Output:

2 4

Think:

1 ❌
2 ✅
3 ❌
4 ✅
5 ❌

13. tap

tap allows you to perform side effects without changing the emitted value.

users$.pipe(
  tap(users => {
    console.log('Users:', users);
  })
).subscribe();

tap is useful for:

Logging
Debugging
Analytics
Updating external state
Observing the pipeline

Avoid using tap to secretly transform data.

Bad:

tap(user => {
  user.name = 'Changed';
})

Prefer transformations through operators such as:

map(user => ({ ...user, name: 'Changed' }))

14. debounceTime

One of the most practical RxJS operators.

Imagine a search box. Without debouncing:

c → ca → cat → cats

You might send four HTTP requests. With:

debounceTime(300)

the application waits until the user stops typing for 300ms.

searchTerm$.pipe(
  debounceTime(300)
)

Conceptually:

c → ca → cat → cats
         ↓ 300ms
      "cats"

Only "cats" continues through the pipeline.

15. distinctUntilChanged

This operator prevents consecutive duplicate values.

of('Angular', 'Angular', 'React', 'React', 'Vue').pipe(
  distinctUntilChanged()
).subscribe(console.log);

Output:

Angular
React
Vue

Very useful for:

Search inputs
Filters
Route parameters
State changes

16. switchMap

switchMap is one of the most important RxJS operators.

Imagine:

User searches:
Angular → Angular RxJS → Angular RxJS operators

Each search creates an HTTP request. You usually don't want an old request to overwrite the latest result. switchMap solves this by switching to the newest inner Observable and unsubscribing from the previous one.

searchTerm$.pipe(
  switchMap(term => this.searchApi(term))
)

Conceptually:

Search A ↓ Request A
Search B ↓ Cancel/Unsubscribe A ↓ Request B
Search C ↓ Cancel/Unsubscribe B ↓ Request C

This makes switchMap ideal for:

Search
Autocomplete
Route parameter changes
Refreshing data
User-driven queries

17. mergeMap

mergeMap subscribes to inner Observables concurrently.

ids$.pipe(
  mergeMap(id => this.getUser(id))
);

Conceptually:

ID 1 → Request 1
ID 2 → Request 2
ID 3 → Request 3

All can run concurrently. Use it when previous operations should not be cancelled.

Typical use cases:

Independent HTTP requests
Processing multiple items
Concurrent operations

18. concatMap

concatMap queues inner Observables and runs them sequentially.

actions$.pipe(
  concatMap(action => this.save(action))
);

Conceptually:

Action A ↓ Request A ↓ Complete
Action B ↓ Request B ↓ Complete
Action C ↓ Request C ↓ Complete

Use it when order matters. For example:

Create record A → Update A → Delete A

You may not want these operations running concurrently.

19. exhaustMap

exhaustMap ignores new emissions while the current inner Observable is running.

Example:

submitClick$.pipe(
  exhaustMap(() => this.submitForm())
);

Imagine the user double-clicks:

Click ↓ Request starts
Click ↓ IGNORED
Click ↓ IGNORED
Request completes

This is extremely useful for preventing duplicate submissions.

20. The Four Important Mapping Operators

A useful mental model:

Operator	Behavior
`switchMap`	Cancel previous
`mergeMap`	Run concurrently
`concatMap`	Queue sequentially
`exhaustMap`	Ignore while busy

Think about them like this:

switchMap - Latest wins
mergeMap - Everything runs
concatMap - One by one
exhaustMap - First wins while busy

Choosing the correct flattening operator is one of the most important RxJS skills.

21. catchError

Errors are inevitable. RxJS provides catchError().

Example:

this.http.get<User[]>('/api/users').pipe(
  catchError(error => {
    console.error(error);
    return of([]);
  })
);

Instead of terminating the application flow unexpectedly, the stream returns an empty array.

However, error handling should reflect the application's requirements. Sometimes you should recover:

catchError(() => of([]))

Sometimes you should rethrow:

catchError(error => {
  return throwError(() => error);
})

22. retry

Transient failures can sometimes be retried.

this.http.get('/api/data').pipe(
  retry(3)
);

Conceptually:

Request ↓ Fail ↓ Retry ↓ Fail ↓ Retry ↓ Fail ↓ Retry

But blindly retrying everything is a bad idea. For example, retrying a validation error won't usually solve anything. Retries are more appropriate for transient failures.

23. finalize

finalize runs when the Observable terminates because of completion, error, or unsubscription.

Example:

this.loading = true;
this.http.get('/api/users').pipe(
  finalize(() => {
    this.loading = false;
  })
).subscribe();

This is useful for cleanup:

Start request → loading = true → HTTP request → complete/error/unsubscribe → loading = false

24. Creating Observables

RxJS provides many creation functions.

of

of(1, 2, 3)

Emits: 1 2 3

from

from([1, 2, 3])

Also emits: 1 2 3

But from can convert many iterable or Promise-like sources. For example:

from(fetch('/api/users'))

25. interval

Creates periodic emissions.

interval(1000).subscribe(value => {
  console.log(value);
});

Output:

0 1 2 3 4 ...

This Observable does not naturally complete. Therefore, long-lived subscriptions should be managed carefully.

26. timer

timer(2000)

Emits after two seconds. You can also create repeated emissions:

timer(0, 1000)

Conceptually:

0 ↓ 1 sec
1 ↓ 1 sec
2 ↓ 1 sec
3 ↓ ...

27. Subjects

A Subject is both:

an Observable
an Observer

Example:

const subject$ = new Subject<number>();

subject$.subscribe(value => {
  console.log('A:', value);
});

subject$.subscribe(value => {
  console.log('B:', value);
});

subject$.next(10);
subject$.next(20);

Output:

A: 10
B: 10
A: 20
B: 20

The Subject multicasts values to its subscribers.

28. BehaviorSubject

BehaviorSubject requires an initial value and stores the latest value.

const user$ = new BehaviorSubject<User | null>(null);

When a new subscriber arrives, it immediately receives the current value.

Current value = User A
Subscriber joins ↓ Receives User A

This makes BehaviorSubject useful for representing state.

Example:

private currentUserSubject = new BehaviorSubject<User | null>(null);
currentUser$ = this.currentUserSubject.asObservable();

Then:

this.currentUserSubject.next(user);

29. ReplaySubject

ReplaySubject can replay previous emissions.

const subject$ = new ReplaySubject<number>(2);

subject$.next(1);
subject$.next(2);
subject$.next(3);

subject$.subscribe(value => {
  console.log(value);
});

Output:

2 3

because we configured it to replay the last two values.

30. combineLatest

Sometimes you need the latest value from multiple streams.

Example:

combineLatest([user$, settings$, permissions$])

Conceptually:

User ────────┐
Settings ────┼──> combineLatest
Permissions ─┘

It emits when one source emits, after every source has emitted at least once.

Useful for:

Filters
UI state
User preferences
Multiple dependent inputs

31. forkJoin

forkJoin waits for all supplied Observables to complete and then emits their final values.

Perfect for independent HTTP requests:

forkJoin({
  users: this.getUsers(),
  products: this.getProducts(),
  orders: this.getOrders()
})

Conceptually:

Users ────────┐
Products ─────┼──> forkJoin → result
Orders ───────┘

Important distinction:

forkJoin is generally for "wait until all complete."
combineLatest is generally for "react whenever the latest values change."

32. withLatestFrom

Sometimes one Observable should trigger the operation while reading the latest value from another.

Example:

submit$.pipe(
  withLatestFrom(formValue$)
)

Think:

submit$ ────────────────┐
       ↓ withLatestFrom ↑
formValue$ ─────────────┘

The submit event is the trigger.

33. RxJS and Angular

RxJS is deeply integrated into Angular. Common examples include:

HttpClient
ActivatedRoute
Router events
Reactive Forms
Event streams
NgRx
WebSocket

Read on DEV Community ↗ ← Back to News

RxJS in Practice: Reactive Programming, Operators, Patterns, and Real-World Examples