如何使用 Play 2.5 在分块响应中将 Anorm 大型查询结果流式传输到客户端

Question

我有一个非常大的结果集（60k+ 记录列），我从数据库中提取并使用 Anorm 进行解析（尽管我可以使用 play 的默认数据访问模块 returns 一个 ResultSet，如果需要的话）。我需要将这些结果直接转换并流式传输到客户端（无需将它们保存在内存中的大列表中），然后将它们直接下载到客户端机器上的文件中。

我一直在参考 ScalaStream 2.5.x Play 文档中 Chunked Responses 部分中演示的内容。我在执行那里显示的 "getDataStream" 部分时遇到问题。

我还参考了 ScalaAnorm 2.5.x Play 文档中 Streaming Results 和 Iteratee 部分中演示的内容。我已经尝试将结果作为枚举器进行管道传输，就像这里返回的一样：

 val resultsEnumerator = Iteratees.from(SQL"SELECT * FROM Test", SqlParser.str("colName"))

进入

val dataContent = Source.fromPublisher(Streams.enumeratorToPublisher(resultsEnumerator))
Ok.chunked(dataContent).withHeaders(("ContentType","application/x-download"),("Content-disposition","attachment; filename=myDataFile.csv"))

但是结果file/content是空的。

而且我找不到任何关于如何在数据服务中转换函数的示例代码或参考，returns 像这样：

@annotation.tailrec
def go(c: Option[Cursor], l: List[String]): List[String] = c match {
  case Some(cursor) => {
    if (l.size == 10000000) l // custom limit, partial processing
    else {
      go(cursor.next, l :+ cursor.row[String]("VBU_NUM"))
    }
  }
  case _ => l
}

val sqlString = s"select colName FROM ${tableName} WHERE ${whereClauseStr}"

val results : Either[List[Throwable], List[String]] = SQL(sqlString).withResult(go(_, List.empty[String]))
results

变成我可以传递给 Ok.chunked() 的东西。

所以基本上我的问题是，我应该如何将每个从数据库中获取的记录提供给一个流，我可以对其进行转换并作为可以下载到文件的分块响应发送给客户端？

我不想为此使用 Slick。但是我可以使用不使用 Anorm 的解决方案，只使用 returns 原始 java.sql.ResultSet 对象的 play dbApi 对象并使用它。

Answer 1

在参考了 Anorm Akka Support 文档并进行了大量试验和错误之后，我能够实现我想要的解决方案。我必须添加这些依赖项

"com.typesafe.play" % "anorm_2.11" % "2.5.2",
"com.typesafe.play" % "anorm-akka_2.11" % "2.5.2",
"com.typesafe.akka" %% "akka-stream" % "2.4.4"

通过 build.sbt Play 2.5 文件。

我实现了类似的东西

//...play imports
import anorm.SqlParser._
import anorm._

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{Sink, Source}

...

private implicit val akkaActorSystem = ActorSystem("MyAkkaActorSytem")
private implicit val materializer = ActorMaterializer()

def streamedAnormResultResponse() = Action {
  implicit val connection = db.getConnection()

  val parser: RowParser[...] = ...
  val sqlQuery: SqlQuery = SQL("SELECT * FROM table")

  val source: Source[Map[String, Any] = AkkaStream.source(sqlQuery, parser, ColumnAliaser.empty).alsoTo(Sink.onComplete({
    case Success(v) =>
      connection.close()
    case Failure(e) =>
      println("Info from the exception: " + e.getMessage)
      connection.close()
  }))

  Ok.chunked(source)
}

如何使用 Play 2.5 在分块响应中将 Anorm 大型查询结果流式传输到客户端

How to stream Anorm large query results to client in chunked response with Play 2.5

scala

anorm

akka-stream

playframework-2.5