http => akka 流 => http

Question

我想使用 akka 流来将一些 json 网络服务连接在一起。我想知道从http请求生成流并将块流传输到另一个的最佳方法。 有没有办法定义这样的图表并运行它而不是下面的代码？ 到目前为止，我尝试过这种方式，但不确定它是否真的在流式传输：

override def receive: Receive = {
   case GetTestData(p, id) =>
     // Get the data and pipes it to itself through a message as recommended
     // https://doc.akka.io/docs/akka-http/current/client-side/request-level.html
     http.singleRequest(HttpRequest(uri = uri.format(p, id)))
       .pipeTo(self)

   case HttpResponse(StatusCodes.OK, _, entity, _) =>
     val initialRes = entity.dataBytes.via(JsonFraming.objectScanner(Int.MaxValue)).map(bStr => ChunkStreamPart(bStr.utf8String))

     // Forward the response to next job and pipes the request response to dedicated actor
     http.singleRequest(HttpRequest(
       method = HttpMethods.POST,
       uri = "googl.cm/flow",
       entity = HttpEntity.Chunked(ContentTypes.`application/json`, 
       initialRes)
     ))


   case resp @ HttpResponse(code, _, _, _) =>
     log.error("Request to test job failed, response code: " + code)
     // Discard the flow to avoid backpressure
     resp.discardEntityBytes()

   case _ => log.warning("Unexpected message in TestJobActor")
 }

Answer 1

这应该是一个与您的receive 等效的图表：

Http()
.cachedHostConnectionPool[Unit](uri.format(p, id))
.collect {
  case (Success(HttpResponse(StatusCodes.OK, _, entity, _)), _) =>
    val initialRes = entity.dataBytes
      .via(JsonFraming.objectScanner(Int.MaxValue))
      .map(bStr => ChunkStreamPart(bStr.utf8String))
    Some(initialRes)

  case (Success(resp @ HttpResponse(code, _, _, _)), _) =>
    log.error("Request to test job failed, response code: " + code)
    // Discard the flow to avoid backpressure
    resp.discardEntityBytes()
    None
}
.collect {
  case Some(initialRes) => initialRes
}
.map { initialRes =>
  (HttpRequest(
     method = HttpMethods.POST,
     uri = "googl.cm/flow",
     entity = HttpEntity.Chunked(ContentTypes.`application/json`, initialRes)
   ),
   ())
}
.via(Http().superPool[Unit]())

this的类型是Flow[(HttpRequest, Unit), (Try[HttpResponse], Unit), HostConnectionPool]，其中Unit是一个关联ID，如果你想知道哪个请求对应到达的响应，可以使用，HostConnectionPool物化值可以用来关闭与主机的连接。只有cachedHostConnectionPool 会返回这个具体化的值，superPool 可能会自行处理（尽管我没有检查过）。无论如何，我建议您在关闭应用程序时使用Http().shutdownAllConnectionPools()，除非您出于某种原因需要。根据我的经验，它更不容易出错（例如忘记关机）。

您也可以使用 Graph DSL，来表达相同的图形：

val graph = Flow.fromGraph(GraphDSL.create() { implicit b =>
import GraphDSL.Implicits._

    val host1Flow = b.add(Http().cachedHostConnectionPool[Unit](uri.format(p, id)))
    val host2Flow = b.add(Http().superPool[Unit]())

    val toInitialRes = b.add(
      Flow[(Try[HttpResponse], Unit)]
        .collect {
          case (Success(HttpResponse(StatusCodes.OK, _, entity, _)), _) =>
            val initialRes = entity.dataBytes
              .via(JsonFraming.objectScanner(Int.MaxValue))
              .map(bStr => ChunkStreamPart(bStr.utf8String))
            Some(initialRes)

          case (Success(resp @ HttpResponse(code, _, _, _)), _) =>
            log.error("Request to test job failed, response code: " + code)
            // Discard the flow to avoid backpressure
            resp.discardEntityBytes()
            None
        }
    )

    val keepOkStatus = b.add(
      Flow[Option[Source[HttpEntity.ChunkStreamPart, Any]]]
        .collect {
          case Some(initialRes) => initialRes
        }
    )

    val toOtherHost = b.add(
      Flow[Source[HttpEntity.ChunkStreamPart, Any]]
        .map { initialRes =>
          (HttpRequest(
             method = HttpMethods.POST,
             uri = "googl.cm/flow",
             entity = HttpEntity.Chunked(ContentTypes.`application/json`, initialRes)
           ),
           ())
        }
    )

    host1Flow ~> toInitialRes ~> keepOkStatus ~> toOtherHost ~> host2Flow

    FlowShape(host1Flow.in, host2Flow.out)
})