使用 Akka Http 转换 Slick Streaming 数据并发送分块响应

Question

目的是从数据库中流式传输数据，对这块数据执行一些计算（此计算返回某个案例类的 Future）并将这些数据作为分块响应发送给用户。目前，我能够在不执行任何计算的情况下流式传输数据并发送响应。但是，我无法执行此计算然后流式传输结果。

这是我实现的路线。

def streamingDB1 =
path("streaming-db1") {
  get {
    val src = Source.fromPublisher(db.stream(getRds))
    complete(src)
  }
}

函数 getRds 返回映射到案例类的表的行（使用 slick）。现在考虑将每一行作为输入并返回另一个案例类的 Future 的函数 compute。像

def compute(x: Tweet) : Future[TweetNew] = ?

如何在变量 src 上实现这个函数，并将这个计算的分块响应（作为流）发送给用户。

Answer 1

您可以使用mapAsync 转换源代码：

val src =
  Source.fromPublisher(db.stream(getRds))
        .mapAsync(parallelism = 3)(compute)

complete(src)

根据需要调整并行度。

---

请注意，您可能需要配置Slick documentation 中提到的一些设置：

注意：某些数据库系统可能需要以某种方式设置会话参数以支持流式传输，而无需在客户端的内存中一次缓存所有数据。例如，PostgreSQL 需要.withStatementParameters(rsType = ResultSetType.ForwardOnly, rsConcurrency = ResultSetConcurrency.ReadOnly, fetchSize = n)（具有所需的页面大小n）和.transactionally 才能进行正确的流式传输。

因此，例如，如果您使用的是 PostgreSQL，那么您的 Source 可能如下所示：

val src =
  Source.fromPublisher(
    db.stream(
      getRds.withStatementParameters(
        rsType = ResultSetType.ForwardOnly,
        rsConcurrency = ResultSetConcurrency.ReadOnly,
        fetchSize = 10
      ).transactionally
    )
  ).mapAsync(parallelism = 3)(compute)

Answer 2

您需要有一种方法来编组 TweetNew，并且如果您发送长度为 0 的块，客户端可能会关闭连接。

此代码适用于 curl：

case class TweetNew(str: String)

def compute(string: String) : Future[TweetNew] = Future {
  TweetNew(string)
}

val route = path("hello") {
  get {
    val byteString: Source[ByteString, NotUsed] = Source.apply(List("t1", "t2", "t3"))
      .mapAsync(2)(compute)
      .map(tweet => ByteString(tweet.str + "\n"))
    complete(HttpEntity(ContentTypes.`text/plain(UTF-8)`, byteString))
  }
}