Stream a Play's Enumerator to S3

ReactiveUploadS3.scala

    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)
      }

    }