This is an automated email from the ASF dual-hosted git repository.
He-Pin 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 f60702a4de refactor: replace manual stream copy loops with
transferTo/readAllBytes (#3146)
f60702a4de is described below
commit f60702a4debd6fc9f0a4ded5a2502939ecc10c65
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Tue Jun 23 17:17:38 2026 +0800
refactor: replace manual stream copy loops with transferTo/readAllBytes
(#3146)
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]
