Мониторинг замкнутого графика Akka поток

Question

Если я создал в Акко потоке RunningGraph, как я могу знать (с внешней стороны)

1. , когда все узлы будут отменены из-за завершение?
2. , когда все узлы были остановлены из-за ошибки?

Answer 1

Я не думаю, что есть способ сделать это для произвольного графа, но если у вас есть свой граф под контролем, вам просто нужно подключить приемники мониторинга к выходу каждого узла, который может выйти из строя или завершить (эти узлы, которые имеют, по меньшей мере, один выход), например:

import akka.actor.Status 

// obtain graph parts (this can be done inside the graph building as well) 
val source: Source[Int, NotUsed] = ... 
val flow: Flow[Int, String, NotUsed] = ... 
val sink: Sink[String, NotUsed] = ... 

// create monitoring actors 
val aggregate = actorSystem.actorOf(Props[Aggregate]) 
val sourceMonitorActor = actorSystem.actorOf(Props(new Monitor("source", aggregate))) 
val flowMonitorActor = actorSystem.actorOf(Props(new Monitor("flow", aggregate))) 

// create the graph 
val graph = GraphDSL.create() { implicit b => 
    import GraphDSL._ 

    val sourceMonitor = b.add(Sink.actorRef(sourceMonitorActor, Status.Success(()))), 
    val flowMonitor = b.add(Sink.actorRef(flowMonitorActor, Status.Success(()))) 

    val bc1 = b.add(Broadcast[Int](2)) 
    val bc2 = b.add(Broadcast[String](2)) 

    // main flow 
    source ~> bc1 ~> flow ~> bc2 ~> sink 

    // monitoring branches 
    bc1 ~> sourceMonitor 
    bc2 ~> flowMonitor 

    ClosedShape 
} 

// run the graph 
RunnableGraph.fromGraph(graph).run() 

class Monitor(name: String, aggregate: ActorRef) extends Actor { 
    override def receive: Receive = { 
    case Status.Success(_) => aggregate ! s"$name completed successfully" 
    case Status.Failure(e) => aggregate ! s"$name completed with failure: ${e.getMessage}" 
    case _ => 
    } 
} 

class Aggregate extends Actor { 
    override def receive: Receive = { 
    case s: String => println(s) 
    } 
} 

также можно создать только один актер мониторинга и использовать его во всех поглотителей мониторинга, но в этом случае вы не сможете легко различать между потоками, которые потерпели неудачу.

И также существует метод watchTermination() по источникам и потокам, который позволяет материализовать будущее, которое заканчивается вместе с потоком в этой точке. Я думаю, что это может быть трудно использовать с GraphDSL, но с регулярными методами потока это может выглядеть следующим образом:

import akka.Done 
import akka.actor.Status 
import akka.pattern.pipe 

val monitor = actorSystem.actorOf(Props[Monitor]) 
source 
    .watchTermination()((f, _) => f pipeTo monitor) 
    .via(flow).watchTermination((f, _) => f pipeTo monitor) 
    .to(sink) 
    .run() 

class Monitor extends Actor { 
    override def receive: Receive = { 
    case Done => println("stream completed") 
    case Status.Failure(e) => println(s"stream failed: ${e.getMessage}") 
    } 
} 

Вы можете преобразовать будущность обжигающе его значение для актера, чтобы различать между потоками.