Event-Stream

Der Event-Stream ist ein systemweiter Bus für One-to-Many-Messaging, das nicht in die Eltern-Kind- oder Sender-Empfänger-Form passt. Jeder Actor kann eine Klasse von Event oder einen kind-diskriminierten Type abonnieren; jeder Code kann publishen; der Bus matcht und tellt das Event an jeden Subscriber.

import { Actor, ActorSystem } from 'actor-ts';

class UserLoggedIn {
  constructor(public readonly userId: string) {}
}

class AuditLogger extends Actor<UserLoggedIn> {
  override preStart(): void {
    this.context.system.eventStream.subscribe(this.context.self, UserLoggedIn);
  }
  override onReceive(event: UserLoggedIn): void {
    this.log.info(`user ${event.userId} logged in`);
  }
}

class MetricsCollector extends Actor<UserLoggedIn> {
  override preStart(): void {
    this.context.system.eventStream.subscribe(this.context.self, UserLoggedIn);
  }
  override onReceive(event: UserLoggedIn): void {
    this.incrementCounter('logins');
  }
}

const system = ActorSystem.create('demo');
system.spawnAnonymous(AuditLogger);
system.spawnAnonymous(MetricsCollector);

// Überall — auch außerhalb eines Actors:
system.eventStream.publish(new UserLoggedIn('user-42'));
// → beide Subscriber empfangen das Event

Der Publisher weiß nicht, wie viele Subscriber existieren, und die Subscriber wissen nicht, wer publisht. Das ist der ganze Sinn — lose Kopplung für Querschnitt-Anliegen (Logging, Metriken, Audit, Fan-Out-Alerts).

Die API

// A channel is named by a class, by an EventKey, or by the bare kind string.
type EventChannel<TEvent> =
  | (abstract new (...args: any[]) => TEvent)
  | EventKey<TEvent>
  | KindOf<TEvent>;

class EventStream {
  subscribe<TEvent>(
    subscriber: ActorRef,
    channel: EventChannel<TEvent>,
    predicate?: (event: TEvent) => boolean,
  ): boolean;

  unsubscribe<TEvent>(subscriber: ActorRef, channel?: EventChannel<TEvent>): boolean;

  publish(event: object): void;
}

Drei Operationen:

• subscribe(ref, channel) — Interesse registrieren. Der Channel ist eine Klasse, ein EventKey oder ein kind-String. Gibt true zurück, wenn ein neues Abo hinzugefügt wurde, false, wenn ein Duplikat ignoriert wurde — und wirft, wenn der Channel unbrauchbar ist (siehe die Caution unten).
• unsubscribe(ref, channel?) — Abos entfernen. Mit einem Channel: nur diesen; ohne: jedes Abo für diese Ref.
• publish(event) — Fire-and-Forget. Der Bus läuft seine Abo-Liste durch, matcht den Channel jedes Abos und tellt jeden Match.

Klassen-Channels matchen via instanceof, Subklassen-Instanzen erreichen also Base-Class-Subscriber:

class SystemEvent {}
class UserLoggedIn extends SystemEvent {}
class UserLoggedOut extends SystemEvent {}

eventStream.subscribe(auditor, SystemEvent);   // fängt beides
eventStream.subscribe(loginCounter, UserLoggedIn);   // fängt nur die In-Events

Das macht hierarchische Event-Taxonomien einfach: Subscriber wählen ihr Level.

Kind-Channels

Messages empfiehlt plain Objects ({ kind: 'x', n: 1 }) statt Klassen — und ein plain Type hat keinen Konstruktor, den man subscribe übergeben könnte. Benenne ihn stattdessen über sein kind. Ein Type und ein const gleichen Namens geben ihm die Aufrufform, die eine Klasse geschenkt bekommt:

import { EventKey } from 'actor-ts';

export type UserLoggedInEvent = {
  readonly kind: 'user-logged-in';
  readonly userId: string;
};
export const UserLoggedInEvent = EventKey.of<UserLoggedInEvent>('user-logged-in');

class AuditLogger extends Actor<UserLoggedInEvent> {
  override preStart(): void {
    this.context.system.eventStream.subscribe(this.context.self, UserLoggedInEvent);
  }
  override onReceive(event: UserLoggedInEvent): void {
    this.log.info(`user ${event.userId} logged in`);
  }
}

system.eventStream.publish({ kind: 'user-logged-in', userId: 'user-42' });

Der Key trägt den Event-Type, ein gegen ihn geschriebenes Predicate sieht also die echte Form, ohne dass du den Parameter annotierst:

eventStream.subscribe(auditor, UserLoggedInEvent, (event) => event.userId !== 'system');

Der nackte kind-String ist die Kurzform, und er kostet den Typ — TEvent hat nichts, woraus es inferiert werden könnte, das Predicate sieht also unknown. Das Typargument anzugeben holt die Typisierung zurück und macht den String selbst gegen das kind des Types prüfbar:

eventStream.subscribe(auditor, 'user-logged-in');                     // predicate sees unknown
eventStream.subscribe<UserLoggedInEvent>(auditor, 'user-logged-in');  // typed, and the string is checked

Daraus folgen zwei Identitätsregeln:

• Ein Key und sein String sind derselbe Channel. Beide Wege zu abonnieren deduppt, und jede Form unsubscribed die andere — auch ein frisch gebautes EventKey.of('user-logged-in').
• Eine Klasse und ein kind sind zwei Channels, selbst wenn die Instanzen der Klasse dieses kind tragen. Sie selektieren überlappende Events, genau wie eine Basisklasse und ihre Subklasse — ein Actor mit beiden Abos bekommt also zwei Zustellungen pro Publish.

Ein kind-Channel matcht genau ein kind

Eine Klassenhierarchie schenkt dir Familien — abonniere die abstrakte Basis und du bekommst alles darunter. Kinds haben keine Basis, es gibt also keine Möglichkeit, “jedes billing.*-Event” zu sagen: abonniere pro kind einzeln, oder gib der Familie stattdessen eine Klassenhierarchie.

Der leere String wird rundheraus abgelehnt — er benennt kein Event, und er liest sich für den, der ihn geschrieben hat, wie “alles”. Beachte außerdem, dass der nackte String dich nicht vor einem Tippfehler schützt, der Key hingegen schon: ein falsch geschriebenes kind kompiliert anstandslos und matcht dann still niemals.

Predicate-Filterung

eventStream.subscribe(
  metricsActor,
  HttpResponse,
  (event) => event.status >= 500,   // nur 5xx-Responses liefern
);

Für hochfrequente Channels (Cluster-Events, jede HTTP-Response, jeder Metric-Tick) ist Filtern auf der Bus-Seite günstiger als Filtern innerhalb des onReceive jedes Subscribers.

Drei Details:

• Kein Dedup für Predicate-tragende Abos. Ohne Predicate deduppt der Bus pro (subscriber, channel)-Paar — subscribe erneut aufzurufen, ist ein No-Op. Mit einem Predicate fügt jeder Aufruf ein neues Abo hinzu, weil Predicate-Funktionen keinen Identitätsvertrag haben. “Diesen Filter ersetzen” heißt unsubscribe und dann erneut subscribe.
• Ein throwendes Predicate wird als “kein Match” für diese Auslieferung behandelt. Andere Subscriber sind unberührt; das Abo bleibt aktiv. Eine Warnung wird über den System-Logger geloggt.
• Predicates laufen zur Publish-Zeit, auf dem Stack des Publishers. Halte sie schnell und rein — ein schwerer Predicate verlangsamt jedes Publish.

Wann den Event-Stream verwenden

Drei gute Passungen:

1. Querschnitt-Beobachtung — Logging, Metriken, Audit. Viele Subscriber, von denen der Publisher keinen kennen soll.
2. Systemweite Benachrichtigungen — “der Cluster hat ein Mitglied gewonnen”, “der Cache wurde geflusht”. Die Cluster-Extension publisht ihre eigenen Events hier; du kannst abonnieren und reagieren.
3. Entkoppeln von Actor-Verdrahtung in Tests — statt eine Spy-Ref durch fünf Schichten zu legen, publishe auf dem Stream und lass den Spy direkt abonnieren.

Wann NICHT verwenden

Ein spezifischer Empfänger = verwende eine Ref

Wenn genau ein Actor die Nachricht erhalten soll, halte seine ActorRef und tell ihm. Der Event-Stream ist für Fan-Out; ihn als “ich will die Ref nicht tippen”-Abkürzung zu verwenden, fügt Runtime-Kosten hinzu (jedes Publish läuft die Abo-Liste durch) und verschleiert die Absicht. Refs sind der Default; der Bus ist die Ausnahme.

Hochvolumige In-Band-Daten

Sensor-Lesungen mit 10 kHz durch den Event-Stream zu streamen, liefert jede Lesung an jeden Subscriber synchron innerhalb von publish. Für Bulk-In-Band-Daten verwende direkte Refs (eine Mailbox pro Quelle) oder DistributedPubSub für Cluster-scoped Publish mit Topic-Routing. Der Bus ist ein Koordinationswerkzeug, keine Daten-Pipeline.

Gestoppte Subscriber werden freigegeben

eventStream.subscribe(ref, Event);
ref.stop();
eventStream.publish(new Event());   // → wird ref nicht zugestellt

Ein stoppender Actor gibt seine Subscriptions im Zuge der Terminierung ab, ein kurzlebiger Subscriber braucht also kein unsubscribe in postStop. Melde trotzdem ab, wenn ein lebender Actor das Interesse verliert — das kann der Bus nicht wissen.

Ein Event kann einen Subscriber noch erreichen, der sich während der Zustellung genau dieses Events abgemeldet hat: die Empfängermenge steht fest, wenn publish aufgerufen wird, eine Abmeldung wirkt also ab dem nächsten Event. Es landet in /deadLetters wie jede andere Nachricht an einen gestoppten Actor.

Eingebaute Events, die abonniert werden können

Das Framework publisht mehrere Event-Typen, die du abonnieren kannst:

Event	Woher es kommt
DeadLetter	Jede an /deadLetters geroutete Nachricht wird auch publisht. Nützlich für “Alarme auf verlorene Nachrichten”.
MemberUp, MemberRemoved, MemberUnreachable, …	Konkrete Cluster-Mitgliedschaftsevents, publisht von der Cluster-Extension. Abonniere jede Klasse direkt — ClusterEvent ist ein Union-Typ, keine abonnierbare Basisklasse. Siehe Cluster.
ReachabilityChanged	Der Failure Detector der publishenden Node hat einen Peer aus den Augen verloren oder wiedergesehen — eine lokale Beobachtung, anders als MemberUnreachable, das auch per Gossip angekommen sein kann. Siehe Failure Detector.
Broker-Events	BrokerActor-Subklassen publishen Connect/Disconnect/Lag-Events.

Alles, was das Framework selbst publisht, ist eine Klasse — Klassen-Channels bleiben also der Weg zu allem oben. Kind-Channels sind für deine eigenen Anwendungs-Events.

Ein Cluster-Event fehlt hier bewusst: CurrentClusterState geht ausschließlich an den Listener, der danach gefragt hat — per cluster.subscribe(listener, { replayMode: 'snapshot' }). Es sagt, wo dieser eine Subscriber startet, und das ist für niemanden sonst eine Neuigkeit, also erreicht es diesen Bus nie.

Siehe die Docs jeder Extension für das volle Event-Vokabular, das sie publisht.

Wie es weitergeht

• DistributedPubSub — cluster-weites Publish/Subscribe mit Topic-Routing; komplementär zum lokalen Event-Stream.
• Actor-System — system.eventStream ist, wo du den Bus erreichst.
• Nachrichten — die Form-Regeln (unveränderlich, keine Method-Refs) gelten genauso für Events, und die dort dokumentierte kind-Konvention ist direkt abonnierbar.

Die EventStream-API-Referenz deckt die volle Schnittstelle ab.