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

Реактивное программирование в JavaScript (Node.js) с библиотекой RxJS

Реактивное программирование — это парадигма, ориентированная на асинхронные потоки данных и распространение изменений. В экосистеме Node.js эта концепция реализуется через библиотеку RxJS (Reactive Extensions for JavaScript). Вместо традиционной работы с колбэками или промисами, RxJS предлагает работу с Observable — объектами, которые представляют поток данных во времени.

Основные концепции

В основе RxJS лежат три ключевые сущности:

• Observable — источник данных, который может генерировать значения с течением времени (события мыши, HTTP-запросы, чтение файлов).
• Observer — потребитель, который подписывается на Observable и реагирует на три типа событий: next (новое значение), error (ошибка), complete (завершение потока).
• Operators — чистые функции для трансформации, фильтрации и комбинирования потоков (например, map, filter, mergeMap).

Установка и настройка

Для работы с RxJS в Node.js установите пакет через npm:

npm install rxjs

Подключение в файле:

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

Создание Observable

Простейший пример — создание потока чисел:

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

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

// Подписка
stream$.subscribe({
    next: value => console.log('Получено:', value),
    error: err => console.error('Ошибка:', err),
    complete: () => console.log('Поток завершен')
});

// Вывод: Получено: 1, Получено: 2, Получено: 3, Поток завершен

Операторы: map и filter

Операторы позволяют обрабатывать данные в потоке без создания побочных эффектов. Пример с фильтрацией четных чисел и преобразованием:

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), // только четные
    map(x => x * 10)          // умножаем на 10
).subscribe(result => console.log(result));

// Вывод: 20, 40, 60

Работа с асинхронными операциями

RxJS отлично подходит для обработки событий ввода/вывода. Например, чтение файла с использованием from:

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

// Чтение файла как Promise
const readFilePromise = (filePath) => {
    return new Promise((resolve, reject) => {
        fs.readFile(filePath, 'utf8', (err, data) => {
            if (err) reject(err);
            else resolve(data);
        });
    });
};

// Создание Observable из Promise
const fileStream$ = from(readFilePromise('./data.txt'));

fileStream$.pipe(
    map(content => content.toUpperCase())
).subscribe({
    next: data => console.log('Содержимое файла:', data),
    error: err => console.error('Ошибка чтения:', err)
});

Комбинирование потоков

Одна из сильных сторон RxJS — возможность комбинировать несколько потоков. Пример с merge (объединение потоков):

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

// Поток A: каждые 1 секунду
const streamA$ = interval(1000).pipe(
    take(3),
    map(x => `A: ${x}`)
);

// Поток B: каждые 1.5 секунды
const streamB$ = interval(1500).pipe(
    take(3),
    map(x => `B: ${x}`)
);

// Объединяем потоки
merge(streamA$, streamB$).subscribe(value => console.log(value));

// Примерный вывод:
// A: 0 (через 1 сек)
// B: 0 (через 1.5 сек)
// A: 1 (через 2 сек)
// A: 2 (через 3 сек)
// B: 1 (через 3 сек)
// B: 2 (через 4.5 сек)

Обработка ошибок

В реактивном программировании ошибки обрабатываются как часть потока. Оператор catchError позволяет перехватить ошибку и продолжить выполнение:

const { throw