如何在流中捕获将来失败的异常

Question

我正在尝试创建一个流来轮询休息服务并解组json对象。

我创建了一个 source.tick，它每 5 秒执行一次 http 请求。如果此操作成功，则 HttpResponse 将包含 OK。否则，服务不可用。结果将发送给参与者。请参阅以下代码：

def poll(pollActor: ActorRef) {
    val source = Source.tick(0.seconds, 3.seconds, HttpRequest(uri = Uri(path = Path("/posts/1"))))
    val flow = Http().outgoingConnectionHttps("jsonplaceholder1.typicode.com").mapAsync(1) {
      case HttpResponse(StatusCodes.OK, _, entity, _) =>
        Unmarshal(entity).to[Item]
      case resp @ HttpResponse(code, _, _, _) =>
        log.warning("Request failed, response code: " + code)
        Future.failed(new Exception)
    }
    source.via(flow).runWith(Sink.actorRef[Equals](pollActor,akka.actor.Status.Success(())))
}

参与者将从流中接收结果，如以下代码所示：

def receive = {
    case k : Item => println(k)
    case f : Failure => {
      println("We failed: " + f)
    }
}

我应该在哪里以及如何处理未来引发的异常？

Answer 1

解决此问题的一种方法是使失败成为流的显式部分。

    val flow = Http().outgoingConnectionHttps("jsonplaceholder1.typicode.com").mapAsync(1) {
      case HttpResponse(StatusCodes.OK, _, entity, _) =>
        Unmarshal(entity).to[Item].map(Right(_))
      case resp @ HttpResponse(code, _, _, _) =>
        Future.successful(Left(MyDomainFailure("some meaningful message/data")))
    }

请注意，现在流的类型是

Flow[HttpRequest, Either[MyDomainFailure, Item], Future[OutgoingConnection]] 

这具有清晰度的附加值，使下游阶段意识到故障并迫使他们处理它(好吧，在这种情况下不是真的，因为你正在使用一个参与者。如果你停留在流的领域内，你将被迫处理它们(。

    def receive = {
      case Right(item) => println(item)
      case Left(failure) => {
        println("We failed: " + failure.msg)
      }
    }

Answer 2

这是我

使用的修复程序，虽然它不会产生异常，但HttpResponse中包含的Failure只是在接收函数中匹配。

def poll(pollActor: ActorRef) {
  val source = Source.tick(0.seconds, 3.seconds, HttpRequest(uri = Uri(path = Path("/posts/1"))))
  val flow = Http().outgoingConnectionHttps("jsonplaceholder1.typicode.com").mapAsync(1) {
    // Where able to reach the API.
    case HttpResponse(StatusCodes.OK, _, entity, _) =>
      // Unmarshal the json response.
      Unmarshal(entity).to[Item]
    // Failed to reach the API.
    case HttpResponse(code, _, _, _) =>
      Future.successful(code)
  }
    source.via(flow).runWith(Sink.actorRef[Any](pollActor,akka.actor.Status.Success(())))
}

在这里，我们匹配HttpResponse产生的Failure。

def receive = {
  case item: Item => println(item)
  case failure: Failure => {
    log.warning("Request failed, response code: " + failure)
  }
}