4

我有一个依赖 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")

好像我错过了一些东西,但看不到什么。感谢您的任何指示。

4

2 回答 2

3

这听起来像是一个问题,一旦实施了 Akka Streams 的监督,就会得到解决。Akka Streams 仍处于“预实验阶段”,因此该功能尚未实现,但肯定会很快包含在内。

// 在撰写此评论时,当前版本为 1.0-M2(预览里程碑)。

于 2015-01-09T11:33:14.523 回答
3

我缺少的是mapAsync

将上述函数更改为:

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

这样,期货就被解开,异常会按预期停止程序。

于 2015-01-21T10:04:15.737 回答