This is an automated email from the ASF dual-hosted git repository. He-Pin pushed a commit to branch refactor/transferTo in repository https://gitbox.apache.org/repos/asf/pekko.git
commit d7471fe0742f12702e2674d2fe90fa0f562af390 Author: θιΈ£ <[email protected]> AuthorDate: Tue Jun 23 11:49:41 2026 +0800 refactor: replace manual stream copy loops with transferTo/readAllBytes Motivation: Seven GZIP decompress methods and one streamToBytes utility use manual @tailrec read loops with buffer + read() + write() to copy streams. JDK 9 provides InputStream.transferTo() and readAllBytes() which do the same thing internally. Modification: - Replace 7 identical readChunk() tail-recursive loops with in.transferTo(out) in decompress methods across cluster, cluster-sharding, cluster-tools, cluster-metrics, distributed-data, serialization-jackson, serialization-jackson3 modules. - Replace persistence streamToBytes manual loop with inputStream.readAllBytes(). - Remove unused @tailrec imports from 5 files. - Remove unused ByteArrayOutputStream import from package.scala. Result: Net reduction of ~79 lines. Functionally identical β transferTo uses an internal 8192-byte buffer (vs original 4096) which only affects chunking granularity, not correctness. Tests: sbt "distributed-data/compile" "cluster/compile" "cluster-tools/compile" "cluster-sharding/compile" "cluster-metrics/compile" "serialization-jackson/compile" "serialization-jackson3/compile" "persistence/compile" β all passed References: Refs #3136 --- .../cluster/metrics/protobuf/MessageSerializer.scala | 12 +----------- .../protobuf/ClusterShardingMessageSerializer.scala | 12 +----------- .../protobuf/DistributedPubSubMessageSerializer.scala | 12 +----------- .../cluster/protobuf/ClusterMessageSerializer.scala | 12 +----------- .../cluster/ddata/protobuf/SerializationSupport.scala | 12 +----------- .../pekko/persistence/serialization/package.scala | 19 +++---------------- .../serialization/jackson/JacksonSerializer.scala | 11 +---------- .../serialization/jackson3/JacksonSerializer.scala | 11 +---------- 8 files changed, 10 insertions(+), 91 deletions(-) diff --git a/cluster-metrics/src/main/scala/org/apache/pekko/cluster/metrics/protobuf/MessageSerializer.scala b/cluster-metrics/src/main/scala/org/apache/pekko/cluster/metrics/protobuf/MessageSerializer.scala index 434f61aa7a..c0baa709d0 100644 --- a/cluster-metrics/src/main/scala/org/apache/pekko/cluster/metrics/protobuf/MessageSerializer.scala +++ b/cluster-metrics/src/main/scala/org/apache/pekko/cluster/metrics/protobuf/MessageSerializer.scala @@ -17,7 +17,6 @@ import java.{ lang => jl } import java.io.{ ByteArrayOutputStream, NotSerializableException, ObjectOutputStream } import java.util.zip.{ GZIPInputStream, GZIPOutputStream } -import scala.annotation.tailrec import scala.collection.immutable import scala.jdk.CollectionConverters._ @@ -80,16 +79,7 @@ class MessageSerializer(val system: ExtendedActorSystem) extends SerializerWithS def decompress(bytes: Array[Byte]): Array[Byte] = { val in = new GZIPInputStream(new UnsynchronizedByteArrayInputStream(bytes)) val out = new ByteArrayOutputStream() - val buffer = new Array[Byte](BufferSize) - - @tailrec def readChunk(): Unit = in.read(buffer) match { - case -1 => () - case n => - out.write(buffer, 0, n) - readChunk() - } - - try readChunk() + try in.transferTo(out) finally in.close() out.toByteArray } diff --git a/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/protobuf/ClusterShardingMessageSerializer.scala b/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/protobuf/ClusterShardingMessageSerializer.scala index 2afa4e7303..01254975a6 100644 --- a/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/protobuf/ClusterShardingMessageSerializer.scala +++ b/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/protobuf/ClusterShardingMessageSerializer.scala @@ -18,7 +18,6 @@ import java.io.NotSerializableException import java.util.zip.GZIPInputStream import java.util.zip.GZIPOutputStream -import scala.annotation.tailrec import scala.collection.immutable import scala.concurrent.duration._ import scala.jdk.CollectionConverters._ @@ -644,16 +643,7 @@ private[pekko] class ClusterShardingMessageSerializer(val system: ExtendedActorS private def decompress(bytes: Array[Byte]): Array[Byte] = { val in = new GZIPInputStream(new UnsynchronizedByteArrayInputStream(bytes)) val out = new ByteArrayOutputStream() - val buffer = new Array[Byte](BufferSize) - - @tailrec def readChunk(): Unit = in.read(buffer) match { - case -1 => () - case n => - out.write(buffer, 0, n) - readChunk() - } - - try readChunk() + try in.transferTo(out) finally in.close() out.toByteArray } diff --git a/cluster-tools/src/main/scala/org/apache/pekko/cluster/pubsub/protobuf/DistributedPubSubMessageSerializer.scala b/cluster-tools/src/main/scala/org/apache/pekko/cluster/pubsub/protobuf/DistributedPubSubMessageSerializer.scala index b4a3a1a14e..b578c1f31c 100644 --- a/cluster-tools/src/main/scala/org/apache/pekko/cluster/pubsub/protobuf/DistributedPubSubMessageSerializer.scala +++ b/cluster-tools/src/main/scala/org/apache/pekko/cluster/pubsub/protobuf/DistributedPubSubMessageSerializer.scala @@ -18,7 +18,6 @@ import java.io.NotSerializableException import java.util.zip.GZIPInputStream import java.util.zip.GZIPOutputStream -import scala.annotation.tailrec import scala.collection.immutable.TreeMap import scala.jdk.CollectionConverters._ @@ -100,16 +99,7 @@ private[pekko] class DistributedPubSubMessageSerializer(val system: ExtendedActo private def decompress(bytes: Array[Byte]): Array[Byte] = { val in = new GZIPInputStream(new UnsynchronizedByteArrayInputStream(bytes)) val out = new ByteArrayOutputStream() - val buffer = new Array[Byte](BufferSize) - - @tailrec def readChunk(): Unit = in.read(buffer) match { - case -1 => () - case n => - out.write(buffer, 0, n) - readChunk() - } - - try readChunk() + try in.transferTo(out) finally in.close() out.toByteArray } diff --git a/cluster/src/main/scala/org/apache/pekko/cluster/protobuf/ClusterMessageSerializer.scala b/cluster/src/main/scala/org/apache/pekko/cluster/protobuf/ClusterMessageSerializer.scala index 13cf1a09e0..d25e4d620c 100644 --- a/cluster/src/main/scala/org/apache/pekko/cluster/protobuf/ClusterMessageSerializer.scala +++ b/cluster/src/main/scala/org/apache/pekko/cluster/protobuf/ClusterMessageSerializer.scala @@ -16,7 +16,6 @@ package org.apache.pekko.cluster.protobuf import java.io.ByteArrayOutputStream import java.util.zip.{ GZIPInputStream, GZIPOutputStream } -import scala.annotation.tailrec import scala.collection.immutable import scala.concurrent.duration.Deadline import scala.jdk.CollectionConverters._ @@ -168,16 +167,7 @@ final class ClusterMessageSerializer(val system: ExtendedActorSystem) def decompress(bytes: Array[Byte]): Array[Byte] = { val in = new GZIPInputStream(new UnsynchronizedByteArrayInputStream(bytes)) val out = new ByteArrayOutputStream() - val buffer = new Array[Byte](BufferSize) - - @tailrec def readChunk(): Unit = in.read(buffer) match { - case -1 => () - case n => - out.write(buffer, 0, n) - readChunk() - } - - try readChunk() + try in.transferTo(out) finally in.close() out.toByteArray } diff --git a/distributed-data/src/main/scala/org/apache/pekko/cluster/ddata/protobuf/SerializationSupport.scala b/distributed-data/src/main/scala/org/apache/pekko/cluster/ddata/protobuf/SerializationSupport.scala index c526e0af0a..b14cf6eb2d 100644 --- a/distributed-data/src/main/scala/org/apache/pekko/cluster/ddata/protobuf/SerializationSupport.scala +++ b/distributed-data/src/main/scala/org/apache/pekko/cluster/ddata/protobuf/SerializationSupport.scala @@ -17,7 +17,6 @@ import java.io.ByteArrayOutputStream import java.util.zip.GZIPInputStream import java.util.zip.GZIPOutputStream -import scala.annotation.tailrec import scala.collection.immutable.TreeMap import scala.jdk.CollectionConverters._ @@ -78,16 +77,7 @@ trait SerializationSupport { def decompress(bytes: Array[Byte]): Array[Byte] = { val in = new GZIPInputStream(new UnsynchronizedByteArrayInputStream(bytes)) val out = new ByteArrayOutputStream() - val buffer = new Array[Byte](BufferSize) - - @tailrec def readChunk(): Unit = in.read(buffer) match { - case -1 => () - case n => - out.write(buffer, 0, n) - readChunk() - } - - try readChunk() + try in.transferTo(out) finally in.close() out.toByteArray } diff --git a/persistence/src/main/scala/org/apache/pekko/persistence/serialization/package.scala b/persistence/src/main/scala/org/apache/pekko/persistence/serialization/package.scala index 650ca27c74..72c8aaa6c0 100644 --- a/persistence/src/main/scala/org/apache/pekko/persistence/serialization/package.scala +++ b/persistence/src/main/scala/org/apache/pekko/persistence/serialization/package.scala @@ -13,26 +13,13 @@ package org.apache.pekko.persistence -import java.io.{ ByteArrayOutputStream, InputStream } +import java.io.InputStream package object serialization { /** * Converts an input stream to a byte array. */ - def streamToBytes(inputStream: InputStream): Array[Byte] = { - val len = 16384 - val buf = new Array[Byte](len) - val out = new ByteArrayOutputStream - - @scala.annotation.tailrec - def copy(): Array[Byte] = { - val n = inputStream.read(buf, 0, len) - if (n != -1) { - out.write(buf, 0, n); copy() - } else out.toByteArray - } - - copy() - } + def streamToBytes(inputStream: InputStream): Array[Byte] = + inputStream.readAllBytes() } diff --git a/serialization-jackson/src/main/scala/org/apache/pekko/serialization/jackson/JacksonSerializer.scala b/serialization-jackson/src/main/scala/org/apache/pekko/serialization/jackson/JacksonSerializer.scala index 7c3aac2030..f749005a2f 100644 --- a/serialization-jackson/src/main/scala/org/apache/pekko/serialization/jackson/JacksonSerializer.scala +++ b/serialization-jackson/src/main/scala/org/apache/pekko/serialization/jackson/JacksonSerializer.scala @@ -536,16 +536,7 @@ import pekko.util.OptionVal if (isGZipped(bytes)) { val in = new GZIPInputStream(new UnsynchronizedByteArrayInputStream(bytes)) val out = new ByteArrayOutputStream() - val buffer = new Array[Byte](BufferSize) - - @tailrec def readChunk(): Unit = in.read(buffer) match { - case -1 => () - case n => - out.write(buffer, 0, n) - readChunk() - } - - try readChunk() + try in.transferTo(out) finally in.close() out.toByteArray } else { diff --git a/serialization-jackson3/src/main/scala/org/apache/pekko/serialization/jackson3/JacksonSerializer.scala b/serialization-jackson3/src/main/scala/org/apache/pekko/serialization/jackson3/JacksonSerializer.scala index 3ca46a78d6..43a5633be5 100644 --- a/serialization-jackson3/src/main/scala/org/apache/pekko/serialization/jackson3/JacksonSerializer.scala +++ b/serialization-jackson3/src/main/scala/org/apache/pekko/serialization/jackson3/JacksonSerializer.scala @@ -537,16 +537,7 @@ import pekko.util.OptionVal if (isGZipped(bytes)) { val in = new GZIPInputStream(new UnsynchronizedByteArrayInputStream(bytes)) val out = new ByteArrayOutputStream() - val buffer = new Array[Byte](BufferSize) - - @tailrec def readChunk(): Unit = in.read(buffer) match { - case -1 => () - case n => - out.write(buffer, 0, n) - readChunk() - } - - try readChunk() + try in.transferTo(out) finally in.close() out.toByteArray } else { --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
