Essential RxJS Operators for Modern JavaScript Development

RxJS (Reactive Extensions for JavaScript) is a powerful library for handling asynchronous data streams using observables. At its core are operators — pure functions that enable you to manipulate, transform, and combine data streams in elegant ways. Here are the most important RxJS operators every developer should master.

Creation Operators

of()

Creates an observable from a sequence of values.

import { of } from 'rxjs';
of(1, 2, 3, 4, 5).subscribe(value => console.log(value));
// Output: 1, 2, 3, 4, 5

from()

Converts arrays, promises, or other observables into an observable stream.

import { from } from 'rxjs';
from([1, 2, 3]).subscribe(value => console.log(value));
from(fetch('/api/data')).subscribe(response => console.log(response));

interval()

Creates an observable that emits sequential numbers at specified intervals.

import { interval } from 'rxjs';
interval(1000).subscribe(value => console.log(`Tick: ${value}`));

Transformation Operators

map()

The most fundamental operator — transforms each emitted value using a provided function.

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

switchMap()

Maps each value to an observable, then flattens the result while canceling previous inner observables.

import { fromEvent } from 'rxjs';
import { switchMap } from 'rxjs/operators';
fromEvent(searchInput, 'input')
  .pipe(
    switchMap(event => this.searchService.search(event.target.value))
  )
  .subscribe(results => console.log(results));

mergeMap() (flatMap)

Maps each value to an observable and merges all inner observables concurrently.

import { of } from 'rxjs';
import { mergeMap, delay } from 'rxjs/operators';
of(1, 2, 3)
  .pipe(
    mergeMap(x => of(x).pipe(delay(1000)))
  )
  .subscribe(value => console.log(value));

concatMap()

Maps each value to an observable and concatenates them sequentially (waits for each to complete).

import { of } from 'rxjs';
import { concatMap, delay } from 'rxjs/operators';
of(1, 2, 3)
  .pipe(
    concatMap(x => of(x * 10).pipe(delay(1000)))
  )
  .subscribe(value => console.log(value));

Filtering Operators

filter()

Emits only values that pass a specified condition.

import { of } from 'rxjs';
import { filter } from 'rxjs/operators';
of(1, 2, 3, 4, 5, 6)
  .pipe(filter(x => x % 2 === 0))
  .subscribe(value => console.log(value));
// Output: 2, 4, 6

debounceTime()

Emits a value only after a specified time has passed without another emission.

import { fromEvent } from 'rxjs';
import { debounceTime } from 'rxjs/operators';
fromEvent(searchInput, 'input')
  .pipe(debounceTime(300))
  .subscribe(event => console.log(event.target.value));

distinctUntilChanged()

Emits only when the current value is different from the previous value.

import { of } from 'rxjs';
import { distinctUntilChanged } from 'rxjs/operators';
of(1, 1, 2, 2, 3, 1)
  .pipe(distinctUntilChanged())
  .subscribe(value => console.log(value));
// Output: 1, 2, 3, 1

take()

Emits only the first n values.

import { interval } from 'rxjs';
import { take } from 'rxjs/operators';
interval(1000)
  .pipe(take(3))
  .subscribe(value => console.log(value));
// Output: 0, 1, 2 (then completes)

takeUntil()

Emits values until another observable emits.

import { interval, fromEvent } from 'rxjs';
import { takeUntil } from 'rxjs/operators';
const stop$ = fromEvent(document, 'click');
interval(1000)
  .pipe(takeUntil(stop$))
  .subscribe(value => console.log(value));

Get senthil kumar’s stories in your inbox

Join Medium for free to get updates from this writer.

Remember me for faster sign in

Combination Operators

combineLatest()

Combines multiple observables and emits when any of them emits.

import { combineLatest, of } from 'rxjs';
import { delay } from 'rxjs/operators';
const obs1$ = of('A', 'B').pipe(delay(1000));
const obs2$ = of(1, 2).pipe(delay(500));
combineLatest([obs1$, obs2$])
  .subscribe(([letter, number]) => console.log(letter, number));

merge()

Merges multiple observables into one by subscribing to all simultaneously.

import { merge, interval } from 'rxjs';
import { map } from 'rxjs/operators';
const obs1$ = interval(1000).pipe(map(x => `First: ${x}`));
const obs2$ = interval(1500).pipe(map(x => `Second: ${x}`));
merge(obs1$, obs2$).subscribe(value => console.log(value));

startWith()

Emits a specified value before the source observable’s values.

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

Error Handling Operators

catchError()

Catches errors and returns a new observable or throws an error.

import { of, throwError } from 'rxjs';
import { catchError, mergeMap } from 'rxjs/operators';
of(1, 2, 3)
  .pipe(
    mergeMap(x => x === 2 ? throwError('Error!') : of(x)),
    catchError(err => of('Caught error: ' + err))
  )
  .subscribe(value => console.log(value));

retry()

Retries the source observable a specified number of times on error.

import { ajax } from 'rxjs/ajax';
import { retry } from 'rxjs/operators';
ajax('/api/data')
  .pipe(retry(3))
  .subscribe(
    response => console.log(response),
    error => console.log('Failed after 3 retries')
  );

Utility Operators

tap()

Performs side effects without modifying the emitted values.

import { of } from 'rxjs';
import { tap, map } from 'rxjs/operators';
of(1, 2, 3)
  .pipe(
    tap(x => console.log('Before:', x)),
    map(x => x * 2),
    tap(x => console.log('After:', x))
  )
  .subscribe();

finalize()

Executes a function when the observable completes or errors.

import { of } from 'rxjs';
import { finalize } from 'rxjs/operators';
of(1, 2, 3)
  .pipe(
    finalize(() => console.log('Observable completed'))
  )
  .subscribe(value => console.log(value));

Real-World Example: Search Feature

Here’s how these operators work together in a practical search implementation:

import { fromEvent } from 'rxjs';
import { 
  debounceTime, 
  distinctUntilChanged, 
  filter, 
  switchMap, 
  tap, 
  catchError 
} from 'rxjs/operators';
import { of } from 'rxjs';
const searchInput = document.getElementById('search');
const searchResults$ = fromEvent(searchInput, 'input').pipe(
  map(event => event.target.value),
  filter(term => term.length > 2), // Only search for terms longer than 2 chars
  debounceTime(300), // Wait 300ms after user stops typing
  distinctUntilChanged(), // Only search if the term changed
  tap(term => console.log('Searching for:', term)),
  switchMap(term => 
    this.searchService.search(term).pipe(
      catchError(error => {
        console.error('Search failed:', error);
        return of([]); // Return empty results on error
      })
    )
  )
);
searchResults$.subscribe(results => {
  displaySearchResults(results);
});

Key Takeaways

Understanding these operators will give you the foundation to build complex reactive applications. Remember:

• Creation operators help you create observables from various sources
• Transformation operators modify the data flowing through your streams
• Filtering operators control which values are emitted
• Combination operators merge multiple streams together
• Error handling operators make your applications resilient
• Utility operators provide helpful side-effect capabilities

Master these operators, and you’ll be well-equipped to handle any reactive programming challenge in JavaScript. The key is to think in terms of data streams and use operators to shape those streams according to your application’s needs.

Next Steps

• Practice combining multiple operators in operator chains
• Explore advanced operators like scan(), buffer(), and window()
• Learn about custom operators and how to create your own
• Study real-world patterns like the async pipe in Angular
• Understand marble diagrams for visualizing operator behavior

Happy reactive programming!