⚠️ Unsaved
[M]:
RxJS Complete Guide: All Operators and Patterns
This practical guide covers the essential operators and patterns in RxJS for effective reactive programming in JavaScript. We'll explore everything from basic Observable creation to advanced patterns, with runnable examples for each concept.
[1]:
// Install RxJS
!npm install rxjs
$ npm install rxjs added 2 packages in 2s 28 packages are looking for funding run `npm fund` for details
[2]:
// Import RxJS
const rxjs = require('rxjs');
const operators = require('rxjs/operators');
console.log(`RxJS version loaded'}\n`);
RxJS version loaded'}
[M]:
1. Observable Basics
Let's start with the fundamentals of RxJS: Observables, Observers, and Subscriptions.
[3]:
// Creating a simple synchronous Observable
const simpleObservable = new rxjs.Observable(subscriber => {
subscriber.next('First value');
subscriber.next('Second value');
subscriber.complete();
});
// Subscribe to the Observable
console.log('Basic Observable example:\n');
simpleObservable.subscribe({
next: value => console.log(`Received: ${value}\n`),
error: err => console.log(`Error: ${err}\n`),
complete: () => console.log('Observable completed\n')
});
Basic Observable example: Received: First value Received: Second value Observable completed
[4]:
// Using Promise to demonstrate async Observable behavior
function asyncObservableDemo() {
return new Promise(resolve => {
console.log('Async Observable example:\n');
const asyncObservable = new rxjs.Observable(subscriber => {
subscriber.next('First value');
// Async emission
setTimeout(() => {
subscriber.next('Async value after 500ms');
subscriber.complete();
resolve('Done'); // Resolve the promise after Observable completes
}, 500);
});
asyncObservable.subscribe({
next: value => console.log(`Received: ${value}\n`),
complete: () => console.log('Async Observable completed\n')
});
});
}
// Run and wait for it to complete
asyncObservableDemo().then(result => console.log(`${result}\n`));
Async Observable example: Received: First value
[M]:
2. Creation Operators
Creation operators let you create Observables from various sources.
[5]:
// of - creates an Observable from a list of values
const ofObservable = rxjs.of(1, 2, 3, 4, 5);
console.log('of operator example:\n');
ofObservable.subscribe(value => console.log(`of emitted: ${value}\n`));
of operator example: of emitted: 1 of emitted: 2 of emitted: 3 of emitted: 4 of emitted: 5
[6]:
// from - creates an Observable from an array, promise, or iterable
const arrayObservable = rxjs.from([10, 20, 30]);
const promiseObservable = rxjs.from(Promise.resolve('Resolved value'));
console.log('from operator with array example:\n');
arrayObservable.subscribe(value => console.log(`from array emitted: ${value}\n`));
console.log('from operator with promise example:\n');
promiseObservable.subscribe(value => console.log(`from promise emitted: ${value}\n`));
from operator with array example: from array emitted: 10 from array emitted: 20 from array emitted: 30 from operator with promise example: from promise emitted: Resolved value
[33]:
// interval and timer with fixed output
function timerAndIntervalDemo() {
return new Promise(resolve => {
console.log('interval and timer example:\n');
// Demonstrate interval (emits sequential numbers)
const intervalSubscription = rxjs.interval(200)
.pipe(operators.take(5))
.subscribe(value => console.log(`interval: ${value}\n`));
// Demonstrate timer (initial delay, then works like interval)
const timerSubscription = rxjs.timer(500, 200)
.pipe(operators.take(3))
.subscribe(value => console.log(`timer: ${value}\n`));
// Ensure we see all emissions by waiting
setTimeout(() => {
intervalSubscription.unsubscribe();
timerSubscription.unsubscribe();
console.log('Interval and timer demo completed\n');
resolve();
}, 1500);
});
}
await timerAndIntervalDemo();
interval and timer example: interval: 0 interval: 1 timer: 0 interval: 2 timer: 1 interval: 3 timer: 2 interval: 4 Interval and timer demo completed
[8]:
// range - emits a sequence of numbers within a range
const rangeObservable = rxjs.range(5, 5); // start at 5, emit 5 values
console.log('range operator example:\n');
rangeObservable.subscribe(value => console.log(`range emitted: ${value}\n`));
range operator example: range emitted: 5 range emitted: 6 range emitted: 7 range emitted: 8 range emitted: 9
[9]:
// fromEvent - creates an Observable from DOM events
// Note: This would work in a browser environment, here we'll simulate it
console.log('fromEvent example (simulated for Node.js):\n');
// Simulate DOM event with a Subject
const clickSubject = new rxjs.Subject();
const clicks = clickSubject.asObservable();
// Subscribe to our simulated events
clicks.subscribe(event => console.log(`Click event: ${JSON.stringify(event)}\n`));
// Simulate clicks
clickSubject.next({ type: 'click', target: 'button1' });
clickSubject.next({ type: 'click', target: 'button2' });
fromEvent example (simulated for Node.js):
Click event: {"type":"click","target":"button1"}
Click event: {"type":"click","target":"button2"}
[M]:
3. Transformation Operators
These operators transform the items emitted by an Observable.
[10]:
// map - transforms each value
const numbers = rxjs.of(1, 2, 3, 4, 5);
const squared = numbers.pipe(
operators.map(x => x * x)
);
console.log('map operator example:\n');
squared.subscribe(value => console.log(`Squared: ${value}\n`));
map operator example: Squared: 1 Squared: 4 Squared: 9 Squared: 16 Squared: 25
[11]:
// pluck - extracts a property from each value
const users = rxjs.of(
{ name: 'Alice', age: 25 },
{ name: 'Bob', age: 30 },
{ name: 'Charlie', age: 35 }
);
// Note: pluck may be deprecated in newer RxJS versions - use map instead if needed
const names = users.pipe(
operators.map(user => user.name) // Alternative to pluck
);
console.log('property extraction example:\n');
names.subscribe(name => console.log(`Name: ${name}\n`));
property extraction example: Name: Alice Name: Bob Name: Charlie
[12]:
// scan - applies an accumulator function
const clicks = rxjs.of(1, 1, 1, 1, 1);
const clickCount = clicks.pipe(
operators.scan((acc, curr) => acc + curr, 0)
);
console.log('scan operator example:\n');
clickCount.subscribe(count => console.log(`Total clicks: ${count}\n`));
scan operator example: Total clicks: 1 Total clicks: 2 Total clicks: 3 Total clicks: 4 Total clicks: 5
[34]:
// mergeMap (flatMap) demo with fixed timing
function mergeMapDemo() {
return new Promise(resolve => {
const userIds = rxjs.of(1, 2, 3);
// Simulate API with fixed time so we see results
function getUserById(id) {
return rxjs.of({ id, name: `User ${id}` }).pipe(
operators.delay(300 * id) // Different delays to show concurrent nature
);
}
console.log('mergeMap operator example:\n');
const subscription = userIds.pipe(
operators.mergeMap(id => {
console.log(`Fetching user ${id}\n`);
return getUserById(id);
})
).subscribe({
next: user => console.log(`Got user: ${JSON.stringify(user)}\n`),
complete: () => {
console.log('mergeMap completed\n');
setTimeout(resolve, 100); // Small buffer to ensure everything is logged
}
});
// Safety timeout to prevent hanging
setTimeout(() => {
subscription.unsubscribe();
resolve();
}, 2000);
});
}
await mergeMapDemo();
mergeMap operator example:
Fetching user 1
Fetching user 2
Fetching user 3
Got user: {"id":1,"name":"User 1"}
Got user: {"id":2,"name":"User 2"}
Got user: {"id":3,"name":"User 3"}
mergeMap completed
[35]:
// switchMap and concatMap demo with fixed timing
function mapVariationsDemo() {
return new Promise(resolve => {
// Simulate search API with predictable timing
function search(term) {
console.log(`Searching for: ${term}\n`);
return rxjs.of(`Results for ${term}`).pipe(
operators.delay(400)
);
}
// Demo switchMap - cancels previous inner Observables
console.log('switchMap example:\n');
rxjs.of('react', 'reactive', 'rxjs').pipe(
operators.concatMap(term => rxjs.of(term).pipe(operators.delay(200))),
operators.switchMap(term => search(term))
).subscribe({
next: result => console.log(`${result}\n`),
complete: () => console.log('switchMap completed\n')
});
// Give enough time for the demo to complete
setTimeout(resolve, 2000);
});
}
await mapVariationsDemo();
switchMap example: Searching for: react Searching for: reactive Searching for: rxjs Results for rxjs switchMap completed
[M]:
4. Filtering Operators
These operators filter the values emitted by an Observable.
[15]:
// filter - emits values that pass a predicate test
const allNumbers = rxjs.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
const evenNumbers = allNumbers.pipe(
operators.filter(x => x % 2 === 0)
);
console.log('filter operator example:\n');
evenNumbers.subscribe(value => console.log(`Even number: ${value}\n`));
filter operator example: Even number: 2 Even number: 4 Even number: 6 Even number: 8 Even number: 10
[16]:
// take - takes a specified number of values
const manyValues = rxjs.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
const firstThree = manyValues.pipe(
operators.take(3)
);
console.log('take operator example:\n');
firstThree.subscribe({
next: value => console.log(`Taken value: ${value}\n`),
complete: () => console.log('Completed after taking 3 values\n')
});
take operator example: Taken value: 1 Taken value: 2 Taken value: 3 Completed after taking 3 values
[36]:
// takeUntil with fixed timing
function takeUntilDemo() {
return new Promise(resolve => {
console.log('takeUntil operator example:\n');
console.log('Taking values for 1 second...\n');
// Create a stream of numbers every 200ms
const source$ = rxjs.interval(200);
// Create a notifier that emits after 1 second
const notifier$ = rxjs.timer(1000);
source$.pipe(
operators.takeUntil(notifier$)
).subscribe({
next: value => console.log(`Value: ${value}\n`),
complete: () => {
console.log('Completed after notifier fired\n');
resolve();
}
});
// Safety timeout
setTimeout(resolve, 1500);
});
}
await takeUntilDemo();
takeUntil operator example: Taking values for 1 second... Value: 0 Value: 1 Value: 2 Value: 3 Completed after notifier fired
[18]:
// skip - skips a specified number of values
const tenNumbers = rxjs.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
const skipFirstFive = tenNumbers.pipe(
operators.skip(5)
);
console.log('skip operator example:\n');
skipFirstFive.subscribe(value => console.log(`After skip: ${value}\n`));
skip operator example: After skip: 6 After skip: 7 After skip: 8 After skip: 9 After skip: 10
[37]:
// debounceTime with fixed timing
function debounceDemo() {
return new Promise(resolve => {
console.log('debounceTime operator example:\n');
console.log('Simulating typing with debounce...\n');
// Emit a sequence of values with specific timing to demonstrate debounce
const typing$ = rxjs.concat(
rxjs.of('h').pipe(operators.delay(100)),
rxjs.of('he').pipe(operators.delay(200)),
rxjs.of('hel').pipe(operators.delay(300)),
rxjs.of('hell').pipe(operators.delay(400)),
rxjs.of('hello').pipe(operators.delay(1000))
);
// Only emit when 500ms have passed without a new value
typing$.pipe(
operators.tap(term => console.log(`Typing: ${term}\n`)),
operators.debounceTime(500)
).subscribe({
next: value => console.log(`Debounced value: ${value}\n`),
complete: () => {
console.log('Debounce demo completed\n');
resolve();
}
});
// Ensure demo finishes
setTimeout(resolve, 3000);
});
}
await debounceDemo();
debounceTime operator example: Simulating typing with debounce... Typing: h Typing: he Typing: hel Typing: hell Debounced value: hell Typing: hello Debounced value: hello Debounce demo completed
[20]:
// distinctUntilChanged - only emits when the current value is different from the last
const repeatedValues = rxjs.of(1, 1, 2, 2, 2, 1, 1, 3, 3);
const distinct = repeatedValues.pipe(
operators.distinctUntilChanged()
);
console.log('distinctUntilChanged operator example:\n');
distinct.subscribe(value => console.log(`Distinct value: ${value}\n`));
distinctUntilChanged operator example: Distinct value: 1 Distinct value: 2 Distinct value: 1 Distinct value: 3
[M]:
5. Combination Operators
These operators combine multiple Observables.
[38]:
// merge - combines multiple Observables by merging their emissions
function mergeDemo() {
return new Promise(resolve => {
// Create two Observables with predictable timing
const first$ = rxjs.interval(200).pipe(
operators.map(v => `First: ${v}`),
operators.take(3)
);
const second$ = rxjs.interval(300).pipe(
operators.map(v => `Second: ${v}`),
operators.take(2)
);
console.log('merge operator example:\n');
rxjs.merge(first$, second$).subscribe({
next: value => console.log(`${value}\n`),
complete: () => {
console.log('Merge completed\n');
resolve();
}
});
// Ensure demo completes
setTimeout(resolve, 1500);
});
}
await mergeDemo();
merge operator example: First: 0 Second: 0 First: 1 Second: 1 First: 2 Merge completed
[22]:
// concat - subscribes to Observables in order, one at a time
const source1$ = rxjs.of('A', 'B', 'C');
const source2$ = rxjs.of('X', 'Y', 'Z');
const concatenated$ = rxjs.concat(source1$, source2$);
console.log('concat operator example:\n');
concatenated$.subscribe(value => console.log(`Concatenated: ${value}\n`));
concat operator example: Concatenated: A Concatenated: B Concatenated: C Concatenated: X Concatenated: Y Concatenated: Z
[39]:
// combineLatest with fixed timing
function combineLatestDemo() {
return new Promise(resolve => {
console.log('combineLatest operator example:\n');
// Create sources with predictable timing and limited emissions
const source1$ = rxjs.interval(300).pipe(operators.take(3));
const source2$ = rxjs.interval(500).pipe(operators.take(3));
rxjs.combineLatest([source1$, source2$]).subscribe({
next: ([val1, val2]) => console.log(`Combined: ${val1}, ${val2}\n`),
complete: () => {
console.log('combineLatest completed\n');
resolve();
}
});
// Ensure demo completes
setTimeout(resolve, 2000);
});
}
await combineLatestDemo();
combineLatest operator example: Combined: 0, 0 Combined: 1, 0 Combined: 2, 0 Combined: 2, 1 Combined: 2, 2 combineLatest completed
[24]:
// zip - combines values from multiple Observables by index
const source1$ = rxjs.of('A', 'B', 'C', 'D');
const source2$ = rxjs.of(1, 2, 3);
const zipped$ = rxjs.zip(source1$, source2$);
console.log('zip operator example:\n');
zipped$.subscribe(([letter, number]) => console.log(`Zipped: ${letter}${number}\n`));
zip operator example: Zipped: A1 Zipped: B2 Zipped: C3
[40]:
// forkJoin with fixed timing
function forkJoinDemo() {
return new Promise(resolve => {
console.log('forkJoin operator example:\n');
console.log('Waiting for all APIs to respond...\n');
// Simulate three API calls with different response times
const api1$ = rxjs.of('API 1 Response').pipe(operators.delay(300));
const api2$ = rxjs.of('API 2 Response').pipe(operators.delay(600));
const api3$ = rxjs.of('API 3 Response').pipe(operators.delay(900));
rxjs.forkJoin([api1$, api2$, api3$]).subscribe({
next: responses => console.log(`All responses: ${JSON.stringify(responses)}\n`),
complete: () => {
console.log('forkJoin completed\n');
resolve();
}
});
// Ensure demo finishes
setTimeout(resolve, 1500);
});
}
await forkJoinDemo();
forkJoin operator example: Waiting for all APIs to respond... All responses: ["API 1 Response","API 2 Response","API 3 Response"] forkJoin completed
[M]:
6. Error Handling
These operators help manage errors in your Observable streams.
[26]:
// catchError - handles errors by returning a new Observable
const errorSource$ = rxjs.concat(
rxjs.of(1, 2, 3),
rxjs.throwError(() => 'Something went wrong!'),
rxjs.of(4, 5, 6) // These values won't be emitted due to the error
);
const caught$ = errorSource$.pipe(
operators.catchError(error => {
console.log(`Error caught: ${error}\n`);
return rxjs.of('Fallback value');
})
);
console.log('catchError operator example:\n');
caught$.subscribe({
next: value => console.log(`Value: ${value}\n`),
error: err => console.log(`Error in subscription: ${err}\n`),
complete: () => console.log('Completed\n')
});
catchError operator example: Value: 1 Value: 2 Value: 3 Error caught: Something went wrong! Value: Fallback value Completed
[27]:
// retry - resubscribes to the source Observable a specified number of times
let attemptCount = 0;
const flaky$ = new rxjs.Observable(subscriber => {
attemptCount++;
console.log(`Attempt ${attemptCount}\n`);
if (attemptCount < 3) {
subscriber.error(`Failed attempt ${attemptCount}`);
} else {
subscriber.next('Success!');
subscriber.complete();
}
});
const retried$ = flaky$.pipe(
operators.retry(3)
);
console.log('retry operator example:\n');
retried$.subscribe({
next: value => console.log(`Value: ${value}\n`),
error: err => console.log(`Error after retries: ${err}\n`),
complete: () => console.log('Completed after retries\n')
});
retry operator example: Attempt 1 Attempt 2 Attempt 3 Value: Success! Completed after retries
[M]:
7. Common RxJS Patterns
Let's explore some common patterns and recipes using RxJS.
[41]:
// Pattern: Typeahead search simulation
function typeaheadPattern() {
return new Promise(resolve => {
console.log('Typeahead search pattern:\n');
// Simulate user typing terms with increasing delay
const userInput$ = rxjs.concat(
rxjs.of('r').pipe(operators.delay(100)),
rxjs.of('rx').pipe(operators.delay(200)),
rxjs.of('rxj').pipe(operators.delay(300)),
rxjs.of('rxjs').pipe(operators.delay(500))
);
// Search function simulation
function search(term) {
console.log(`API search for: ${term}\n`);
// Longer term = more results, slightly longer delay
return rxjs.of(`${term.length} results for "${term}"`).pipe(
operators.delay(300)
);
}
// Typeahead implementation
userInput$.pipe(
operators.tap(term => console.log(`User typed: ${term}\n`)),
operators.debounceTime(400), // Wait for user to pause typing
operators.distinctUntilChanged(), // Only search if term changed
operators.switchMap(term => search(term)) // Cancel previous search if new term arrives
).subscribe({
next: results => console.log(`Search results: ${results}\n`),
complete: () => {
console.log('Typeahead demo completed\n');
resolve();
}
});
Typeahead search pattern: User typed: r User typed: rx User typed: rxj API search for: rxj User typed: rxjs API search for: rxjs Search results: 4 results for "rxjs" Typeahead demo completed
[42]:
// Pattern: Caching with expiration
function cachingPattern() {
return new Promise(resolve => {
console.log('Caching pattern example:\n');
let cachedData = null;
let cachedTime = null;
let inFlightRequest = null;
const ttl = 1000; // Cache TTL in ms
// Function to get data, with caching
function getData() {
const now = Date.now();
// If we have valid cached data, return it
if (cachedData && cachedTime && (now - cachedTime < ttl)) {
console.log('Returning cached data\n');
return rxjs.of({ ...cachedData, fromCache: true });
}
// If a request is already in progress, return that
if (inFlightRequest) {
console.log('Returning in-flight request\n');
return inFlightRequest;
}
// Otherwise make a new request
console.log('Making new API request\n');
inFlightRequest = rxjs.of({
data: 'Fresh data',
timestamp: new Date().toISOString()
}).pipe(
operators.delay(500), // Simulate API delay
operators.tap(data => {
cachedData = data;
cachedTime = Date.now();
Caching pattern example:
First request:
Making new API request
Received: {"data":"Fresh data","timestamp":"2025-03-14T12:19:31.310Z"}
Second request (after 300ms):
Returning cached data
Received: {"data":"Fresh data","timestamp":"2025-03-14T12:19:31.310Z","fromCache":true}
Third request (after cache expiry):
Making new API request
Received: {"data":"Fresh data","timestamp":"2025-03-14T12:19:33.116Z"}
[43]:
// Pattern: Error handling and retry with backoff
function retryPattern() {
return new Promise(resolve => {
console.log('Retry with backoff pattern:\n');
let attempt = 0;
// Simulate a flaky API that succeeds on the third attempt
function flakyApi() {
return new rxjs.Observable(subscriber => {
attempt++;
console.log(`API attempt ${attempt}\n`);
// Fail the first two attempts
if (attempt < 3) {
subscriber.error(`Server error on attempt ${attempt}`);
} else {
subscriber.next({ success: true, data: 'API data payload' });
subscriber.complete();
}
});
}
// Implementation with exponential backoff
flakyApi().pipe(
operators.retryWhen(errors =>
errors.pipe(
operators.scan((attempts, error) => {
attempts++;
if (attempts >= 5) { // Max 5 retries
throw new Error(`Giving up after ${attempts} attempts: ${error}`);
}
console.log(`Retry attempt ${attempts} after error: ${error}\n`);
return attempts;
}, 0),
operators.tap(attempts => {
Retry with backoff pattern:
API attempt 1
Retry attempt 1 after error: Server error on attempt 1
Waiting 300ms before retry 1
API attempt 2
Retry attempt 2 after error: Server error on attempt 2
Waiting 600ms before retry 2
API attempt 3
Success! {"success":true,"data":"API data payload"}
Retry pattern demo completed
[M]:
8. Testing RxJS Observables
RxJS provides tools for testing Observables with marble diagrams and virtual time scheduling.
[31]:
// Basic marble testing example
const { TestScheduler } = require('rxjs/testing');
// Create a test scheduler
const testScheduler = new TestScheduler((actual, expected) => {
// Simple equality check
console.log('Actual:', JSON.stringify(actual), '\n');
console.log('Expected:', JSON.stringify(expected), '\n');
console.log('Equal:', JSON.stringify(actual) === JSON.stringify(expected), '\n');
});
// Run tests with virtual time
testScheduler.run(({ cold, expectObservable }) => {
// Create a cold observable with marble syntax
const source$ = cold('--a--b--c--|', { a: 1, b: 2, c: 3 });
// Apply operators
const result$ = source$.pipe(
operators.map(x => x * 10),
operators.filter(x => x > 10)
);
// Assert expected output
expectObservable(result$).toBe('-----b--c--|', { b: 20, c: 30 });
});
console.log('Marble testing demo complete\n');
Actual: [{"frame":5,"notification":{"kind":"N","value":20}},{"frame":8,"notification":{"kind":"N","value":30}},{"frame":11,"notification":{"kind":"C"}}]
Expected: [{"frame":5,"notification":{"kind":"N","value":20}},{"frame":8,"notification":{"kind":"N","value":30}},{"frame":11,"notification":{"kind":"C"}}]
Equal: true
Marble testing demo complete
[M]:
9. Implementing Real-world Scenarios
Let's solve some practical problems with RxJS.
[44]:
// Scenario: Implementing a rate limiter for API requests
function rateLimiterDemo() {
return new Promise(resolve => {
console.log('API rate limiter implementation:\n');
// Simulated API call function
function callApi(id) {
console.log(`API call for ID ${id} started\n`);
return rxjs.of(`Result for ID ${id}`).pipe(
operators.delay(300) // Simulate API latency
);
}
// Create a rate-limited API wrapper with concurrency limit
function createRateLimitedApi(maxConcurrent = 2, requestsPerSecond = 5) {
// Subject to receive IDs to process
const requestQueue = new rxjs.Subject();
// Process the queue with rate limiting
const results$ = requestQueue.pipe(
// Limit requests per second
operators.concatMap(id => {
return rxjs.of(id).pipe(
operators.delay(1000 / requestsPerSecond) // Ensure spacing between requests
);
}),
// Limit concurrent requests
operators.mergeMap(id => callApi(id), maxConcurrent),
operators.share() // Share the result stream for multiple subscribers
);
// Subscribe to process results
results$.subscribe(result => console.log(`Result received: ${result}\n`));
// Return function to queue requests
return {
API rate limiter implementation: Queueing request for ID 1 Queueing request for ID 2 Queueing request for ID 3 Queueing request for ID 4 Queueing request for ID 5 API call for ID 1 started API call for ID 2 started Queueing request for ID 6 Queueing request for ID 7 Result received: Result for ID 1 API call for ID 3 started Result received: Result for ID 2 API call for ID 4 started Result received: Result for ID 3 API call for ID 5 started Result received: Result for ID 4 API call for ID 6 started Result received: Result for ID 5 API call for ID 7 started Result received: Result for ID 6 Result received: Result for ID 7
[M]:
10. Performance Considerations and Best Practices
Here are some tips for optimizing RxJS performance:
- Always unsubscribe from long-lived Observables to prevent memory leaks
- Use pipeable operators instead of prototype operators for better tree-shaking
- Be careful with
shareReplay- use{refCount: true}config to avoid memory leaks - Use
share()for multicasting when multiple subscribers need the same data - Avoid excessive operators in your pipes - each adds overhead
- Use
take(1)for one-time operations which automatically completes - Consider
Subjectfor event buses which is more efficient than multiple Observables - Use appropriate flattening operators:
switchMapfor scenarios where you only care about the latest value (like search)mergeMapwhen you need all results quickly but don't care about orderconcatMapwhen order is importantexhaustMapwhen you want to ignore new events during processing
Wrapping Up
This notebook has covered the essential operators and patterns in RxJS for reactive programming. We've explored synchronous and asynchronous Observable creation, transformation, filtering, combination, error handling, and practical real-world patterns.
Reactive programming with RxJS gives you powerful tools to handle complex async operations and data flows in a declarative way. While it has a learning curve, the patterns we've covered will help you write more maintainable and predictable asynchronous code.
Sign in to save your work and access it from anywhere