Akka Streams / HTTP：从响应中获取原始请求答案

【问题标题】：Akka Streams / HTTP: Get original request from responseAkka Streams / HTTP：从响应中获取原始请求
【发布时间】：2018-05-14 14:21:55
【问题描述】：

我有一个 Akka Streams 源，它通过一个流程并发布一个 HTTP 请求：

source.map(toRequest)
  .via(Http().outgoingConnection(host))
  .map(toMessage)

假设toRequest 方法将字符串映射到HttpRequest，toMessage 方法将HttpResponse 映射到下游处理所需的消息类。假设消息类需要包含一些原始信息。

是否可以从HttpResponse 获取原始HttpRequest？如果没有，有什么办法可以保留一些原始信息？

【问题讨论】：

标签： scala akka akka-stream akka-http

【解决方案1】：

一种方法是使用客户端 API 的 Future-based variant 和一个自定义案例类，该类包含您要向下游传播的信息。例如：

case class Message(request: HttpRequest, response: HttpResponse)

source
  .map(toRequest)
  .mapAsync(parallelism = 3) { request => // adjust the level of parallelism as needed
    Http().singleRequest(request).map(response => Message(request, response))
  }
  // continue processing; at this point you have a Source[Message, _]

【讨论】：

【解决方案2】：

您可以使用您的图形 API 绕过您的 HttpRequest。一个例子是：

object Main extends App {

  implicit val as = ActorSystem("Test")
  implicit val m = ActorMaterializer()
  implicit val ec = as.dispatcher

  def src1 = Source.fromIterator(() => List(1, 2, 3).toIterator)

  val srcGraph = Source.fromGraph(GraphDSL.create() { implicit builder: GraphDSL.Builder[NotUsed] =>
    import GraphDSL.Implicits._

    // Create one flow to prepend the 'Number' string to the input integer
    def flow1 = Flow[Int].map { el => s"Number $el"  }

    // Create a broadcast stage 
    val broadcast = builder.add(Broadcast[Int](2))
    val zip       = builder.add(Zip[Int, String]())

    src1 ~> broadcast.in

    // The 0 port sends the int to the zip stage directly
    broadcast.out(0) ~>          zip.in0
    broadcast.out(1) ~> flow1 ~> zip.in1

    SourceShape(zip.out)
  })

  Source.fromGraph(srcGraph).runForeach(println(_))
}

graph api 提供了许多选项来做这样的事情。

【讨论】：