01 · Problem
Wofür braucht man das?
Asynchrone oder mehrfache Werte sollen produziert, transformiert und unabhängig abbestellt werden können.
02 · Denkmodell
Das mentale Modell
Jede Subscription besitzt ihren eigenen closed-Zustand und Teardown. Ein Operator erstellt lediglich ein neues Observable, das die Quelle abonniert.
SimpleObservable, Subscription sowie map/filter existieren, aber ein Root-Export fehlt. isUnsubscribed liegt global auf dem Observable, sodass ein Unsubscribe alle gegenwärtigen und zukünftigen Subscriber blockiert.
03 · Kontrollfluss
Was passiert in welcher Reihenfolge?
- Producer erhält einen sicheren Observer pro Subscription.
- next liefert Werte nur solange diese Subscription offen ist.
- error oder complete schließen sie genau einmal.
- unsubscribe führt den Teardown idempotent aus.
- Operatoren leiten Werte und Fehler an ihre eigenen Subscriber weiter.
04 · Bauteile
Die entscheidenden Verträge
SimpleObservable
Source-Klasse für subscribe und pipe ohne Package-Root-Export.
SimpleSubscription
Kapselt einen Teardown und einen Unsubscribe-Zustand.
map / filter
any-typisierte Operatoren auf einem Quellobservable.
done
Identitätsoperator ohne terminale Wirkung.
05 · Build it yourself
Selbst implementieren
Implementiere zuerst Subscription-Lifecycle und sichere Observer, bevor Operatoren hinzukommen.
- Observer mit next, error und complete definieren.
- Pro subscribe einen lokalen closed-Schalter und idempotenten Teardown erzeugen.
- Operator-Callbackfehler in den Error-Kanal übersetzen.
type Observer<T> = {
next(value: T): void;
error(error: unknown): void;
complete(): void;
};
class Subscription {
private closed = false;
constructor(private teardown: () => void) {}
unsubscribe() {
if (this.closed) return;
this.closed = true;
this.teardown();
}
}06 · Verifizieren
Was du testen solltest
- Unsubscribe eines Subscribers beeinflusst keinen zweiten Subscriber.
- Teardown läuft auch bei wiederholtem Unsubscribe genau einmal.
- Nach error oder complete werden keine weiteren Signale geliefert.
07 · Grenzen
Kompromisse und Stolperfallen
- Cold Observables isolieren Subscriber, starten Producer aber mehrfach.
- Hot Multicasting teilt Arbeit und benötigt gemeinsamen Lifecycle sowie Backpressure-Regeln.
- Eine kleine API ist lehrreich, ersetzt nicht Scheduling und Operatorbreite von RxJS.
Die vorhandenen Tests liegen im Source, werden vom Package-Testscript aber nicht ausgeführt.
08 · Weiterdenken