scala - Play 2.x : Reactive file upload with Iteratees

Question

我将从问题开始:How to use Scala API's Iteratee将文件上传到云存储(在我的情况下是 Azure Blob 存储，但我认为它现在不是最重要的)

背景:

我需要将输入分成大约 1 MB 的块，以便将大型媒体文件 (300 MB+) 存储为 Azure 的 BlockBlobs .不幸的是，我的 Scala 知识仍然很差(我的项目是基于 Java 的，Scala 在其中的唯一用途是上传 Controller )。

我试过这个代码:Why makes calling error or done in a BodyParser's Iteratee the request hang in Play Framework 2.0? (作为 Input Iteratee ) - 它工作得很好，但每个 Element我可以使用的大小为 8192 字节，因此对于将数百兆字节的文件发送到云来说太小了。

我必须说这对我来说是一种全新的方法，很可能我误解了一些东西(不想说我误解了一切；>)

我将感谢任何提示或链接，这将帮助我解决该主题。如果有任何类似用法的样本，这将是我了解这个想法的最佳选择。

Answer 1

基本上，您首先需要将输入重新分块为更大的块，1024 * 1024 字节。

首先让我们有一个 Iteratee这将消耗多达 1m 的字节(可以将最后一个块变小)

val consumeAMB = 
  Traversable.takeUpTo[Array[Byte]](1024*1024) &>> Iteratee.consume()

使用它，我们可以构造一个 Enumeratee (适配器)将使用名为 grouped 的 API 重新组合块:

val rechunkAdapter:Enumeratee[Array[Byte],Array[Byte]] =
  Enumeratee.grouped(consumeAMB)

这里分组使用 Iteratee确定在每个块中放入多少。为此，它使用了我们的消耗 AMB。这意味着结果是 Enumeratee将输入重新分块到 Array[Byte] 1MB。

现在我们需要写 BodyParser ，这将使用 Iteratee.foldM发送每个字节块的方法:

val writeToStore: Iteratee[Array[Byte],_] =
  Iteratee.foldM[Array[Byte],_](connectionHandle){ (c,bytes) => 
    // write bytes and return next handle, probable in a Future
  }

foldM 传递一个状态并在其传递的函数中使用它 (S,Input[Array[Byte]]) => Future[S]返回一个新的 future 状态。 foldM 不会再次调用该函数，直到 Future已完成并且有一个可用的输入块。

正文解析器将重新分块输入并将其推送到存储中:

BodyParser( rh => (rechunkAdapter &>> writeToStore).map(Right(_)))

Returning a Right 表示您在正文解析结束时返回正文(恰好是此处的处理程序)。