Zum Inhalt springen
actor-ts
Deutsch

ProducerController

ProducerController umwickelt die ausgehenden Nachrichten eines Senders, vergibt Sequence Numbers und hält sie, bis sie vom Consumer geACKt sind.

import { ProducerController, ProducerControllerOptions } from 'actor-ts';


const producerControllerOptions = ProducerControllerOptions.create<OrderEvent>()
  .withProducerId('orders')                     // stabile Identität
  .withConsumer(consumerRef)
  .withWindowSize(16)                           // Flow-Control-Window
  .withResendTimeout(500);
const producer = system.spawn(
  () => new ProducerController<OrderEvent>(producerControllerOptions),
  'producer',
);


producer.tell({ kind: 'reliable-delivery.send', body: { orderId: 'o-1' } });

Intern:

  • Vergibt Seq 1 für die erste Nachricht, 2 für die nächste, …
  • Sendet jede via Wrapped Tell an den Consumer.
  • Hält sie in einem Buffer, bis sie geACKt sind.
  • Retransmittiert, wenn binnen resendTimeout kein ACK kommt.

Konfiguration

Abschnitt betitelt „Konfiguration“
type ProducerControllerOptionsType<T> = {
  consumer:          ActorRef<Delivery<T>>;
  resendTimeout?:  number;       // Default 500
  windowSize?:       number;       // Default 16
  producerId?:       string;       // wird auto-generiert, wenn weggelassen
};
FeldZweck
consumerDie ConsumerController-Ref, an die zugestellt wird.
resendTimeoutWie lange auf ein ACK gewartet wird, bevor retransmittiert wird.
windowSizeFlow-Control-Window — pausiert das Queuing nach N unbestätigten In-Flight-Nachrichten.
producerIdStabiler Identifier, mit dem der Consumer über Restarts hinweg dedupliziert. Wird auto-generiert, wenn weggelassen; bei persistierten Producern willst Du einen stabilen String, damit Recovery matcht.

Backpressure

Abschnitt betitelt „Backpressure“
windowSize: 16

Der Producer hält bis zu 16 unbestätigte Nachrichten in Flight. Darüber hinaus werden eingehende { kind: 'reliable-delivery.send' }-Nachrichten eingereiht — das tell des Callers ist erfolgreich, aber das eigentliche Senden pausiert, bis ACKs den Slot freigeben.

Es gibt kein Signal zurück an den Caller, wenn dieses Queuing passiert: ein reliable-delivery.send ist Fire-and-Forget — der Producer antwortet nie darauf, es gibt also kein ask-basiertes “Ich habe sie gesendet”-ACK, auf das man warten könnte.

Für Sender, die wissen müssen, ob eine Nachricht durchgekommen ist, übergibst du einen confirm-Callback beim Send. Er feuert, sobald der Consumer diese Nachricht per ACK bestätigt — mit null bei Erfolg oder einem Error, wenn der Producer stoppt, während die Nachricht noch in der Warteschlange liegt:

producer.tell({
  kind: 'reliable-delivery.send',
  body: { orderId: 'o-1' },
  confirm: (err) => {
    if (err) console.error('delivery failed', err);
    else console.log('delivered and acked');
  },
});

Um daraus echtes Backpressure zu machen, richte dein Sende-Tempo nach diesen Bestätigungen aus — halte die nächste Charge zurück, bis frühere Nachrichten bestätigt sind.

Retransmission

Abschnitt betitelt „Retransmission“

Wenn ein ACK nicht binnen resendTimeout eintrifft:

Seq 5 zum Zeitpunkt t=0 gesendet
ACK für Seq 5 kommt nie
bei t=500 → Seq 5 retransmitten
bei t=1000 → Seq 5 retransmitten
... bis das ACK kommt oder der Producer stoppt

Retransmits nutzen dieselbe Seq wie das Original. Der Consumer dedupliziert anhand der Seq.

State + Restart

Abschnitt betitelt „State + Restart“
// Ohne Persistenz:
Producer crasht → Buffer weg → unbestätigte Nachrichten sind verloren
// Beim Restart setzt das Senden bei Seq 1 wieder ein (kein Resume)


// Mit ProducerController + PersistentActor-Wrapper:
Producer crasht → Recovery baut Buffer + last-acked-Seq wieder auf
// Beim Restart setzt das Senden bei der richtigen Seq fort

Für volle Durability den Producer in ein PersistentActor-Muster einwickeln — ausgehende Nachrichten persistieren, bevor gesendet wird; beim Restart unbestätigte neu abspielen.

Das einfache ProducerController des Frameworks ist in-memory — ausreichend für ephemere Streams. Für durable Streams legst du Persistenz darauf.

Mehrere Consumer

Abschnitt betitelt „Mehrere Consumer“
// Ein Producer pro Consumer:
const p1 = system.spawnAnonymous(() => new ProducerController({ consumer: c1Ref, ... }));
const p2 = system.spawnAnonymous(() => new ProducerController({ consumer: c2Ref, ... }));

Jeder ProducerController ist 1:1 mit einem ConsumerController. Für Fan-out senden mehrere Producer an verschiedene Consumer.

Für routing-basiertes Fan-out (ein logischer Stream auf N Consumer aufgeteilt) würdest du einen eigenen Router darüber schreiben.

Cluster-aware

Abschnitt betitelt „Cluster-aware“

Die consumer-Ref des ProducerController kann auf einen Remote-Consumer zeigen (anderer Cluster-Node). Der Cluster-Transport serialisiert Envelopes; Retransmissions funktionieren gleich.

Der Producer hält den Buffer lokal — bei einem Crash des Producer-Hosts sind diese Nachrichten verloren, sofern nicht persistiert.

Vergleich mit rohem tell

Abschnitt betitelt „Vergleich mit rohem tell“
ref.tell(msg):                            ProducerController:
- Keine Seq, kein ACK, kein Retransmit      - Vertrag zuverlässige Zustellung
- Verloren bei totem Empfänger              - Übersteht transiente Fehler
- Sub-Mikrosekunden-Kosten                  - ~50µs Overhead pro Nachricht

Nimm den Controller nur für Streams, in denen Verlust inakzeptabel ist; rohes tell für alles andere.

Wohin als Nächstes

Abschnitt betitelt „Wohin als Nächstes“