You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pekko.apache.org by fa...@apache.org on 2026/09/22 09:46:19 UTC
(pekko-connectors) branch main updated: File: use int for the Java API zip compression level (#1873)
This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-connectors.git
The following commit(s) were added to refs/heads/main by this push:
new 8d6ef47de File: use int for the Java API zip compression level (#1873)
8d6ef47de is described below
commit 8d6ef47de323b7761f4ac9d81ec4ff7a5285830b
Author: PJ Fanning <pj...@users.noreply.github.com>
AuthorDate: Tue Sep 22 10:46:13 2026 +0100
File: use int for the Java API zip compression level (#1873)
Motivation:
Archive.zip(deflateCompression) added in #1829 exposes scala.Option[Int]
in the javadsl. A Java caller has to write
Archive.zip(Some$.MODULE$.apply(Deflater.NO_COMPRESSION))
with the boxing that implies. Every other javadsl in the repo uses
java.util.Optional or a plain value for this kind of parameter
(LogRotatorSink, s3/javadsl/S3, GoogleSettings).
The overload is new in 2.0.0 and has not appeared in a release, so the
signature can still be corrected.
Neither DSL validated the level either. An out-of-range value reached
ZipOutputStream.setLevel inside preStart, so the stream failed at
materialization with "invalid compression level" rather than at the call
site.
Modification:
The javadsl overload now takes a plain Int; the compression domain is
0-9 plus Deflater.DEFAULT_COMPRESSION, so Optional adds nothing over an
int. Both overloads delegate to a private zipFlow that keeps the
Option-based scaladsl call.
The scaladsl keeps Option[Int], which is idiomatic there, and gains a
require that rejects a level outside 0-9 and DEFAULT_COMPRESSION, so an
invalid value fails at the call site with a message naming the bound.
Result:
Java callers write Archive.zip(Deflater.NO_COMPRESSION). An invalid level
is rejected when the flow is constructed.
Tests:
Added two tests to the Java ArchiveTest: one builds a zip via
Archive.zip(Deflater.NO_COMPRESSION) and round-trips it through
ArchiveHelper.unzip, which is the call the old signature made awkward
from Java; the other asserts levels 10 and -2 are rejected. Added
matching scaladsl tests for the rejection and for
Deflater.DEFAULT_COMPRESSION being accepted. Java tests: 6 total, 0
failed. Scala: 26 total, 0 failed. Ran file/mimaReportBinaryIssues
(clean), the module scalafmt checks and javafmtCheckAll.
References:
Follow-up to #1829.
---
.../stream/connectors/file/javadsl/Archive.scala | 22 ++++++----
.../stream/connectors/file/scaladsl/Archive.scala | 19 ++++++--
file/src/test/java/docs/javadsl/ArchiveTest.java | 50 ++++++++++++++++++++++
.../src/test/scala/docs/scaladsl/ArchiveSpec.scala | 18 ++++++++
4 files changed, 98 insertions(+), 11 deletions(-)
diff --git a/file/src/main/scala/org/apache/pekko/stream/connectors/file/javadsl/Archive.scala b/file/src/main/scala/org/apache/pekko/stream/connectors/file/javadsl/Archive.scala
index 386cebe25..58e822241 100644
--- a/file/src/main/scala/org/apache/pekko/stream/connectors/file/javadsl/Archive.scala
+++ b/file/src/main/scala/org/apache/pekko/stream/connectors/file/javadsl/Archive.scala
@@ -32,23 +32,29 @@ 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.
+ *
+ * @param deflateCompression compression level, 0-9, where 0 is no compression and 9 is maximum compression.
+ * Pass `Deflater.DEFAULT_COMPRESSION` (-1) to use the default compression level of the underlying library.
* @since 2.0.0
*/
def zip(
+ deflateCompression: Int): Flow[Pair[ArchiveMetadata, Source[ByteString, NotUsed]], ByteString, NotUsed] =
+ zipFlow(Some(deflateCompression))
+
+ /**
+ * Flow for compressing multiple files into one ZIP file, using the default
+ * compression level of the underlying library.
+ */
+ def zip(): Flow[Pair[ArchiveMetadata, Source[ByteString, NotUsed]], ByteString, NotUsed] =
+ zipFlow(None)
+
+ private def zipFlow(
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(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.
*/
diff --git a/file/src/main/scala/org/apache/pekko/stream/connectors/file/scaladsl/Archive.scala b/file/src/main/scala/org/apache/pekko/stream/connectors/file/scaladsl/Archive.scala
index 01def3ca2..9fe5b8f19 100644
--- a/file/src/main/scala/org/apache/pekko/stream/connectors/file/scaladsl/Archive.scala
+++ b/file/src/main/scala/org/apache/pekko/stream/connectors/file/scaladsl/Archive.scala
@@ -22,6 +22,7 @@ import pekko.util.ByteString
import java.io.File
import java.nio.charset.{ Charset, StandardCharsets }
+import java.util.zip.Deflater
/**
* Scala API.
@@ -30,16 +31,28 @@ 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.
+ * `Deflater.DEFAULT_COMPRESSION` (-1) is also accepted. If not specified, the default compression
+ * level of the underlying library is used.
* @since 2.0.0
*/
def zip(
- deflateCompression: Option[Int]): Flow[(ArchiveMetadata, Source[ByteString, Any]), ByteString, NotUsed] =
+ deflateCompression: Option[Int]): Flow[(ArchiveMetadata, Source[ByteString, Any]), ByteString, NotUsed] = {
+ deflateCompression.foreach { level =>
+ require(
+ level == Deflater.DEFAULT_COMPRESSION ||
+ (level >= Deflater.NO_COMPRESSION &&
+ level <= Deflater.BEST_COMPRESSION),
+ s"deflateCompression must be between ${Deflater.NO_COMPRESSION} and ${Deflater.BEST_COMPRESSION}, " +
+ s"or ${Deflater.DEFAULT_COMPRESSION} for the default, was $level")
+ }
ZipArchiveManager.zipFlow(deflateCompression)
+ }
/**
- * Flow for compressing multiple files into one ZIP file.
+ * Flow for compressing multiple files into one ZIP file, using the default
+ * compression level of the underlying library.
*/
def zip(): Flow[(ArchiveMetadata, Source[ByteString, Any]), ByteString, NotUsed] =
zip(None)
diff --git a/file/src/test/java/docs/javadsl/ArchiveTest.java b/file/src/test/java/docs/javadsl/ArchiveTest.java
index c57416f4b..3c3c07cb9 100644
--- a/file/src/test/java/docs/javadsl/ArchiveTest.java
+++ b/file/src/test/java/docs/javadsl/ArchiveTest.java
@@ -16,6 +16,7 @@ package docs.javadsl;
import static org.apache.pekko.util.ByteString.emptyByteString;
import static org.hamcrest.CoreMatchers.*;
import static org.hamcrest.MatcherAssert.*;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import java.io.File;
import java.net.URISyntaxException;
@@ -28,6 +29,7 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.TimeUnit;
+import java.util.zip.Deflater;
import org.apache.pekko.Done;
import org.apache.pekko.NotUsed;
import org.apache.pekko.actor.ActorSystem;
@@ -137,6 +139,54 @@ public class ArchiveTest {
new File("logo.zip").delete();
}
+ @Test
+ public void flowShouldCreateZIPArchiveWithCompressionLevel() throws Exception {
+ ByteString fileContent1 = readFileAsByteString(getFileFromResource("pekko.svg"));
+ ByteString fileContent2 = readFileAsByteString(getFileFromResource("foundation.svg"));
+
+ Source<ByteString, NotUsed> source1 = toSource(fileContent1);
+ Source<ByteString, NotUsed> source2 = toSource(fileContent2);
+
+ Pair<ArchiveMetadata, Source<ByteString, NotUsed>> pair1 =
+ Pair.create(ArchiveMetadata.create("pekko.svg"), source1);
+ Pair<ArchiveMetadata, Source<ByteString, NotUsed>> pair2 =
+ Pair.create(ArchiveMetadata.create("foundation.svg"), source2);
+
+ Source<Pair<ArchiveMetadata, Source<ByteString, NotUsed>>, NotUsed> source =
+ Source.from(List.of(pair1, pair2));
+
+ // the compression level is a plain int, so it is callable from Java
+ // without wrapping it in a scala.Option
+ Sink<ByteString, CompletionStage<IOResult>> fileSink =
+ FileIO.toPath(Paths.get("logo-stored.zip"));
+ CompletionStage<IOResult> ioResult =
+ source.via(Archive.zip(Deflater.NO_COMPRESSION)).runWith(fileSink, system);
+
+ ioResult.toCompletableFuture().get(3, TimeUnit.SECONDS);
+
+ Map<String, ByteString> inputFiles =
+ new HashMap<String, ByteString>() {
+ {
+ put("pekko.svg", fileContent1);
+ put("foundation.svg", fileContent2);
+ }
+ };
+
+ ByteString resultFileContent = readFileAsByteString(Paths.get("logo-stored.zip"));
+ Map<String, ByteString> unzip = archiveHelper.unzip(resultFileContent);
+
+ assertThat(inputFiles, is(unzip));
+
+ // cleanup
+ new File("logo-stored.zip").delete();
+ }
+
+ @Test
+ public void zipShouldRejectInvalidCompressionLevel() {
+ assertThrows(IllegalArgumentException.class, () -> Archive.zip(10));
+ assertThrows(IllegalArgumentException.class, () -> Archive.zip(-2));
+ }
+
@Test
public void flowShouldCreateTARArchive() throws Exception {
Path filePath1 = getFileFromResource("pekko.svg");
diff --git a/file/src/test/scala/docs/scaladsl/ArchiveSpec.scala b/file/src/test/scala/docs/scaladsl/ArchiveSpec.scala
index 0383c76c8..4fe768b62 100644
--- a/file/src/test/scala/docs/scaladsl/ArchiveSpec.scala
+++ b/file/src/test/scala/docs/scaladsl/ArchiveSpec.scala
@@ -135,6 +135,24 @@ class ArchiveSpec
archiveHelper.unzip(pekkoZipped.futureValue).asScala shouldBe inputFiles
}
+ "reject an out-of-range compression level" in {
+ an[IllegalArgumentException] should be thrownBy Archive.zip(Some(10))
+ an[IllegalArgumentException] should be thrownBy Archive.zip(Some(-2))
+ }
+
+ "accept the Deflater default compression level" in {
+ val inputFiles = generateInputFiles(5, 100)
+ val inputStream = filesToStream(inputFiles)
+ val zipFlow = Archive.zip(Some(Deflater.DEFAULT_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)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org