我有一个依赖 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")
好像我错过了一些东西,但看不到什么。感谢您的任何指示。