Akka-Streams - Shutdown-Stream mit Gruppierung ohne Datenverlust

Question

Ich habe eine Quelle, die Elemente und eine Senke gruppiert, die eine Batch-Anfrage macht, Ich benutze KillSwitch, um den Graph zu einem beliebigen Zeitpunkt herunterfahren zu können. Das Problem, dass die Datensätze der letzten unvollständigen Batch dass Quellenausgänge sind verloren, wenn switch.shutdown() ist

val source = Source.tick(10.millis, 10.millis, "tick").grouped(500) 

val (switch, _) = source.viaMat(KillSwitches.single)(Keep.right) 
.toMat(sink)(Keep.both).run() 

Thread.sleep(3000) // wait some arbitrary time 

switch.shutdown() 

ein Weg ‚auszuspülen‘, um die unvollständige Batch Gibt es genannt wird, wenn Abschaltung geschieht?

Answer 1

Das Verhalten des Kill-Schalter Abschaltung ist Positions, wie pro ihre docs

Nach dem Aufruf von [[UniqueKillSwitch # shutdown()]] der laufenden Instanz der [[Graph]] von [[FlowShape ]], die materialisiert auf die [[UniqueKillSwitch]] wird seine Downstream und brechen ihre upstream (außer wenn abgeschlossen oder fehlgeschlagen bereits in diesem Fall der Befehl wird ignoriert).

Siehe auch weitere Dokumentation here.

Jetzt wird die Stufe grouped eine teilweise gefüllte Gruppe nur zum Abschlusszeitpunkt ausgeben, aber nicht, wenn sie abgebrochen wird.

Dies bedeutet, dass die unter dem Diagramm (vor Kill gruppiert) verhält sich wie Sie

beobachtet

val switch = 
    Source.tick(10.millis, 175.millis, "tick") 
      .grouped(10) 
      .viaMat(KillSwitches.single)(Keep.right) 
      .toMat(Sink.foreach(println))(Keep.left) 
      .run() 

während das Diagramm unten (nach Kill gruppiert) wird Teilgruppen stromabwärts bei Beendigung

emittieren

val switch = 
    Source.tick(10.millis, 175.millis, "tick") 
      .viaMat(KillSwitches.single)(Keep.right) 
      .grouped(10) 
      .toMat(Sink.foreach(println))(Keep.left) 
      .run()