播放2.x:使用Iteratees上传活动文件

Question

我将从问题开始:如何使用Scala API Iteratee将文件上传到云存储(在我的情况下是Azure Blob存储,但我认为它现在不是最重要的)

背景:

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

我尝试使用这段代码:为什么调用错误或在BodyParser的Iteratee中完成请求在Play Framework 2.0中挂起？(作为Input Iteratee) - 它工作得很好,但Element我可以使用的每个大小为8192字节,因此它太小,无法向云端发送一百兆字节的文件.

我必须说这对我来说是一种全新的方法,而且很可能是我误解了一些东西(不想告诉我,我误解了一切;>)

我会感谢任何提示或链接,这将有助于我这个主题.如果有任何相似用途的样本,那么对我来说这是最好的选择.

Answer 1

基本上你首先需要的是rechunk输入作为更大的块,1024*1024字节.

首先让我们有一个Iteratee消耗高达1米的字节(确定最后一个块更小)

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

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

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

这里分组使用a Iteratee来确定每个块中放入多少.它使用我们的consumeAMB.这意味着结果是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(_)))

返回一个右表示你在正文解析结束时返回一个正文(这恰好是这里的处理程序).