scala - Akka Streams 与 Akka HTTP 服务器和客户端-6ren

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

转载作者：行者123 更新时间：2023-12-02 02:55:19

25

4

我正在尝试在我的 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] -> 流程[请求，响应]。现在，如果我的 Akka Stream 知识对我有用，我需要使用 Flow.fromGraph 来完成这样的事情，但是，这就是我陷入困境的地方。

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

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

<小时/>

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

最佳答案

您不需要使用Flow.fromGraph。相反，使用 flatMapConcat 的单个 Flow 将起作用:

//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!"))    
})

然后可以使用此流绑定(bind)到传入请求:

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

而且一切都没有 future ......

关于scala - Akka Streams 与 Akka HTTP 服务器和客户端，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/34253591/

25

4

0

文章推荐： c# - 对象处置异常 : The CancellationTokenSource has been disposed

文章推荐： opengl - OpenGL坐标系是左手坐标系还是右手坐标系？

文章推荐： java - 如何为 JAVA 中的泛型类型提供默认值

文章推荐： c++ - getaddrinfo在程序中调用assert

scala - 使用 Akka/Akka Streams/Akka HTTP 时出现 Akka 版本冲突
我正在尝试使用 Scala 2.12.3、sbt 0.13.6 以及 build.sbt 文件中的以下库依赖项构建一个项目: libraryDependencies ++= Seq( "com.t
akka - Akka 消息传递如何工作
我根本没有任何 Akka 经验。我想知道 Akka 消息传递如何在 JVM 内以及 JVM 之间工作。 JVM 中的消息是一些类似 POJO 的对象吗？ JVM之间的通信是否需要任何类型的JMS(如服
akka - Akka EventBus可以与远程参与者一起使用吗？
Akka的EventBus是否可以与远程 Actor 一起使用？据我所知，它本身不支持此功能。任何人都可以确认吗？看起来有可能对提供相似功能的Actor进行编码。例如。在订阅远程服务器上的Even
akka - Akka 中集群和远程处理的区别
我在单个 JVM 上使用 Akka 开发了我的应用程序。现在我想在多台机器上分配工作负载。我开始阅读文档并感到困惑。有两种方法可以通过集群和远程处理使 Akka 应用程序分发。我不明白两者之间的区
akka - 将 Akka HTTP 连接到 Akka 流
我想使用 Akka HTTP 构建一个连接到现有接收器(带有 Kafka react 流)的 REST 服务，但我不知道如何将 HTTP 流链接到 Akka 流接收器... 我应该选择使用 Flows
akka - 线程模型和 AKKA
我在某处听说默认情况下是一个 Actor 系统，这意味着它的 ExecutorService/Dispatcher 正在创建一个非 Deamon 线程池来运行 Actor。如果确实如此，那将解释我所经
akka - Akka 中的存储和控制感知邮箱
在我的应用程序中，我有一个角色需要在等待某些操作完成时存储消息，同时它需要支持高优先级消息(控制消息)。 stash trait 需要一个 Dequeue邮箱类型，我找不到控制感知出队邮箱是否有意义。
akka - Akka.NET 和原始 Akka 可以使用 Remoting 进行通信吗？
Akka.NET 和原始 Akka 可以使用 Remoting 进行通信吗？换句话说，Akka 可以用于连接系统中的 JVM 和 CLR 吗？最佳答案这个问题在 akka.net Github
akka - Akka Decider 是否可以访问完整的故障场景？
Akka 新手。创建一个扩展 SupervisorStrategy 的新 Scala 类为我提供了以下模板: class MySupervisorStrategy extends Supervisor
akka - 创建用于并行处理集合元素的 Akka 流
我正在尝试为包含并行处理流的 Akka 流定义一个图(我正在使用 Akka.NET，但这应该无关紧要)。想象一个订单的数据源，每个订单由一个订单 ID 和一个产品列表(订单商品)组成。工作流程如下:
akka - 如何取消 Akka Actor ？
我有一个 akka actor(worker)接收请求并回复它。请求处理可能需要 3-60 分钟。来电者(也是 Actor )目前正在使用 !!!并等待 future.get，但是如果需要，可以更改
akka - 我应该如何在 Akka 持久性中构建持久性参与者？
我应该如何在 Akka 持久化 (Eventsourcing/CQRS) 中构建我的 Actor？分层平行我的电子商务应用程序中有这些域对象用户 - 用户可以创建帐户商店 - 用户可以创建商
akka - 如何将多个参与者作为源附加到 Akka 流？
我正在尝试构建和运行一个 akka 流(在 Java DSL 中)，以 2 个 actor 作为源，然后是一个合并结点，然后是 1 个接收器: Source src1 = Source.act
akka - 如何在 Akka 中监督集群单例？
我正在尝试监督 Akka Actor ，更具体地说是 Cluster Singleton创建使用 ClusterSingletonManager .我试图更好地控制异常、日志和 Actor 的生命周期
akka - 何时使用各种 Akka 邮箱类型
我试图了解何时何地使用不同的 built-in Akka mailboxes以及何时适合自己滚动。但是，该页面上的任何地方都没有解释“ 有界邮箱 ”实际上是什么，或者它的行为方式与无界邮箱有何不同。此
akka - 在 Akka 中等待多个结果
在Akka中等待多个actor的结果的正确方法是什么？ Principles of Reactive Programming Coursera 类(class)有一个带有复制键值存储的练习。无需深入研
akka - 崩溃后如何重新创建动态创建的持久化 AKKA actor
我正在为一个项目评估 Akka，我正在尝试弄清楚我是否可以通过将参与者状态保存在高可用数据存储中来使用 Akka-Persistence 实现服务的高可用性。 (我不打算使用 Akka-Cluster
akka-cluster - Akka 集群状态弱起来
我阅读了 Akka 文档并找到了这个 As mentioned before, if a node is unreachable then gossip convergence is not poss
akka - 使用 akka 流有条件地跳过流
我正在使用 akka 流，并且我有一段我需要有条件地跳过的图表，因为流程无法处理某些值。具体来说，我有一个接受字符串并发出 http 请求的流，但是当字符串为空时，服务器无法处理这种情况。但我只需要返
akka - 提高 Akka 远程吞吐量
我们正在考虑使用 Akka 进行客户端服务器通信，并尝试对数据传输进行基准测试。目前我们正在尝试发送一百万条消息，其中每条消息都是一个具有 8 个字符串字段的案例类。目前，我们正在努力获得可接受的性

首页

博学

6Ren·AI

商城

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