Can you do more to reason about your failures?
I'm not a huge fan of the Go language, but, they avoid throwing exceptions
at all costs. There's benefits to be very conservative about throwing
exceptions, only throwing them in truly exceptional cases (IE to indicate
there's a bug in your code, a thread was interrupted, those kinds of things)
On Mon, Oct 23, 2017 at 12:25 PM Andrew <subo...@gmail.com> wrote:
> Perhaps, but I think that would be really tedious to implement in a
> real-life implementation, because it's not just one aspect of the graph
> that can fail. I'd have to construct a lot of different failed types to
> float downstream and handle failures at each junction of the graph. In
> cases where I'm using third party stream libraries (e.g. alpakka), I would
> not be able to wrap the code with error handling. I suppose I could use
> source.recover{...} in these situations but again it seems very tedious. If
> there is not a clean solution I'd lean towards creating a graph per request
> but would prefer a shared graph.
>
> On Mon, Oct 23, 2017 at 9:46 AM Tim Harper <timchar...@gmail.com> wrote:
>
>> Have you considered wrapping your erroneous request in a Try ?
>>
>>
>> val (queue: SourceQueueWithComplete[String], source: Source[String,
>> NotUsed]) =
>> sourceQueue
>> .via(Flow[String].map { num =>
>> println(s"flow handling ${num}")
>> if (num == "1") {
>>
>>
>> Failure(new Exception())
>> } else {
>> Success(num)
>> }
>> })
>>
>> .map {
>> case Success(v) => v
>> case Failure(ex) => "error"
>> }
>>
>> .toMat(BroadcastHub.sink(bufferSize = 1))(Keep.both).run()
>>
>>
>>
>>
>> On Saturday, October 21, 2017 at 7:25:27 AM UTC-6, sub...@gmail.com
>> wrote:
>>>
>>> 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 a topic in the
>> Google Groups "Akka User List" group.
>> To unsubscribe from this topic, visit
>> https://groups.google.com/d/topic/akka-user/v3vFDox-jLg/unsubscribe.
>> To unsubscribe from this group and all its topics, send an email to
>> akka-user+unsubscr...@googlegroups.com.
>> To post to this group, send email to akka-user@googlegroups.com.
>> Visit this group at https://groups.google.com/group/akka-user.
>> For more options, visit https://groups.google.com/d/optout.
>>
> --
> >>>>>>>>>> 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 a topic in the
> Google Groups "Akka User List" group.
> To unsubscribe from this topic, visit
> https://groups.google.com/d/topic/akka-user/v3vFDox-jLg/unsubscribe.
> To unsubscribe from this group and all its topics, send an email to
> akka-user+unsubscr...@googlegroups.com.
> To post to this group, send email to akka-user@googlegroups.com.
> Visit this group at https://groups.google.com/group/akka-user.
> For more options, visit https://groups.google.com/d/optout.
>
--
>>>>>>>>>> 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 akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
