Почему исключения не останавливаются поток потока Akka

Question

У меня есть поток, который зависит от ответов API. Когда ответы не соответствуют тому, что я ожидаю, генерируется исключение. Эта стратегия хорошо работает для Spray и для прямого тестирования методов с specs2.

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

Это мой поток:

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

Моя стратегия для этого является использование map фьючерсов.

так:

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

Это потенциальный метатель исключение в совершенно другом месте:

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

Кажется, я что-то не хватает, но не может видеть, что. Спасибо за любые указатели.

Answer 1

Это звучит как проблема, которая будет рассмотрена после того, как будет реализован Supervision for Akka Streams. Потоки Akka по-прежнему «пред-экспериментальные», так что функция еще не реализована, но, безусловно, планируется вскоре включить.

// На момент написания настоящего комментария текущая версия 1.0-M2 (предварительный этап).

Answer 2

Дело в том, что я отсутствовал был mapAsync.

изменения выше функцию:

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

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