如何将解组器应用于HttpRequest->HttpResponse流

Question

我在我的java项目中使用akka框架来处理HTTP服务，这就是为什么我使用javaDSL来处理akka流。所以，我的故事在下面。我创建了http缓存池（在我的演员中）：

val pool = Http.get(getContext().system()).cachedHostConnectionPool(
            "https://somehost.com",
            ActorMaterializer.create(getContext().system())
    )

然后我基于HttpRequest创建了一个源：

val source = Source.single(HttpRequest.GET("/resources/123/"))

然后我创建了解组器，用于将HttpReporte转换为精确的实体：

val unmarshaller = Jackson.unmarshaller(objectMapper, ResourceEntity.class)

下一步是通过池传递源，以处理我的actor中的响应：

source.via(pool)
      // TODO: unmarshalling
      .to(Sink.actorRef(getSelf(), new FlowCompletedMessage()))
      .run(ActorMaterializer.create(getContext().system()));

问题是如何将解组器应用到这个流中？我找不到任何例子，我觉得自己就像一个试图把正方形塞进圆孔的孩子...

Answer 1

大致如下：

source.via(pool)
      .mapAsync(response => Unmarshal(response).to[T])
      .to(Sink.actorRef(getSelf(), new FlowCompletedMessage()))
      .run(ActorMaterializer.create(getContext().system()));