Reactive Programming in JavaScript (Node.js): RxJS & Observables

Reactive Programming in JavaScript (Node.js) with the RxJS Library

Reactive programming is a paradigm focused on asynchronous data streams and the propagation of changes. In the Node.js ecosystem, this concept is implemented through the RxJS library (Reactive Extensions for JavaScript). Instead of traditional work with callbacks or promises, RxJS offers work with Observable — objects that represent a stream of data over time.

Core Concepts

At the heart of RxJS are three key entities:

• Observable — a data source that can generate values over time (mouse events, HTTP requests, file reads).
• Observer — a consumer that subscribes to an Observable and reacts to three types of events: next (new value), error (error), complete (stream completion).
• Operators — pure functions for transforming, filtering, and combining streams (e.g., map, filter, mergeMap).

Installation and Setup

To work with RxJS in Node.js, install the package via npm:

npm install rxjs

Importing in a file:

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

Creating an Observable

The simplest example — creating a stream of numbers:

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

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

// Subscriptionstream$.subscribe({    next: value => console.log('Received:', value),    error: err => console.error('Error:', err),    complete: () => console.log('Stream completed')});

// Output: Received: 1, Received: 2, Received: 3, Stream completed

Operators: map and filter

Operators allow you to process data in a stream without creating side effects. Example with filtering even numbers and transforming:

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

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

numbers$.pipe(    filter(x => x % 2 === 0), // only even    map(x => x * 10)          // multiply by 10).subscribe(result => console.log(result));

// Output: 20, 40, 60

Working with Asynchronous Operations

RxJS is great for handling input/output events. For example, reading a file using from:

const { from } = require('rxjs');const { map } = require('rxjs/operators');const fs = require('fs');const path = require('path');

// Reading a file as a Promiseconst readFilePromise = (filePath) => {    return new Promise((resolve, reject) => {        fs.readFile(filePath, 'utf8', (err, data) => {            if (err) reject(err);            else resolve(data);        });    });};

// Creating an Observable from a Promiseconst fileStream$ = from(readFilePromise('./data.txt'));

fileStream$.pipe(    map(content => content.toUpperCase())).subscribe({    next: data => console.log('File content:', data),    error: err => console.error('Read error:', err)});

Combining Streams

One of the strong points of RxJS is the ability to combine multiple streams. Example with merge (merging streams):

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

// Stream A: every 1 secondconst streamA$ = interval(1000).pipe(    take(3),    map(x => `A: ${x}`));

// Stream B: every 1.5 secondsconst streamB$ = interval(1500).pipe(    take(3),    map(x => `B: ${x}`));

// Merging