add asInputStream to ByteString (#1085)

* add getInputStream to ByteString * scala 2 updates * rename as asInputStream * make methods final * use different asInputStream implementations on different subclasses Co-Authored-By: João Ferreira <[email protected]> * Update ByteBufferBackedInputStream.scala scalafmt * remove default asInputStream impl * add some tests - more needed * Create bytestring-inputstream.excludes * scalafmt * Update bytestring-inputstream.excludes * Update ByteStringInputStreamSpec.scala * Update ByteStringInputStreamSpec.scala * Update ByteStringInputStreamSpec.scala * add benchmark * Update ByteString_asInputStream_Benchmark.scala * change to iterator earlier in chain --------- Co-authored-by: João Ferreira <[email protected]>
apache · Feb 5, 2024 · ef628ea · ef628ea
1 parent e26c6f6
commit ef628ea
Show file tree

Hide file tree

Showing 7 changed files with 256 additions and 13 deletions.
diff --git a/actor-tests/src/test/scala/org/apache/pekko/util/ByteStringInputStreamSpec.scala b/actor-tests/src/test/scala/org/apache/pekko/util/ByteStringInputStreamSpec.scala
@@ -0,0 +1,81 @@
+/*
+ * 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.util
+
+import java.io.{ ByteArrayOutputStream, InputStream, OutputStream }
+import java.nio.charset.StandardCharsets
+
+import org.apache.pekko
+import pekko.util.ByteString.{ ByteString1, ByteString1C, ByteStrings }
+
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
+
+class ByteStringInputStreamSpec extends AnyWordSpec with Matchers {
+  "ByteString1" must {
+    "support asInputStream" in {
+      ByteString1.empty.asInputStream.read() shouldEqual -1
+      ByteString1.empty.asInputStream.read(Array.empty) shouldEqual -1
+      toUtf8String(ByteString1.empty.asInputStream) shouldEqual ""
+      toUtf8String(ByteString1.fromString("abc").asInputStream) shouldEqual "abc"
+    }
+  }
+  "ByteString1C" must {
+    "support asInputStream" in {
+      toUtf8String(ByteString1C.fromString("").asInputStream) shouldEqual ""
+      toUtf8String(ByteString1C.fromString("abc").asInputStream) shouldEqual "abc"
+      val bytes = "abcdef".getBytes(StandardCharsets.US_ASCII)
+      toUtf8String(ByteString.fromArray(bytes, 1, 3).asInputStream) shouldEqual "bcd"
+    }
+  }
+  "ByteStrings" must {
+    "support asInputStream" in {
+      val empty = ByteStrings(ByteString1.fromString(""), ByteString1.fromString(""))
+      empty.asInputStream.read() shouldEqual -1
+      empty.asInputStream.read(Array.empty) shouldEqual -1
+      toUtf8String(empty.asInputStream) shouldEqual ""
+      val abc = ByteStrings(ByteString1.fromString("a"), ByteString1.fromString("bc"))
+      toUtf8String(abc.asInputStream) shouldEqual "abc"
+    }
+  }
+
+  private def toUtf8String(input: InputStream): String =
+    new String(toByteArray(input), StandardCharsets.UTF_8)
+
+  private def toByteArray(input: InputStream): Array[Byte] = {
+    val output = new ByteArrayOutputStream
+    try {
+      copy(input, output)
+      output.toByteArray
+    } finally {
+      output.close()
+    }
+  }
+
+  private def copy(input: InputStream, output: OutputStream): Int = {
+    val buffer = new Array[Byte](4096)
+    var count = 0
+    var n = input.read(buffer)
+    while (n != -1) {
+      output.write(buffer, 0, n)
+      count += n
+      n = input.read(buffer)
+    }
+    count
+  }
+}
diff --git a/actor/src/main/mima-filters/1.0.x.backwards.excludes/bytestring-inputstream.excludes b/actor/src/main/mima-filters/1.0.x.backwards.excludes/bytestring-inputstream.excludes
@@ -0,0 +1,19 @@
+# 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.
+
+# Add ByteString.asInputStream
+ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.pekko.util.ByteString.asInputStream")
diff --git a/actor/src/main/scala-2.12/org/apache/pekko/util/ByteString.scala b/actor/src/main/scala-2.12/org/apache/pekko/util/ByteString.scala
@@ -13,14 +13,15 @@
 
 package org.apache.pekko.util
 
-import java.io.{ ObjectInputStream, ObjectOutputStream }
+import java.io.{ ByteArrayInputStream, InputStream, ObjectInputStream, ObjectOutputStream, SequenceInputStream }
 import java.lang.{ Iterable => JIterable }
 import java.nio.{ ByteBuffer, ByteOrder }
 import java.nio.charset.{ Charset, StandardCharsets }
 import java.util.Base64
 
 import scala.annotation.{ tailrec, varargs }
 import scala.collection.IndexedSeqOptimized
+import scala.collection.JavaConverters._
 import scala.collection.generic.CanBuildFrom
 import scala.collection.immutable
 import scala.collection.immutable.{ IndexedSeq, VectorBuilder }
@@ -271,6 +272,8 @@ object ByteString {
     }
 
     override def toArrayUnsafe(): Array[Byte] = bytes
+
+    override def asInputStream: InputStream = new ByteArrayInputStream(bytes)
   }
 
   /** INTERNAL API: ByteString backed by exactly one array, with start / end markers */
@@ -436,6 +439,9 @@ object ByteString {
       if (startIndex == 0 && length == bytes.length) bytes
       else toArray
     }
+
+    override def asInputStream: InputStream =
+      new ByteArrayInputStream(bytes, startIndex, length)
   }
 
   private[pekko] object ByteStrings extends Companion {
@@ -566,6 +572,9 @@ object ByteString {
 
     def asByteBuffers: scala.collection.immutable.Iterable[ByteBuffer] = bytestrings.map { _.asByteBuffer }
 
+    override def asInputStream: InputStream =
+      new SequenceInputStream(bytestrings.iterator.map(_.asInputStream).asJavaEnumeration)
+
     def decodeString(charset: String): String = compact.decodeString(charset)
 
     def decodeString(charset: Charset): String = compact.decodeString(charset)
@@ -827,6 +836,15 @@ sealed abstract class ByteString extends IndexedSeq[Byte] with IndexedSeqOptimiz
    */
   def toArrayUnsafe(): Array[Byte] = toArray
 
+  /**
+   * Return the bytes in this ByteString as an InputStream.
+   *
+   * @return the bytes in this ByteString accessible as an InputStream
+   * @see [[asByteBuffer]]
+   * @since 1.1.0
+   */
+  def asInputStream: InputStream
+
   override def foreach[@specialized U](f: Byte => U): Unit = iterator.foreach(f)
 
   private[pekko] def writeToOutputStream(os: ObjectOutputStream): Unit
@@ -886,7 +904,6 @@ sealed abstract class ByteString extends IndexedSeq[Byte] with IndexedSeqOptimiz
    * all fragments. Will always have at least one entry.
    */
   def getByteBuffers(): JIterable[ByteBuffer] = {
-    import scala.collection.JavaConverters.asJavaIterableConverter
     asByteBuffers.asJava
   }
 

diff --git a/actor/src/main/scala-2.13/org/apache/pekko/util/ByteString.scala b/actor/src/main/scala-2.13/org/apache/pekko/util/ByteString.scala
@@ -13,17 +13,17 @@
 
 package org.apache.pekko.util
 
-import java.io.{ ObjectInputStream, ObjectOutputStream }
+import java.io.{ ByteArrayInputStream, InputStream, ObjectInputStream, ObjectOutputStream, SequenceInputStream }
 import java.lang.{ Iterable => JIterable }
 import java.nio.{ ByteBuffer, ByteOrder }
 import java.nio.charset.{ Charset, StandardCharsets }
 import java.util.Base64
-import scala.annotation.{ tailrec, varargs }
+import scala.annotation.{ nowarn, tailrec, varargs }
 import scala.collection.{ immutable, mutable }
 import scala.collection.immutable.{ IndexedSeq, IndexedSeqOps, StrictOptimizedSeqOps, VectorBuilder }
 import scala.collection.mutable.{ Builder, WrappedArray }
+import scala.jdk.CollectionConverters._
 import scala.reflect.ClassTag
-import scala.annotation.nowarn
 
 object ByteString {
 
@@ -278,6 +278,7 @@ object ByteString {
 
     override def toArrayUnsafe(): Array[Byte] = bytes
 
+    override def asInputStream: InputStream = new ByteArrayInputStream(bytes)
   }
 
   /** INTERNAL API: ByteString backed by exactly one array, with start / end markers */
@@ -448,6 +449,9 @@ object ByteString {
       if (startIndex == 0 && length == bytes.length) bytes
       else toArray
     }
+
+    override def asInputStream: InputStream =
+      new ByteArrayInputStream(bytes, startIndex, length)
   }
 
   private[pekko] object ByteStrings extends Companion {
@@ -578,6 +582,9 @@ object ByteString {
 
     def asByteBuffers: scala.collection.immutable.Iterable[ByteBuffer] = bytestrings.map { _.asByteBuffer }
 
+    override def asInputStream: InputStream =
+      new SequenceInputStream(bytestrings.iterator.map(_.asInputStream).asJavaEnumeration)
+
     def decodeString(charset: String): String = compact.decodeString(charset)
 
     def decodeString(charset: Charset): String = compact.decodeString(charset)
@@ -876,6 +883,15 @@ sealed abstract class ByteString
    */
   def toArrayUnsafe(): Array[Byte] = toArray
 
+  /**
+   * Return the bytes in this ByteString as an InputStream.
+   *
+   * @return the bytes in this ByteString accessible as an InputStream
+   * @see [[asByteBuffer]]
+   * @since 1.1.0
+   */
+  def asInputStream: InputStream
+
   override def foreach[@specialized U](f: Byte => U): Unit = iterator.foreach(f)
 
   private[pekko] def writeToOutputStream(os: ObjectOutputStream): Unit
@@ -931,9 +947,7 @@ sealed abstract class ByteString
    * Java API: Returns an Iterable of read-only ByteBuffers that directly wraps this ByteStrings
    * all fragments. Will always have at least one entry.
    */
-  @nowarn
   def getByteBuffers(): JIterable[ByteBuffer] = {
-    import scala.collection.JavaConverters.asJavaIterableConverter
     asByteBuffers.asJava
   }
 

diff --git a/actor/src/main/scala-3/org/apache/pekko/util/ByteString.scala b/actor/src/main/scala-3/org/apache/pekko/util/ByteString.scala
@@ -13,7 +13,7 @@
 
 package org.apache.pekko.util
 
-import java.io.{ ObjectInputStream, ObjectOutputStream }
+import java.io.{ ByteArrayInputStream, InputStream, ObjectInputStream, ObjectOutputStream, SequenceInputStream }
 import java.lang.{ Iterable => JIterable }
 import java.nio.{ ByteBuffer, ByteOrder }
 import java.nio.charset.{ Charset, StandardCharsets }
@@ -23,10 +23,9 @@ import scala.annotation.{ tailrec, varargs }
 import scala.collection.{ immutable, mutable }
 import scala.collection.immutable.{ IndexedSeq, IndexedSeqOps, StrictOptimizedSeqOps, VectorBuilder }
 import scala.collection.mutable.{ Builder, WrappedArray }
+import scala.jdk.CollectionConverters._
 import scala.reflect.ClassTag
 
-import scala.annotation.nowarn
-
 object ByteString {
 
   /**
@@ -279,6 +278,8 @@ object ByteString {
     }
 
     override def toArrayUnsafe(): Array[Byte] = bytes
+
+    override def asInputStream: InputStream = new ByteArrayInputStream(bytes)
   }
 
   /** INTERNAL API: ByteString backed by exactly one array, with start / end markers */
@@ -449,6 +450,9 @@ object ByteString {
       if (startIndex == 0 && length == bytes.length) bytes
       else toArray
     }
+
+    override def asInputStream: InputStream =
+      new ByteArrayInputStream(bytes, startIndex, length)
   }
 
   private[pekko] object ByteStrings extends Companion {
@@ -579,6 +583,9 @@ object ByteString {
 
     def asByteBuffers: scala.collection.immutable.Iterable[ByteBuffer] = bytestrings.map { _.asByteBuffer }
 
+    override def asInputStream: InputStream =
+      new SequenceInputStream(bytestrings.iterator.map(_.asInputStream).asJavaEnumeration)
+
     def decodeString(charset: String): String = compact.decodeString(charset)
 
     def decodeString(charset: Charset): String = compact.decodeString(charset)
@@ -876,6 +883,15 @@ sealed abstract class ByteString
    */
   def toArrayUnsafe(): Array[Byte] = toArray
 
+  /**
+   * Return the bytes in this ByteString as an InputStream.
+   *
+   * @return the bytes in this ByteString accessible as an InputStream
+   * @see [[asByteBuffer]]
+   * @since 1.1.0
+   */
+  def asInputStream: InputStream
+
   override def foreach[@specialized U](f: Byte => U): Unit = iterator.foreach(f)
 
   private[pekko] def writeToOutputStream(os: ObjectOutputStream): Unit
@@ -931,9 +947,7 @@ sealed abstract class ByteString
    * Java API: Returns an Iterable of read-only ByteBuffers that directly wraps this ByteStrings
    * all fragments. Will always have at least one entry.
    */
-  @nowarn
   def getByteBuffers(): JIterable[ByteBuffer] = {
-    import scala.collection.JavaConverters.asJavaIterableConverter
     asByteBuffers.asJava
   }
 

diff --git a/bench-jmh/README.md b/bench-jmh/README.md
@@ -9,7 +9,7 @@ Pekko uses [sbt-jmh](https://github.com/sbt/sbt-jmh) to integrate [Java Microben
 ```shell
 sbt shell
 pekko > project bench-jmh
-sbt:bench-jmh> Jmh/run -i 3 -wi 3 -f 1 .*ActorCreationBenchmark
+sbt:pekko-bench-jmh> Jmh/run -i 3 -wi 3 -f 1 .*ActorCreationBenchmark
 ```
 
 or execute in one-line command