Hi,
I'm using a Source.queue with BroadcastHub to implement a pattern where a
web request can add an item to the queue, attach to the graph and get a
result. The problem I'm stuck on is that I'm not sure how to handle errors
without failing the graph. Example:
Without the supervision strategy I see getting the following output
flow handling 1
Result 1 Success(Some(error))
Result 2 Success(None)
The graph stops after the exception and "2" is never processed
With the supervision, the graph recovers and processes "2" but never sees
"error"
flow handling 1
Graph failed java.lang.Exception.. resuming
Result 1 Failure(java.util.concurrent.TimeoutException: No elements passed
in the last 1 second.)
flow handling 2
Result 2 Success(Some(2))
What I'd like to see is
Result 1 Success(Some(error))
Result 2 Success(Some(2))
Is there a way I can recover the graph on a failure but also see the value
returned from the "recover" function. Also, why does the graph still fail
the graph when there is a recover? Thanks
import akka.NotUsed
import akka.actor.ActorSystem
import akka.stream._
import akka.stream.scaladsl.{BroadcastHub, Flow, Keep, Sink, Source,
SourceQueueWithComplete}
import scala.concurrent.Await
import scala.util.Try
object QueueGraph extends App {
implicit val system = ActorSystem()
implicit val materializer = ActorMaterializer()
val sourceQueue: Source[String, SourceQueueWithComplete[String]] =
Source.queue[String](bufferSize = 2, OverflowStrategy.backpressure)
val decider: Supervision.Decider = {
case e: Exception =>
println(s"Graph failed ${e}.. resuming")
Supervision.Resume
}
val (queue: SourceQueueWithComplete[String], source: Source[String, NotUsed])
=
sourceQueue
.via(Flow[String].map { num =>
println(s"flow handling ${num}")
if (num == "1") {
throw new Exception()
} else {
num
}
})
.recover{case e: Exception => "error"}
// If supervisor is removed, the recover catches the exception and returns
"error" but the graph fails and no longer processes anything. With supervisor
"error" is never emitted
// .withAttributes(ActorAttributes.supervisionStrategy(decider))
.toMat(BroadcastHub.sink(bufferSize = 1))(Keep.both).run()
import scala.concurrent.duration._
// find error
val f = source
.filter(_ == "error")
.idleTimeout(1.second)
.toMat(Sink.headOption)(Keep.right).run()
queue.offer("1")
println(s"Result 1 ${Try(Await.result(f, Duration.Inf))}")
queue.offer("2")
val f2 = source
.filter(_ == "2")
.idleTimeout(1.second)
.toMat(Sink.headOption)(Keep.right).run()
println(s"Result 2 ${Try(Await.result(f2, Duration.Inf))}")
}
--
>>>>>>>>>> Read the docs: http://akka.io/docs/
>>>>>>>>>> Check the FAQ:
>>>>>>>>>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
---
You received this message because you are subscribed to the Google Groups "Akka
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
To post to this group, send email to [email protected].
Visit this group at https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
