Programación reactiva en JavaScript (Node.js): librería RxJS

Programación reactiva en JavaScript (Node.js) con 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 JavaScript (Node.js), una de las librerías más potentes para implementar este paradigma es RxJS (Reactive Extensions for JavaScript). Proporciona herramientas para trabajar con Observable (secuencias observables), permitiendo a los desarrolladores procesar eficientemente eventos, solicitudes HTTP, temporizadores y otras fuentes de datos asíncronas.

Conceptos principales de RxJS

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

• Observable (Objeto Observable) — representa un flujo de datos que puede generar valores a lo largo del tiempo.
• Observer (Observador) — objeto que se suscribe al Observable y reacciona a la llegada de nuevos datos (next), errores (error) o la finalización del flujo (complete).
• Subscription (Suscripción) — resultado de llamar al método subscribe(). Permite gestionar el ciclo de vida de la suscripción (por ejemplo, cancelarla).

También juegan un papel importante los operadores — funciones puras que permiten transformar, filtrar, combinar y componer flujos de datos.

Instalación y configuración

Para usar RxJS en un proyecto Node.js, basta con instalar el paquete a través de npm:

npm install rxjs

Una vez instalada, la librería está lista para usar. Importe los componentes necesarios:

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

Creación de Observable

Existen varias formas de crear flujos de datos. Veamos las básicas:

1. Creación a partir de un array

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

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

Resultado de la ejecución:

Recibido: 1Recibido: 2Recibido: 3Recibido: 4Recibido: 5Flujo completado

2. Generación periódica de valores

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

const timer$ = interval(1000).pipe(    take(4) // limitamos a 4 valores);

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

Salida (con intervalo de 1 segundo):

Tick 0Tick 1Tick 2Tick 3Temporizador completado

Trabajo con operadores

Los operadores son el corazón de RxJS. Permiten transformar datos sobre la marcha. Veamos dos operadores populares:

• map — transforma cada valor del flujo.
• filter — solo deja pasar los valores que cumplen una condición.

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), // dejamos solo los pares    map(value => value * 10)          // multiplicamos cada uno por 10).subscribe({    next: value => console.log(`Resultado: ${value}`),    complete: () => console.log('Listo')});

Resultado:

Resultado: 20Resultado: 40Resultado: 60Listo

Manejo de errores en flujos reactivos

RxJS proporciona mecanismos integrados para el manejo de errores. Por ejemplo, el operador catchError permite interceptar una excepción y continuar la ejecución:

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

const error$ = throwError(() =>