remove ByteStringInputStream

http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/hpack/HeaderDecompression.scala

@@ -92,8 +92,9 @@ private[http2] final class HeaderDecompression(masterHeaderParser: HttpHeaderPar
             }
           }
         }
+        val bis = payload.asInputStream
         try {
-          decoder.decode(ByteStringInputStream(payload), Receiver)
+          decoder.decode(bis, Receiver)
           decoder.endHeaderBlock() // TODO: do we have to check the result here?
 
           push(eventsOut, ParsedHeadersFrame(streamId, endStream, headers.result(), prioInfo))
@@ -102,7 +103,7 @@ private[http2] final class HeaderDecompression(masterHeaderParser: HttpHeaderPar
             // this is signalled by the decoder when it failed, we want to react to this by rendering a GOAWAY frame
             fail(eventsOut,
               new Http2Compliance.Http2ProtocolException(ErrorCode.COMPRESSION_ERROR, "Decompression failed."))
-        }
+        } finally bis.close()
       }
 
       object Idle extends State {

http-tests/src/test/scala/org/apache/pekko/http/scaladsl/coding/CoderSpec.scala

@@ -13,11 +13,10 @@
 
 package org.apache.pekko.http.scaladsl.coding
 
-import java.io.{ ByteArrayInputStream, ByteArrayOutputStream, InputStream, OutputStream }
+import java.io.{ ByteArrayOutputStream, InputStream, OutputStream }
 import java.util.concurrent.ThreadLocalRandom
 import java.util.zip.DataFormatException
 
-import scala.annotation.nowarn
 import scala.annotation.tailrec
 import scala.concurrent.Await
 import scala.concurrent.ExecutionContext.Implicits.global
@@ -36,7 +35,6 @@ import pekko.util.ByteString
 import org.scalatest.Inspectors
 import org.scalatest.wordspec.AnyWordSpec
 
-@nowarn("msg=deprecated .* is internal API")
 abstract class CoderSpec extends AnyWordSpec with CodecSpecSupport with Inspectors {
   protected def Coder: Coder
   protected def newDecodedInputStream(underlying: InputStream): InputStream
@@ -191,7 +189,7 @@ abstract class CoderSpec extends AnyWordSpec with CodecSpecSupport with Inspecto
 
   def streamDecode(bytes: ByteString): ByteString = {
     val output = new ByteArrayOutputStream()
-    val input = newDecodedInputStream(new ByteArrayInputStream(bytes.toArray))
+    val input = newDecodedInputStream(bytes.asInputStream)
 
     val buffer = new Array[Byte](500)
     @tailrec def copy(from: InputStream, to: OutputStream): Unit = {

http2-tests/src/test/scala/org/apache/pekko/http/impl/engine/http2/Http2FrameHpackSupport.scala

@@ -16,7 +16,6 @@ package org.apache.pekko.http.impl.engine.http2
 import scala.collection.immutable.VectorBuilder
 
 import org.apache.pekko
-import pekko.http.impl.engine.http2.hpack.ByteStringInputStream
 import pekko.http.scaladsl.model._
 import pekko.http.scaladsl.model.headers.RawHeader
 import pekko.http.shaded.com.twitter.hpack._
@@ -59,16 +58,17 @@ trait Http2FrameHpackSupport extends Http2FrameProbeDelegator with Http2FrameSen
   val decoder = new Decoder(Http2Protocol.InitialMaxHeaderListSize, Http2Protocol.InitialMaxHeaderTableSize)
 
   def decodeHeaders(bytes: ByteString): Seq[(String, String)] = {
-    val bis = ByteStringInputStream(bytes)
     val hs = new VectorBuilder[(String, String)]()
-
-    decoder.decode(bis,
-      new HeaderListener {
-        def addHeader(name: String, value: String, parsedValue: AnyRef, sensitive: Boolean): AnyRef = {
-          hs += name -> value
-          parsedValue
-        }
-      })
+    val bis = bytes.asInputStream
+    try
+      decoder.decode(bis,
+        new HeaderListener {
+          def addHeader(name: String, value: String, parsedValue: AnyRef, sensitive: Boolean): AnyRef = {
+            hs += name -> value
+            parsedValue
+          }
+        })
+    finally bis.close()
     hs.result()
   }
   def decodeHeadersToResponse(bytes: ByteString): HttpResponse =