You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pekko.apache.org by he...@apache.org on 2026/06/23 03:49:56 UTC
(pekko) branch refactor/transferTo created (now d7471fe074)
This is an automated email from the ASF dual-hosted git repository.
He-Pin pushed a change to branch refactor/transferTo
in repository https://gitbox.apache.org/repos/asf/pekko.git
at d7471fe074 refactor: replace manual stream copy loops with transferTo/readAllBytes
This branch includes the following new commits:
new d7471fe074 refactor: replace manual stream copy loops with transferTo/readAllBytes
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org
(pekko) 01/01: refactor: replace manual stream copy loops with transferTo/readAllBytes
Posted by he...@apache.org.
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: θιΈ£ <he...@alibaba-inc.com>
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: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org