Akka Stream + Akka Http - 获取错误请求答案

【问题标题】：Akka Stream + Akka Http - Get Request on ErrorAkka Stream + Akka Http - 获取错误请求
【发布时间】：2016-08-15 01:51:34
【问题描述】：

我有以下流效果很好：

source
  .map(x => HttpRequest(uri = x.rawRequest))
  .via(Http().outgoingConnection(host, port))
  .to(Sink.actorRef(myActor, IsDone))
  .run()

还有一个简单的actor来处理流完成时的响应状态和最终消息：

/**
  * A simple actor to count how many rows have been processed
  * in the complete process given a http status
  *
  * It also finish the main thread upon a message of type [[IsDone]] is received
  */
class MyActor extends Actor with ActorLogging {

  var totalProcessed = 0

  def receive = LoggingReceive {

    case response: HttpResponse =>

      if(response.status.isSuccess()) {
        totalProcessed = totalProcessed + 1
      } else if(response.status.isFailure()) {
        log.error(s"Http response error: ${response.status.intValue()} - ${response.status.reason()}")
      } else {
        log.error(s"Error: ${response.status.intValue()} - ${response.status.reason()}")
      }

    case IsDone =>
      println(s"total processed: $totalProcessed")
      sys.exit()
  }
}

case object IsDone

我不知道这是否是计算事物和处理响应状态的最佳方法，但到目前为止它正在工作。

问题是如何以我可以知道哪个请求导致特定错误的方式将原始请求传递给参与者。

我的演员可能会期待以下内容：

case (request: String, response: HttpResponse) =>

但是如何传递我在管道开始时拥有的信息？

我在想这样map：

source
  .map(x => (HttpRequest(uri = x.rawRequest), x.rawRequest))

但我不知道如何触发 Http 流。

有什么建议吗？

【问题讨论】：

尝试使用主机连接池而不是显式打开每个请求的传出连接。该模型需要每个请求的任意标识符，然后在响应中返回，以便您可以正确关联请求和响应
嗨@cmbaxter 你是想用这个例子吗？ doc.akka.io/docs/akka/2.4.4/scala/http/client-side/… 但用字符串代替？

标签： scala akka akka-stream akka-http

【解决方案1】：

在@cmbaxter 的帮助下，我可以使用以下代码解决我的问题：

val poolClientFlow = Http().cachedHostConnectionPool[String](host, port)

source
  .map(url => HttpRequest(uri = url) -> url)
  .via(poolClientFlow)
  .to(Sink.actorRef(myActor, IsDone))
  .run()

现在我的演员能够收到这个：

case (respTry: Try[HttpResponse], request: String) =>

【讨论】：