Akka Streams 与 Akka HTTP 服务器和客户端

Question

我正在尝试在我的 Akka Http 服务器上创建一个端点，它使用外部服务告诉用户它的 IP 地址（我知道这可以更容易地执行，但我这样做是为了挑战）。

最上层不使用流的代码是这样的：

implicit val system = ActorSystem()
implicit val materializer = ActorMaterializer()

val requestHandler: HttpRequest => Future[HttpResponse] = {
  case HttpRequest(GET, Uri.Path("/"), _, _, _) =>
    Http().singleRequest(HttpRequest(GET, Uri("http://checkip.amazonaws.com/"))).flatMap { response =>
      response.entity.dataBytes.runFold(ByteString(""))(_ ++ _) map { string =>
        HttpResponse(entity = HttpEntity(MediaTypes.`text/html`,
          "<html><body><h1>" + string.utf8String + "</h1></body></html>"))
      }
    }

  case _: HttpRequest =>
    Future(HttpResponse(404, entity = "Unknown resource!"))
}

Http().bindAndHandleAsync(requestHandler, "localhost", 8080)

它工作正常。然而，作为一个挑战，我想限制自己只使用流（没有 Future's）。

这是我想用于这种方法的布局： Source[Request] -> Flow[Request, Request] -> Flow[Request, Response] ->Flow[Response, Response] 并容纳 404 路由，也 Source[Request] -> Flow[Request, Response]。现在，如果我的 Akka Stream 知识对我有用，我需要为这样的事情使用 Flow.fromGraph，然而，这就是我被困的地方。

在 Future 中，我可以为各种端点做一个简单的地图和平面地图，但在流中，这意味着将 Flow 分成多个 Flow，我不太确定我该怎么做.我考虑过使用 UnzipWith 和 Options 或通用广播。

如能提供有关此主题的任何帮助，我们将不胜感激。

我不知道是否有必要这样做？ -- http://doc.akka.io/docs/akka-stream-and-http-experimental/2.0-M2/scala/stream-customize.html

Answer 1

您不需要使用 Flow.fromGraph。相反，使用 flatMapConcat 的单一流程将起作用：

//an outgoing connection flow
val checkIPFlow = Http().outgoingConnection("checkip.amazonaws.com")

//converts the final html String to an HttpResponse
def byteStrToResponse(byteStr : ByteString) = 
  HttpResponse(entity = new Default(MediaTypes.`text/html`,
                                    byteStr.length,
                                    Source.single(byteStr)))

val reqResponseFlow = Flow[HttpRequest].flatMapConcat[HttpResponse]( _ match {
  case HttpRequest(GET, Uri.Path("/"), _, _, _) =>
    Source.single(HttpRequest(GET, Uri("http://checkip.amazonaws.com/")))
          .via(checkIPFlow)
          .mapAsync(1)(_.entity.dataBytes.runFold(ByteString(""))(_ ++ _))
          .map("<html><body><h1>" + _.utf8String + "</h1></body></html>")
          .map(ByteString.apply)
          .map(byteStrToResponse)

  case _ =>
    Source.single(HttpResponse(404, entity = "Unknown resource!"))    
})

此流程随后可用于绑定传入请求：

Http().bindAndHandle(reqResponseFlow, "localhost", 8080)

而且都没有 Futures...

Akka Streams 与 Akka HTTP 服务器和客户端

Akka Streams with Akka HTTP Server and Client

scala

akka

akka-stream

akka-http