Programação Reativa em JavaScript (Node.js): RxJS & Observables

Programação Reativa em JavaScript (Node.js) com a biblioteca RxJS

Programação reativa é um paradigma orientado a fluxos de dados assíncronos e propagação de mudanças. No ecossistema Node.js, este conceito é implementado através da biblioteca RxJS (Reactive Extensions for JavaScript). Em vez do trabalho tradicional com callbacks ou promises, o RxJS oferece trabalho com Observable — objetos que representam um fluxo de dados ao longo do tempo.

Conceitos principais

Na base do RxJS estão três entidades-chave:

• Observable — fonte de dados que pode gerar valores ao longo do tempo (eventos de mouse, requisições HTTP, leitura de arquivos).
• Observer — consumidor que se inscreve no Observable e reage a três tipos de eventos: next (novo valor), error (erro), complete (término do fluxo).
• Operators — funções puras para transformação, filtragem e combinação de fluxos (por exemplo, map, filter, mergeMap).

Instalação e configuração

Para trabalhar com RxJS no Node.js, instale o pacote via npm:

npm install rxjs

Importação no arquivo:

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

Criação de Observable

Exemplo mais simples — criação de um fluxo de números:

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

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

// Inscriçãostream$.subscribe({    next: value => console.log('Recebido:', value),    error: err => console.error('Erro:', err),    complete: () => console.log('Fluxo concluído')});

// Saída: Recebido: 1, Recebido: 2, Recebido: 3, Fluxo concluído

Operadores: map e filter

Operadores permitem processar dados no fluxo sem criar efeitos colaterais. Exemplo com filtragem de números pares e transformação:

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

// Saída: 20, 40, 60

Trabalhando com operações assíncronas

RxJS é excelente para processar eventos de entrada/saída. Por exemplo, leitura de arquivo usando from:

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

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

// Criação de Observable a partir de Promiseconst fileStream$ = from(readFilePromise('./data.txt'));

fileStream$.pipe(    map(content => content.toUpperCase())).subscribe({    next: data => console.log('Conteúdo do arquivo:', data),    error: err => console.error('Erro de leitura:', err)});

Combinação de fluxos

Um dos pontos fortes do RxJS é a capacidade de combinar múltiplos fluxos. Exemplo com merge (junção de fluxos):

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

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

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

// Combina