Reaktive Programmierung

Ein deklaratives Programmierparadigma, das mit Datenströmen arbeitet: Man beschreibt einmal, wie auf eintreffende Ereignisse reagiert wird, und Änderungen breiten sich automatisch aus. In Angular geschieht das mit RxJS-Observables und seit Angular 16 zusätzlich mit Signals.

Code-Beispiele aus Quelle - Angular Mastery Workshop (Folie 258–272), von den Folienbildern abgeschrieben.

Die Idee

Der Kurs definiert reaktive Programmierung als „deklaratives Paradigma, das sich mit Datenströmen und der Ausbreitung von Änderungen beschäftigt“ und fragt gleich selbst: „Und was heisst das jetzt?“ (Folie 258).

Beispiel: Klicks entprellen (Folie 259). Der klassische Weg ist imperativ und ausführlich. Man muss den Zustand (die aktuelle Timeout-ID, die entprellte Funktion) selbst verwalten und speichern:

function debounce(fn, timeout) {
    let timeoutId;

    return function(...args) {
        if (timeoutId) {
            clearTimeout(timeoutId);
        }
        timeoutId = setTimeout(() => fn(...args), timeout);
    }
}

const debouncedClick = debounce(() => console.log('clicked'), 300);

window.addEventListener('click', debouncedClick);

Reaktiv ist es deklarativ und ohne eigenen Zustand:

import { fromEvent } from 'rxjs';
import { debounceTime } from 'rxjs/operators';

fromEvent(window, 'click')
  .pipe(debounceTime(300))
  .subscribe(() => console.log('clicked'))

Merkmale (Folie 260): deklarativ (der Strom wird einmal definiert und reagiert während der ganzen Lebensdauer der Komponente oder des Services), funktional (der ganze Zustand steckt im Strom, Nebeneffekte werden ausdrücklich behandelt) und Aufrufe werden gemeinsam statt einzeln behandelt, also Drosseln, Entprellen und Abbrechen ohne zusätzlichen Zustand.

Der Kernsatz: Ein Observable ist eine asynchrone Sammlung von Werten (Folie 263). So wie ein Array Werte im Raum nebeneinander hält, liefert ein Observable Werte in der Zeit nacheinander. Deshalb lassen sich darauf dieselben Operationen anwenden wie auf Arrays (map, filter, reduce, siehe Modernes JavaScript).

Ein Operator verbindet Ströme. merge(stream1, stream2) macht aus den Strömen 20–40–60–80–100 und 1–1 einen Strom mit allen Ereignissen in zeitlicher Reihenfolge: 20–40–60–1–80–100–1 (Marble-Diagramm, Folie 264).

Kalte und heisse Ströme

(Folie 265)

Abonnieren

Abonnieren (subscribe) ist wie der Aufruf einer Funktion: Erst dann läuft ein kalter Strom. Der Observer bekommt jeden Wert, einen Fehler und das Ende des Stroms (Folie 266–267):

// Handler inline
this.userService.loadUsers().subscribe(
    users => console.log('Stream emited next value', users),
    error => console.error('Stream error', error),
    () => console.log('Stream completed')
)

// Observer als Objekt
const observer = {
    next: users => console.log('Stream emited next value', users),
    error: error => console.error('Stream error', error),
    complete: () => console.log('Stream completed')
}
this.userService.loadUsers().subscribe(observer);

// nur die Ereignisse, die interessieren
const observer = {
    complete: () => console.log('Stream completed')
}
this.userService.loadUsers().subscribe(observer);

Regeln aus dem Kurs

1. Nie ein subscribe in einem subscribe. Stattdessen einen Flattening-Operator nehmen (Folie 268):

// schlecht: Abonnement im Abonnement
createOrder(order: Order) {
    this.orderService.create(order)
        .subscribe(createdOrder => {
            this.customerService
                .addOrderToCustomer(this.customer, createdOrder.id)
                .subscribe(updatedCustomer => {
                    this.customer = updatedCustomer;
                });
        });
}

// gut: Flattening-Operator
createOrder(order: Order) {
    this.orderService.create(order)
        .pipe(
            concatMap(createdOrder => this.customerService
                .addOrderToCustomer(this.customer, createdOrder.id))
        )
        .subscribe(updatedCustomer => {
            this.customer = updatedCustomer;
        });
}

(Auf der Folie steht this,customer mit Komma statt Punkt.)

2. Wenn möglich mit | async im Template abonnieren. Das beendet das Abonnement automatisch (Folie 269):

// explizites subscribe … und ein Speicherleck
@Component({
    template: `
        <user-item *ngFor="let user of users" [user]="user">
        </user-item>
    `
})
export class UserListComponent implements OnInit {

    users: User[];

    constructor(private userService: UserService) {}

    ngOnInit() {
        this.userService.getUsers().subscribe(users => {
            this.users = users;
        });
    }
}

// besser: Service-Strom durchreichen und | async verwenden
@Component({
    template: `
        <user-item *ngFor="let user of users | async" [user]="user">
        </user-item>
    `
})
export class UserListComponent implements OnInit {

    users: Observable<User[]>;      // auf Folie 269 steht User[], das ist falsch; Folie 272 hat es richtig

    constructor(private userService: UserService) {}

    ngOnInit() {
        this.users = this.userService.users;
    }
}

3. Immer abmelden. Sonst entstehen Speicherlecks, die in kleinen Apps unbemerkt bleiben und mit dem Wachstum stark zunehmen (Folie 234, 270–271):

// kein Abmelden … führt zu Speicherleck
ngOnInit() {
    this.route.queryParams.subscribe(params => /* ... */);
}

// besser, aber imperativ: Subscription merken und in ngOnDestroy abmelden
ngOnInit() {
    this.subscription = this.route.queryParams.subscribe(params => /* ... */);
}

ngOnDestroy() {
    if (this.subscription) {
        this.subscription.unsubscribe();
    }
}

// empfohlen: ein destroy-Subject für beliebig viele Ströme
destroy = new Subject<void>();

ngOnInit() {
    this.route.queryParams
        .pipe(takeUntil(this.destroy))
        .subscribe(params => /* ... */);
}

ngOnDestroy() {
    this.destroy.next();
    this.destroy.complete();
}

4. Fehler früh abfangen mit catchError, möglichst an der Grenze zum Backend (Folie 289, siehe Angular-Backend-Kommunikation).

Welche Operatoren es gibt und wann man welchen nimmt, steht bei RxJS.

Signals: die zweite Form der Reaktivität

Das Timer-Beispiel aus dem Kurs sähe mit Signals etwa so aus (Einordnung Claude, nicht aus der Quelle):

import { Component, signal, computed } from '@angular/core';
import { toSignal } from '@angular/core/rxjs-interop';
import { timer } from 'rxjs';

@Component({
  selector: 'my-org-timer',
  template: `<p>Elapsed {{ elapsed() }} seconds ({{ minutes() }} min)</p>`
})
export class TimerComponent {
  elapsed = toSignal(timer(0, 1000), { initialValue: 0 });   // Abmelden übernimmt Angular
  minutes = computed(() => Math.floor(this.elapsed() / 60));
}

Verwandt

• RxJS – die Bibliothek mit Observables und Operatoren
• Angular-Komponente – Lebenszyklus und Change Detection
• State Management – Zustand als Strom oder als Signal
• Modernes JavaScript – Promises und async/await als einfachere Vorstufe