Reaktive Programmierung in JavaScript (Node.js): RxJS & Observables

Reaktive Programmierung in JavaScript (Node.js) mit der Bibliothek RxJS

Reaktive Programmierung ist ein Paradigma, das sich auf asynchrone Datenströme und die Verbreitung von Änderungen konzentriert. Im Node.js-Ökosystem wird dieses Konzept durch die Bibliothek RxJS (Reactive Extensions for JavaScript) umgesetzt. Anstelle der traditionellen Arbeit mit Callbacks oder Promises bietet RxJS die Arbeit mit Observable – Objekten, die einen Datenstrom im Zeitverlauf darstellen.

Grundlegende Konzepte

RxJS basiert auf drei zentralen Entitäten:

• Observable – eine Datenquelle, die im Zeitverlauf Werte erzeugen kann (Mausereignisse, HTTP-Anfragen, Dateilesen).
• Observer – ein Konsument, der ein Observable abonniert und auf drei Arten von Ereignissen reagiert: next (neuer Wert), error (Fehler), complete (Abschluss des Stroms).
• Operators – reine Funktionen zur Transformation, Filterung und Kombination von Strömen (z. B. map, filter, mergeMap).

Installation und Einrichtung

Um RxJS in Node.js zu verwenden, installieren Sie das Paket über npm:

npm install rxjs

Einbindung in die Datei:

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

Erstellen eines Observable

Ein einfaches Beispiel – Erstellen eines Zahlenstroms:

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('Erhalten:', value),    error: err => console.error('Fehler:', err),    complete: () => console.log('Strom abgeschlossen')});

// Ausgabe: Erhalten: 1, Erhalten: 2, Erhalten: 3, Strom abgeschlossen

Operatoren: map und filter

Operatoren ermöglichen die Verarbeitung von Daten im Strom ohne Nebenwirkungen. Beispiel mit Filterung gerader Zahlen und 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), // nur gerade    map(x => x * 10)          // mit 10 multiplizieren).subscribe(result => console.log(result));

// Ausgabe: 20, 40, 60

Arbeit mit asynchronen Operationen

RxJS eignet sich hervorragend für die Verarbeitung von Ein-/Ausgabe-Ereignissen. Beispiel zum Lesen einer Datei mit from:

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

// Datei als Promise lesenconst readFilePromise = (filePath) => {    return new Promise((resolve, reject) => {        fs.readFile(filePath, 'utf8', (err, data) => {            if (err) reject(err);            else resolve(data);        });    });};

// Observable aus Promise erstellenconst fileStream$ = from(readFilePromise('./data.txt'));

fileStream$.pipe(    map(content => content.toUpperCase())).subscribe({    next: data => console.log('Dateiinhalt:', data),    error: err => console.error('Lesefehler:', err)});

Kombinieren von Strömen

Eine der Stärken von RxJS ist die Möglichkeit, mehrere Ströme zu kombinieren. Beispiel mit merge (Zusammenführung von Strömen):

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

// Strom A: alle 1 Sekundeconst streamA$ = interval(1000).pipe(    take(3),    map(x => `A: ${x}`));

// Strom B: alle 1,5 Sekundenconst streamB$ = interval(1500).pipe(    take(3),    map(x => `B: ${x}`));

// Zusammenführen