Akka流与Akka HTTP服务器和客户端
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)
这是我认为我应该用于这种方法的布局:source[Request]->flow[Request,Request]->flow[Request,Request]->flow[Request,Response]->flow[Response,Response]并且为了适应404路由，还可以使用source[Request]->flow[Request,Response]。现在，如果我的Akka流知识对我有用的话，我需要使用flow.fromgraph来处理这样的事情，然而，这就是我被困住的地方。
在future中，我可以为不同的endpoint进行简单的映射和flatMap，但在流中，这意味着将流划分为多个流，我不太确定该如何进行。我想过使用UnzipWith和Options或通用广播。
如能在这方面提供任何帮助，将不胜感激。
如果有必要的话我就不知道了？--http://doc.akka.io/docs/akka-stream-and-http-experimental/2.0-M2/scala/stream-customize.html
共有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)
没有未来...
-
前面的章节介绍了所有 Redis 的重要功能组件： 数据结构、数据类型、事务、Lua 环境、事件处理、数据库、持久化， 等等， 但是我们还没有对 Redis 服务器本身做任何介绍。 不过， 服务器本身并没有多少需要介绍的新东西， 因为服务器除了维持服务器状态之外， 最重要的就是将前面介绍过的各个功能模块组合起来， 而这些功能模块在前面的章节里已经介绍过了， 所以本章将焦点放在服务器的初始化过程，
-
问题内容： 我有一个带有两个服务器端进程的程序。一台服务器只是将ArrayList发送给客户端。另一个服务器首先从客户端获取一个字符串，然后找到与该ID对应的正确记录，然后将记录发回。 我在第二个服务器进程中遇到问题。请参阅下面的println语句，其中显示“卡在此处”。那就是程序挂起的地方。 客户端： 问题答案： 您需要在两端的ObjectInputStream之前创建ObjectOutputS
-
问题内容： 我尝试使用以下代码从服务器到客户端发送文件和目录列表。服务器正在从客户端接收消息，但我不知道服务器是否没有发送回结果或客户端是否不接受结果。 服务器端： 问题答案： 据我所见，您在客户端上做的同时在服务器上做。从服务器发送的字符串中没有行尾字符，因此客户端将永远无法完成。执行outqw.println（）或添加到要发送的内容的末尾。话虽这么说，很难用一堆注释掉的东西来浏览未格式化的代码
-
我正在使用spring cloud Eureka配置一个应用程序。我在8761端口启动我的discovery应用程序，并在“http：//localhost:8761”中到达控制台。 所以，我启动了我的客户端应用程序，它出现在eureka控制台的“应用程序”页面中。 第一个问题：我的客户机在properties config中使用了“server.port=0”，所以tomcat端口是随机启动的。
-
似乎服务器拒绝了wireshark输出中的tls协商，但我从代码中看不出原因。它是基于工作的代码，只是它被否决了，因此我用新的API更新。代码是开始。需要使用真实的证书。有人知道为什么服务器发送tcp FIN，ack吗？ 我有以下服务器代码： 23 16.856111 sonymobi_7f:55:af intelcor_25:1d:fc ARP 42 10.1.10.100在84:c7:ea:7
-
我想在一些计算机之间建立点对点连接，这样用户就可以在没有外部服务器的情况下聊天和交换文件。我最初的想法如下： 我在服务器上制作了一个中央服务器插座，所有应用程序都可以连接到该插座。此ServerSocket跟踪已连接的套接字（客户端），并将新连接的客户端的IP和端口提供给所有其他客户端。每个客户端都会创建一个新的ServerSocket，所有客户端都可以连接到它。 换句话说：每个客户端都有一个Se