Programmation Réactive en JavaScript (Node.js) : RxJS & Observables

Programmation Réactive en JavaScript (Node.js) avec la bibliothèque RxJS

La programmation réactive est un paradigme orienté vers les flux de données asynchrones et la propagation des changements. Dans l'écosystème Node.js, ce concept est implémenté via la bibliothèque RxJS (Reactive Extensions for JavaScript). Au lieu du travail traditionnel avec des callbacks ou des promesses, RxJS propose de travailler avec des Observable — des objets qui représentent un flux de données dans le temps.

Concepts fondamentaux

À la base de RxJS se trouvent trois entités clés :

• Observable — une source de données qui peut générer des valeurs au fil du temps (événements de souris, requêtes HTTP, lecture de fichiers).
• Observer — un consommateur qui s'abonne à un Observable et réagit à trois types d'événements : next (nouvelle valeur), error (erreur), complete (fin du flux).
• Operators — des fonctions pures pour transformer, filtrer et combiner des flux (par exemple, map, filter, mergeMap).

Installation et configuration

Pour travailler avec RxJS dans Node.js, installez le paquet via npm :

npm install rxjs

Importation dans le fichier :

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

Création d'un Observable

Exemple simple — création d'un flux de nombres :

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

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

// Abonnementstream$.subscribe({    next: value => console.log('Reçu :', value),    error: err => console.error('Erreur :', err),    complete: () => console.log('Flux terminé')});

// Sortie : Reçu : 1, Reçu : 2, Reçu : 3, Flux terminé

Opérateurs : map et filter

Les opérateurs permettent de traiter les données dans le flux sans créer d'effets secondaires. Exemple avec filtrage des nombres pairs et transformation :

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), // uniquement les pairs    map(x => x * 10)          // multiplier par 10).subscribe(result => console.log(result));

// Sortie : 20, 40, 60

Travail avec les opérations asynchrones

RxJS est parfait pour traiter les événements d'entrée/sortie. Par exemple, la lecture d'un fichier avec from :

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

// Lecture de fichier en tant que Promiseconst readFilePromise = (filePath) => {    return new Promise((resolve, reject) => {        fs.readFile(filePath, 'utf8', (err, data) => {            if (err) reject(err);            else resolve(data);        });    });};

// Création d'un Observable à partir d'une Promiseconst fileStream$ = from(readFilePromise('./data.txt'));

fileStream$.pipe(    map(content => content.toUpperCase())).subscribe({    next: data => console.log('Contenu du fichier :', data),    error: err => console.error('Erreur de lecture :', err)});

Combinaison de flux

L'un des points forts de RxJS est la possibilité de combiner plusieurs flux. Exemple avec merge (fusion de flux) :

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

// Flux A : toutes les 1 secondeconst streamA$ = interval(1000).pipe(    take(3),    map(x => `A : ${x}`));

// Flux B : toutes les 1,5 secondesconst streamB$ = interval(1500).pipe(    take(3),    map(x => `B : ${x}`));

// Fusion