播放播放2.5时在运行中创建的zip文件以及带有反压的akka​​流 [英] stream a zip created on the fly with play 2.5 and akka stream with backpressure

问题描述

我想使用带有反压的akka​​流，在播放框架2.5中流式传输动态创建的zip(而不是将其完全存储在内存中).这是我的代码，带有即时创建的小zip(16KB).当客户端下载与操作相关联的url时，下载不会开始.

I would like to stream a zip created on the fly (without putting it entirely in memory) with the play-framework 2.5 using akka stream with backpressure. Here my code, with a small zip created on the fly(16KB).When the client download the url associated with the action, the download does not start.

import java.util.zip.{ ZipEntry, ZipOutputStream, GZIPOutputStream }
import akka.stream.scaladsl._
import akka.util.ByteString
import play.api.mvc._
import scala.concurrent.duration._
import java.io.{ BufferedOutputStream, ByteArrayOutputStream }
import scala.concurrent.{ Promise, Future }
import akka.stream.OverflowStrategy
class ZipController extends Controller {

  def getStreamedZip = Action {
    val source: Source[ByteString, java.io.OutputStream] = StreamConverters.asOutputStream()
    val result = source.mapMaterializedValue(x => {
      val zip = new ZipOutputStream(x)
      (0 to 100).map { i =>
        zip.putNextEntry(new ZipEntry("test-zip/README-" + i + ".txt"))
        zip.write("This is the line:\n".map(_.toByte).toArray)
        zip.closeEntry()
      }
      zip.close
      x
    })
    Ok.chunked(result).withHeaders(
      "Content-Type" -> "application/zip",
      "Content-Disposition" -> "attachment; filename=test.zip"
    )
  }


}

基本上，我想在1 GB内存服务器上流式传输2GB的zip文件.并且此zip文件将包含大约15MB的文件.是否可以在不将每个文件都完全加载到内存的情况下编写zip文件?如果说3个客户端以1MB/秒的速度下载zip.这些下载大约需要多少内存?预先谢谢你.

Basically I want to stream a zip file of 2GB on a 1 GB memory server. And this zip will be composed of files of about 15MB. Is it possible to write the zip without loading entirely each file in memory ? If let say 3 clients download the zip at 1MB/second speed. Approximatively how much memory these downloads will take ? Thank you in advance.

推荐答案

此处是源自

Here is an implementation sourced at https://gist.github.com/kirked/412b5156f94419e71ce4a84ec1d54761

/* License: MIT */

import com.typesafe.scalalogging.slf4j.StrictLogging
import java.io.{ByteArrayOutputStream, InputStream, IOException}
import java.util.zip.{ZipEntry, ZipOutputStream}
import play.api.libs.iteratee.{Enumeratee, Enumerator}
import scala.concurrent.{Future, ExecutionContext}

/**
 * Play iteratee-based reactive zip-file generation.
 */
object ZipEnumerator extends StrictLogging {

  /**
   * A source to zip.
   * 
   * @param  filepath The zip-file path at which to store the data.
   * @param  stream   The data stream provider.
   */
  case class Source(filepath: String, stream: () => Future[Option[InputStream]])

  /**
   * Given sources, returns an Enumerator that feeds a zip-file of the source contents.
   */
  def apply(sources: Iterable[Source])(implicit ec: ExecutionContext): Enumerator[Array[Byte]] = {
    val resolveSources: Enumerator[ResolvedSource] = Enumerator.unfoldM(sources) { sources =>
      sources.headOption match {
        case None                                 => Future(None)
        case Some(Source(filepath, futureStream)) =>
          futureStream().map { _.map(stream => (sources.tail, ResolvedSource(filepath, stream)) ) }
      }
    }

    val buffer = new ZipBuffer(8192)

    val writeCentralDirectory = Enumerator.generateM(Future {
      if (buffer.isClosed) None
      else {
        buffer.close
        Some(buffer.bytes)
      }
    })

    resolveSources &> zipeach(buffer) andThen writeCentralDirectory
  }


  private def zipeach(buffer: ZipBuffer)(implicit ec: ExecutionContext): Enumeratee[ResolvedSource, Array[Byte]] = {
    Enumeratee.mapConcat[ResolvedSource] { source =>
      buffer.zipStream.putNextEntry(new ZipEntry(source.filepath))
      var done = false

      def entryDone: Unit = {
        done = true
        buffer.zipStream.closeEntry
        source.stream.close
      }

      def restOfStream: Stream[Array[Byte]] = {
        if (done) Stream.empty
        else {
          while (!done && !buffer.full) {
            try {
              val byte = source.stream.read
              if (byte == -1) entryDone
              else buffer.zipStream.write(byte)
            }
            catch {
              case e: IOException =>
                logger.error(s"reading/zipping stream [${source.filepath}]", e)
                entryDone
            }
          }
          buffer.bytes #:: restOfStream
        }
      }

      restOfStream
    }
  }


  private case class ResolvedSource(filepath: String, stream: InputStream)


  private class ZipBuffer(capacity: Int) {
    private val buf = new ByteArrayOutputStream(capacity)
    private var closed = false
    val zipStream = new ZipOutputStream(buf)

    def close(): Unit = {
      if (!closed) {
        closed = true
        reset
        zipStream.close   // writes central directory
      }
    }

    def isClosed = closed

    def reset: Unit = buf.reset

    def full: Boolean = buf.size >= capacity

    def bytes: Array[Byte] = {
      val result = buf.toByteArray
      reset
      result
    }
  }
}

用法看起来像这样:

val s3 = ...
val sources = items.map(item => ZipEnumerator.Source(item.filename, { () => s3.getInputStream(item.storagePath) }))
Ok.chunked(ZipEnumerator(sources))(play.api.http.Writeable.wBytes).withHeaders(
            CONTENT_TYPE -> "application/zip",
            CONTENT_DISPOSITION -> s"attachment; filename=MyFiles.zip; filename*=UTF-8''My%20Files.zip"
          )

这篇关于播放播放2.5时在运行中创建的zip文件以及带有反压的akka​​流的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！