BrokerActor-Basisklasse

Jeder Protokoll-Actor in actor-ts/io/broker — KafkaActor, MqttActor, NatsActor usw. — erbt von BrokerActor. Die Basisklasse besitzt den gemeinsamen Lifecycle: Verbindungszustandsautomat, Reconnect-mit-Backoff, Outbound-Buffer, Subscriber-Fan-out, Publishen von Lifecycle-Events.

BrokerActor (die abstrakte Basis) besitzt:

• Lifecycle-Zustandsautomat — disconnected ↔ connecting ↔ connected ↔ disconnecting.
• Outbound-Buffer — Nachrichten, die vor stehender Verbindung gesendet werden.
• Reconnect-Schleife — exponentielles Backoff bei Verbindungsverlust.
• Subscriber-Tracking — Fan-out für eingehende Events.
• Gewünschte Subscriptions — Protokoll-Subscriptions, die jede einzelne Verbindung überleben und bei jedem Reconnect neu aufgebaut werden.

Subklassen implementieren drei Protokoll-Hooks:

Hook	Wann aufgerufen
connectImplementation	Die protokollspezifische Verbindung öffnen.
disconnectImplementation	Sie schließen — beim Stoppen und vor jedem Re-Connect-Versuch.
dispatchOutgoing(envelope)	Eine einzelne gepufferte Nachricht auf die Leitung senden.

Subklassen implementieren die drei Protokoll-Hooks; den Rest macht die Basisklasse. Diese Seite dokumentiert das, was geteilt wird. Für protokollspezifische Details siehe die Seiten pro Protokoll.

Der Zustandsautomat

Vier Zustände:

• disconnected — initial; aktuell nicht verbunden.
• connecting — connectImplementation läuft.
• connected — Verbindung steht; Nachrichten fließen.
• disconnecting — disconnectImplementation läuft.

Ein disconnected → connecting → Fehler löst eine Reconnect-Schleife aus: Backoff + Retry bis Erfolg oder bis maxAttempts erschöpft sind.

Jeder Re-Connect-Versuch startet auf sauberem Grund: die Basisklasse ruft zuerst disconnectImplementation auf, wenn ein früherer Versuch irgendetwas geöffnet hat — eine Subklasse baut also nie eine neue Verbindung auf den Handles der toten auf. Damit ist disconnectImplementation ein idempotenter Kontrakt: er darf auf einer bereits toten Verbindung aufgerufen werden und muss ausschließlich die lebenden Handles verwerfen, niemals die gewünschten Subscriptions (siehe unten).

Subklassen-Kontrakt

abstract class BrokerActor<S, Command, P, Subscription = never> extends Actor<Command> {
  // Subklassen implementieren:
  protected abstract configKey(): string;
  protected abstract builtInDefaultOptions(): Partial<S>;
  protected abstract readOptionsFromConfig(config: Config): Partial<S>;
  protected abstract requiredOptions(): ReadonlyArray<keyof S>;
  protected abstract endpointLabel(): string;

  protected abstract connectImplementation(): Promise<void>;
  protected abstract disconnectImplementation(): Promise<void>;
  protected abstract dispatchOutgoing(envelope: OutboundEnvelope<P>): Promise<void>;
}

Drei Kategorien:

• Settings-Glue (configKey, builtInDefaultOptions, readOptionsFromConfig, requiredOptions, endpointLabel) — beschreibt, wie sich die Settings aus den drei Ebenen zusammensetzen (Konstruktor + HOCON + Defaults) und wie validiert wird.
• Protokoll-Hooks (connectImplementation, disconnectImplementation, dispatchOutgoing) — die protokollspezifische Arbeit.

endpointLabel ist die menschenlesbare Verbindungs-Identität (“amqp://localhost:5672”, “kafka-cluster-1”), die in Log-Zeilen und Lifecycle-Events auftaucht.

Was die Basisklasse für dich erledigt

Outbound-Buffering

this.enqueueOutbound(payload);

Subklassen rufen this.enqueueOutbound(payload) zum Senden auf; es gibt true zurück, wenn die Nachricht gesendet oder gepuffert wurde, false, wenn sie verworfen wurde. Die Basisklasse:

• Wenn connected — verpackt payload in ein OutboundEnvelope und ruft sofort dispatchOutgoing(envelope) auf.
• Wenn disconnected (oder connecting) — puffert bis zu outboundBuffer Envelopes (Default 1000). Beim Reconnect wird der Buffer der Reihe nach geleert.

Bei Überlauf verwirft die Basisklasse immer das älteste gepufferte Envelope (FIFO) und publisht ein BrokerBufferOverflow- Event auf dem Event-Stream — es wird nie geworfen. Mit outboundBuffer: 0 ist das Puffern deaktiviert: die Nachricht wird verworfen und ein BrokerNotConnected-Event publisht.

Subscriber-Tracking

this.onReceive(message) {
  // `Terminated` erreicht onReceive, gehört aber nicht zur getypten
  // Command-Union — warum dieser Arm deiner ist, steht unten.
  if (message instanceof Terminated) {
    this.pruneTerminatedSubscriber(message.actor);
    return;
  }
  match(message)
    .with({ kind: 'subscribe' }, (m) => this.onSubscribe(m))
    .with({ kind: 'unsubscribe' }, (m) => this.onUnsubscribe(m))
    .exhaustive();
}

private onSubscribe(message: SubscribeCommand): void {
  this.subscribeRef(message.topic, message.subscriber);
}

private onUnsubscribe(message: UnsubscribeCommand): void {
  this.unsubscribeRef(message.topic, message.subscriber);
}

subscribeRef(topic, ref) registriert ref als interessiert an den eingehenden Nachrichten von topic und death-watcht die Ref.

Der Terminated-Arm ist nicht optional

subscribeRef startet die Death-Watch, aber die Basisklasse kann für dich nicht darauf reagieren: onReceive ist abstrakt, jede Unterklasse besitzt also ihr eigenes Dispatch, und die Basis sieht nie eine Nachricht. Ohne den Arm oben bleibt ein gestoppter Subscriber in jedem Topic, das er hielt, und kostet bei jedem Fan-out ein tell in die Dead Letters — das war ein echter Defekt (#1111), kein hypothetischer.

pruneTerminatedSubscriber(ref) entfernt den Subscriber aus allen Topics und liefert zurück, ob überhaupt etwas entfernt wurde — so kannst du den Tod eines Subscribers von dem jedes anderen beobachteten Actors unterscheiden. Es macht kein unwatch: die Zelle hat die Watch bereits fallen gelassen, als sie Terminated zugestellt hat.

unsubscribeRef vergleicht über den Pfad der Ref statt über das Ref-Objekt — eine andere Ref auf denselben Actor unsubscribed also trotzdem.

Wenn das Protokoll eine eingehende Nachricht pusht, ruft die Subklasse auf:

this.fanOutToTopic(topic, inboundMessage);

Die Basisklasse stellt sie an jeden Subscriber für dieses Topic zu.

Desired Subscriptions

Das Subscriber-Tracking oben spielt innerhalb des Actor-Systems — welche lokalen Refs welches Topic wollen. Eine gewünschte Subscription ist die andere Hälfte: die Subscription, die der Actor auf dem Broker hält und die bei jedem Verbindungsaufbau neu gesetzt werden muss.

Die Basisklasse hält diese Menge getrennt von den lebenden Handles, damit sie jede einzelne Verbindung überlebt:

// Vormerken (und sofort setzen, falls verbunden).
await this.rememberSubscription('orders.new', target);
// Entfernen — auf dem Broker und aus der gewünschten Menge.
await this.forgetSubscription('orders.new');

Subklassen liefern drei Hooks:

Hook	Zweck
initialSubscriptions()	In den Options deklarierte Subscriptions. Einmalig vor dem ersten Connect in die gewünschte Menge übernommen.
applySubscription(key, subscription)	Eine Subscription auf der lebenden Verbindung setzen. Muss einen bereits lebenden Key tolerieren.
revokeSubscription(key)	Eine abbauen. Optional — mehrere Protokolle können eine einzelne Subscription nur mitsamt dem ganzen Consumer aufgeben.

und spielen die ganze Menge aus connectImplementation heraus erneut ein:

protected async connectImplementation(): Promise<void> {
  this.connection = await MyClient.connect(this.options.url);
  await this.applyDesiredSubscriptions();   // konfiguriert + zur Laufzeit
}

Getrieben wird das Wiedereinspielen von der Subklasse und nicht von der Basisklasse, weil der richtige Punkt im Handshake protokollspezifisch ist — kafkajs etwa will jedes subscribe vor consumer.run haben.

Was das bringt:

• Eine zur Laufzeit hinzugefügte Subscription überlebt einen Reconnect — sie ist nicht bloß ein Aufruf auf einer Verbindung, die gerade stirbt.
• Ein subscribe, das während einer Trennung eintrifft, wird vorgemerkt und beim nächsten Connect gesetzt, statt verworfen zu werden.
• Das Übernehmen aus den Options passiert nur einmal, ein unsubscribe zur Laufzeit wird also nicht vom nächsten Reconnect wiederbelebt.
• Ein erneutes rememberSubscription auf einem lebenden Key baut ihn zuerst ab, damit eine geänderte Nutzlast (etwa ein anderer Ziel-Actor) tatsächlich greift.
• Eine Subscription, die nicht gesetzt werden kann, wird als Warnung geloggt, und der Rest der Menge geht trotzdem durch. Ein schlechtes Subject reißt die Verbindung nicht mit — hinterlässt den Actor aber auch nicht stillschweigend verbunden-und-taub.

MqttActor ist älter als dieser Mechanismus und behält seine eigene, reichhaltigere Registry (QoS pro Topic, mehrere Targets pro Pattern, Deathwatch auf jedem) — mit den gleichen Reconnect-Garantien.

Reconnect-mit-Backoff

reconnect: {
  initialDelayMs: 200,
  maxDelayMs:     30_000,
  factor:         2,
  maxAttempts:    Infinity,   // der Default — endlos wiederholen
}

Pro Actor konfigurierbar (oder reconnect: false, um Auto-Reconnect ganz zu deaktivieren). Jeder Versuch wartet min(initialDelayMs * factor^(attempt - 1), maxDelayMs) — reines exponentielles Backoff, ohne Jitter.

Jeder Versuch löst BrokerReconnectAttempt auf dem Event-Stream aus; nachdem maxAttempts erschöpft sind (sofern endlich), löst BrokerReconnectFailed aus und der Actor bleibt getrennt.

Lifecycle-Events

Publisht auf system.eventStream:

Event	Wann
BrokerConnected	Ein connectImplementation war erfolgreich.
BrokerDisconnected	Ein disconnectImplementation lief oder eine Verbindung ist fehlgeschlagen.
BrokerReconnectAttempt	Ein Reconnect-Versuch startet gerade.
BrokerReconnectFailed	maxAttempts erschöpft.
BrokerBufferOverflow	Der Outbound-Buffer hat ein Envelope verworfen.
BrokerNotConnected	Gesendet ohne Verbindung.

Subscribiere, um jeden Broker-Actor einheitlich zu beobachten:

system.eventStream.subscribe(monitorRef, BrokerConnected);
system.eventStream.subscribe(monitorRef, BrokerDisconnected);

Die Events enthalten actorPath — so unterscheidest du Events verschiedener Broker-Actors im System.

Auflösung der Settings

1. builtInDefaultOptions()           ← niedrigste Priorität (immer angewendet)
2. readOptionsFromConfig()    ← HOCON-Überschreibungen
3. Konstruktor-Argument        ← höchste Priorität (pro Instanz)

preStart mergt die drei Ebenen, validiert gegen requiredOptions() und legt das Ergebnis für den Rest des Actor- Lebens unter this.options ab.

Fehlende Pflicht-Settings führen zu einem frühen Fehler-Throw in preStart — der Actor durchläuft den Fehlerpfad des Supervisors, bevor er überhaupt versucht, sich zu verbinden.

Einen eigenen Protokoll-Actor schreiben

import { BrokerActor, type OutboundEnvelope, type BrokerCommonOptionsType } from 'actor-ts';

interface MyProtocolOptionsType extends BrokerCommonOptionsType {
  readonly url: string;
}

class MyProtocolActor extends BrokerActor<MyProtocolOptionsType, Command, MyPayload> {
  private connection: MyClient | null = null;

  protected configKey() { return 'actor-ts.io.broker.my-protocol'; }
  protected builtInDefaultOptions() { return { /* ... */ }; }
  protected readOptionsFromConfig(c) { /* HOCON parsen */ return {}; }
  protected requiredOptions() { return ['url'] as const; }
  protected endpointLabel() { return this.options.url; }

  protected async connectImplementation(): Promise<void> {
    this.connection = await MyClient.connect(this.options.url);
    this.connection.onMessage((m) => this.fanOutToTopic(m.topic, m));
  }

  // Wird beim Stoppen UND vor jedem Re-Connect-Versuch aufgerufen —
  // idempotent und auch auf einer bereits toten Verbindung sicher.
  protected async disconnectImplementation(): Promise<void> {
    await this.connection?.close();
    this.connection = null;
  }

  protected async dispatchOutgoing(env: OutboundEnvelope<MyPayload>): Promise<void> {
    await this.connection!.send(env.payload);
  }
}

Den Rest übernimmt die Basisklasse. Die meisten Drittanbieter- Clients (kafkajs, nats.js usw.) haben eine eventbasierte Message-Receive-API, die sich sauber auf this.fanOutToTopic(...) abbilden lässt.

Wohin als Nächstes

• I/O-Übersicht — das große Bild: welche Protokolle ausgeliefert werden.
• Kafka / MQTT / NATS / usw. — Seiten pro Protokoll.
• Event-Stream — wo Lifecycle-Events publisht werden.
• Backoff-Policy — ein eigenständiges Backoff-Primitiv (die Reconnect-Schleife rechnet selbst).