file: add zip compression level

file/src/main/scala/org/apache/pekko/stream/connectors/file/impl/archive/FileByteStringSeparators.scala

@@ -35,10 +35,10 @@ import pekko.util.ByteString
     ByteString(endFileWord)
 
   def isStartingByteString(b: ByteString): Boolean =
-    b.utf8String.startsWith(startFileWord)
+    b.size >= 7 && b.slice(0, 7).utf8String == startFileWord
 
   def isEndingByteString(b: ByteString): Boolean =
-    b.utf8String == endFileWord
+    b.size == 5 && b.utf8String == endFileWord
 
   def getPathFromStartingByteString(b: ByteString): String = {
     val splitted = b.utf8String.split(separator)

file/src/main/scala/org/apache/pekko/stream/connectors/file/impl/archive/ZipArchiveFlow.scala

@@ -26,13 +26,18 @@ import pekko.util.{ ByteString, ByteStringBuilder }
  * INTERNAL API
  */
 @InternalApi private[file] final class ZipArchiveFlowStage(
-    val shape: FlowShape[ByteString, ByteString]) extends GraphStageLogic(shape) {
+    val shape: FlowShape[ByteString, ByteString],
+    deflateCompression: Option[Int] = None) extends GraphStageLogic(shape) {
 
   private def in = shape.in
   private def out = shape.out
 
   private val builder = new ByteStringBuilder()
   private val zip = new ZipOutputStream(builder.asOutputStream)
+
+  override def preStart(): Unit =
+    deflateCompression.foreach(l => zip.setLevel(l))
+
   private var emptyStream = true
 
   setHandler(
@@ -88,7 +93,8 @@ import pekko.util.{ ByteString, ByteStringBuilder }
 /**
  * INTERNAL API
  */
-@InternalApi private[file] final class ZipArchiveFlow extends GraphStage[FlowShape[ByteString, ByteString]] {
+@InternalApi private[file] final class ZipArchiveFlow(deflateCompression: Option[Int] = None)
+    extends GraphStage[FlowShape[ByteString, ByteString]] {
 
   val in: Inlet[ByteString] = Inlet(Logging.simpleName(this) + ".in")
   val out: Outlet[ByteString] = Outlet(Logging.simpleName(this) + ".out")
@@ -99,5 +105,5 @@ import pekko.util.{ ByteString, ByteStringBuilder }
   override val shape: FlowShape[ByteString, ByteString] = FlowShape(in, out)
 
   override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
-    new ZipArchiveFlowStage(shape)
+    new ZipArchiveFlowStage(shape, deflateCompression)
 }

file/src/main/scala/org/apache/pekko/stream/connectors/file/impl/archive/ZipArchiveManager.scala

@@ -25,8 +25,9 @@ import pekko.util.ByteString
  */
 @InternalApi private[file] object ZipArchiveManager {
 
-  def zipFlow(): Flow[(ArchiveMetadata, Source[ByteString, Any]), ByteString, NotUsed] = {
-    val archiveZipFlow = new ZipArchiveFlow()
+  def zipFlow(
+      deflateCompression: Option[Int] = None): Flow[(ArchiveMetadata, Source[ByteString, Any]), ByteString, NotUsed] = {
+    val archiveZipFlow = new ZipArchiveFlow(deflateCompression)
     Flow[(ArchiveMetadata, Source[ByteString, Any])]
       .flatMapConcat {
         case (metadata, stream) =>

file/src/main/scala/org/apache/pekko/stream/connectors/file/javadsl/Archive.scala

@@ -32,12 +32,22 @@ object Archive {
 
   /**
    * Flow for compressing multiple files into one ZIP file.
+   * @param deflateCompression optional compression level, 0-9, where 0 is no compression and 9 is maximum compression.
+   * If not specified, the default compression level of the underlying library will be used.
+   * @since 2.0.0
    */
-  def zip(): Flow[Pair[ArchiveMetadata, Source[ByteString, NotUsed]], ByteString, NotUsed] =
+  def zip(
+      deflateCompression: Option[Int]): Flow[Pair[ArchiveMetadata, Source[ByteString, NotUsed]], ByteString, NotUsed] =
     Flow
       .create[Pair[ArchiveMetadata, Source[ByteString, NotUsed]]]()
       .map(func(pair => (pair.first, pair.second.asScala)))
-      .via(scaladsl.Archive.zip().asJava)
+      .via(scaladsl.Archive.zip(deflateCompression).asJava)
+
+  /**
+   * Flow for compressing multiple files into one ZIP file.
+   */
+  def zip(): Flow[Pair[ArchiveMetadata, Source[ByteString, NotUsed]], ByteString, NotUsed] =
+    zip(None)
 
   /**
    * Flow for reading ZIP files.

file/src/main/scala/org/apache/pekko/stream/connectors/file/scaladsl/Archive.scala

@@ -28,11 +28,21 @@ import java.nio.charset.{ Charset, StandardCharsets }
  */
 object Archive {
 
+  /**
+   * Flow for compressing multiple files into one ZIP file.
+   * @param deflateCompression optional compression level, 0-9, where 0 is no compression and 9 is maximum compression.
+   * If not specified, the default compression level of the underlying library will be used.
+   * @since 2.0.0
+   */
+  def zip(
+      deflateCompression: Option[Int]): Flow[(ArchiveMetadata, Source[ByteString, Any]), ByteString, NotUsed] =
+    ZipArchiveManager.zipFlow(deflateCompression)
+
   /**
    * Flow for compressing multiple files into one ZIP file.
    */
   def zip(): Flow[(ArchiveMetadata, Source[ByteString, Any]), ByteString, NotUsed] =
-    ZipArchiveManager.zipFlow()
+    zip(None)
 
   /**
    * Flow for reading ZIP files.

file/src/test/scala/docs/scaladsl/ArchiveSpec.scala

@@ -15,6 +15,7 @@ package docs.scaladsl
 
 import java.io._
 import java.nio.file.{ Files, Path, Paths }
+import java.util.zip.Deflater
 import org.apache.pekko
 import pekko.actor.ActorSystem
 import pekko.stream.connectors.file.ArchiveMetadata
@@ -121,6 +122,19 @@ class ArchiveSpec
         archiveHelper.unzip(pekkoZipped.futureValue).asScala shouldBe inputFiles
       }
 
+      "archive files with compression flag" in {
+        val inputFiles = generateInputFiles(5, 100)
+        val inputStream = filesToStream(inputFiles)
+        val zipFlow = Archive.zip(Some(Deflater.NO_COMPRESSION))
+
+        val pekkoZipped: Future[ByteString] =
+          inputStream
+            .via(zipFlow)
+            .runWith(Sink.fold(ByteString.empty)(_ ++ _))
+
+        archiveHelper.unzip(pekkoZipped.futureValue).asScala shouldBe inputFiles
+      }
+
       "unarchive files" in {
         val inputFiles = generateInputFiles(5, 100)
         val inputStream = filesToStream(inputFiles)

file/src/test/scala/org/apache/pekko/stream/connectors/file/impl/archive/ZipArchiveFlowTest.scala

@@ -20,6 +20,7 @@ import pekko.stream.scaladsl.Keep
 import pekko.stream.testkit.scaladsl.{ TestSink, TestSource }
 import pekko.testkit.TestKit
 import pekko.util.ByteString
+import java.util.zip.Deflater
 import org.scalatest.BeforeAndAfterAll
 import org.scalatest.wordspec.AnyWordSpecLike
 
@@ -50,6 +51,27 @@ class ZipArchiveFlowTest
         downstream.expectComplete()
       }
     }
+
+    "compression flag given and stream ends" should {
+      "emit element only when downstream requests" in {
+        val (upstream, downstream) =
+          TestSource[ByteString]()
+            .via(new ZipArchiveFlow(Some(Deflater.NO_COMPRESSION)))
+            .toMat(TestSink())(Keep.both)
+            .run()
+
+        upstream.sendNext(FileByteStringSeparators.createStartingByteString("test"))
+        upstream.sendNext(ByteString(1))
+        upstream.sendNext(FileByteStringSeparators.createEndingByteString())
+        upstream.sendComplete()
+
+        downstream.request(2)
+        downstream.expectNextN(2)
+        downstream.request(1)
+        downstream.expectNextN(1)
+        downstream.expectComplete()
+      }
+    }
   }
 
   override def afterAll(): Unit = {