ByteString.toArrayUnsafe method for zero copy transformation of bytestrings (#30262)

akka-actor-tests/src/test/scala/akka/util/ByteStringSpec.scala

@@ -1007,6 +1007,18 @@ class ByteStringSpec extends AnyWordSpec with Matchers with Checkers {
         deserialize(serialize(original)) shouldEqual original
       }
     }
+
+    "unsafely wrap and unwrap bytes" in {
+      // optimal case
+      val bytes = Array.fill[Byte](100)(7)
+      val bs = ByteString.fromArrayUnsafe(bytes)
+      val bytes2 = bs.toArrayUnsafe()
+      (bytes2 should be).theSameInstanceAs(bytes)
+
+      val combinedBs = bs ++ bs
+      val combinedBytes = combinedBs.toArrayUnsafe()
+      combinedBytes should ===(bytes ++ bytes)
+    }
   }
 
   "A ByteStringIterator" must {

akka-actor/src/main/scala-2.12/akka/util/ByteString.scala

@@ -258,6 +258,7 @@ object ByteString {
       buffer.putByteArrayUnsafe(bytes)
     }
 
+    override def toArrayUnsafe(): Array[Byte] = bytes
   }
 
   /** INTERNAL API: ByteString backed by exactly one array, with start / end markers */
@@ -273,6 +274,7 @@ object ByteString {
 
     def readFromInputStream(is: ObjectInputStream): ByteString1 =
       ByteString1C.readFromInputStream(is).toByteString1
+
   }
 
   /**
@@ -417,6 +419,11 @@ object ByteString {
     }
 
     protected def writeReplace(): AnyRef = new SerializationProxy(this)
+
+    override def toArrayUnsafe(): Array[Byte] = {
+      if (startIndex == 0 && length == bytes.length) bytes
+      else toArray
+    }
   }
 
   private[akka] object ByteStrings extends Companion {
@@ -787,6 +794,25 @@ sealed abstract class ByteString extends IndexedSeq[Byte] with IndexedSeqOptimiz
   override def copyToArray[B >: Byte](xs: Array[B], start: Int, len: Int): Unit =
     iterator.copyToArray(xs, start, len)
 
+  /**
+   * Unsafe API: Use only in situations you are completely confident that this is what
+   * you need, and that you understand the implications documented below.
+   *
+   * If the ByteString is backed by a single array it is returned without any copy. If it is backed by a rope
+   * of multiple ByteString instances a new array will be allocated and the contents will be copied
+   * into it before returning it.
+   *
+   * This method of exposing the bytes of a ByteString can save one array
+   * copy and allocation in the happy path scenario and which can lead to better performance,
+   * however it also means that one MUST NOT modify the returned in array, or unexpected
+   * immutable data structure contract-breaking behavior will manifest itself.
+   *
+   * This API is intended for users who need to pass the byte array to some other API, which will
+   * only read the bytes and never mutate then. For all other intents and purposes, please use the usual
+   * toArray method - which provide the immutability guarantees by copying the backing array.
+   */
+  def toArrayUnsafe(): Array[Byte] = toArray
+
   override def foreach[@specialized U](f: Byte => U): Unit = iterator.foreach(f)
 
   private[akka] def writeToOutputStream(os: ObjectOutputStream): Unit

akka-actor/src/main/scala-2.13/akka/util/ByteString.scala

@@ -9,13 +9,11 @@ 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.{ immutable, mutable }
 import scala.collection.immutable.{ IndexedSeq, IndexedSeqOps, StrictOptimizedSeqOps, VectorBuilder }
 import scala.collection.mutable.{ Builder, WrappedArray }
 import scala.reflect.ClassTag
-
 import scala.annotation.nowarn
 
 object ByteString {
@@ -271,6 +269,8 @@ object ByteString {
       toCopy
     }
 
+    override def toArrayUnsafe(): Array[Byte] = bytes
+
   }
 
   /** INTERNAL API: ByteString backed by exactly one array, with start / end markers */
@@ -436,6 +436,11 @@ object ByteString {
     }
 
     protected def writeReplace(): AnyRef = new SerializationProxy(this)
+
+    override def toArrayUnsafe(): Array[Byte] = {
+      if (startIndex == 0 && length == bytes.length) bytes
+      else toArray
+    }
   }
 
   private[akka] object ByteStrings extends Companion {
@@ -843,6 +848,25 @@ sealed abstract class ByteString
   override def copyToArray[B >: Byte](xs: Array[B], start: Int, len: Int): Int =
     throw new UnsupportedOperationException("Method copyToArray is not implemented in ByteString")
 
+  /**
+   * Unsafe API: Use only in situations you are completely confident that this is what
+   * you need, and that you understand the implications documented below.
+   *
+   * If the ByteString is backed by a single array it is returned without any copy. If it is backed by a rope
+   * of multiple ByteString instances a new array will be allocated and the contents will be copied
+   * into it before returning it.
+   *
+   * This method of exposing the bytes of a ByteString can save one array
+   * copy and allocation in the happy path scenario and which can lead to better performance,
+   * however it also means that one MUST NOT modify the returned in array, or unexpected
+   * immutable data structure contract-breaking behavior will manifest itself.
+   *
+   * This API is intended for users who need to pass the byte array to some other API, which will
+   * only read the bytes and never mutate then. For all other intents and purposes, please use the usual
+   * toArray method - which provide the immutability guarantees by copying the backing array.
+   */
+  def toArrayUnsafe(): Array[Byte] = toArray
+
   override def foreach[@specialized U](f: Byte => U): Unit = iterator.foreach(f)
 
   private[akka] def writeToOutputStream(os: ObjectOutputStream): Unit

akka-actor/src/main/scala-3/akka/util/ByteString.scala

@@ -271,6 +271,7 @@ object ByteString {
       toCopy
     }
 
+    override def toArrayUnsafe(): Array[Byte] = bytes
   }
 
   /** INTERNAL API: ByteString backed by exactly one array, with start / end markers */
@@ -436,6 +437,11 @@ object ByteString {
     }
 
     protected def writeReplace(): AnyRef = new SerializationProxy(this)
+
+    override def toArrayUnsafe(): Array[Byte] = {
+      if (startIndex == 0 && length == bytes.length) bytes
+      else toArray
+    }
   }
 
   private[akka] object ByteStrings extends Companion {
@@ -842,6 +848,25 @@ sealed abstract class ByteString
   override def copyToArray[B >: Byte](xs: Array[B], start: Int, len: Int): Int =
     throw new UnsupportedOperationException("Method copyToArray is not implemented in ByteString")
 
+  /**
+   * Unsafe API: Use only in situations you are completely confident that this is what
+   * you need, and that you understand the implications documented below.
+   *
+   * If the ByteString is backed by a single array it is returned without any copy. If it is backed by a rope
+   * of multiple ByteString instances a new array will be allocated and the contents will be copied
+   * into it before returning it.
+   *
+   * This method of exposing the bytes of a ByteString can save one array
+   * copy and allocation in the happy path scenario and which can lead to better performance,
+   * however it also means that one MUST NOT modify the returned in array, or unexpected
+   * immutable data structure contract-breaking behavior will manifest itself.
+   *
+   * This API is intended for users who need to pass the byte array to some other API, which will
+   * only read the bytes and never mutate then. For all other intents and purposes, please use the usual
+   * toArray method - which provide the immutability guarantees by copying the backing array.
+   */
+  def toArrayUnsafe(): Array[Byte] = toArray
+
   override def foreach[@specialized U](f: Byte => U): Unit = iterator.foreach(f)
 
   private[akka] def writeToOutputStream(os: ObjectOutputStream): Unit