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.git
The following commit(s) were added to refs/heads/main by this push:
new b88cae3d47 Add zstd compression and decompression for pekko streams
(heap buffer APIs) (#3639)
b88cae3d47 is described below
commit b88cae3d47c8457e8bc49558e52d7be4f120ca9f
Author: PJ Fanning <[email protected]>
AuthorDate: Fri Oct 23 21:34:10 2026 +0100
Add zstd compression and decompression for pekko streams (heap buffer APIs)
(#3639)
* Add zstd compression and decompression for pekko streams
* Use zstd-jni heap buffer streaming APIs for zstd compression
Motivation:
zstd-jni 1.5.7-23 added ZstdCompressCtx.compressByteBufferStream and
ZstdDecompressCtx.decompressByteBufferStream, which accept array-backed
heap ByteBuffers (luben/zstd-jni#324). The zstd stages no longer need to
copy every element into fixed size direct buffers.
Modification:
- Upgrade zstd-jni to 1.5.7-24.
- Rewrite ZstdCompressor on ZstdCompressCtx and ZstdDecompressor on
ZstdDecompressCtx using heap buffers.
- Add ZstdBuffers to feed ByteString fragments to zstd and to hand
output arrays to ByteStrings without copying where possible.
- Fail zstdDecompress on truncated input, emit chunks of at most
maxBytesPerChunk and pass the dictionary offset, length and level to
ZstdDictCompress in the right order.
- Revert the ByteBufferCleaner visibility change, it is no longer used.
Result:
zstd compression and decompression avoid direct buffers and most array
copies, and detect truncated streams.
Tests:
- sbt "stream-tests/testOnly org.apache.pekko.stream.io.compression.*
org.apache.pekko.stream.scaladsl.CompressionSpec": passed
- sbt "stream/mimaReportBinaryIssues": passed
References:
Refs #2404, Refs #2409, Refs luben/zstd-jni#324
* Add ByteString.asByteBuffersUnsafe for zero-copy heap buffers
Motivation:
zstd-jni needs array-backed buffers, but asByteBuffers returns read-only
buffers with no accessible array and code outside ByteString can only get
at a sliced fragment's bytes via toArrayUnsafe, which copies.
Modification:
Add private[pekko] ByteString.asByteBuffersUnsafe, which wraps each
non-empty fragment's backing array (with its start index and length)
without copying, and use it for the zstd input buffers.
Result:
zstd compression and decompression no longer copy sliced or fragmented
input ByteStrings.
Tests:
- sbt "actor-tests/testOnly org.apache.pekko.util.ByteString*": passed
- sbt "stream-tests/testOnly org.apache.pekko.stream.io.compression.*
org.apache.pekko.stream.scaladsl.CompressionSpec": passed
- sbt "actor/mimaReportBinaryIssues" "stream/mimaReportBinaryIssues": passed
References:
Refs #2409, Refs luben/zstd-jni#324
* Make the zstd-jni dependency of pekko-stream optional
Motivation:
zstd-jni is a 6.5 MB jar of native libraries that most pekko-stream
users do not need. The Compression objects called zstd-jni during their
initialization, so gzip and deflate users loaded the native library and
would fail without zstd-jni on the classpath.
Modification:
- Declare zstd-jni as Optional for pekko-stream (and as an optional
OSGi import), add it to the stream-tests test classpath.
- Replace the zstd-jni calls for the zstd constants in the scaladsl and
javadsl Compression objects with libzstd's values, documented in code.
- Document the dependency in the zstd operator docs.
Result:
pekko-stream does not pull in zstd-jni, and the other compression
operators work without it.
Tests:
- sbt "stream-tests/testOnly org.apache.pekko.stream.io.compression.*
org.apache.pekko.stream.scaladsl.CompressionSpec": passed,
including a gzip/deflate round trip in a class loader without zstd-jni
and a check that the constants match zstd-jni
- sbt "stream/mimaReportBinaryIssues": passed
- sbt "stream/makePom" "stream/packageBin": zstd-jni is optional in the
POM and the OSGi Import-Package
References:
Refs #2409, Refs luben/zstd-jni#324
* Add JMH benchmark for gzip and zstd compression flows
Motivation:
There were no benchmarks for the stream compression operators.
Modification:
Add CompressionBenchmark comparing gzip and zstd compression and
decompression of 16 MiB of JSON-like text in 1 KiB and 64 KiB elements,
and add zstd-jni to bench-jmh (it is optional in pekko-stream).
Result:
The zstd operators can be benchmarked against gzip.
Tests:
- sbt "bench-jmh/Jmh/run -wi 3 -w 2s -i 5 -r 2s -f 1
.*CompressionBenchmark.*": ran
References:
Refs #2409
* Only hand over nearly full zstd output buffers without copying
Motivation:
ZstdBuffers.takeOutput handed the output array over to the emitted
ByteString once it was half full, so e.g. 64 KiB decompressed chunks kept
a 128 KiB array alive and every chunk allocated a new 128 KiB array.
Modification:
Hand the array over only when at least 7/8 of it is used, otherwise copy
the output into an exact-size array and reuse the buffer.
Result:
Emitted ByteStrings retain at most 1/8 unused space.
Tests:
- sbt "stream-tests/testOnly org.apache.pekko.stream.io.compression.*
org.apache.pekko.stream.scaladsl.CompressionSpec": passed
References:
Refs #2409
* Remove the zstd-jni default from ZstdDecompressor
Motivation:
ZstdDecompressor's maxBytesPerChunk defaulted to Zstd.blockSizeMax(), the
only
zstd default not taken from the documented Compression constants. The gzip
and
deflate decompressor stages have no default, and every caller passes the
value.
Modification:
Make maxBytesPerChunk a required parameter of ZstdDecompressor.
Result:
ZstdDecompressor matches the other decompressor stages, and callers pick the
chunk size explicitly (Compression.zstdDecompress already does).
Tests:
sbt "stream/scalafmt" "stream-tests/testOnly
org.apache.pekko.stream.io.compression.*"
(145 succeeded, 0 failed)
References:
Review comment
https://github.com/apache/pekko/pull/3639#issuecomment-6347243585
* Keep the zstd-free class loader open in CompressionWithoutZstdSpec
Closing the URLClassLoader right after the nested ActorSystem terminated
let a late dispatcher shutdown task fail with NoClassDefFoundError, which
jvm-exit-on-fatal-error turned into an exit of the forked stream-tests JVM.
Leave the loader open and disable jvm-exit-on-fatal-error for that system.
* Tidy zstd test idioms
Motivation:
Review nits: an[ClassNotFoundException] should read
a[ClassNotFoundException],
and charset names should not be passed as strings.
Modification:
Use a[ClassNotFoundException] in CompressionWithoutZstdSpec and
StandardCharsets.UTF_8 in ZstdSpec.
Result:
No behaviour change.
Tests:
sbt "stream-tests/testOnly *ZstdSpec *CompressionWithoutZstdSpec" (24
passed)
References:
None
* Fix zstd decompression of chunk-aligned output and address review
Motivation:
ZstdDecompressor failed valid streams whose decompressed size is an exact
multiple of maxBytesPerChunk. When the output buffer filled exactly at the
end
of a frame, outputPending stayed true and one more decompress call was made.
zstd then waited for a new frame and the stage failed with "Truncated zstd
stream" after all data had been emitted. The default chunk is 128 KiB, so
any
128 KiB-aligned payload was affected. The review also found an incomplete
operator index, inaccurate operator docs, String dictionaries encoded with
the
platform default charset, and a Java source under src/main/scala.
Modification:
- Only treat output as pending while the frame is incomplete: a completed
frame
is fully flushed.
- Add the zstd operators to the generated stream operator index.
- Document the zstd overloads and autoFlush, place Dependency before
Signature,
note that zstdDecompress does not support dictionaries, and fix the
zstdDecompress error description.
- Encode String dictionaries as UTF-8 in the scaladsl and javadsl
ZstdDictionary.
- Move javadsl ZstdDictionary.java to src/main/java and make its field
private.
- Use imports instead of fully qualified names in new code.
Result:
Chunk-aligned and concatenated zstd streams decompress correctly, and String
dictionaries give the same bytes on every node.
Tests:
- New ZstdSpec test for chunk-aligned single and concatenated frames (fails
without the fix with "Truncated zstd stream").
- New ZstdSpec test for UTF-8 String dictionaries.
- sbt "actor-tests/testOnly org.apache.pekko.util.ByteStringSpec" (246
passed)
- sbt "stream-tests/testOnly org.apache.pekko.stream.io.compression.*" (149
passed)
- sbt "actor/mimaReportBinaryIssues" "stream/mimaReportBinaryIssues"
- scalafmt and javafmtCheck (JDK 17)
References:
Review
https://github.com/apache/pekko/pull/3639#pullrequestreview-5624956465
* Add dictionary support to zstdDecompress
Motivation:
Compression.zstd accepts a dictionary, but zstdDecompress did not, so output
compressed with a dictionary could not be decompressed with Pekko Streams.
Modification:
- Add zstdDecompress(maxBytesPerChunk, dictionary) to the scaladsl and
javadsl
Compression APIs.
- ZstdDecompressor builds one ZstdDictDecompress per stage, which digests
the
dictionary (or its offset/length slice, without copying it in Java) once,
and
loads it by reference into the decompression context of every
materialization. The dictionary level only applies to compression and is
ignored. A dictionary with only one of offset and length is rejected when
the
flow is created.
- Document the new overload.
Result:
Data compressed with a zstd dictionary can be decompressed with
zstdDecompress.
Tests:
- New ZstdSpec tests: round trips with a whole dictionary and a dictionary
slice
through the scaladsl and javadsl APIs, using incompressible data so the
frames
must reference the dictionary; decompressing without the dictionary
fails; an
invalid offset/length combination is rejected.
- sbt "stream-tests/testOnly org.apache.pekko.stream.io.compression.*
org.apache.pekko.stream.scaladsl.CompressionSpec"
(158 passed)
- sbt "stream/mimaReportBinaryIssues" "bench-jmh/compile"
References:
https://github.com/apache/pekko/pull/3639
* feat: mark the zstd compression API as ApiMayChange
Motivation:
The zstd API is new in 2.0.0 and is likely to change: limiting the
compression and decompression windows (RFC 9659) needs new parameters
once zstd-jni exposes them, and the scaladsl ZstdDictionary case class
cannot gain fields without breaking binary compatibility.
Modification:
- Mark the zstd and zstdDecompress overloads and the Zstd* constants in
scaladsl and javadsl Compression as @ApiMayChange.
- Mark the scaladsl ZstdDictionary class and companion, and the javadsl
ZstdDictionary class, as @ApiMayChange.
- Document toImpl as INTERNAL API in both ZstdDictionary classes, since
the javadsl one is public to Java callers.
Result:
The zstd API can still evolve after 2.0.0.
Tests:
- Annotations and docs only, no behaviour change
- sbt "stream/compile" "stream/javafmtCheck"
- scalafmt on the changed Scala files
References:
https://github.com/apache/pekko/pull/3639
* feat: add zstd compression and decompression settings, with windowLog
Motivation:
Every new zstd option needed another Compression.zstd or zstdDecompress
overload in both DSLs. Limiting the compression window (RFC 9659 caps the
HTTP zstd content-coding at windowLog 23) would have added a fourth
compression overload, and the decompression window limit is next.
Modification:
- Add ZstdCompressionSettings (level, autoFlush, windowLog, dictionary) and
ZstdDecompressionSettings (maxBytesPerChunk, dictionary) in
org.apache.pekko.stream, shared by both DSLs like RestartSettings. They
are
final classes with private constructors and withX methods, so fields can
be added compatibly. withDictionary accepts the scaladsl and the javadsl
ZstdDictionary. Both are @ApiMayChange.
- Add zstd(settings) and zstdDecompress(settings) to both DSLs. Keep zstd,
zstd(level, dictionary) and zstdDecompress(maxBytesPerChunk), and remove
the unreleased zstd(level, dictionary, autoFlush) and
zstdDecompress(maxBytesPerChunk, dictionary) overloads.
- ZstdCompressor sets ZstdCompressCtx.setWindowLog when windowLog is not 0.
Add the ZstdMinWindowLog and ZstdMaxWindowLog constants.
- Update the zstd and zstdDecompress operator pages.
Result:
The zstd window can be limited for peers such as HTTP clients, and new zstd
options no longer need new overloads.
Tests:
- New ZstdSpec tests: a frame compressed at level 22 is rejected by a
decoder
limited to windowLog 23 and accepted with withWindowLog(23), from both
DSLs and with a dictionary; settings defaults, validation and withX
methods. Existing dictionary and auto-flush tests use the settings.
- sbt "stream-tests/testOnly ...ZstdSpec ...ZstdAutoFlushSpec
...CompressionWithoutZstdSpec" (62 passed, 1 canceled by an existing assume)
- sbt "++3.3.8" "stream/compile" "stream-tests/Test/compile"
- sbt "stream/javafmtCheck", "stream/headerCreate", scalafmt on changed
files
References:
https://github.com/apache/pekko/pull/3639
Refs #3674
* fix: release zstd compressor native resources on setup failure and close
Motivation:
Review of #3639 found that ZstdCompressor could leak native memory: a
failing dictionary setup (e.g. a negative slice offset) threw after the
ZstdCompressCtx was allocated, and the stage never got to close it. A
ZstdDictCompress was never closed explicitly, so its memory waited for
finalization and leaked with finalization disabled. Neither has a Cleaner.
Modification:
- Close the context if setting the level, window log or dictionary fails,
and close a ZstdDictCompress if loading it into the context fails.
- Keep the ZstdDictCompress and close it after the context in close().
- Make the javadsl ZstdDictionary final.
Result:
The zstd compressor releases its native memory on every path.
Tests:
- New ZstdSpec tests: the context is closed after a failed setup, and both
dictionary-with-level variants are closed with the compressor. Both fail
without the fix.
- sbt "stream-tests/testOnly ...ZstdSpec ...ZstdAutoFlushSpec
...CompressionWithoutZstdSpec ...CompressionSpec"
- sbt "++3.3.8" "stream-tests/Test/compile"
- sbt "stream/javafmtCheck", scalafmt on the changed Scala files
References:
https://github.com/apache/pekko/pull/3639
* fix: free the zstd decompression dictionary with its context
Motivation:
ZstdDecompressor digested the dictionary once into a ZstdDictDecompress
shared by every materialization. It could not be closed when one
materialization stopped and has no Cleaner, so its native memory was only
freed by finalization, and never with finalization disabled. This is the
decompression side of the ZstdDictCompress review finding on #3639.
Modification:
- Load the dictionary bytes into each materialization's ZstdDecompressCtx,
which owns the digested dictionary and frees it when the stage closes
the context.
- Copy the dictionary bytes, or the slice, when the operator is created, and
validate the slice bounds there.
- Close the context if loading the dictionary fails, since postStop is not
called then.
Result:
The zstd decompressor no longer relies on finalization to free dictionary
memory. Each materialization digests the dictionary itself, as the
compressor already does for dictionaries without a level.
Tests:
- New ZstdSpec tests: slices outside the dictionary are rejected when the
operator is created, and the operator decompresses with a copy of the
dictionary, for more than one materialization. The native memory itself
is not observable from the tests.
- sbt "stream-tests/testOnly ...ZstdSpec ...ZstdAutoFlushSpec
...CompressionWithoutZstdSpec ...CompressionSpec" (73 passed, 1 canceled by an
existing assume)
- scalafmt on the changed files
References:
https://github.com/apache/pekko/pull/3639
* fix: validate zstd compression dictionary slices and allow them without a
level
Motivation:
Review of #3639: ZstdCompressor did not check that a dictionary slice is
within the dictionary, so a bad offset or length failed in zstd-jni with
a less clear error. A slice without a level, which the Java
ZstdDictionary constructors allow and zstdDecompress accepts, was
rejected by the compressor.
Modification:
- Share the slice bounds check between ZstdCompressor and
ZstdDecompressor in the internal ZstdDictionaryImpl object.
- ZstdCompressor checks the slice before digesting it, and loads a copy
of the slice into the context when no level is set.
Result:
Out-of-bounds dictionary slices fail with the same
IllegalArgumentException for compression and decompression, and a slice
without a level can be used for compression.
Tests:
- sbt "stream-tests/testOnly
org.apache.pekko.stream.io.compression.ZstdSpec
org.apache.pekko.stream.io.compression.ZstdAutoFlushSpec
org.apache.pekko.stream.io.compression.CompressionWithoutZstdSpec": 76 passed
- With the ZstdCompressor change reverted, the new slice tests fail
References:
Refs #3639
* fix: hold the scaladsl ZstdDictionary bytes in a ByteString
Motivation:
Review of #3639: the scaladsl ZstdDictionary is a case class with an
Array[Byte] field, so it was mutable and its equals and hashCode
compared the array by reference.
Modification:
- The dictionary field is now an immutable ByteString. The Array[Byte]
factory methods copy the array.
- toImpl hands the operators a copy of the bytes, so whoever holds the
impl cannot modify the dictionary.
Result:
The scaladsl ZstdDictionary is immutable and has value semantics.
Tests:
- sbt "stream-tests/testOnly
org.apache.pekko.stream.io.compression.ZstdSpec
org.apache.pekko.stream.io.compression.ZstdAutoFlushSpec
org.apache.pekko.stream.io.compression.CompressionWithoutZstdSpec
org.apache.pekko.stream.scaladsl.CompressionSpec": 81 passed
- sbt "++3.3" "stream-tests/Test/compile": ok
References:
Refs #3639
* fix: make the javadsl ZstdDictionary immutable
Motivation:
The javadsl ZstdDictionary had setters and held the caller's byte[],
so it could be changed after an operator was created from it, and it
compared by reference. The scaladsl ZstdDictionary was already made
immutable.
Modification:
- Final fields, with the bytes held in a ByteString. The byte[]
constructors copy the array.
- Remove the setters. getDictionary returns the ByteString.
- Add equals, hashCode and toString.
- toImpl hands the operators a copy of the bytes.
Result:
The javadsl ZstdDictionary is immutable and has value semantics, the
same as the scaladsl one.
Tests:
- sbt "stream/javafmt" (JDK 17)
- sbt "stream-tests/testOnly
org.apache.pekko.stream.io.compression.ZstdSpec
org.apache.pekko.stream.io.compression.ZstdAutoFlushSpec
org.apache.pekko.stream.io.compression.CompressionWithoutZstdSpec
org.apache.pekko.stream.scaladsl.CompressionSpec": 83 passed
- sbt "++3.3" "stream-tests/Test/compile": ok
References:
Refs #3639
---------
Co-authored-by: Matthew de Detrich <[email protected]>
---
.../org/apache/pekko/util/ByteStringSpec.scala | 30 ++
.../scala/org/apache/pekko/util/ByteString.scala | 30 ++
.../pekko/stream/io/CompressionBenchmark.scala | 108 +++++++
.../paradox/stream/operators/Compression/zstd.md | 57 ++++
.../stream/operators/Compression/zstdDecompress.md | 43 +++
docs/src/main/paradox/stream/operators/index.md | 4 +
project/Dependencies.scala | 7 +-
project/OSGi.scala | 2 +-
.../pekko/stream/io/compression/CoderSpec.scala | 7 +-
.../compression/CompressionWithoutZstdSpec.scala | 91 ++++++
.../pekko/stream/io/compression/DeflateSpec.scala | 2 +-
.../pekko/stream/io/compression/GzipSpec.scala | 4 +-
.../stream/io/compression/ZstdAutoFlushSpec.scala | 31 ++
.../pekko/stream/io/compression/ZstdSpec.scala | 343 +++++++++++++++++++++
.../pekko/stream/scaladsl/CompressionSpec.scala | 15 +-
.../stream/javadsl/compression/ZstdDictionary.java | 194 ++++++++++++
.../pekko/stream/ZstdCompressionSettings.scala | 121 ++++++++
.../pekko/stream/ZstdDecompressionSettings.scala | 92 ++++++
.../stream/impl/io/compression/ZstdBuffers.scala | 56 ++++
.../impl/io/compression/ZstdCompressor.scala | 142 +++++++++
.../impl/io/compression/ZstdDecompressor.scala | 129 ++++++++
.../impl/io/compression/ZstdDictionaryImpl.scala | 37 +++
.../apache/pekko/stream/javadsl/Compression.scala | 104 +++++++
.../apache/pekko/stream/scaladsl/Compression.scala | 114 +++++++
.../scaladsl/compression/ZstdDictionary.scala | 86 ++++++
25 files changed, 1838 insertions(+), 11 deletions(-)
diff --git
a/actor-tests/src/test/scala/org/apache/pekko/util/ByteStringSpec.scala
b/actor-tests/src/test/scala/org/apache/pekko/util/ByteStringSpec.scala
index 271d982c0e..9e45d4dd51 100644
--- a/actor-tests/src/test/scala/org/apache/pekko/util/ByteStringSpec.scala
+++ b/actor-tests/src/test/scala/org/apache/pekko/util/ByteStringSpec.scala
@@ -2277,6 +2277,36 @@ class ByteStringSpec extends AnyWordSpec with Matchers
with Checkers {
}
}
+ "asByteBuffersUnsafe" in {
+ check { (a: ByteString) =>
+ a.asByteBuffersUnsafe.foldLeft(ByteString.empty) { (bs, bb) =>
+ bs ++ ByteString(bb)
+ } == a
+ }
+ check { (a: ByteString) =>
+ a.asByteBuffersUnsafe.forall(bb => bb.hasArray && !bb.isReadOnly &&
bb.hasRemaining)
+ }
+ check { (a: ByteString) =>
+ a.asByteBuffersUnsafe.size == a.asByteBuffers.count(_.hasRemaining)
+ }
+ }
+
+ "asByteBuffersUnsafe wraps the backing arrays without copying" in {
+ val array = Array[Byte](0, 1, 2, 3, 4, 5)
+ val compact = ByteString1C(array)
+ compact.asByteBuffersUnsafe.map(_.array()).toSeq should ===(Seq(array))
+
+ val sliced = ByteString1(array, 1, 3)
+ val slicedBuffer = sliced.asByteBuffersUnsafe.head
+ (slicedBuffer.array() should be).theSameInstanceAs(array)
+ (slicedBuffer.position(), slicedBuffer.limit()) should ===((1, 4))
+
+ val other = Array[Byte](6, 7)
+ val fragmented = sliced ++ ByteString1C(other) ++ ByteString1(array,
4, 2)
+ fragmented.asByteBuffersUnsafe.map(_.array()).toSeq should
===(Seq(array, other, array))
+ ByteString.empty.asByteBuffersUnsafe shouldBe empty
+ }
+
"toString should start with ByteString(" in {
check { (bs: ByteString) =>
bs.toString.startsWith("ByteString(")
diff --git a/actor/src/main/scala/org/apache/pekko/util/ByteString.scala
b/actor/src/main/scala/org/apache/pekko/util/ByteString.scala
index ee94979efe..0aa17cf7e9 100644
--- a/actor/src/main/scala/org/apache/pekko/util/ByteString.scala
+++ b/actor/src/main/scala/org/apache/pekko/util/ByteString.scala
@@ -201,6 +201,9 @@ object ByteString {
override def asByteBuffers:
scala.collection.immutable.Iterable[ByteBuffer] = List(asByteBuffer)
+ private[pekko] override def asByteBuffersUnsafe:
immutable.Iterable[ByteBuffer] =
+ if (isEmpty) Nil else List(ByteBuffer.wrap(bytes))
+
override def decodeString(charset: String): String =
if (isEmpty) "" else new String(bytes, charset)
@@ -618,6 +621,12 @@ object ByteString {
def asByteBuffers: scala.collection.immutable.Iterable[ByteBuffer] =
List(asByteBuffer)
+ private[pekko] override def asByteBuffersUnsafe:
immutable.Iterable[ByteBuffer] =
+ if (isEmpty) Nil else List(byteBufferUnsafe)
+
+ /** INTERNAL API: see [[ByteString.asByteBuffersUnsafe]] */
+ private[pekko] def byteBufferUnsafe: ByteBuffer = ByteBuffer.wrap(bytes,
startIndex, length)
+
override def decodeString(charset: String): String =
if (isEmpty) ""
else new String(bytes, startIndex, length, charset)
@@ -1187,6 +1196,12 @@ object ByteString {
def asByteBuffers: scala.collection.immutable.Iterable[ByteBuffer] =
List(first.asByteBuffer, second.asByteBuffer)
+ private[pekko] override def asByteBuffersUnsafe:
immutable.Iterable[ByteBuffer] = {
+ val builder = new VectorBuilder[ByteString1]
+ addFragmentsTo(builder)
+ builder.result().collect { case fragment if fragment.nonEmpty =>
fragment.byteBufferUnsafe }
+ }
+
override def asInputStream: InputStream =
new SequenceInputStream(Iterator(first.asInputStream,
second.asInputStream).asJavaEnumeration)
@@ -1753,6 +1768,9 @@ object ByteString {
def asByteBuffers: scala.collection.immutable.Iterable[ByteBuffer] =
bytestrings.map { _.asByteBuffer }
+ private[pekko] override def asByteBuffersUnsafe:
immutable.Iterable[ByteBuffer] =
+ bytestrings.collect { case fragment if fragment.nonEmpty =>
fragment.byteBufferUnsafe }
+
override def asInputStream: InputStream =
new
SequenceInputStream(bytestrings.iterator.map(_.asInputStream).asJavaEnumeration)
@@ -2826,6 +2844,18 @@ object ByteString {
*/
def asByteBuffers: immutable.Iterable[ByteBuffer]
+ /**
+ * INTERNAL API
+ *
+ * Returns array-backed ByteBuffers that wrap the non-empty fragments of
this ByteString without
+ * copying, for APIs that need heap buffers with an accessible array
(read-only buffers have no
+ * accessible array). Unlike [[asByteBuffers]] the buffers are writable and
expose the backing
+ * arrays, so callers must never write to them nor leak them. The bytes of
each fragment are
+ * between the position and the limit of its buffer. Empty for an empty
ByteString.
+ */
+ private[pekko] def asByteBuffersUnsafe: immutable.Iterable[ByteBuffer] =
+ if (isEmpty) Nil else List(ByteBuffer.wrap(toArrayUnsafe()))
+
/**
* Java API: Returns an Iterable of read-only ByteBuffers that directly
wraps this ByteStrings
* all fragments. Will always have at least one entry.
diff --git
a/bench-jmh/src/main/scala/org/apache/pekko/stream/io/CompressionBenchmark.scala
b/bench-jmh/src/main/scala/org/apache/pekko/stream/io/CompressionBenchmark.scala
new file mode 100644
index 0000000000..c9bbd7ce65
--- /dev/null
+++
b/bench-jmh/src/main/scala/org/apache/pekko/stream/io/CompressionBenchmark.scala
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.io
+
+import java.util.concurrent.TimeUnit
+
+import scala.collection.immutable
+import scala.concurrent.Await
+import scala.concurrent.duration._
+import scala.util.Random
+
+import org.openjdk.jmh.annotations._
+
+import org.apache.pekko
+import pekko.Done
+import pekko.NotUsed
+import pekko.actor.ActorSystem
+import pekko.stream.SystemMaterializer
+import pekko.stream.scaladsl.{ Compression, Flow, Sink, Source }
+import pekko.util.ByteString
+
+import com.typesafe.config.ConfigFactory
+
+/**
+ * Compresses and decompresses 16 MiB of JSON-like text split into `chunkSize`
elements.
+ */
+@State(Scope.Benchmark)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+@BenchmarkMode(Array(Mode.AverageTime))
+class CompressionBenchmark {
+
+ private val TotalBytes = 16 * 1024 * 1024
+
+ implicit val system: ActorSystem = ActorSystem(
+ "CompressionBenchmark",
+ ConfigFactory.parseString("""
+ pekko.loglevel = "OFF"
+ pekko.stdout-loglevel = "OFF"
+ """).withFallback(ConfigFactory.load()))
+
+ @Param(Array("gzip", "zstd"))
+ var algorithm = ""
+
+ @Param(Array("1024", "65536"))
+ var chunkSize = 0
+
+ private var chunks: immutable.Seq[ByteString] = _
+ private var compressedChunks: immutable.Seq[ByteString] = _
+ private var compressFlow: Flow[ByteString, ByteString, NotUsed] = _
+ private var decompressFlow: Flow[ByteString, ByteString, NotUsed] = _
+
+ @Setup
+ def setup(): Unit = {
+ SystemMaterializer(system).materializer
+
+ val random = new Random(42)
+ val builder = ByteString.newBuilder
+ var i = 0
+ while (builder.length < TotalBytes) {
+ builder ++= ByteString(
+
s"""{"id":$i,"name":"user-${random.nextInt(10000)}","score":${random.nextDouble()},"tags":["a","b","${random
+ .alphanumeric
+ .take(8)
+ .mkString}"]}
+""")
+ i += 1
+ }
+ chunks = builder.result().take(TotalBytes).grouped(chunkSize).toVector
+
+ algorithm match {
+ case "gzip" =>
+ compressFlow = Compression.gzip
+ decompressFlow = Compression.gzipDecompress()
+ case "zstd" =>
+ compressFlow = Compression.zstd
+ decompressFlow = Compression.zstdDecompress()
+ }
+ compressedChunks =
Await.result(Source(chunks).via(compressFlow).runWith(Sink.seq), 1.minute)
+ val ratio = TotalBytes.toDouble / compressedChunks.map(_.length).sum
+ println(f"$algorithm chunkSize=$chunkSize compression ratio $ratio%.2f")
+ }
+
+ @TearDown
+ def shutdown(): Unit = Await.result(system.terminate(), 5.seconds)
+
+ @Benchmark
+ def compress(): Done =
+ Await.result(Source(chunks).via(compressFlow).runWith(Sink.ignore),
1.minute)
+
+ @Benchmark
+ def decompress(): Done =
+
Await.result(Source(compressedChunks).via(decompressFlow).runWith(Sink.ignore),
1.minute)
+}
diff --git a/docs/src/main/paradox/stream/operators/Compression/zstd.md
b/docs/src/main/paradox/stream/operators/Compression/zstd.md
new file mode 100644
index 0000000000..209bf29f76
--- /dev/null
+++ b/docs/src/main/paradox/stream/operators/Compression/zstd.md
@@ -0,0 +1,57 @@
+# Compression.zstd
+
+Creates a flow that zstd-compresses a stream of ByteStrings.
+
+@ref[Compression operators](../index.md#compression-operators)
+
+## Dependency
+
+The zstd operators use [zstd-jni](https://github.com/luben/zstd-jni), an
optional dependency of Pekko Streams
+that has to be added explicitly:
+
+@@dependency[sbt,Maven,Gradle] {
+ group=com.github.luben
+ artifact=zstd-jni
+ version="$zstd_jni_version$"
+}
+
+## Signature
+
+@apidoc[Compression.zstd](stream.*.Compression$) {
scala="#zstd:org.apache.pekko.stream.scaladsl.Flow[org.apache.pekko.util.ByteString,org.apache.pekko.util.ByteString,org.apache.pekko.NotUsed]"
java="#zstd()" }
+@apidoc[Compression.zstd](stream.*.Compression$) {
scala="#zstd(level:Int,dictionary:Option[org.apache.pekko.stream.scaladsl.compression.ZstdDictionary]):org.apache.pekko.stream.scaladsl.Flow[org.apache.pekko.util.ByteString,org.apache.pekko.util.ByteString,org.apache.pekko.NotUsed]"
java="#zstd(int,java.util.Optional)" }
+@apidoc[Compression.zstd](stream.*.Compression$) {
scala="#zstd(settings:org.apache.pekko.stream.ZstdCompressionSettings):org.apache.pekko.stream.scaladsl.Flow[org.apache.pekko.util.ByteString,org.apache.pekko.util.ByteString,org.apache.pekko.NotUsed]"
java="#zstd(org.apache.pekko.stream.ZstdCompressionSettings)" }
+
+## Description
+
+Creates a flow that zstd-compresses a stream of ByteStrings. Note that the
compressor
+will SYNC_FLUSH after every @apidoc[util.ByteString] so that it is guaranteed
that every @apidoc[util.ByteString]
+coming out of the flow can be fully decompressed without waiting for
additional data. This may
+come at a compression performance cost for very small chunks.
+
+Use the `zstd(level, dictionary)` overload to control the compression level
and, optionally, to compress with a
+dictionary. Output compressed with a dictionary can only be decompressed with
the same dictionary, see
+@ref[zstdDecompress](zstdDecompress.md).
+
+The `zstd(settings)` overload takes a
@apidoc[ZstdCompressionSettings](stream.ZstdCompressionSettings), which sets
the level and dictionary as well as:
+
+* `withAutoFlush`: if `true` (the default), the compressor flushes after every
single element in the stream. If it is
+ `false`, the compressor only emits output when the compression algorithm
produces it, which can improve the
+ compression ratio for small chunks, but an emitted `ByteString` may not be
fully decompressible until more data
+ arrives.
+* `withWindowLog`: limits the window size to `2^windowLog` bytes. The
decompressor has to buffer a window of this size,
+ and since the size of a stream is not known up front, a frame declares the
default window of its compression level,
+ which is 128 MiB (`windowLog` 27) for the highest levels. Set `windowLog`
when the output is decompressed by a peer
+ that limits the window, for example to 23 (8 MiB) for the HTTP `zstd`
content-coding, as required by
+ [RFC 9659](https://www.rfc-editor.org/rfc/rfc9659). The default of 0 uses
the default of the compression level.
+
+## Reactive Streams semantics
+
+@@@div { .callout }
+
+**emits** when the compression algorithm produces output for the received
`ByteString`
+
+**backpressures** when downstream backpressures
+
+**completes** when upstream completes
+
+@@@
diff --git
a/docs/src/main/paradox/stream/operators/Compression/zstdDecompress.md
b/docs/src/main/paradox/stream/operators/Compression/zstdDecompress.md
new file mode 100644
index 0000000000..d7e0774f59
--- /dev/null
+++ b/docs/src/main/paradox/stream/operators/Compression/zstdDecompress.md
@@ -0,0 +1,43 @@
+# Compression.zstdDecompress
+
+Creates a flow that zstd-decompresses a stream of ByteStrings.
+
+@ref[Compression operators](../index.md#compression-operators)
+
+## Dependency
+
+The zstd operators use [zstd-jni](https://github.com/luben/zstd-jni), an
optional dependency of Pekko Streams
+that has to be added explicitly:
+
+@@dependency[sbt,Maven,Gradle] {
+ group=com.github.luben
+ artifact=zstd-jni
+ version="$zstd_jni_version$"
+}
+
+## Signature
+
+@apidoc[Compression.zstdDecompress](stream.*.Compression$) {
scala="#zstdDecompress(maxBytesPerChunk:Int):org.apache.pekko.stream.scaladsl.Flow[org.apache.pekko.util.ByteString,org.apache.pekko.util.ByteString,org.apache.pekko.NotUsed]"
java="#zstdDecompress(int)" }
+@apidoc[Compression.zstdDecompress](stream.*.Compression$) {
scala="#zstdDecompress(settings:org.apache.pekko.stream.ZstdDecompressionSettings):org.apache.pekko.stream.scaladsl.Flow[org.apache.pekko.util.ByteString,org.apache.pekko.util.ByteString,org.apache.pekko.NotUsed]"
java="#zstdDecompress(org.apache.pekko.stream.ZstdDecompressionSettings)" }
+
+## Description
+
+Creates a flow that zstd-decompresses a stream of ByteStrings. If the input is
truncated or corrupt this operator
+fails with a `com.github.luben.zstd.ZstdException`.
+
+The `zstdDecompress(settings)` overload takes a
@apidoc[ZstdDecompressionSettings](stream.ZstdDecompressionSettings), which
sets the maximum chunk size
+and, with `withDictionary`, a dictionary to decompress data that was
compressed with one. It must be the same
+dictionary (including any offset and length) that was used for compression;
the dictionary level only applies to
+compression and is ignored.
+
+## Reactive Streams semantics
+
+@@@div { .callout }
+
+**emits** when the decompression algorithm produces output for the received
`ByteString` (the emitted `ByteString` is of `maxBytesPerChunk` maximum length)
+
+**backpressures** when downstream backpressures
+
+**completes** when upstream completes
+
+@@@
diff --git a/docs/src/main/paradox/stream/operators/index.md
b/docs/src/main/paradox/stream/operators/index.md
index b2bc7b616f..c1df767850 100644
--- a/docs/src/main/paradox/stream/operators/index.md
+++ b/docs/src/main/paradox/stream/operators/index.md
@@ -370,6 +370,8 @@ Flow operators to (de)compress.
|Compression|<a name="gzip"></a>@ref[gzip](Compression/gzip.md)|Creates a flow
that gzip-compresses a stream of ByteStrings. |
|Compression|<a
name="gzipdecompress"></a>@ref[gzipDecompress](Compression/gzipDecompress.md)|Creates
a flow that gzip-decompresses a stream of ByteStrings. |
|Compression|<a
name="inflate"></a>@ref[inflate](Compression/inflate.md)|Creates a flow that
deflate-decompresses a stream of ByteStrings. |
+|Compression|<a name="zstd"></a>@ref[zstd](Compression/zstd.md)|Creates a flow
that zstd-compresses a stream of ByteStrings.|
+|Compression|<a
name="zstddecompress"></a>@ref[zstdDecompress](Compression/zstdDecompress.md)|Creates
a flow that zstd-decompresses a stream of ByteStrings.|
## Error handling
@@ -629,5 +631,7 @@ For more background see the @ref[Error Handling in
Streams](../stream-error.md)
* [zipWith](Source-or-Flow/zipWith.md)
* [zipWithIndex](Source-or-Flow/zipWithIndex.md)
* [zipWithN](Source/zipWithN.md)
+* [zstd](Compression/zstd.md)
+* [zstdDecompress](Compression/zstdDecompress.md)
@@@
diff --git a/project/Dependencies.scala b/project/Dependencies.scala
index a4d46c3159..55980dad0f 100644
--- a/project/Dependencies.scala
+++ b/project/Dependencies.scala
@@ -383,11 +383,13 @@ object Dependencies {
Docs.gson,
Provided.levelDB)
- lazy val benchJmh = l ++= Seq(logback, Provided.levelDB,
Provided.levelDBNative, Compile.jctools)
+ lazy val benchJmh = l ++= Seq(logback, Provided.levelDB,
Provided.levelDBNative, Compile.jctools, zstdJni)
// pekko stream
- lazy val stream = l ++= Seq[sbt.ModuleID](reactiveStreams, jspecify,
TestDependencies.scalatest)
+ // zstd-jni is optional, users of the zstd compression operators need to add
it themselves
+ lazy val stream =
+ l ++= Seq[sbt.ModuleID](reactiveStreams, jspecify, zstdJni % Optional,
TestDependencies.scalatest)
lazy val streamTyped = l ++= Seq[sbt.ModuleID](jspecify)
@@ -398,6 +400,7 @@ object Dependencies {
TestDependencies.junit)
lazy val streamTests = l ++= Seq(
+ zstdJni % Test,
TestDependencies.scalatest,
TestDependencies.scalatestScalaCheck,
TestDependencies.junit,
diff --git a/project/OSGi.scala b/project/OSGi.scala
index d54563cb5b..61f8b464ef 100644
--- a/project/OSGi.scala
+++ b/project/OSGi.scala
@@ -100,7 +100,7 @@ object OSGi {
lazy val stream =
exports(
packages = Seq("org.apache.pekko.stream.*"),
- imports = Seq(scalaParsingCombinatorImport()))
+ imports = Seq(scalaParsingCombinatorImport(),
optionalResolution("com.github.luben.zstd.*")))
lazy val streamTestkit = exports(Seq("org.apache.pekko.stream.testkit.*"))
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/io/compression/CoderSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/io/compression/CoderSpec.scala
index af7191b0d7..abbf778fbc 100644
---
a/stream-tests/src/test/scala/org/apache/pekko/stream/io/compression/CoderSpec.scala
+++
b/stream-tests/src/test/scala/org/apache/pekko/stream/io/compression/CoderSpec.scala
@@ -15,11 +15,11 @@ package org.apache.pekko.stream.io.compression
import java.io.{ ByteArrayOutputStream, InputStream, OutputStream }
import java.util.concurrent.ThreadLocalRandom
-import java.util.zip.DataFormatException
import scala.annotation.tailrec
import scala.concurrent.Await
import scala.concurrent.duration._
+import scala.reflect.ClassTag
import scala.util.control.NoStackTrace
import org.apache.pekko
@@ -31,7 +31,8 @@ import pekko.util.ByteString
import org.scalatest.Inspectors
import org.scalatest.wordspec.AnyWordSpec
-abstract class CoderSpec(codecName: String) extends AnyWordSpec with
CodecSpecSupport with Inspectors {
+abstract class CoderSpec[CorruptInputException: ClassTag](codecName: String)
extends AnyWordSpec with CodecSpecSupport
+ with Inspectors {
import CompressionTestingTools._
protected def newCompressor(): Compressor
@@ -85,7 +86,7 @@ abstract class CoderSpec(codecName: String) extends
AnyWordSpec with CodecSpecSu
"throw an error on corrupt input" in {
(the[RuntimeException] thrownBy {
ourDecode(corruptContent)
- }).ultimateCause should be(a[DataFormatException])
+ }).ultimateCause should be(a[CorruptInputException])
}
}
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/io/compression/CompressionWithoutZstdSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/io/compression/CompressionWithoutZstdSpec.scala
new file mode 100644
index 0000000000..e92f7dc0f3
--- /dev/null
+++
b/stream-tests/src/test/scala/org/apache/pekko/stream/io/compression/CompressionWithoutZstdSpec.scala
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.io.compression
+
+import java.io.File
+import java.net.URLClassLoader
+import java.util.concurrent.TimeUnit
+
+import scala.concurrent.Await
+import scala.concurrent.duration._
+
+import org.apache.pekko
+import pekko.actor.ActorSystem
+import pekko.stream.javadsl
+import pekko.stream.scaladsl.{ Compression, Sink, Source }
+import pekko.util.ByteString
+
+import com.typesafe.config.ConfigFactory
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
+
+/**
+ * zstd-jni is an optional dependency of pekko-stream, so the other
compression operators must
+ * work without it on the classpath.
+ */
+class CompressionWithoutZstdSpec extends AnyWordSpec with Matchers {
+
+ "Compression" should {
+ "support gzip and deflate without zstd-jni on the classpath" in {
+ // stream-tests run in a forked JVM, so java.class.path is the test
classpath
+ val urls = System
+ .getProperty("java.class.path")
+ .split(File.pathSeparator)
+ .filterNot(new File(_).getName.startsWith("zstd-jni"))
+ .map(new File(_).toURI.toURL)
+ // not closed: dispatcher shutdown tasks of the terminated ActorSystem
may still load classes from it
+ val loader = new URLClassLoader(urls, ClassLoader.getPlatformClassLoader)
+ a[ClassNotFoundException] should be thrownBy
loader.loadClass("com.github.luben.zstd.Zstd")
+
+ val roundTrip =
loader.loadClass(classOf[CompressionWithoutZstdRoundTrip].getName)
+ val thread = Thread.currentThread()
+ val contextClassLoader = thread.getContextClassLoader
+ thread.setContextClassLoader(loader)
+ val result =
+ try
roundTrip.getMethod("run").invoke(roundTrip.getDeclaredConstructor().newInstance())
+ finally thread.setContextClassLoader(contextClassLoader)
+ result should ===("gzip:hello deflate:hello javadsl:hello zstd-level:3")
+ }
+ }
+}
+
+/** Loaded and run in a class loader without zstd-jni by
[[CompressionWithoutZstdSpec]]. */
+class CompressionWithoutZstdRoundTrip {
+ def run(): String = {
+ // use this class loader (without zstd-jni) rather than one found from the
context or call stack
+ val classLoader = getClass.getClassLoader
+ // a fatal error in this nested ActorSystem must fail the test, not exit
the forked test JVM
+ val config = ConfigFactory
+ .parseString("pekko.jvm-exit-on-fatal-error = off")
+ .withFallback(ConfigFactory.load(classLoader))
+ val system = ActorSystem("CompressionWithoutZstd", Some(config),
Some(classLoader), None)
+ try {
+ def roundTrip(source: Source[ByteString, Any]): String =
+ Await.result(source.runWith(Sink.fold(ByteString.empty)(_ ++
_))(pekko.stream.Materializer(system)), 10.seconds)
+ .utf8String
+ val data = Source.single(ByteString("hello"))
+ val gzip =
roundTrip(data.via(Compression.gzip).via(Compression.gzipDecompress()))
+ val deflate =
roundTrip(data.via(Compression.deflate).via(Compression.inflate()))
+ val javaDeflate =
roundTrip(data.via(javadsl.Compression.deflate).via(javadsl.Compression.inflate(1024)))
+ s"gzip:$gzip deflate:$deflate javadsl:$javaDeflate
zstd-level:${Compression.ZstdDefaultCompressionLevel}"
+ } finally {
+ system.terminate()
+ Await.ready(system.whenTerminated, FiniteDuration(10, TimeUnit.SECONDS))
+ }
+ }
+}
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/io/compression/DeflateSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/io/compression/DeflateSpec.scala
index 90a7f3551b..d63b29ea92 100644
---
a/stream-tests/src/test/scala/org/apache/pekko/stream/io/compression/DeflateSpec.scala
+++
b/stream-tests/src/test/scala/org/apache/pekko/stream/io/compression/DeflateSpec.scala
@@ -21,7 +21,7 @@ import pekko.stream.impl.io.compression.{ Compressor,
DeflateCompressor }
import pekko.stream.scaladsl.{ Compression, Flow }
import pekko.util.ByteString
-class DeflateSpec extends CoderSpec("deflate") {
+class DeflateSpec extends CoderSpec[DataFormatException]("deflate") {
import CompressionTestingTools._
protected def newCompressor(): Compressor = new DeflateCompressor
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/io/compression/GzipSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/io/compression/GzipSpec.scala
index 177bbc0974..19d56902a7 100644
---
a/stream-tests/src/test/scala/org/apache/pekko/stream/io/compression/GzipSpec.scala
+++
b/stream-tests/src/test/scala/org/apache/pekko/stream/io/compression/GzipSpec.scala
@@ -15,14 +15,14 @@ package org.apache.pekko.stream.io.compression
import java.io.{ InputStream, OutputStream }
import java.nio.charset.StandardCharsets
-import java.util.zip.{ GZIPInputStream, GZIPOutputStream, ZipException }
+import java.util.zip.{ DataFormatException, GZIPInputStream, GZIPOutputStream,
ZipException }
import org.apache.pekko
import pekko.stream.impl.io.compression.{ Compressor, GzipCompressor }
import pekko.stream.scaladsl.{ Compression, Flow }
import pekko.util.ByteString
-class GzipSpec extends CoderSpec("gzip") {
+class GzipSpec extends CoderSpec[DataFormatException]("gzip") {
import CompressionTestingTools._
protected def newCompressor(): Compressor = new GzipCompressor
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/io/compression/ZstdAutoFlushSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/io/compression/ZstdAutoFlushSpec.scala
new file mode 100644
index 0000000000..0e8e35a0c8
--- /dev/null
+++
b/stream-tests/src/test/scala/org/apache/pekko/stream/io/compression/ZstdAutoFlushSpec.scala
@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.io.compression
+
+import org.apache.pekko
+import pekko.stream.ZstdCompressionSettings
+import pekko.stream.scaladsl.{ Compression, Flow }
+import pekko.util.ByteString
+
+class ZstdAutoFlushSpec extends ZstdSpec {
+ override protected val encoderFlow: Flow[ByteString, ByteString, Any] =
+ Compression.zstd(ZstdCompressionSettings().withAutoFlush(false))
+
+ override protected val autoFlush: Boolean = false
+
+}
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/io/compression/ZstdSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/io/compression/ZstdSpec.scala
new file mode 100644
index 0000000000..d9025524dc
--- /dev/null
+++
b/stream-tests/src/test/scala/org/apache/pekko/stream/io/compression/ZstdSpec.scala
@@ -0,0 +1,343 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.io.compression
+
+import java.io.{ ByteArrayInputStream, IOException, InputStream, OutputStream }
+import java.nio.charset.StandardCharsets
+
+import scala.collection.mutable
+import scala.concurrent.duration._
+import scala.util.Random
+
+import com.github.luben.zstd.{
+ Zstd,
+ ZstdCompressCtx,
+ ZstdException,
+ ZstdInputStream,
+ ZstdInputStreamNoFinalizer,
+ ZstdOutputStream
+}
+
+import org.apache.pekko
+import pekko.stream.{ ZstdCompressionSettings, ZstdDecompressionSettings }
+import pekko.stream.impl.io.compression.{ Compressor, ZstdCompressor }
+import pekko.stream.javadsl.{ Compression => JavaCompression }
+import pekko.stream.javadsl.compression.{ ZstdDictionary => JavaZstdDictionary
}
+import pekko.stream.scaladsl.{ Compression, Flow, Source }
+import pekko.stream.scaladsl.compression.ZstdDictionary
+import pekko.util.ByteString
+
+class ZstdSpec extends CoderSpec[ZstdException]("zstd") {
+ import CompressionTestingTools._
+
+ override protected def newCompressor(): Compressor = new ZstdCompressor
+
+ override protected def encoderFlow: Flow[ByteString, ByteString, Any] =
Compression.zstd
+
+ override protected def decoderFlow(maxBytesPerChunk: Int): Flow[ByteString,
ByteString, Any] =
+ Compression.zstdDecompress(maxBytesPerChunk)
+
+ override protected def newDecodedInputStream(underlying: InputStream):
InputStream =
+ new ZstdInputStream(underlying)
+
+ override protected def newEncodedOutputStream(underlying: OutputStream):
OutputStream =
+ new ZstdOutputStream(underlying)
+
+ override def extraTests(): Unit = {
+ "decode concatenated compressions" in {
+ ourDecode(Seq(encode("Hello, "), encode("dear "), encode("User!")).join)
should readAs("Hello, dear User!")
+ }
+ "throw an error on truncated input" in {
+ val ex = the[RuntimeException] thrownBy
ourDecode(streamEncode(smallTextBytes).dropRight(5))
+ ex.ultimateCause.getMessage should equal("Truncated zstd stream")
+ }
+ "throw an error if compressed data is just missing the end of the frame"
in {
+ val ex = the[RuntimeException] thrownBy
ourDecode(newCompressor().compressAndFlush(smallTextBytes))
+ ex.ultimateCause.getMessage should equal("Truncated zstd stream")
+ }
+ "decompress input whose size is an exact multiple of maxBytesPerChunk" in {
+ val input = ByteString(Array.fill(4096)(1.toByte))
+ val compressed = ourEncode(input)
+ for (chunkSize <- Seq(4096, 2048, 1024, 4095, 4097)) {
+ withClue(s"maxBytesPerChunk = $chunkSize: ") {
+
Source.single(compressed).via(decoderFlow(chunkSize)).join.awaitResult(3.seconds)
should ===(input)
+ Source(List(compressed, compressed))
+ .via(decoderFlow(chunkSize))
+ .join
+ .awaitResult(3.seconds) should ===(input ++ input)
+ }
+ }
+ }
+ "compress and decompress multi-fragment ByteStrings" in {
+ // slices of a single array, so the fragments do not cover their whole
backing arrays
+ def fragmented(bytes: ByteString): ByteString =
+ ByteString(0.toByte +:
bytes.toArray).drop(1).grouped(100).foldLeft(ByteString.empty)(_ ++ _)
+ val input = fragmented(ByteString(largeText, "UTF8"))
+ val compressed = ourEncode(input)
+ streamDecode(compressed) should equal(input)
+ ourDecode(fragmented(compressed)) should equal(input)
+ }
+ "define the same constants as zstd-jni" in {
+ Compression.ZstdMinCompressionLevel should
===(Zstd.minCompressionLevel())
+ Compression.ZstdMaxCompressionLevel should
===(Zstd.maxCompressionLevel())
+ Compression.ZstdDefaultCompressionLevel should
===(Zstd.defaultCompressionLevel())
+ Compression.ZstdDecompressMaxBlockSize should ===(Zstd.blockSizeMax())
+ Compression.ZstdMinWindowLog should ===(Zstd.windowLogMin())
+ Compression.ZstdMaxWindowLog should ===(Zstd.windowLogMax())
+ }
+ "limit the compression window with windowLog" in {
+ val input = ByteString(Array.fill(1 << 16)(1.toByte))
+ def compress(flow: Flow[ByteString, ByteString, Any]): ByteString =
+ Source.single(input).via(flow).join.awaitResult(3.seconds)
+ // a decoder that rejects frames with a window larger than 2^windowLogMax
+ def decodeWithLongMax(compressed: ByteString, windowLogMax: Int):
ByteString = {
+ val in = new ZstdInputStreamNoFinalizer(new
ByteArrayInputStream(compressed.toArray))
+ try {
+ in.setLongMax(windowLogMax)
+ ByteString(in.readAllBytes())
+ } finally in.close()
+ }
+
+ val highest =
ZstdCompressionSettings().withLevel(Compression.ZstdMaxCompressionLevel)
+
+ // the streamed size is unknown, so the frame declares the default
window of the level, 2^27 for level 22
+ val unlimited = compress(Compression.zstd(highest))
+ an[IOException] should be thrownBy decodeWithLongMax(unlimited, 23)
+
+ val limited = compress(Compression.zstd(highest.withWindowLog(23)))
+ decodeWithLongMax(limited, 23) should ===(input)
+ ourDecode(limited) should ===(input)
+ an[IOException] should be thrownBy decodeWithLongMax(limited, 22)
+
decodeWithLongMax(compress(JavaCompression.zstd(highest.withWindowLog(23)).asScala),
23) should ===(input)
+
+ val withDictionary =
+
compress(Compression.zstd(highest.withWindowLog(23).withAutoFlush(false).withDictionary(
+ ZstdDictionary(smallTextBytes))))
+ Zstd.decompress(withDictionary.toArray, smallTextBytes.toArray,
input.size) should ===(input.toArray)
+ }
+ "validate the compression settings" in {
+ val settings = ZstdCompressionSettings()
+ settings.level should ===(Compression.ZstdDefaultCompressionLevel)
+ settings.autoFlush should ===(true)
+ settings.windowLog should ===(0)
+ settings.dictionaryImpl should ===(None)
+ (ZstdCompressionSettings.create() should be).theSameInstanceAs(settings)
+
+ for (windowLog <- Seq(-1, Compression.ZstdMinWindowLog - 1,
Compression.ZstdMaxWindowLog + 1))
+ an[IllegalArgumentException] should be thrownBy
settings.withWindowLog(windowLog)
+ for (level <- Seq(Compression.ZstdMinCompressionLevel - 1,
Compression.ZstdMaxCompressionLevel + 1))
+ an[IllegalArgumentException] should be thrownBy
settings.withLevel(level)
+ an[IllegalArgumentException] should be thrownBy
+ Compression.zstd(Compression.ZstdMaxCompressionLevel + 1, None)
+
+ val updated =
settings.withLevel(19).withAutoFlush(false).withWindowLog(23)
+ .withDictionary(new JavaZstdDictionary(smallTextBytes.toArray))
+ (updated.level, updated.autoFlush, updated.windowLog) should ===((19,
false, 23))
+ updated.dictionaryImpl.map(_.dictionary.toSeq) should
===(Some(smallTextBytes.toSeq))
+ updated.withoutDictionary.dictionaryImpl should ===(None)
+ // the default settings are not changed
+ settings.level should ===(Compression.ZstdDefaultCompressionLevel)
+ }
+ "validate the decompression settings" in {
+ val settings = ZstdDecompressionSettings()
+ settings.maxBytesPerChunk should
===(Compression.ZstdDecompressMaxBlockSize)
+ settings.dictionaryImpl should ===(None)
+ (ZstdDecompressionSettings.create() should
be).theSameInstanceAs(settings)
+ an[IllegalArgumentException] should be thrownBy
settings.withMaxBytesPerChunk(0)
+ settings.withMaxBytesPerChunk(1024).maxBytesPerChunk should ===(1024)
+
settings.withDictionary(ZstdDictionary(smallTextBytes)).withoutDictionary.dictionaryImpl
should ===(None)
+ }
+ "encode String dictionaries as UTF-8" in {
+ val text = "dictionary é中"
+ val utf8 = text.getBytes(StandardCharsets.UTF_8)
+ ZstdDictionary(text).dictionary should ===(ByteString(utf8))
+ new JavaZstdDictionary(text).getDictionary should ===(ByteString(utf8))
+ }
+ "give the Java dictionary value semantics" in {
+ val dictBytes = smallTextBytes.toArray
+ val dictionary = new JavaZstdDictionary(dictBytes,
Compression.ZstdDefaultCompressionLevel)
+ dictionary should ===(new JavaZstdDictionary(dictBytes.clone(),
Compression.ZstdDefaultCompressionLevel))
+ dictionary.hashCode should ===(
+ new JavaZstdDictionary(smallTextBytes,
Compression.ZstdDefaultCompressionLevel).hashCode)
+ dictionary should not be new JavaZstdDictionary(dictBytes)
+ // the array is copied, so changing it does not affect the dictionary
+ java.util.Arrays.fill(dictBytes, 0.toByte)
+ dictionary.getDictionary should ===(smallTextBytes)
+ // and so is the array handed to the operators
+ java.util.Arrays.fill(dictionary.toImpl.dictionary, 0.toByte)
+ dictionary.getDictionary should ===(smallTextBytes)
+ }
+ "give the Scala dictionary value semantics" in {
+ val dictBytes = smallTextBytes.toArray
+ val dictionary = ZstdDictionary(dictBytes,
Compression.ZstdDefaultCompressionLevel)
+ dictionary should ===(ZstdDictionary(dictBytes.clone(),
Compression.ZstdDefaultCompressionLevel))
+ dictionary.hashCode should ===(ZstdDictionary(smallTextBytes,
Compression.ZstdDefaultCompressionLevel).hashCode)
+ dictionary should not be ZstdDictionary(dictBytes)
+ // the array is copied, so changing it does not affect the dictionary
+ java.util.Arrays.fill(dictBytes, 0.toByte)
+ dictionary.dictionary should ===(smallTextBytes)
+ // and so is the array handed to the operators
+ java.util.Arrays.fill(dictionary.toImpl.dictionary, 0.toByte)
+ dictionary.dictionary should ===(smallTextBytes)
+ }
+ "decompress with a dictionary" in {
+ // incompressible input that is only found in the dictionary, so the
frame must reference the dictionary
+ val dictBytes = new Array[Byte](4096)
+ new Random(42).nextBytes(dictBytes)
+ val input = ByteString(dictBytes)
+ val compressed = Source
+ .single(input)
+ .via(Compression.zstd(Compression.ZstdDefaultCompressionLevel,
Some(ZstdDictionary(dictBytes))))
+ .join
+ .awaitResult(3.seconds)
+ compressed.size should be < input.size / 10
+ // the dictionary level only applies to compression
+ for (dictionary <-
+ Seq(ZstdDictionary(dictBytes), ZstdDictionary(dictBytes,
Compression.ZstdMaxCompressionLevel))) {
+ Source
+ .single(compressed)
+
.via(Compression.zstdDecompress(ZstdDecompressionSettings().withDictionary(dictionary)))
+ .join
+ .awaitResult(3.seconds) should ===(input)
+ }
+ Source
+ .single(compressed)
+ .via(JavaCompression.zstdDecompress(
+ ZstdDecompressionSettings.create().withDictionary(new
JavaZstdDictionary(dictBytes))))
+ .join
+ .awaitResult(3.seconds) should ===(input)
+
+ val ex = the[RuntimeException] thrownBy ourDecode(compressed)
+ ex.ultimateCause shouldBe a[ZstdException]
+ }
+ "decompress with a dictionary slice" in {
+ val content = new Array[Byte](4096)
+ new Random(42).nextBytes(content)
+ val dictBytes = "padding".getBytes(StandardCharsets.UTF_8) ++ content
+ val dictionary = ZstdDictionary(dictBytes,
Compression.ZstdDefaultCompressionLevel, 7, content.length)
+ val input = ByteString(content)
+ val compressed = new ZstdCompressor(dictionary =
Some(dictionary.toImpl)).compressAndFinish(input)
+ compressed.size should be < input.size / 10
+
+ Source
+ .single(compressed)
+
.via(Compression.zstdDecompress(ZstdDecompressionSettings().withDictionary(dictionary)))
+ .join
+ .awaitResult(3.seconds) should ===(input)
+ }
+ "reject a decompression dictionary slice outside the dictionary" in {
+ for ((offset, length) <- Seq((-1, 4), (0, -1), (1,
smallTextBytes.length)))
+ withClue(s"offset $offset, length $length: ") {
+ an[IllegalArgumentException] should be thrownBy
Compression.zstdDecompress(
+ ZstdDecompressionSettings().withDictionary(
+ ZstdDictionary(smallTextBytes.toArray,
Compression.ZstdDefaultCompressionLevel, offset, length)))
+ }
+ }
+ "decompress with a copy of the dictionary" in {
+ val dictBytes = new Array[Byte](4096)
+ new Random(42).nextBytes(dictBytes)
+ val input = ByteString(dictBytes)
+ val compressed = new ZstdCompressor(dictionary =
Some(ZstdDictionary(dictBytes).toImpl)).compressAndFinish(input)
+ val decompress =
Compression.zstdDecompress(ZstdDecompressionSettings().withDictionary(ZstdDictionary(dictBytes)))
+ // changing the array after the operator is created does not affect it
+ java.util.Arrays.fill(dictBytes, 0.toByte)
+ for (_ <- 1 to 2) // every materialization loads the dictionary into its
own context
+ Source.single(compressed).via(decompress).join.awaitResult(3.seconds)
should ===(input)
+ }
+ "reject a decompression dictionary with only one of offset and length" in {
+ val dictionary = new JavaZstdDictionary(smallTextBytes.toArray, null:
Integer, Integer.valueOf(1), null: Integer)
+ an[IllegalArgumentException] should be thrownBy
+
JavaCompression.zstdDecompress(ZstdDecompressionSettings.create().withDictionary(dictionary))
+ }
+ "close the compression context when the compressor setup fails" in {
+ // the context is recorded outside of the subclass, whose fields are
only set after the constructor
+ val contexts = mutable.Buffer.empty[ZstdCompressCtx]
+ def compressor(dictionary: ZstdDictionary): ZstdCompressor =
+ new ZstdCompressor(dictionary = Some(dictionary.toImpl)) {
+ override private[pekko] def newCompressCtx(): ZstdCompressCtx = {
+ val ctx = super.newCompressCtx()
+ contexts += ctx
+ ctx
+ }
+ }
+
+ val valid = compressor(ZstdDictionary(smallTextBytes,
Compression.ZstdDefaultCompressionLevel))
+ contexts.head.setLevel(Compression.ZstdDefaultCompressionLevel) // still
open
+ valid.close()
+ an[IllegalStateException] should be thrownBy
contexts.head.setLevel(Compression.ZstdDefaultCompressionLevel)
+
+ contexts.clear()
+ an[Exception] should be thrownBy compressor(
+ ZstdDictionary(smallTextBytes,
Compression.ZstdDefaultCompressionLevel, -1, smallTextBytes.length))
+ contexts should have size 1
+ an[IllegalStateException] should be thrownBy
contexts.head.setLevel(Compression.ZstdDefaultCompressionLevel)
+ }
+ "close the compression dictionary with the compressor" in {
+ val dictBytes = ("padding" + smallText *
4).getBytes(StandardCharsets.UTF_8)
+ def loadInto(compressor: ZstdCompressor): Unit = {
+ val ctx = new ZstdCompressCtx()
+ try ctx.loadDict(compressor.compressDictionary.get)
+ finally ctx.close()
+ }
+ for (dictionary <- Seq(
+ ZstdDictionary(dictBytes, Compression.ZstdDefaultCompressionLevel),
+ ZstdDictionary(dictBytes, Compression.ZstdDefaultCompressionLevel,
7, dictBytes.length - 7))) {
+ val compressor = new ZstdCompressor(dictionary =
Some(dictionary.toImpl))
+ compressor.compressAndFinish(smallTextBytes)
+ loadInto(compressor) // still open
+ compressor.close()
+ // a closed dictionary can no longer be used
+ an[IllegalStateException] should be thrownBy loadInto(compressor)
+ }
+ // a dictionary without a level is loaded into the context and needs no
separate close
+ new ZstdCompressor(dictionary =
Some(ZstdDictionary(dictBytes).toImpl)).compressDictionary should ===(None)
+ }
+ "compress with a dictionary slice" in {
+ val dictBytes = ("padding" + smallText *
4).getBytes(StandardCharsets.UTF_8)
+ val dictionary = ZstdDictionary(dictBytes,
Compression.ZstdDefaultCompressionLevel, 7, dictBytes.length - 7)
+ val compressed = new ZstdCompressor(dictionary =
Some(dictionary.toImpl)).compressAndFinish(smallTextBytes)
+
+ val decompressed = Zstd.decompress(compressed.toArray,
dictBytes.drop(7), smallTextBytes.length)
+ ByteString(decompressed) should readAs(smallText)
+ }
+ "compress with a dictionary slice without a level" in {
+ val dictBytes = ("padding" + smallText *
4).getBytes(StandardCharsets.UTF_8)
+ val dictionary =
+ new JavaZstdDictionary(dictBytes, null: Integer, Integer.valueOf(7),
Integer.valueOf(dictBytes.length - 7))
+ val compressed = new ZstdCompressor(dictionary =
Some(dictionary.toImpl)).compressAndFinish(smallTextBytes)
+
+ val decompressed = Zstd.decompress(compressed.toArray,
dictBytes.drop(7), smallTextBytes.length)
+ ByteString(decompressed) should readAs(smallText)
+ }
+ "reject a compression dictionary slice outside the dictionary" in {
+ val dictBytes = smallTextBytes.toArray
+ for {
+ level <- Seq[Integer](null,
Integer.valueOf(Compression.ZstdDefaultCompressionLevel))
+ (offset, length) <- Seq((-1, 4), (0, -1), (1, dictBytes.length))
+ } withClue(s"level $level, offset $offset, length $length: ") {
+ val dictionary = new JavaZstdDictionary(dictBytes, level,
Integer.valueOf(offset), Integer.valueOf(length))
+ val ex = the[IllegalArgumentException] thrownBy new
ZstdCompressor(dictionary = Some(dictionary.toImpl))
+ ex.getMessage should include("must be within the dictionary")
+ }
+ }
+ "reject a compression dictionary with only one of offset and length" in {
+ val dictionary = new JavaZstdDictionary(smallTextBytes.toArray, null:
Integer, Integer.valueOf(1), null: Integer)
+ an[IllegalArgumentException] should be thrownBy new
ZstdCompressor(dictionary = Some(dictionary.toImpl))
+ }
+ }
+}
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/CompressionSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/CompressionSpec.scala
index 0b358da9f4..41f5af5a23 100644
---
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/CompressionSpec.scala
+++
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/CompressionSpec.scala
@@ -16,8 +16,7 @@ package org.apache.pekko.stream.scaladsl
import java.nio.charset.StandardCharsets
import org.apache.pekko
-import pekko.stream.impl.io.compression.DeflateCompressor
-import pekko.stream.impl.io.compression.GzipCompressor
+import pekko.stream.impl.io.compression.{ DeflateCompressor, GzipCompressor,
ZstdCompressor }
import pekko.stream.testkit.StreamSpec
import pekko.util.ByteString
@@ -27,6 +26,8 @@ class CompressionSpec extends StreamSpec {
def deflate(s: String): ByteString = new
DeflateCompressor().compressAndFinish(ByteString(s))
+ def zstd(s: String): ByteString = new
ZstdCompressor().compressAndFinish(ByteString(s))
+
val data = "hello world"
"Gzip decompression" must {
@@ -47,4 +48,14 @@ class CompressionSpec extends StreamSpec {
res.futureValue should ===(data)
}
}
+
+ "Zstd decompression" must {
+ "be able to decompress a zstd stream" in {
+ val source =
+
Source.single(zstd(data)).via(Compression.zstdDecompress()).map(_.decodeString(StandardCharsets.UTF_8))
+
+ val res = source.runFold("")(_ + _)
+ res.futureValue should ===(data)
+ }
+ }
}
diff --git
a/stream/src/main/java/org/apache/pekko/stream/javadsl/compression/ZstdDictionary.java
b/stream/src/main/java/org/apache/pekko/stream/javadsl/compression/ZstdDictionary.java
new file mode 100644
index 0000000000..12bcb621b0
--- /dev/null
+++
b/stream/src/main/java/org/apache/pekko/stream/javadsl/compression/ZstdDictionary.java
@@ -0,0 +1,194 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.javadsl.compression;
+
+import org.apache.pekko.annotation.ApiMayChange;
+import org.apache.pekko.annotation.InternalApi;
+import org.apache.pekko.stream.impl.io.compression.ZstdDictionaryImpl;
+import org.apache.pekko.util.ByteString;
+import org.apache.pekko.util.OptionalUtil;
+import scala.Option;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Objects;
+import java.util.OptionalInt;
+
+/**
+ * Configuration class based on the official C zstd library to specify
dictionary options for
+ * compression.
+ *
+ * @see <a
+ *
href="https://github.com/facebook/zstd?tab=readme-ov-file#dictionary-compression-how-to";>Zstd
+ * dictionary readme</a>
+ * @see <a
href="https://facebook.github.io/zstd/zstd_manual.html#Chapter10";>Zstd C
manual</a>
+ * @since 2.0.0
+ */
+@ApiMayChange
+public final class ZstdDictionary {
+ private final ByteString dictionary;
+ private final OptionalInt level;
+ private final OptionalInt offset;
+ private final OptionalInt length;
+
+ private ZstdDictionary(
+ ByteString dictionary, OptionalInt level, OptionalInt offset,
OptionalInt length) {
+ this.dictionary = Objects.requireNonNull(dictionary, "dictionary");
+ this.level = level;
+ this.offset = offset;
+ this.length = length;
+ }
+
+ private static OptionalInt optionalInt(Integer value) {
+ return value == null ? OptionalInt.empty() : OptionalInt.of(value);
+ }
+
+ // the byte[] constructors copy the array, so later changes to it do not
affect the dictionary
+
+ public ZstdDictionary(byte[] dictionary) {
+ this(ByteString.fromArray(dictionary));
+ }
+
+ public ZstdDictionary(byte[] dictionary, int level) {
+ this(ByteString.fromArray(dictionary), level);
+ }
+
+ public ZstdDictionary(byte[] dictionary, Integer level) {
+ this(ByteString.fromArray(dictionary), level);
+ }
+
+ public ZstdDictionary(byte[] dictionary, int level, int offset, int length) {
+ this(ByteString.fromArray(dictionary), level, offset, length);
+ }
+
+ public ZstdDictionary(byte[] dictionary, Integer level, Integer offset,
Integer length) {
+ this(ByteString.fromArray(dictionary), level, offset, length);
+ }
+
+ public ZstdDictionary(ByteString dictionary) {
+ this(dictionary, OptionalInt.empty(), OptionalInt.empty(),
OptionalInt.empty());
+ }
+
+ public ZstdDictionary(ByteString dictionary, int level) {
+ this(dictionary, OptionalInt.of(level), OptionalInt.empty(),
OptionalInt.empty());
+ }
+
+ public ZstdDictionary(ByteString dictionary, Integer level) {
+ this(dictionary, optionalInt(level), OptionalInt.empty(),
OptionalInt.empty());
+ }
+
+ public ZstdDictionary(ByteString dictionary, int level, int offset, int
length) {
+ this(dictionary, OptionalInt.of(level), OptionalInt.of(offset),
OptionalInt.of(length));
+ }
+
+ public ZstdDictionary(ByteString dictionary, Integer level, Integer offset,
Integer length) {
+ this(dictionary, optionalInt(level), optionalInt(offset),
optionalInt(length));
+ }
+
+ public ZstdDictionary(String dictionary) {
+ this(ByteString.fromString(dictionary, StandardCharsets.UTF_8));
+ }
+
+ public ZstdDictionary(String dictionary, int level) {
+ this(ByteString.fromString(dictionary, StandardCharsets.UTF_8), level);
+ }
+
+ public ZstdDictionary(String dictionary, Integer level) {
+ this(ByteString.fromString(dictionary, StandardCharsets.UTF_8), level);
+ }
+
+ public ZstdDictionary(String dictionary, int level, int offset, int length) {
+ this(ByteString.fromString(dictionary, StandardCharsets.UTF_8), level,
offset, length);
+ }
+
+ public ZstdDictionary(String dictionary, Integer level, Integer offset,
Integer length) {
+ this(ByteString.fromString(dictionary, StandardCharsets.UTF_8), level,
offset, length);
+ }
+
+ public ByteString getDictionary() {
+ return dictionary;
+ }
+
+ public OptionalInt getLevel() {
+ return level;
+ }
+
+ public OptionalInt getOffset() {
+ return offset;
+ }
+
+ public OptionalInt getLength() {
+ return length;
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (this == other) return true;
+ if (!(other instanceof ZstdDictionary)) return false;
+ ZstdDictionary that = (ZstdDictionary) other;
+ return dictionary.equals(that.dictionary)
+ && level.equals(that.level)
+ && offset.equals(that.offset)
+ && length.equals(that.length);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(dictionary, level, offset, length);
+ }
+
+ @Override
+ public String toString() {
+ return "ZstdDictionary("
+ + dictionary.size()
+ + " bytes, level="
+ + level
+ + ", offset="
+ + offset
+ + ", length="
+ + length
+ + ")";
+ }
+
+ /** INTERNAL API */
+ @InternalApi
+ public ZstdDictionaryImpl toImpl() {
+ // a copy, so that whoever holds the impl cannot modify the dictionary
+ final byte[] bytes = dictionary.toArray();
+ return new ZstdDictionaryImpl() {
+ @Override
+ public byte[] dictionary() {
+ return bytes;
+ }
+
+ @Override
+ public Option<Object> level() {
+ return OptionalUtil.convertOptionalToScala(level);
+ }
+
+ @Override
+ public Option<Object> offset() {
+ return OptionalUtil.convertOptionalToScala(offset);
+ }
+
+ @Override
+ public Option<Object> length() {
+ return OptionalUtil.convertOptionalToScala(length);
+ }
+ };
+ }
+}
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/ZstdCompressionSettings.scala
b/stream/src/main/scala/org/apache/pekko/stream/ZstdCompressionSettings.scala
new file mode 100644
index 0000000000..e3781d84fb
--- /dev/null
+++
b/stream/src/main/scala/org/apache/pekko/stream/ZstdCompressionSettings.scala
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream
+
+import org.apache.pekko
+import pekko.annotation.{ ApiMayChange, InternalApi }
+import pekko.stream.impl.io.compression.ZstdDictionaryImpl
+import pekko.stream.scaladsl.Compression
+
+/**
+ * Settings for the zstd compression operators, `Compression.zstd` in the
Scala and Java DSLs.
+ *
+ * @since 2.0.0
+ */
+@ApiMayChange
+final class ZstdCompressionSettings private (
+ val level: Int,
+ val autoFlush: Boolean,
+ val windowLog: Int,
+ dictionary: Option[ZstdDictionaryImpl]) {
+
+ require(
+ level >= Compression.ZstdMinCompressionLevel && level <=
Compression.ZstdMaxCompressionLevel,
+ s"level must be between ${Compression.ZstdMinCompressionLevel} and
${Compression.ZstdMaxCompressionLevel}, " +
+ s"was $level")
+ require(
+ windowLog == 0 || (windowLog >= Compression.ZstdMinWindowLog && windowLog
<= Compression.ZstdMaxWindowLog),
+ s"windowLog must be 0 or between ${Compression.ZstdMinWindowLog} and
${Compression.ZstdMaxWindowLog}, " +
+ s"was $windowLog")
+
+ /**
+ * The compression level, must be greater or equal to
`Compression.ZstdMinCompressionLevel` and less than or equal
+ * to `Compression.ZstdMaxCompressionLevel`. Defaults to
`Compression.ZstdDefaultCompressionLevel`.
+ */
+ def withLevel(level: Int): ZstdCompressionSettings = copy(level = level)
+
+ /**
+ * If true (the default) the compressor flushes after every single element
in the stream, so that every emitted
+ * `ByteString` can be fully decompressed without waiting for more data. If
false, the compressor only emits output
+ * when the compression algorithm produces it, which can improve the
compression ratio for small elements.
+ */
+ def withAutoFlush(autoFlush: Boolean): ZstdCompressionSettings =
copy(autoFlush = autoFlush)
+
+ /**
+ * The log2 of the maximum window size, which a decompressor needs to
buffer. 0 (the default) uses the default of
+ * the compression level, otherwise it must be greater or equal to
`Compression.ZstdMinWindowLog` and less than or
+ * equal to `Compression.ZstdMaxWindowLog`. For example, RFC 9659 limits the
HTTP `zstd` content-coding to a window
+ * log of 23 (8 MiB).
+ */
+ def withWindowLog(windowLog: Int): ZstdCompressionSettings = copy(windowLog
= windowLog)
+
+ /**
+ * Scala API: compress with a dictionary. The output can only be
decompressed with the same dictionary.
+ */
+ def withDictionary(dictionary: scaladsl.compression.ZstdDictionary):
ZstdCompressionSettings =
+ copy(dictionary = Some(dictionary.toImpl))
+
+ /**
+ * Java API: compress with a dictionary. The output can only be decompressed
with the same dictionary.
+ */
+ def withDictionary(dictionary: javadsl.compression.ZstdDictionary):
ZstdCompressionSettings =
+ copy(dictionary = Some(dictionary.toImpl))
+
+ /**
+ * Compress without a dictionary, the default.
+ */
+ def withoutDictionary: ZstdCompressionSettings = copy(dictionary = None)
+
+ /**
+ * INTERNAL API
+ */
+ @InternalApi
+ private[pekko] def dictionaryImpl: Option[ZstdDictionaryImpl] = dictionary
+
+ override def toString: String =
+
s"ZstdCompressionSettings(level=$level,autoFlush=$autoFlush,windowLog=$windowLog,dictionary=${dictionary.isDefined})"
+
+ private def copy(
+ level: Int = level,
+ autoFlush: Boolean = autoFlush,
+ windowLog: Int = windowLog,
+ dictionary: Option[ZstdDictionaryImpl] = dictionary):
ZstdCompressionSettings =
+ new ZstdCompressionSettings(level, autoFlush, windowLog, dictionary)
+}
+
+/**
+ * @since 2.0.0
+ */
+@ApiMayChange
+object ZstdCompressionSettings {
+
+ private val Default =
+ new ZstdCompressionSettings(Compression.ZstdDefaultCompressionLevel,
autoFlush = true, windowLog = 0, None)
+
+ /**
+ * Scala API: the default settings, the default compression level with auto
flush, the default window size of the
+ * level and no dictionary
+ */
+ def apply(): ZstdCompressionSettings = Default
+
+ /**
+ * Java API: the default settings, the default compression level with auto
flush, the default window size of the
+ * level and no dictionary
+ */
+ def create(): ZstdCompressionSettings = Default
+}
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/ZstdDecompressionSettings.scala
b/stream/src/main/scala/org/apache/pekko/stream/ZstdDecompressionSettings.scala
new file mode 100644
index 0000000000..20191833ad
--- /dev/null
+++
b/stream/src/main/scala/org/apache/pekko/stream/ZstdDecompressionSettings.scala
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream
+
+import org.apache.pekko
+import pekko.annotation.{ ApiMayChange, InternalApi }
+import pekko.stream.impl.io.compression.ZstdDictionaryImpl
+import pekko.stream.scaladsl.Compression
+
+/**
+ * Settings for the zstd decompression operators, `Compression.zstdDecompress`
in the Scala and Java DSLs.
+ *
+ * @since 2.0.0
+ */
+@ApiMayChange
+final class ZstdDecompressionSettings private (
+ val maxBytesPerChunk: Int,
+ dictionary: Option[ZstdDictionaryImpl]) {
+
+ require(maxBytesPerChunk > 0, s"maxBytesPerChunk must be greater than 0, was
$maxBytesPerChunk")
+
+ /**
+ * The maximum length of the emitted `ByteString`s. Defaults to
`Compression.ZstdDecompressMaxBlockSize`.
+ */
+ def withMaxBytesPerChunk(maxBytesPerChunk: Int): ZstdDecompressionSettings =
+ copy(maxBytesPerChunk = maxBytesPerChunk)
+
+ /**
+ * Scala API: decompress with a dictionary, which must be the one used for
compression. Its level is ignored.
+ */
+ def withDictionary(dictionary: scaladsl.compression.ZstdDictionary):
ZstdDecompressionSettings =
+ copy(dictionary = Some(dictionary.toImpl))
+
+ /**
+ * Java API: decompress with a dictionary, which must be the one used for
compression. Its level is ignored.
+ */
+ def withDictionary(dictionary: javadsl.compression.ZstdDictionary):
ZstdDecompressionSettings =
+ copy(dictionary = Some(dictionary.toImpl))
+
+ /**
+ * Decompress without a dictionary, the default.
+ */
+ def withoutDictionary: ZstdDecompressionSettings = copy(dictionary = None)
+
+ /**
+ * INTERNAL API
+ */
+ @InternalApi
+ private[pekko] def dictionaryImpl: Option[ZstdDictionaryImpl] = dictionary
+
+ override def toString: String =
+
s"ZstdDecompressionSettings(maxBytesPerChunk=$maxBytesPerChunk,dictionary=${dictionary.isDefined})"
+
+ private def copy(
+ maxBytesPerChunk: Int = maxBytesPerChunk,
+ dictionary: Option[ZstdDictionaryImpl] = dictionary):
ZstdDecompressionSettings =
+ new ZstdDecompressionSettings(maxBytesPerChunk, dictionary)
+}
+
+/**
+ * @since 2.0.0
+ */
+@ApiMayChange
+object ZstdDecompressionSettings {
+
+ private val Default = new
ZstdDecompressionSettings(Compression.ZstdDecompressMaxBlockSize, None)
+
+ /**
+ * Scala API: the default settings, chunks of at most
`Compression.ZstdDecompressMaxBlockSize` bytes and no dictionary
+ */
+ def apply(): ZstdDecompressionSettings = Default
+
+ /**
+ * Java API: the default settings, chunks of at most
`Compression.ZstdDecompressMaxBlockSize` bytes and no dictionary
+ */
+ def create(): ZstdDecompressionSettings = Default
+}
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/impl/io/compression/ZstdBuffers.scala
b/stream/src/main/scala/org/apache/pekko/stream/impl/io/compression/ZstdBuffers.scala
new file mode 100644
index 0000000000..a5634ff233
--- /dev/null
+++
b/stream/src/main/scala/org/apache/pekko/stream/impl/io/compression/ZstdBuffers.scala
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.impl.io.compression
+
+import java.nio.ByteBuffer
+
+import org.apache.pekko
+import pekko.annotation.InternalApi
+import pekko.util.ByteString
+
+/**
+ * INTERNAL API
+ *
+ * zstd-jni accepts array-backed heap ByteBuffers for streaming
(de)compression, these helpers
+ * move data between ByteStrings and such buffers while avoiding array copies
where possible.
+ */
+@InternalApi private[compression] object ZstdBuffers {
+
+ /**
+ * The fragments of `input` as heap buffers that wrap the backing arrays
without copying, so
+ * that a multi-fragment ByteString does not need to be compacted. zstd only
reads from them.
+ */
+ def inputBuffers(input: ByteString): Iterator[ByteBuffer] =
input.asByteBuffersUnsafe.iterator
+
+ /**
+ * Takes the bytes written to `output` (from 0 until its position) as a
ByteString. When the
+ * buffer is nearly full (at least 7/8 used) its array is handed over
without copying and a new
+ * buffer is returned for further output. Otherwise the bytes are copied
into an exact-size
+ * array, so that emitted ByteStrings do not retain much unused space, and
the buffer is reused.
+ */
+ def takeOutput(output: ByteBuffer): (ByteString, ByteBuffer) = {
+ val produced = output.position()
+ val capacity = output.capacity()
+ if (produced >= capacity - (capacity >> 3))
+ (ByteString.fromArrayUnsafe(output.array(), 0, produced),
ByteBuffer.allocate(capacity))
+ else {
+ output.clear()
+ (ByteString.fromArray(output.array(), 0, produced), output)
+ }
+ }
+}
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/impl/io/compression/ZstdCompressor.scala
b/stream/src/main/scala/org/apache/pekko/stream/impl/io/compression/ZstdCompressor.scala
new file mode 100644
index 0000000000..8e6b9234e4
--- /dev/null
+++
b/stream/src/main/scala/org/apache/pekko/stream/impl/io/compression/ZstdCompressor.scala
@@ -0,0 +1,142 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.impl.io.compression
+
+import java.nio.ByteBuffer
+
+import scala.util.control.NonFatal
+
+import com.github.luben.zstd.{ EndDirective, ZstdCompressCtx, ZstdDictCompress
}
+
+import org.apache.pekko
+import pekko.annotation.InternalApi
+import pekko.stream.scaladsl.Compression
+import pekko.util.{ ByteString, ByteStringBuilder }
+
+/** INTERNAL API */
+@InternalApi private[pekko] object ZstdCompressor {
+ private val OutputBufferSize = 65536
+ private val EmptyInput = ByteBuffer.allocate(0)
+}
+
+/**
+ * INTERNAL API
+ *
+ * @param windowLog the log2 of the maximum window size, 0 uses the default of
the compression level
+ */
+@InternalApi private[pekko] class ZstdCompressor(
+ compressionLevel: Int = Compression.ZstdDefaultCompressionLevel,
+ dictionary: Option[ZstdDictionaryImpl] = None,
+ windowLog: Int = 0) extends Compressor {
+ import ZstdCompressor._
+
+ private val compressCtx = newCompressCtx()
+
+ // A dictionary with a level is digested into a ZstdDictCompress, which has
to be closed after the
+ // context. Neither has a Cleaner, and the stage cannot close them in
postStop if the compressor is
+ // not created, so close them here if the setup fails.
+ private val compressDict: Option[ZstdDictCompress] =
+ try {
+ compressCtx.setLevel(compressionLevel)
+ if (windowLog != 0) compressCtx.setWindowLog(windowLog)
+ dictionary.flatMap(loadDictionary)
+ } catch {
+ case NonFatal(e) =>
+ compressCtx.close()
+ throw e
+ }
+
+ // zstd-jni supports heap ByteBuffers for streaming compression, so no
direct buffers are needed
+ private var outputBuffer = ByteBuffer.allocate(OutputBufferSize)
+
+ /** For tests, to observe that the context is closed. */
+ private[pekko] def newCompressCtx(): ZstdCompressCtx = new ZstdCompressCtx()
+
+ /** For tests, to observe that the dictionary is closed. */
+ private[pekko] def compressDictionary: Option[ZstdDictCompress] =
compressDict
+
+ private def loadDictionary(dict: ZstdDictionaryImpl):
Option[ZstdDictCompress] =
+ (dict.level, dict.length, dict.offset) match {
+ case (None, None, None) =>
+ compressCtx.loadDict(dict.dictionary)
+ None
+ case (Some(dictLevel), None, None) =>
+ Some(loadDictCompress(new ZstdDictCompress(dict.dictionary,
dictLevel)))
+ case (Some(dictLevel), Some(dictLength), Some(dictOffset)) =>
+ ZstdDictionaryImpl.requireSlice(dict.dictionary, dictOffset,
dictLength)
+ Some(loadDictCompress(new ZstdDictCompress(dict.dictionary,
dictOffset, dictLength, dictLevel)))
+ case (None, Some(dictLength), Some(dictOffset)) =>
+ // the context can only load a whole array, so load a copy of the slice
+ ZstdDictionaryImpl.requireSlice(dict.dictionary, dictOffset,
dictLength)
+ compressCtx.loadDict(java.util.Arrays.copyOfRange(dict.dictionary,
dictOffset, dictOffset + dictLength))
+ None
+ case _ =>
+ throw new IllegalArgumentException("ZstdDictionary offset and length
must be set together")
+ }
+
+ private def loadDictCompress(dictCompress: ZstdDictCompress):
ZstdDictCompress =
+ try {
+ compressCtx.loadDict(dictCompress)
+ dictCompress
+ } catch {
+ case NonFatal(e) =>
+ dictCompress.close()
+ throw e
+ }
+
+ override def compress(input: ByteString): ByteString = compressStream(input,
EndDirective.CONTINUE)
+
+ override def flush(): ByteString = compressStream(ByteString.empty,
EndDirective.FLUSH)
+
+ override def finish(): ByteString = compressStream(ByteString.empty,
EndDirective.END)
+
+ override def compressAndFlush(input: ByteString): ByteString =
compressStream(input, EndDirective.FLUSH)
+
+ override def compressAndFinish(input: ByteString): ByteString =
compressStream(input, EndDirective.END)
+
+ override def close(): Unit =
+ try compressCtx.close()
+ finally compressDict.foreach(_.close())
+
+ private def compressStream(input: ByteString, endOp: EndDirective):
ByteString = {
+ val builder = ByteString.newBuilder
+ val inputBuffers = ZstdBuffers.inputBuffers(input)
+ // only the last fragment of the input gets endOp, the ones before it are
just compressed
+ while (inputBuffers.hasNext) {
+ val inputBuffer = inputBuffers.next()
+ compressBuffer(inputBuffer, if (inputBuffers.hasNext)
EndDirective.CONTINUE else endOp, builder)
+ }
+ if (input.isEmpty && endOp != EndDirective.CONTINUE)
+ compressBuffer(EmptyInput, endOp, builder)
+ builder.result()
+ }
+
+ private def compressBuffer(inputBuffer: ByteBuffer, endOp: EndDirective,
builder: ByteStringBuilder): Unit = {
+ var done = false
+ while (!done) {
+ val flushed = compressCtx.compressByteBufferStream(outputBuffer,
inputBuffer, endOp)
+ if (outputBuffer.position() > 0) {
+ val (output, nextBuffer) = ZstdBuffers.takeOutput(outputBuffer)
+ builder ++= output
+ outputBuffer = nextBuffer
+ }
+ // with CONTINUE zstd may keep data in its internal buffers, otherwise
loop until everything is flushed
+ done = if (endOp == EndDirective.CONTINUE) !inputBuffer.hasRemaining
else flushed
+ }
+ }
+}
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/impl/io/compression/ZstdDecompressor.scala
b/stream/src/main/scala/org/apache/pekko/stream/impl/io/compression/ZstdDecompressor.scala
new file mode 100644
index 0000000000..ecabddbe80
--- /dev/null
+++
b/stream/src/main/scala/org/apache/pekko/stream/impl/io/compression/ZstdDecompressor.scala
@@ -0,0 +1,129 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.impl.io.compression
+
+import java.nio.ByteBuffer
+
+import scala.annotation.tailrec
+import scala.util.control.NonFatal
+
+import com.github.luben.zstd.{ Zstd, ZstdDecompressCtx, ZstdException }
+
+import org.apache.pekko
+import pekko.annotation.InternalApi
+import pekko.stream.Attributes
+import pekko.stream.impl.fusing.GraphStages.SimpleLinearGraphStage
+import pekko.stream.stage.{ GraphStageLogic, InHandler, OutHandler }
+import pekko.util.ByteString
+
+/** INTERNAL API */
+@InternalApi private[pekko] object ZstdDecompressor {
+ private val EmptyInput = ByteBuffer.allocate(0)
+ private val NoInput: Iterator[ByteBuffer] = Iterator.empty
+
+ // The level only applies to compression. Each materialization loads the
dictionary into its own
+ // context, which owns the digested dictionary and frees it on close. A
shared ZstdDictDecompress
+ // could not be closed when one materialization stops, and has no Cleaner,
so it would only be freed
+ // by finalization. The bytes are copied so that later changes to the
caller's array have no effect.
+ private def dictionaryBytes(dictionary: ZstdDictionaryImpl): Array[Byte] = {
+ val bytes = dictionary.dictionary
+ (dictionary.offset, dictionary.length) match {
+ case (None, None) => bytes.clone()
+ case (Some(offset), Some(length)) =>
+ ZstdDictionaryImpl.requireSlice(bytes, offset, length)
+ java.util.Arrays.copyOfRange(bytes, offset, offset + length)
+ case _ =>
+ throw new IllegalArgumentException("ZstdDictionary offset and length
must be set together")
+ }
+ }
+}
+
+/** INTERNAL API */
+@InternalApi private[pekko] class ZstdDecompressor(maxBytesPerChunk: Int,
dictionary: Option[ZstdDictionaryImpl])
+ extends SimpleLinearGraphStage[ByteString] {
+ import ZstdDecompressor._
+
+ require(maxBytesPerChunk > 0, "maxBytesPerChunk must be greater than 0")
+
+ private val decompressDict = dictionary.map(dictionaryBytes)
+
+ override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
{
+ new GraphStageLogic(shape) with InHandler with OutHandler {
+
+ private val decompressCtx = new ZstdDecompressCtx()
+ // postStop is not called if creating the logic fails, so close the
context here
+ try decompressDict.foreach(decompressCtx.loadDict)
+ catch {
+ case NonFatal(e) =>
+ decompressCtx.close()
+ throw e
+ }
+
+ // zstd-jni supports heap ByteBuffers for streaming decompression, so no
direct buffers are needed
+ private var outputBuffer = ByteBuffer.allocate(maxBytesPerChunk)
+ private var inputBuffer = EmptyInput
+ // the remaining fragments of the current element, see
ZstdBuffers.inputBuffers
+ private var inputBuffers = NoInput
+ // set when the last call filled the output buffer before the frame
completed, zstd may then
+ // still hold decompressed data. A completed frame is fully flushed, and
another call would make
+ // zstd wait for a new frame and report the stream as incomplete.
+ private var outputPending = false
+ // false while a frame has been started but not fully decompressed and
flushed
+ private var frameComplete = true
+
+ override def onPush(): Unit = {
+ inputBuffers = ZstdBuffers.inputBuffers(grab(in))
+ decompress()
+ }
+
+ override def onPull(): Unit = decompress()
+
+ override def onUpstreamFinish(): Unit =
+ if (!hasInput && !outputPending) finish()
+ else if (isAvailable(out)) decompress()
+ // otherwise the remaining input is decompressed on the next onPull
+
+ private def hasInput: Boolean = {
+ while (!inputBuffer.hasRemaining && inputBuffers.hasNext) inputBuffer
= inputBuffers.next()
+ inputBuffer.hasRemaining
+ }
+
+ @tailrec
+ private def decompress(): Unit =
+ if (hasInput || outputPending) {
+ frameComplete =
decompressCtx.decompressByteBufferStream(outputBuffer, inputBuffer)
+ outputPending = !frameComplete && !outputBuffer.hasRemaining
+ if (outputBuffer.position() > 0) {
+ val (output, nextBuffer) = ZstdBuffers.takeOutput(outputBuffer)
+ outputBuffer = nextBuffer
+ push(out, output)
+ } else decompress()
+ } else if (isClosed(in)) finish()
+ else pull(in)
+
+ private def finish(): Unit =
+ if (frameComplete) completeStage()
+ else failStage(new ZstdException(Zstd.errSrcSizeWrong(), "Truncated
zstd stream"))
+
+ override def postStop(): Unit = decompressCtx.close()
+
+ setHandlers(in, out, this)
+ }
+ }
+
+}
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/impl/io/compression/ZstdDictionaryImpl.scala
b/stream/src/main/scala/org/apache/pekko/stream/impl/io/compression/ZstdDictionaryImpl.scala
new file mode 100644
index 0000000000..0415a9f7ef
--- /dev/null
+++
b/stream/src/main/scala/org/apache/pekko/stream/impl/io/compression/ZstdDictionaryImpl.scala
@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.impl.io.compression
+
+import org.apache.pekko.annotation.{ InternalApi, InternalStableApi }
+
+@InternalStableApi
+private[pekko] abstract class ZstdDictionaryImpl {
+ def dictionary: Array[Byte]
+ def level: Option[Int]
+ def offset: Option[Int]
+ def length: Option[Int]
+}
+
+/** INTERNAL API */
+@InternalApi private[pekko] object ZstdDictionaryImpl {
+ def requireSlice(bytes: Array[Byte], offset: Int, length: Int): Unit =
+ require(
+ offset >= 0 && length >= 0 && offset <= bytes.length - length,
+ s"ZstdDictionary offset [$offset] and length [$length] must be within
the dictionary of " +
+ s"[${bytes.length}] bytes")
+}
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Compression.scala
b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Compression.scala
index f143528da0..a2d82858b7 100644
--- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Compression.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Compression.scala
@@ -13,8 +13,15 @@
package org.apache.pekko.stream.javadsl
+import java.util.Optional
+
+import scala.jdk.OptionConverters.RichOptional
+
import org.apache.pekko
import pekko.NotUsed
+import pekko.annotation.ApiMayChange
+import pekko.stream.{ ZstdCompressionSettings, ZstdDecompressionSettings }
+import pekko.stream.javadsl.compression.ZstdDictionary
import pekko.stream.scaladsl
import pekko.util.ByteString
@@ -104,4 +111,101 @@ object Compression {
def deflate(level: Int, nowrap: Boolean, autoFlush: Boolean):
Flow[ByteString, ByteString, NotUsed] =
scaladsl.Compression.deflate(level, nowrap, autoFlush).asJava
+ // The zstd constants delegate to scaladsl.Compression, which explains why
they are copies of
+ // libzstd's values rather than calls to the optional zstd-jni dependency.
+
+ /**
+ * The minimum compression level supported by zstd
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ final val ZstdMinCompressionLevel: Int =
scaladsl.Compression.ZstdMinCompressionLevel
+
+ /**
+ * The maximum compression level supported by zstd
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ final val ZstdMaxCompressionLevel: Int =
scaladsl.Compression.ZstdMaxCompressionLevel
+
+ /**
+ * The zstd default compression level
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ final val ZstdDefaultCompressionLevel: Int =
scaladsl.Compression.ZstdDefaultCompressionLevel
+
+ /**
+ * The minimum window log supported by zstd compression
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ final val ZstdMinWindowLog: Int = scaladsl.Compression.ZstdMinWindowLog
+
+ /**
+ * The maximum window log supported by zstd compression on 64-bit platforms
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ final val ZstdMaxWindowLog: Int = scaladsl.Compression.ZstdMaxWindowLog
+
+ /**
+ * Creates a flow that zstd-compresses a stream of ByteStrings, with the
default compression level.
+ *
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def zstd: Flow[ByteString, ByteString, NotUsed] =
+ scaladsl.Compression.zstd.asJava
+
+ /**
+ * Same as [[zstd]] with a custom level and an optional dictionary.
+ *
+ * @param level The compression level, must be greater or equal to
[[ZstdMinCompressionLevel]] and less than or equal
+ * to [[ZstdMaxCompressionLevel]]
+ * @param dictionary An optional dictionary that can be used for compression
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def zstd(level: Int, dictionary: Optional[ZstdDictionary]): Flow[ByteString,
ByteString, NotUsed] = {
+ val settings = ZstdCompressionSettings.create().withLevel(level)
+ zstd(dictionary.toScala.fold(settings)(settings.withDictionary))
+ }
+
+ /**
+ * Same as [[zstd]] with the given settings, see
[[pekko.stream.ZstdCompressionSettings]] for the level, flush
+ * mode, window size and dictionary.
+ *
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def zstd(settings: ZstdCompressionSettings): Flow[ByteString, ByteString,
NotUsed] =
+ scaladsl.Compression.zstd(settings).asJava
+
+ /**
+ * The maximum block size used by zstd decompression
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ final val ZstdDecompressMaxBlockSize: Int =
scaladsl.Compression.ZstdDecompressMaxBlockSize
+
+ /**
+ * Creates a flow that zstd-decompresses a stream of ByteStrings.
+ *
+ * @param maxBytesPerChunk The maximum length of the emitted ByteStrings
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def zstdDecompress(maxBytesPerChunk: Int): Flow[ByteString, ByteString,
NotUsed] =
+ scaladsl.Compression.zstdDecompress(maxBytesPerChunk).asJava
+
+ /**
+ * Same as [[zstdDecompress]] with the given settings, see
[[pekko.stream.ZstdDecompressionSettings]] for the chunk
+ * size and dictionary.
+ *
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def zstdDecompress(settings: ZstdDecompressionSettings): Flow[ByteString,
ByteString, NotUsed] =
+ scaladsl.Compression.zstdDecompress(settings).asJava
}
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Compression.scala
b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Compression.scala
index 967ff09a87..d1de529b2f 100644
--- a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Compression.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Compression.scala
@@ -17,7 +17,10 @@ import java.util.zip.Deflater
import org.apache.pekko
import pekko.NotUsed
+import pekko.annotation.ApiMayChange
+import pekko.stream.{ ZstdCompressionSettings, ZstdDecompressionSettings }
import pekko.stream.impl.io.compression._
+import pekko.stream.scaladsl.compression.ZstdDictionary
import pekko.util.ByteString
object Compression {
@@ -104,4 +107,115 @@ object Compression {
*/
def inflate(maxBytesPerChunk: Int, nowrap: Boolean): Flow[ByteString,
ByteString, NotUsed] =
Flow[ByteString].via(new DeflateDecompressor(maxBytesPerChunk,
nowrap)).named("inflate")
+
+ // The zstd constants below are copies of libzstd's values instead of calls
to zstd-jni (e.g.
+ // Zstd.minCompressionLevel()). zstd-jni is an optional dependency: calling
it here would make
+ // initializing this object, and so using gzip or deflate, fail with
NoClassDefFoundError when
+ // zstd-jni is not on the classpath, and would load the native library even
for those users.
+ // ZstdSpec checks that they still match zstd-jni, so a zstd-jni upgrade
that changes one of
+ // them fails the tests.
+
+ /**
+ * The minimum compression level supported by zstd
+ * @since 2.0.0
+ */
+ // ZSTD_minCLevel() is -ZSTD_TARGETLENGTH_MAX, which is ZSTD_BLOCKSIZE_MAX
(1 << 17) in libzstd
+ @ApiMayChange
+ final val ZstdMinCompressionLevel: Int = -(1 << 17)
+
+ /**
+ * The maximum compression level supported by zstd
+ * @since 2.0.0
+ */
+ // ZSTD_maxCLevel() is ZSTD_MAX_CLEVEL in libzstd
+ @ApiMayChange
+ final val ZstdMaxCompressionLevel: Int = 22
+
+ /**
+ * The zstd default compression level
+ * @since 2.0.0
+ */
+ // ZSTD_defaultCLevel() is ZSTD_CLEVEL_DEFAULT in libzstd
+ @ApiMayChange
+ final val ZstdDefaultCompressionLevel: Int = 3
+
+ /**
+ * The minimum window log supported by zstd compression
+ * @since 2.0.0
+ */
+ // ZSTD_WINDOWLOG_MIN in libzstd
+ @ApiMayChange
+ final val ZstdMinWindowLog: Int = 10
+
+ /**
+ * The maximum window log supported by zstd compression on 64-bit platforms
+ * @since 2.0.0
+ */
+ // ZSTD_WINDOWLOG_MAX_64 in libzstd
+ @ApiMayChange
+ final val ZstdMaxWindowLog: Int = 31
+
+ /**
+ * Creates a flow that zstd-compresses a stream of ByteStrings, with the
default compression level.
+ *
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def zstd: Flow[ByteString, ByteString, NotUsed] =
zstd(ZstdCompressionSettings())
+
+ /**
+ * Same as [[zstd]] with a custom level and an optional dictionary.
+ *
+ * @param level The compression level, must be greater or equal to
[[ZstdMinCompressionLevel]] and less than or equal
+ * to [[ZstdMaxCompressionLevel]]
+ * @param dictionary An optional dictionary that can be used for compression
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def zstd(level: Int, dictionary: Option[ZstdDictionary]): Flow[ByteString,
ByteString, NotUsed] = {
+ val settings = ZstdCompressionSettings().withLevel(level)
+ zstd(dictionary.fold(settings)(settings.withDictionary))
+ }
+
+ /**
+ * Same as [[zstd]] with the given settings, see
[[pekko.stream.ZstdCompressionSettings]] for the level, flush
+ * mode, window size and dictionary.
+ *
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def zstd(settings: ZstdCompressionSettings): Flow[ByteString, ByteString,
NotUsed] =
+ CompressionUtils.compressorFlow(
+ () => new ZstdCompressor(settings.level, settings.dictionaryImpl,
settings.windowLog),
+ settings.autoFlush)
+
+ /**
+ * The maximum block size used by zstd decompression
+ * @since 2.0.0
+ */
+ // ZSTD_BLOCKSIZE_MAX in libzstd is 1 << ZSTD_BLOCKSIZELOG_MAX (17), i.e.
128 KiB
+ @ApiMayChange
+ final val ZstdDecompressMaxBlockSize: Int = 1 << 17
+
+ /**
+ * Creates a flow that zstd-decompresses a stream of ByteStrings.
+ *
+ * @param maxBytesPerChunk The maximum length of the emitted ByteStrings
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def zstdDecompress(maxBytesPerChunk: Int = ZstdDecompressMaxBlockSize):
Flow[ByteString, ByteString, NotUsed] =
+
zstdDecompress(ZstdDecompressionSettings().withMaxBytesPerChunk(maxBytesPerChunk))
+
+ /**
+ * Same as [[zstdDecompress]] with the given settings, see
[[pekko.stream.ZstdDecompressionSettings]] for the chunk
+ * size and dictionary.
+ *
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def zstdDecompress(settings: ZstdDecompressionSettings): Flow[ByteString,
ByteString, NotUsed] =
+ Flow[ByteString]
+ .via(new ZstdDecompressor(settings.maxBytesPerChunk,
settings.dictionaryImpl))
+ .named("zstdDecompress")
}
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/compression/ZstdDictionary.scala
b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/compression/ZstdDictionary.scala
new file mode 100644
index 0000000000..2899c87802
--- /dev/null
+++
b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/compression/ZstdDictionary.scala
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.scaladsl.compression
+
+import java.nio.charset.StandardCharsets
+
+import org.apache.pekko
+import pekko.annotation.{ ApiMayChange, InternalApi }
+import pekko.stream.impl.io.compression.ZstdDictionaryImpl
+import pekko.util.ByteString
+
+/**
+ * Configuration class based on the official C zstd library to specify
dictionary options for
+ * compression.
+ *
+ * @see
https://github.com/facebook/zstd?tab=readme-ov-file#dictionary-compression-how-to
+ * @see https://facebook.github.io/zstd/zstd_manual.html#Chapter10
+ * @since 2.0.0
+ */
+@ApiMayChange
+final case class ZstdDictionary private (dictionary: ByteString, level:
Option[Int],
+ offset: Option[Int], length: Option[Int]) { clazz =>
+
+ /**
+ * INTERNAL API
+ */
+ @InternalApi
+ private[pekko] def toImpl: ZstdDictionaryImpl = new ZstdDictionaryImpl {
+ // a copy, so that whoever holds the impl cannot modify the dictionary
+ override val dictionary: Array[Byte] = clazz.dictionary.toArray
+ override val level: Option[Int] = clazz.level
+ override val offset: Option[Int] = clazz.offset
+ override val length: Option[Int] = clazz.length
+ }
+}
+
+/**
+ * @since 2.0.0
+ */
+@ApiMayChange
+object ZstdDictionary {
+
+ // the array overloads copy the array, so later changes to it do not affect
the dictionary
+
+ def apply(dictionary: Array[Byte]): ZstdDictionary =
+ ZstdDictionary(ByteString(dictionary))
+
+ def apply(dictionary: Array[Byte], level: Int): ZstdDictionary =
+ ZstdDictionary(ByteString(dictionary), level)
+
+ def apply(dictionary: Array[Byte], level: Int, offset: Int, length: Int):
ZstdDictionary =
+ ZstdDictionary(ByteString(dictionary), level, offset, length)
+
+ def apply(dictionary: ByteString): ZstdDictionary =
+ ZstdDictionary(dictionary, None, None, None)
+
+ def apply(dictionary: ByteString, level: Int): ZstdDictionary =
+ ZstdDictionary(dictionary, Some(level), None, None)
+
+ def apply(dictionary: ByteString, level: Int, offset: Int, length: Int):
ZstdDictionary =
+ ZstdDictionary(dictionary, Some(level), Some(offset), Some(length))
+
+ def apply(dictionary: String): ZstdDictionary =
+ ZstdDictionary(ByteString(dictionary, StandardCharsets.UTF_8))
+
+ def apply(dictionary: String, level: Int): ZstdDictionary =
+ ZstdDictionary(ByteString(dictionary, StandardCharsets.UTF_8), level)
+
+ def apply(dictionary: String, level: Int, offset: Int, length: Int):
ZstdDictionary =
+ ZstdDictionary(ByteString(dictionary, StandardCharsets.UTF_8), level,
offset, length)
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
