# RxJS in Node.js: Ereignisströme reaktiv verarbeiten URL: https://www.mikebild.dev/de/blog/rxjs-in-node-ereignisstroeme/ ```js let buffer = []; let timer = null; emitter.on('data', line => { const entry = JSON.parse(line); if (entry.level !== 'error') return; buffer.push(entry); if (!timer) { timer = setTimeout(() => { saveErrors(buffer); buffer = []; timer = null; }, 1000); } }); ``` Drei Anforderungen stecken in diesem Handler: parsen, filtern, im Sekundentakt gebündelt wegschreiben. Der Zwischenzustand liegt in zwei Modulvariablen, geprüft wird er nur noch mit Zeitgeber-Attrappen. Kommt morgen ein zweiter Konsument dazu oder soll zusätzlich nach Dienst gruppiert werden, wächst die Verschachtelung weiter – und mit ihr die Zahl der Stellen, an denen der geteilte Zustand kippen kann. RxJS setzt genau hier an: überall dort, wo aus einem Ereignisstrom ein komponierbares Programm werden soll. Wer sich das imperative Fundament schon angesehen hat – etwa im Beitrag zu [EventEmitter und Ereignismodell](https://www.mikebild.dev/de/blog/nodejs-eventemitter-und-ereignismodell/) – erkennt in RxJS dieselbe Idee, nur von der anderen Seite aufgezäumt. ## Von on/emit zu deklarativen Strömen Der `EventEmitter` ist imperativ. Ich sage: „Wenn `data` passiert, führe diese Funktion aus.“ Das ist ein Befehl. Jeder Handler ist eine eigene kleine Insel, und wenn ich mehrere Schritte verketten will, verknüpfe ich sie von Hand über gemeinsamen Zustand. RxJS beschreibt denselben Ereignisstrom deklarativ. Ich sage nicht mehr „tu dies, wenn jenes passiert“, sondern „dieser Strom ist die Quelle, transformiert durch diese Kette von Schritten“. Die Kette ist eine Beschreibung, kein Ablauf. Erst beim `subscribe` wird sie aktiviert. Die Brücke zwischen beiden Welten ist `fromEvent`. Der Operator nimmt einen Node-`EventEmitter` und einen Ereignisnamen und liefert ein Observable: ```js const { EventEmitter } = require('events'); const { fromEvent } = require('rxjs'); const emitter = new EventEmitter(); const data$ = fromEvent(emitter, 'data'); const subscription = data$.subscribe({ next: value => console.log('received', value), }); emitter.emit('data', { level: 'info', message: 'server started' }); ``` `fromEvent(emitter, 'data')` ändert nichts an der Semantik des Emitters. Es registriert bei jedem `subscribe` einen Listener über `addListener` und entfernt ihn bei `unsubscribe` wieder. Das Observable ist damit ein Adapter: Der Emitter bleibt die Quelle, RxJS legt nur eine komponierbare Schicht darüber. Wer den Unterschied zwischen dem Push-Modell von Observables und dem Pull-Modell von Iteratoren vertiefen will, findet das im Beitrag zu [RxJS: Push und Pull](https://www.mikebild.dev/de/blog/rxjs-push-und-pull/). Ein Hinweis zur Konvention: Ich hänge an Variablen, die ein Observable halten, ein `$` an (`data$`). Das ist kein Zwang, hat sich in der RxJS-Welt aber eingebürgert und macht im Code sofort sichtbar, was ein Strom ist und was ein einzelner Wert. ## Die Pipeline: Operatoren als reine Funktionen Der eigentliche Gewinn liegt in den Operatoren. Sie sind reine Funktionen, die ein Observable entgegennehmen und ein neues zurückgeben. Verkettet werden sie über `.pipe(...)`. In RxJS 7 importiere ich alles – Erzeugungsfunktionen wie `fromEvent` und Operatoren wie `map` – direkt aus `'rxjs'`. Der frühere Import aus `'rxjs/operators'` funktioniert noch, ist aber überholt. Ein typisches serverseitiges Beispiel ist Log-Verarbeitung. Ich möchte eingehende Rohzeilen zu Objekten parsen, nur die Fehler behalten und diese pro Sekunde gebündelt wegschreiben, statt bei jeder einzelnen Zeile die Datenbank anzufassen: ```js const { fromEvent } = require('rxjs'); const { map, filter, bufferTime } = require('rxjs'); const logStream$ = fromEvent(logEmitter, 'line').pipe( map(line => JSON.parse(line)), filter(entry => entry.level === 'error'), bufferTime(1000), ); const subscription = logStream$.subscribe({ next: batch => saveErrors(batch), error: err => console.error('stream failed', err), }); ``` Was hier steht, liest sich fast wie eine Beschreibung der Anforderung. `map` transformiert jeden Wert, `filter` lässt nur passende durch, `bufferTime(1000)` sammelt über ein Zeitfenster von einer Sekunde und emittiert dann ein Array. Der `next`-Handler bekommt also nicht einzelne Einträge, sondern jeweils ein Bündel – oder ein leeres Array, wenn im Fenster nichts kam. Jeder Schritt ist für sich verständlich und für sich testbar. Und wenn morgen die Anforderung kommt, zusätzlich nach Dienst zu gruppieren, füge ich einen Operator ein, statt eine verschachtelte Callback-Struktur umzubauen. Ein zweites Muster aus der Praxis: Sensordaten drosseln und dabei einen laufenden Mittelwert bilden. Angenommen, ein Gerät feuert deutlich häufiger, als ich verarbeiten will: ```js const { fromEvent } = require('rxjs'); const { throttleTime, scan } = require('rxjs'); const average$ = fromEvent(sensor, 'reading').pipe( map(event => event.value), throttleTime(500), scan((sum, value, index) => (sum * index + value) / (index + 1), 0), ); average$.subscribe({ next: avg => console.log('running average', avg) }); ``` `throttleTime(500)` lässt einen Wert durch und ignoriert dann für 500 Millisekunden alle weiteren. `scan` verhält sich wie `reduce`, emittiert aber jeden Zwischenstand – so bekomme ich bei jeder durchgelassenen Messung den aktualisierten Durchschnitt. Diese Kombination aus laufender Aggregation und Zeitfenster ist im Repository unter `operators.js` vorgezeichnet, wenn auch dort noch mit der älteren RxJS-6-Importweise. Wenn ich pro Ereignis eine asynchrone Weiterverarbeitung anstoßen will – etwa Details nachladen – hilft `mergeMap`. Es mappt jeden Wert auf ein inneres Observable und flacht die Ergebnisse nebenläufig ab. Der zweite Parameter begrenzt, wie viele innere Verarbeitungen gleichzeitig laufen dürfen: ```js const { mergeMap } = require('rxjs'); const details$ = ids$.pipe( mergeMap(id => fetchDetails(id), 4), ); ``` Mit `mergeMap(id => fetchDetails(id), 4)` laufen höchstens vier Abrufe parallel. Diese Begrenzung ist wichtig, und zwar aus einem Grund, der gleich noch eine eigene Rolle spielt. Der grobe Fluss durch so eine Pipeline sieht immer gleich aus: ```mermaid flowchart LR E[EventEmitter] -->|emit data| F["fromEvent
(Observable)"] F --> M[map] M --> FI[filter] FI --> B[bufferTime] B --> S[Subscriber] ``` ## Ein Subject: aus einem Strom viele Konsumenten machen Bis hierher hatte jeder Strom genau einen Verbraucher. Sobald mehrere Teile der Anwendung denselben Ereignisstrom brauchen, kommt das `Subject` ins Spiel. Ein `Subject` ist zugleich Observer und Observable: Ich kann Werte hineingeben und andere können es abonnieren. Es verteilt jeden Wert an alle Subscriber gleichzeitig – Multicast. ```js const { fromEvent, Subject } = require('rxjs'); const bus = new Subject(); // one source feeds the bus fromEvent(emitter, 'data').subscribe(bus); // multiple consumers share the same stream const loggerSub = bus.subscribe({ next: v => logger.write(v) }); const metricsSub = bus.subscribe({ next: v => metrics.record(v) }); // clean up later loggerSub.unsubscribe(); metricsSub.unsubscribe(); ``` Der Unterschied zum direkten Abonnieren des Quell-Observables ist grundlegend. Würden `logger` und `metrics` beide `fromEvent(emitter, 'data')` abonnieren, hinge am Emitter zweimal ein Listener und die Quelle würde zweimal ausgeführt. Über das `Subject` gibt es eine Ausführung, deren Werte an alle Subscriber verteilt werden. Genau das ist der Kern der Unterscheidung zwischen Unicast und Multicast: ```mermaid flowchart LR P[Producer] --> SUB[Subject] SUB --> C1[Logger-Subscriber] SUB --> C2[Metrics-Subscriber] SUB --> C3[Audit-Subscriber] ``` Neben dem einfachen `Subject` gibt es Varianten, die je nach Anwendungsfall passen: - `BehaviorSubject(seed)` hält den letzten Wert und liefert ihn jedem neuen Subscriber sofort. Praktisch für Zustand, den ein spät hinzukommender Konsument kennen muss. - `ReplaySubject(bufferSize)` puffert die letzten n Werte und spielt sie neuen Subscribern vor. - `AsyncSubject` gibt nur den allerletzten Wert aus, und das erst beim `complete`. Im Repository unter `subject.js` sieht man das Grundprinzip: dieselbe Quelle, mehrere `subscribe`-Aufrufe. Dort steht noch die positionsbasierte `subscribe(next, error, complete)`-Form. In RxJS 7 nutze ich stattdessen durchgängig den Objekt-Observer `subscribe({ next, error, complete })` – dazu gleich mehr bei den Fallstricken. ## Hot und cold: der Unterschied, der Ergebnisse verfälscht Ein Punkt, an dem in Teams zuverlässig Verwirrung entsteht, ist die Frage, wann die Quelle eines Observables entsteht. Ein kaltes Observable erzeugt seinen Producer pro `subscribe`. Jeder Subscriber bekommt eine eigene, frische Ausführung – Unicast. Ein `from([1, 2, 3])` ist kalt: Jeder, der es abonniert, sieht die Werte 1, 2, 3 von vorne. Ein heißes Observable hat seinen Producer außerhalb. Die Werte entstehen unabhängig davon, ob jemand zuhört. Wer spät abonniert, verpasst, was vorher lief – Multicast. Ein `Subject` ist heiß, und ein `fromEvent` über einen bereits laufenden Emitter verhält sich ebenfalls heiß: Die Ereignisse feuern ohnehin, ich klinke mich nur ein. Diese Unterscheidung ist keine Theorie. Wer erwartet, dass „jeder Subscriber frisch startet“, aber tatsächlich einen heißen Strom vor sich hat, bekommt bei spät hinzukommenden Konsumenten schlicht falsche Ergebnisse – fehlende Anfangswerte, eine Aggregation, die mitten im Strom beginnt. Wenn ich will, dass auch ein später Subscriber den letzten Stand oder die Historie sieht, ist genau das der Fall für `BehaviorSubject`, `ReplaySubject` oder den Operator `shareReplay`. ## Die wichtige Abgrenzung: RxJS hat keine echte Backpressure Dieser Punkt gehört früh geklärt, sonst entstehen Erwartungen, die das Werkzeug nicht einlöst. Wer Node-Streams kennt, kennt Backpressure: Ein lesbarer Stream kann pausieren, ein `highWaterMark` begrenzt den Puffer, `pipe` regelt den Fluss zwischen Producer und Konsument automatisch. Der schnelle Producer wird gebremst, wenn der langsame Konsument nicht hinterherkommt. RxJS hat das nicht. Es gibt keinen eingebauten Mechanismus, mit dem ein Subscriber die Quelle zum Langsamerwerden zwingt. Ein `Subject` oder ein `fromEvent` pusht Werte, sobald sie da sind, egal ob die Verarbeitung mitkommt. Was RxJS stattdessen anbietet, sind zeitbasierte und puffernde Operatoren: - `throttleTime(ms)` lässt einen Wert durch und verwirft dann für eine Weile alle weiteren. - `debounceTime(ms)` wartet, bis eine Ruhephase eintritt, und emittiert dann den letzten Wert. - `sampleTime(ms)` greift in festen Abständen den jeweils aktuellen Wert ab. - `bufferTime(ms)` sammelt über ein Fenster und liefert Bündel. - das `concurrent`-Limit von `mergeMap` begrenzt die Zahl paralleler innerer Verarbeitungen. Der entscheidende Unterschied: Diese Operatoren verwerfen oder aggregieren Werte, aber sie bremsen die Quelle nicht. Der Producer feuert unverändert weiter; ich reduziere nur, was hinten ankommt. Für eine Log-Anzeige oder Sensordaten ist das oft genau richtig – ein paar Zwischenwerte wegzuwerfen ist unkritisch. Für eine Verarbeitung, bei der jeder einzelne Wert zählen muss und die Quelle tatsächlich gedrosselt gehört, sind Node-Streams das richtige Werkzeug, nicht RxJS. Man kann beide Welten verbinden. Das Repository zeigt unter `node.js` die Brücke über `rxjs-stream`: ```js const { spawn } = require('child_process'); const { streamToRx } = require('rxjs-stream'); const { map } = require('rxjs'); streamToRx(spawn('ls', ['-lha']).stdout) .pipe(map(chunk => chunk.toString())) .subscribe({ next: text => console.log(text) }); ``` `streamToRx` macht aus einem Node-Readable ein Observable und behält dabei die Stream-Backpressure im unteren Teil. So bekomme ich die Komponierbarkeit von RxJS für die Transformation, ohne die Flusskontrolle des Streams aufzugeben. Wer die Grundlagen des Event-Loops und der Stream-Architektur dahinter auffrischen will, findet sie im [Node.js-Training](https://www.mikebild.dev/de/blog/nodejs-training-event-loop-architektur/). ## Fallstricke, die mir immer wieder begegnen Zwei Fehler tauchen so regelmäßig auf, dass ich sie hier ausdrücklich nenne. Der erste ist der Subscription-Leak. `fromEvent` registriert bei jedem `subscribe` einen Listener am Emitter. Ohne `unsubscribe()` bleibt dieser Listener hängen – auch dann, wenn die Verarbeitung längst nicht mehr gebraucht wird. Das Ergebnis ist ein Memory-Leak und eine Pipeline, die unbemerkt weiterläuft. Die Regel ist einfach: Ich halte die Subscription, die `.subscribe()` zurückgibt, und rufe `unsubscribe()` auf, wenn ich fertig bin. Alternativ begrenze ich den Strom deklarativ über `take(n)` oder `takeUntil(stop$)`, sodass er sich selbst beendet: ```js const { fromEvent, timer } = require('rxjs'); const { takeUntil } = require('rxjs'); const stop$ = timer(60000); fromEvent(emitter, 'data') .pipe(takeUntil(stop$)) .subscribe({ next: v => process(v) }); ``` Hier räumt `takeUntil` nach 60 Sekunden auf, ohne dass ich die Subscription selbst verwalten muss. Bei langlebigen Servern ist eine der beiden Strategien Pflicht, sonst sammeln sich Listener über die Laufzeit an. Der zweite Fallstrick ist das Verwechseln von hot und cold, das ich oben schon beschrieben habe – es äußert sich fast immer als „mein zweiter Subscriber bekommt nicht dieselben Werte“. Die Antwort ist meist ein `Subject` mit passender Variante oder `shareReplay`. Und ein dritter, eher handwerklicher Punkt: Viele Beispiele im Netz und auch die älteren Snippets im Repository nutzen noch RxJS-6-Muster. Konkret heißt das für RxJS 7: - Statt `subscribe(next, error, complete)` mit einzelnen Argumenten den Objekt-Observer `subscribe({ next, error, complete })` verwenden – die positionsbasierte Form ist seit 6.4 überholt. - Statt `tap(fn, errFn)` mit mehreren Callbacks ebenfalls den Objekt-Observer nutzen. - Operatoren aus `'rxjs'` importieren statt aus `'rxjs/operators'`. Diese Anpassungen sind mechanisch, aber wichtig, damit der Code nicht auf einem Muster aufsetzt, das perspektivisch verschwindet. ## Wann RxJS statt rohem EventEmitter? Nicht jede Node-Anwendung braucht RxJS. Ein einzelner `on('data', ...)`-Handler bleibt mit purem `EventEmitter` schlanker und hat keine zusätzliche Abhängigkeit. Ich greife dann zu RxJS, wenn mindestens einer dieser Punkte zutrifft: - Ich muss mehrere Verarbeitungsschritte verketten – transformieren, filtern, aggregieren – und will das nicht über verschachtelte Callbacks und geteilten Zustand lösen. - Ich brauche zeitbasierte Steuerung: drosseln, entprellen, in Fenstern bündeln. - Denselben Ereignisstrom sollen mehrere Konsumenten unabhängig verarbeiten (Multicast über ein `Subject`). - Der Strom soll sich sauber und deklarativ beenden lassen, etwa über `takeUntil`. - Der Verarbeitungscode soll in kleine, für sich testbare Bausteine zerfallen. Umgekehrt bleibe ich beim rohen Emitter oder greife zu Node-Streams, wenn es nur ein einziger simpler Handler ist oder wenn ich echte Backpressure über den ganzen Weg brauche. ## Fazit RxJS verschiebt in Node.js die Perspektive vom Befehl zur Beschreibung. Der `EventEmitter` sagt, was bei jedem Ereignis zu tun ist; RxJS beschreibt den ganzen Ereignisstrom als komponierbare Kette von reinen Funktionen. `fromEvent` ist die Brücke dorthin, ohne dass sich an der Semantik des Emitters etwas ändert, und ein `Subject` macht aus einem Strom viele Konsumenten. Zwei Dinge sollte man dabei nie vergessen. Erstens die Disziplin beim Aufräumen: Ohne `unsubscribe`, `take` oder `takeUntil` bleiben Listener und Verarbeitung am Leben. Zweitens die ehrliche Grenze des Werkzeugs: RxJS steuert Überlast nur zeitbasiert und puffernd, nicht über echte Backpressure. Wer diese beiden Punkte im Kopf behält, bekommt mit RxJS eine sehr angenehme Art, serverseitige Ereignisströme zu verarbeiten – lesbar, testbar und erweiterbar, ohne im Callback-Dickicht zu versinken. ## Weiterführende Quellen - Repository-Beispiele: `/Users/mikebild/Documents/introduction-nodejs/EXAMPLES/RXJS/operators.js`, `.../subject.js`, `.../node.js` - [fromEvent](https://rxjs.dev/api/index/function/fromEvent) - [Subject-Guide](https://rxjs.dev/guide/subject) - [Observable-Guide](https://rxjs.dev/guide/observable) - [Operatoren-Guide](https://rxjs.dev/guide/operators) - [mergeMap](https://rxjs.dev/api/operators/mergeMap) - [throttleTime](https://rxjs.dev/api/index/function/throttleTime) - [bufferTime](https://rxjs.dev/api/operators/bufferTime) - [scan](https://rxjs.dev/api/operators/scan) - [subscribe-Argumente (Deprecation)](https://rxjs.dev/deprecations/subscribe-arguments) - [Glossar und Semantik: hot/cold, unicast/multicast](https://rxjs.dev/guide/glossary-and-semantics) - [Node.js events](https://nodejs.org/docs/latest-v18.x/api/events.html)