使用分块传输编码从 scala Play 服务器流式传输案例类对象答案

【问题标题】：Streaming case class objects from a scala Play server using chunked transfer encoding使用分块传输编码从 scala Play 服务器流式传输案例类对象
【发布时间】：2020-11-23 21:56:39
【问题描述】：

所以，我使用 Play framework 2.7 来设置流媒体服务器。我要做的是流式传输大约 500 个大小相似的自定义案例类对象。

这是生成流的控制器的一部分 -

def generate: Action[AnyContent] = Action {
    val products = (1 to 500).map(Product(_, "some random string")).toList
    Ok.chunked[Product](Source(products))
  }

Product 是我正在使用的自定义案例类。隐式 writable 将此对象反序列化为 json。

这是处理此流的控制器的一部分 -

def process(): Action[AnyContent] = Action.async {
    val request = ws.url(STREAMING_URL).withRequestTimeout(Duration.Inf).withMethod("GET")
    request.stream().flatMap {
      _.bodyAsSource
        .map(_.utf8String)
        .map { x => println(x); x }
        .fold(0) { (acc, _) => acc + 1 }
        .runWith(Sink.last)
        .andThen {
          case Success(v) => println(s"Total count - $v")
          case Failure(_) => println("Error encountered")
        }
    }.map(_ => Ok)
  }

我所期望的是，我的案例类的每个对象都作为单个块传输并同样接收，以便它们可以单独序列化并由接收器使用。这意味着，使用上面的代码，我的期望是我应该收到正好 500 个块，但这个值总是超过这个值。

我可以看到的是，这 500 个对象中恰好有一个是拆分传输的，并且是分 2 个块而不是 1 个块传输的。

这是一个正常的对象，如接收端所见 -

{
  "id" : 494,
  "name" : "some random string"
}

这是一个一分为二的对象-

{
  "id" : 463,
  "name" : "some random strin
g"
}

因此，这不能序列化回我的 Product 案例类的实例。

但是，如果我在发送者控制器中对源进行了某种限制，我会按预期收到块。

例如，当我每秒只流式传输 5 个元素时，这完全可以正常工作 -

def generate: Action[AnyContent] = Action {
    val products = (1 to 500).map(Product(_, "some random string")).toList
    Ok.chunked[Product](Source(products).throttle(5, 1.second))
  }

谁能帮我理解为什么会这样？

【问题讨论】：

只是猜测，但也许没有限制发送方生成元素的速度比发送它们的速度要快，因此它将块打包到极限，并分割一些字符串。通过节流，它可能有足够的时间在每个块中放置一个元素。
此信息可能解释：Another pitfall is that Actor messages can be lost and must be retransmitted in that case. Failure to do so would lead to holes at the receiving side. from doc.akka.io/docs/akka/current/stream/…。我想你将不得不处理背压。
@Felipe 我不认为这是流的问题 - 演员就是这种情况，并且流建立在演员之上以避免这些陷阱。
你确定这不仅仅是println的并发问题吗？您可以尝试改用日志记录，看看它是否还能重现？
不是，因为我也在计算接收方的块数。它们超出了预期。

标签： playframework akka akka-stream akka-http

【解决方案1】：

正如here 所述，有一个JsonFraming 用于将有效的JSON 对象与传入的ByteString 流分开。

在你的情况下，你可以尝试这种方式

  _.bodyAsSource.via(JsonFraming.objectScanner(Int.MaxValue)).map(_.utf8String)

【讨论】：