Play 2.x: реактивная загрузка файлов с помощью Iteratees

По сути, сначала вам нужно перераспределить ввод как более крупные куски, 1024 * 1024 байта.

Сначала давайте создадим Iteratee, который будет потреблять до 1 млн байт (хорошо, чтобы последний фрагмент был меньше)

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

Используя это, мы можем создать Enumeratee (адаптер), который будет перегруппировывать куски, используя API под названием grouped:

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

Здесь grouped использует Iteratee, чтобы определить, сколько помещать в каждый фрагмент. Для этого он использует наш потребляемый AMB. Это означает, что результатом является Enumeratee, который повторно разбивает входные данные на Array[Byte] размером 1 МБ.

Теперь нам нужно написать 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] для возврата нового будущего состояния. foldM не будет вызывать функцию снова до тех пор, пока Future не будет завершено и не появится доступный кусок ввода.

И парсер тела будет перераспределять входные данные и помещать их в хранилище:

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

Возврат права указывает, что вы возвращаете тело к концу синтаксического анализа тела (который здесь является обработчиком).

Если ваша цель — стримить на S3, вот помощник, который я реализовал и протестировал:

def uploadStream(bucket: String, key: String, enum: Enumerator[Array[Byte]])
                (implicit ec: ExecutionContext): Future[CompleteMultipartUploadResult] = {
  import scala.collection.JavaConversions._

  val initRequest = new InitiateMultipartUploadRequest(bucket, key)
  val initResponse = s3.initiateMultipartUpload(initRequest)
  val uploadId = initResponse.getUploadId

  val rechunker: Enumeratee[Array[Byte], Array[Byte]] = Enumeratee.grouped {
    Traversable.takeUpTo[Array[Byte]](5 * 1024 * 1024) &>> Iteratee.consume()
  }

  val uploader = Iteratee.foldM[Array[Byte], Seq[PartETag]](Seq.empty) { case (etags, bytes) =>
    val uploadRequest = new UploadPartRequest()
      .withBucketName(bucket)
      .withKey(key)
      .withPartNumber(etags.length + 1)
      .withUploadId(uploadId)
      .withInputStream(new ByteArrayInputStream(bytes))
      .withPartSize(bytes.length)

    val etag = Future { s3.uploadPart(uploadRequest).getPartETag }
    etag.map(etags :+ _)
  }

  val futETags = enum &> rechunker |>>> uploader

  futETags.map { etags =>
    val compRequest = new CompleteMultipartUploadRequest(bucket, key, uploadId, etags.toBuffer[PartETag])
    s3.completeMultipartUpload(compRequest)
  }.recoverWith { case e: Exception =>
    s3.abortMultipartUpload(new AbortMultipartUploadRequest(bucket, key, uploadId))
    Future.failed(e)
  }

}

добавьте следующее в ваш файл конфигурации

play.http.parser.maxMemoryBuffer=256K

Для тех, кто также пытается найти решение этой проблемы с потоковой передачей, вместо того, чтобы писать полностью новый BodyParser, вы также можете использовать то, что уже реализовано в parse.multipartFormData. Вы можете реализовать что-то вроде ниже, чтобы перезаписать обработчик по умолчанию handleFilePartAsTemporaryFile.

def handleFilePartAsS3FileUpload: PartHandler[FilePart[String]] = {
  handleFilePart {
    case FileInfo(partName, filename, contentType) =>

      (rechunkAdapter &>> writeToS3).map {
        _ =>
          val compRequest = new CompleteMultipartUploadRequest(...)
          amazonS3Client.completeMultipartUpload(compRequest)
          ...
      }
  }
}

def multipartFormDataS3: BodyParser[MultipartFormData[String]] = multipartFormData(handleFilePartAsS3FileUpload)

Я могу сделать эту работу, но я все еще не уверен, передается ли весь процесс загрузки. Я попробовал несколько больших файлов, кажется, что загрузка S3 начинается только тогда, когда весь файл был отправлен со стороны клиента.

Я посмотрел на приведенную выше реализацию парсера и думаю, что все подключено с помощью Iteratee, поэтому файл должен быть потоковым. Если у кого-то есть понимание этого, это будет очень полезно.