scala - streams - ¿Cómo detener bruscamente una secuencia de akka Runnable Graph?

akka streams

Question

No soy capaz de averiguar cómo detener la secuencia de Akka Runnable Graph inmediatamente? ¿Cómo usar killswitch para lograr esto? Han sido solo unos días que comencé las transmisiones de akka. En mi caso, leo líneas de un archivo y hago algunas operaciones de flujo y escritura en el fregadero. Lo que quiero hacer es dejar de leer el archivo inmediatamente cuando quiera, y espero que esto posiblemente detenga todo el gráfico en ejecución. Cualquier idea sobre esto sería muy apreciada.

Gracias por adelantado.

Answer 1

Una forma tiene un servicio o shutdownhookup que puede llamar a un gráfico cancelable

val graph=
    Source.tick(FiniteDuration(0,TimeUnit.SECONDS), FiniteDuration(1,TimeUnit.SECONDS), Random.nextInt).to(Sink.foreach(println))
  val cancellable=graph.run()

  cancellable.cancel

El cancelable.cancel puede ser parte de ActorSystem.registerOnTermination

Answer 2

Desde Akka Streams 2.4.3, hay una forma elegante de detener la transmisión desde el exterior a través de KillSwitch .

Considere el siguiente ejemplo, que detiene la transmisión después de 10 segundos.

object ExampleStopStream extends App {

  implicit val system = ActorSystem("streams")
  implicit val materializer = ActorMaterializer()

  import system.dispatcher

  val source = Source.
    fromIterator(() => Iterator.continually(Random.nextInt(100))).
    delay(500.millis, DelayOverflowStrategy.dropHead)
  val square = Flow[Int].map(x => x * x)
  val sink = Sink.foreach(println)

  val (killSwitch, done) =
    source.via(square).
    viaMat(KillSwitches.single)(Keep.right).
    toMat(sink)(Keep.both).run()

  system.scheduler.scheduleOnce(10.seconds) {
    println("Shutting down...")
    killSwitch.shutdown()
  }

  done.foreach { _ =>
    println("I''m done")
    Await.result(system.terminate(), 1.seconds)
  }

}