Reactive Programming in JavaScript (Node.js): RxJS Library

Reactive Programming in JavaScript (Node.js) with RxJS

Reactive programming is a paradigm focused on asynchronous data streams and the propagation of changes. In the JavaScript (Node.js) ecosystem, one of the most powerful libraries for implementing this paradigm is RxJS (Reactive Extensions for JavaScript). It provides tools for working with Observables (observable sequences), allowing developers to efficiently handle events, HTTP requests, timers, and other asynchronous data sources.

Core Concepts of RxJS

At the heart of RxJS are three key components:

• Observable — represents a data stream that can generate values over time.
• Observer — an object that subscribes to an Observable and reacts to new data (next), errors (error), or stream completion (complete).
• Subscription — the result of calling the subscribe() method. It allows managing the lifecycle of a subscription (e.g., cancelling it).

Also important are operators — pure functions that allow transforming, filtering, merging, and combining data streams.

Installation and Setup

To use RxJS in a Node.js project, simply install the package via npm:

npm install rxjs

After installation, the library is ready to use. Import the necessary components:

const { Observable, of, from, interval } = require('rxjs');const { map, filter, take } = require('rxjs/operators');

Creating an Observable

There are several ways to create data streams. Let's look at the basics:

1. Creating from an array

const { from } = require('rxjs');

const numbers$ = from([1, 2, 3, 4, 5]);numbers$.subscribe({    next: value => console.log(`Received: ${value}`),    complete: () => console.log('Stream completed')});

Execution result:

Received: 1Received: 2Received: 3Received: 4Received: 5Stream completed

2. Periodic value generation

const { interval } = require('rxjs');const { take } = require('rxjs/operators');

const timer$ = interval(1000).pipe(    take(4) // limit to 4 values);

timer$.subscribe({    next: value => console.log(`Tick ${value}`),    complete: () => console.log('Timer completed')});

Output (with a 1-second interval):

Tick 0Tick 1Tick 2Tick 3Timer completed

Working with Operators

Operators are the heart of RxJS. They allow transforming data on the fly. Let's look at two popular operators:

• map — transforms each value in the stream.
• filter — only passes values that satisfy a condition.

const { of } = require('rxjs');const { map, filter } = require('rxjs/operators');

const source$ = of(1, 2, 3, 4, 5, 6);

source$.pipe(    filter(value => value % 2 === 0), // keep only even numbers    map(value => value * 10)          // multiply each by 10).subscribe({    next: value => console.log(`Result: ${value}`),    complete: () => console.log('Done')});

Result:

Result: 20Result: 40Result: 60Done

Error Handling in Reactive Streams

RxJS provides built-in mechanisms for error handling. For example, the catchError operator allows catching an exception and continuing execution:

const { throwError, of } = require('rxjs');const { catchError } = require('rxjs/operators');

const error$ = throwError(() =>