3

我正在尝试在我的 Akka Http Server 上创建一个端点，它使用外部服务告诉用户它的 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 知识对我有用，我需要使用 aFlow.fromGraph来做这样的事情，但是，这就是我卡住的地方。

在 aFuture中，我可以为各种端点做一个简单的 map 和 flatMap，但是在流中这意味着将 Flow 分成多个 Flow，我不太确定我会怎么做。我考虑过使用 UnzipWith 和 Options 或通用广播。

对此主题的任何帮助将不胜感激。

如果这有必要，我不知道？-- http://doc.akka.io/docs/akka-stream-and-http-experimental/2.0-M2/scala/stream-customize.html

4

1 回答 1

5

您不需要使用Flow.fromGraph. 相反，使用的单个 FlowflatMapConcat将起作用：

//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)

而所有没有期货...

于 2015-12-13T21:37:29.067 回答