从基于 json 数组的 akka http 响应中获取实体序列

Question

我正在尝试像 Akka http 客户端一样使用 Flow，但在从 http 响应中的 json 数组获取 case class 对象的 Seq 时遇到了麻烦。

HTTP 响应：

{
 "bars": [],
 "foos": [
   {
    "id": "a7d1ba80-0934-11e9-0ef9-efa612d204a1",
    "type": "manual",
    "color": "green",
   },
   {
    "id": "b7d1ba80-0934-11e9-0ef9-efa612d204a2",
    "type": "semi-manual",
    "color": "white"
   }
 ]
}

案例class是：

case class FooResponse(foos: Seq[Foo])
case class Foo(id: String, type: String, color: String)

我做的客户：

private val flow: Flow[HttpRequest, HttpResponse, Future[Http.OutgoingConnection]] = Http().outgoingConnection(host, port)

def getFoos(): Seq[Foo] = {
  val req = HttpRequest(method = HttpMethods.GET)
  .withUri(Uri("/api/foo")).withHeaders(headers.Accept(MediaRange(MediaTypes.`application/json`)))

  Source
   .single(req)
   .via(flow)
   .map(response => Unmarshal(response.entity).to[FooResponse])
}

因此，我有一个带有 Future[FooResponse] 的源。我怎么能从它 return Seq[Foo] 作为函数结果。

Answer 1

type 是关键字（类型别名），不应用作变量名
您需要在 Future 上调用 onComplete() 方法并等待结果。为此，我们可以在 getFoos() - var fs: Seq[Foo] = Nil 中定义一个占位符。对于未来，我们添加 onComplete(tfr => fs = tfr.get.foos) 然后等待一段时间：Await.result(f, Duration(5000, "millis"))（在本例中最多 5 秒）。最后我们可以 return 占位符中的内容。

这应该可以完成工作：

def getFoos(): Seq[Foo] = {
  var fs: Seq[Foo] = Nil
  val req = HttpRequest(method = HttpMethods.GET)
  .withUri(Uri("/api/foo")).withHeaders(headers.Accept(MediaRange(MediaTypes.`application/json`)))

  val f = Source
   .single(req)
   .via(flow)
   .map(response => Unmarshal(response.entity).to[FooResponse])
  f.onComplete(tfr => fs = tfr.get.foos)
  Await.result(f, Duration(5000, "millis"))
  fs
}

Answer 2

我建议将 getFoos 的 return 类型从 Seq[Foo] 更改为 Future[Seq[Foo]]，以便保持在 Future 的上下文中：

def getFoos(): Future[Seq[Foo]] = {
  val req =
    HttpRequest(method = HttpMethods.GET)
      .withUri(Uri("/api/foo"))
      .withHeaders(headers.Accept(MediaRange(MediaTypes.`application/json`)))

  Source
    .single(req)
    .via(flow)
    .map(response => Unmarshal(response.entity).to[FooResponse])
    .mapAsync(parallelism = 1)(fooResponse => fooResponse.map(_.foos))
    .runWith(Sink.head)
}

此外，由于 type 是 Scala 中的保留关键字，因此在 Foo 情况下需要将其用反引号括起来 class:

case class Foo(id: String, `type`: String, color: String)

从基于 json 数组的 akka http 响应中获取实体序列

Get Seq of entities from akka http response based on json array

json

scala

akka

akka-stream

akka-http