CODEWITHSUNDEEP
scalastreamakkaakka-stream
raam86
raam86

Reputation: 6871

Why do exceptions not stop Akka Stream flow

I have a flow that relies on API responses. When the responses do not conform to what I expect an exception is thrown. This strategy works nicely for Spray and for direct method testing with specs2.

However, when I try to use streams with exception throwing modules the flow simply halts.

This is my flow:

 Source(() => file)
      .via(csvToSeq)
      .via(getFromElastic)
      .via(futureExtrtactor)
      .via(findLocaionOfId)
      .foreach(v => v.map(v => println("foreached", v)))
      .onComplete(_ => system.shutdown())

My strategy for this is using map for futures.

like so:

 val findLocaionOfId = Flow[Future[Seq[(String, JsValue)]]].map(future => future.map(jsSeq => {
      jsSeq.zipWithIndex.flatMap { case (x, i) => x._2.asJsObject.getFields("_source").flatMap(js => {
        js.asJsObject("Couldn't convert").getFields("externalId").map({
          case JsString(str) => {
              (i + 1, i == 0, js)
            }
            else (i, false, js)
          }
          case _ => (i, false, x)
        })
      })
      }
    }))

This is a potential exception thrower in a completely different location:

val encoded_url = URLEncoder.encode(url, "UTF-8")

Seems like I am missing something but can't see what. Thanks for any pointers.

Upvotes: 4

Views: 1653

Answers (2)

raam86
raam86

Reputation: 6871

The thing i was missing was mapAsync.

changing the above function to: 

val findLocaionOfId = Flow[Future[Seq[(String, JsValue)]]].mapAsync({...})

This way the futures are unwrapped and exceptions stop the program as expected.

Upvotes: 3

Konrad 'ktoso' Malawski
Konrad 'ktoso' Malawski

Reputation: 13130

This sounds like an issue that will be addressed once Supervision for Akka Streams is implemented. Akka Streams are still "pre-experimental" so that feature has not been implemented yet, but is definitely planned to be included soon.

// As of writing this comment current version is 1.0-M2 (preview milestone).

Upvotes: 3

Related Questions