识别Akka HttpRequest和HttpRequest？

Question

当使用Akka HttpRequest将请求发送给一个actor时，我无法识别响应。actor将处理收到的每条消息，但它不知道哪个请求用于获得该响应。有什么方法可以识别每个请求并匹配响应呢？

注意:我没有服务器来再次发送请求体的任何部分。

预先感谢

import akka.actor.{ Actor, ActorLogging }
import akka.http.scaladsl.Http
import akka.http.scaladsl.model._
import akka.stream.{ ActorMaterializer, ActorMaterializerSettings }
import akka.util.ByteString

class Myself extends Actor with ActorLogging {

import akka.pattern.pipe
import context.dispatcher

final implicit val materializer: ActorMaterializer = 
       ActorMaterializer(ActorMaterializerSettings(context.system))

def receive = {
  case HttpResponse(StatusCodes.OK, headers, entity, _) =>
    entity.dataBytes.runFold(ByteString(""))(_ ++ _).foreach { body =>
      log.info("Got response, body: " + body.utf8String)
  }
  case resp @ HttpResponse(code, _, _, _) =>
    log.info("Request failed, response code: " + code)
    resp.discardEntityBytes()
  }

}

import akka.actor.{ActorSystem, Props}
import akka.http.scaladsl.Http
import akka.http.scaladsl.model._
import akka.stream.ActorMaterializer

object HttpServerMain extends App {

import akka.pattern.pipe

//  import system.dispatcher
implicit val system = ActorSystem()
implicit val materializer = ActorMaterializer()
// needed for the future flatMap/onComplete in the end
implicit val executionContext = system.dispatcher

val http = Http(system)

val myActor = system.actorOf(Props[MySelf])

http.singleRequest(HttpRequest(uri = "http://akka.io"))
    .pipeTo(myActor)

http.singleRequest(HttpRequest(uri = "http://akka.io/another-request"))
    .pipeTo(myActor)
Thread.sleep(2000)
system.terminate()

Answer 1

我认为你不能直接使用pipeTo做到这一点，因为它本质上只是添加和调用你的未来。一种选择是映射（请求，响应）元组，然后将其发送到actor：

val request = HttpRequest(uri = "http://akka.io")
http.singleRequest(request).map {
  response => myActor ! (request, response)
}

class Myself extends Actor with ActorLogging {
  ...
  def receive = {
    case (request, HttpResponse(StatusCodes.OK, headers, entity, _)) =>
      ...

    case (request, resp @ HttpResponse(code, _, _, _)) =>
      log.info(request.toString)
      ...
  }
}

Answer 2

您可以简单地使用 map 来转换未来，并在通过管道将其传送到 myActor 之前向其添加某种 ID（通常为此目的称为相关 ID）：

http.singleRequest(HttpRequest(uri = "http://akka.io"))
    .map(x => (1, x)).pipeTo(myActor)

您需要更改模式匹配块以取补码：

case (id, HttpResponse(StatusCodes.OK, headers, entity, _)) =>

如果出于某种原因，您不能/不想更改模式匹配块，您可以使用相同的方法，但可以在已完成的请求中添加一个唯一的HTTP头（使用复制），如下所示（如果编译，则不检查）：

// make a unique header name that you are sure will not be
// received from http response:
val correlationHeader: HttpHeader = ... // mycustomheader

// Basically hack the response to add your header:
http.singleRequest(HttpRequest(uri = "http://akka.io"))
    .map(x => x.copy(headers = correlationHeader +: headers)).pipeTo(myActor)

// Now you can check your header to see which response that was:
case HttpResponse(StatusCodes.OK, headers, entity, _) =>
  headers.find(_.is("mycustomheader")).map(_.value).getOrElse("NA")

与前面的选项相比，这更像是一个黑客，因为您正在修改一个响应。