Monitoring a closed graph Akka Stream

Question

If I have created a RunningGraph in Akka Stream, how can I know (from the outside)

1. when all nodes are cancelled due to completion?
2. when all nodes have been stopped due to an error?

Answer 1

I don't think there is a way to do it for an arbitrary graph, but if you have your graph under control, you just need to attach monitoring sinks to the output of each node which can fail or complete (these are nodes which have at least one output), for example:

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)
  }
}

It is also possible to create only one monitoring actor and use it in all monitoring sinks, but in that case you won't be able to differentiate easily between streams which have failed.

And there also is watchTermination() method on sources and flows which allows to materialize a future which terminates together with the flow at this point. I think it may be difficult to use with GraphDSL, but with regular stream methods it could look like this:

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}")
  }
}

You can transform the future before piping its value to the actor to differentiate between streams.