Programación Reactiva en JavaScript (Node.js): RxJS y Observables

Programación Reactiva en JavaScript (Node.js) con la librería RxJS

La programación reactiva es un paradigma orientado a flujos de datos asíncronos y la propagación de cambios. En el ecosistema de Node.js, este concepto se implementa a través de la librería RxJS (Reactive Extensions for JavaScript). En lugar del trabajo tradicional con callbacks o promesas, RxJS ofrece trabajar con Observable — objetos que representan un flujo de datos en el tiempo.

Conceptos principales

En el núcleo de RxJS hay tres entidades clave:

• Observable — fuente de datos que puede generar valores a lo largo del tiempo (eventos del ratón, peticiones HTTP, lectura de archivos).
• Observer — consumidor que se suscribe al Observable y reacciona a tres tipos de eventos: next (nuevo valor), error (error), complete (finalización del flujo).
• Operators — funciones puras para transformar, filtrar y combinar flujos (por ejemplo, map, filter, mergeMap).

Instalación y configuración

Para trabajar con RxJS en Node.js, instale el paquete a través de npm:

npm install rxjs

Importación en el archivo:

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

Creación de un Observable

El ejemplo más simple: creación de un flujo de números:

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

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

// Suscripciónstream$.subscribe({    next: value => console.log('Recibido:', value),    error: err => console.error('Error:', err),    complete: () => console.log('Flujo completado')});

// Salida: Recibido: 1, Recibido: 2, Recibido: 3, Flujo completado

Operadores: map y filter

Los operadores permiten procesar datos en el flujo sin crear efectos secundarios. Ejemplo con filtrado de números pares y transformación:

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), // solo pares    map(x => x * 10)          // multiplicamos por 10).subscribe(result => console.log(result));

// Salida: 20, 40, 60

Trabajo con operaciones asíncronas

RxJS es ideal para manejar eventos de entrada/salida. Por ejemplo, lectura de un archivo usando from:

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

// Lectura de archivo como Promiseconst readFilePromise = (filePath) => {    return new Promise((resolve, reject) => {        fs.readFile(filePath, 'utf8', (err, data) => {            if (err) reject(err);            else resolve(data);        });    });};

// Creación de Observable a partir de Promiseconst fileStream$ = from(readFilePromise('./data.txt'));

fileStream$.pipe(    map(content => content.toUpperCase())).subscribe({    next: data => console.log('Contenido del archivo:', data),    error: err => console.error('Error de lectura:', err)});

Combinación de flujos

Una de las fortalezas de RxJS es la capacidad de combinar múltiples flujos. Ejemplo con merge (unión de flujos):

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

// Flujo A: cada 1 segundoconst streamA$ = interval(1000).pipe(    take(3),    map(x => `A: ${x}`));

// Flujo B: cada 1.5 segundosconst streamB$ = interval(1500).pipe(    take(3),    map(x => `B: ${x}`));

// Combinación