Merge pull request #18 from pjfanning/asInputStream

support pekko 1.1 ByteString.asInputStream

pekko-http-jackson/src/main/scala/com/github/pjfanning/pekkohttpjackson/ByteStringInputStream.scala

@@ -0,0 +1,49 @@
+/*
+ * 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 com.github.pjfanning.pekkohttpjackson
+
+import java.io.{ ByteArrayInputStream, InputStream }
+import java.lang.invoke.{ MethodHandles, MethodType }
+
+import scala.util.Try
+
+import org.apache.pekko
+import pekko.util.ByteString
+import pekko.util.ByteString.ByteString1C
+
+private[pekkohttpjackson] final object ByteStringInputStream {
+  private val byteStringInputStreamMethodTypeOpt = Try {
+    val lookup                = MethodHandles.publicLookup()
+    val inputStreamMethodType = MethodType.methodType(classOf[InputStream])
+    lookup.findVirtual(classOf[ByteString], "asInputStream", inputStreamMethodType)
+  }.toOption
+
+  def apply(bs: ByteString): InputStream = bs match {
+    case cs: ByteString1C =>
+      getInputStreamUnsafe(cs)
+    case _ =>
+      if (byteStringInputStreamMethodTypeOpt.isDefined) {
+        byteStringInputStreamMethodTypeOpt.get.invoke(bs).asInstanceOf[InputStream]
+      } else {
+        getInputStreamUnsafe(bs)
+      }
+  }
+
+  private def getInputStreamUnsafe(bs: ByteString): InputStream =
+    new ByteArrayInputStream(bs.toArrayUnsafe())
+}

pekko-http-jackson/src/main/scala/com/github/pjfanning/pekkohttpjackson/JacksonSupport.scala

@@ -205,7 +205,7 @@ trait JacksonSupport {
       objectMapper: ObjectMapper with ClassTagExtensions = defaultObjectMapper
   ): Unmarshaller[ByteString, A] =
     Unmarshaller { _ => bs =>
-      Future.fromTry(Try(objectMapper.readValue[A](bs.toArrayUnsafe())))
+      Future.fromTry(Try(objectMapper.readValue[A](ByteStringInputStream(bs))))
     }
 
   /**

pekko-http-jsoniter-scala/src/main/scala/com/github/pjfanning/pekkohttpjsoniterscala/ByteStringInputStream.scala

@@ -0,0 +1,47 @@
+/*
+ * 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 com.github.pjfanning.pekkohttpjsoniterscala
+
+import org.apache.pekko.util.ByteString
+import org.apache.pekko.util.ByteString.ByteString1C
+
+import java.io.{ ByteArrayInputStream, InputStream }
+import java.lang.invoke.{ MethodHandles, MethodType }
+import scala.util.Try
+
+private[pekkohttpjsoniterscala] final object ByteStringInputStream {
+  private val byteStringInputStreamMethodTypeOpt = Try {
+    val lookup                = MethodHandles.publicLookup()
+    val inputStreamMethodType = MethodType.methodType(classOf[InputStream])
+    lookup.findVirtual(classOf[ByteString], "asInputStream", inputStreamMethodType)
+  }.toOption
+
+  def apply(bs: ByteString): InputStream = bs match {
+    case cs: ByteString1C =>
+      getInputStreamUnsafe(cs)
+    case _ =>
+      if (byteStringInputStreamMethodTypeOpt.isDefined) {
+        byteStringInputStreamMethodTypeOpt.get.invoke(bs).asInstanceOf[InputStream]
+      } else {
+        getInputStreamUnsafe(bs)
+      }
+  }
+
+  private def getInputStreamUnsafe(bs: ByteString): InputStream =
+    new ByteArrayInputStream(bs.toArrayUnsafe())
+}

pekko-http-jsoniter-scala/src/main/scala/com/github/pjfanning/pekkohttpjsoniterscala/JsoniterScalaSupport.scala

@@ -136,7 +136,7 @@ trait JsoniterScalaSupport {
       codec: JsonValueCodec[A],
       config: ReaderConfig = defaultReaderConfig
   ): Unmarshaller[ByteString, A] =
-    Unmarshaller(_ => bs => Future.fromTry(Try(readFromArray(bs.toArrayUnsafe(), config))))
+    Unmarshaller(_ => bs => Future.fromTry(Try(readFromStream(ByteStringInputStream(bs), config))))
 
   /**
     * HTTP entity => `Source[A, _]`

pekko-http-play-json/src/main/scala/com/github/pjfanning/pekkohttpplayjson/ByteStringInputStream.scala

@@ -0,0 +1,47 @@
+/*
+ * 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 com.github.pjfanning.pekkohttpplayjson
+
+import org.apache.pekko.util.ByteString
+import org.apache.pekko.util.ByteString.ByteString1C
+
+import java.io.{ ByteArrayInputStream, InputStream }
+import java.lang.invoke.{ MethodHandles, MethodType }
+import scala.util.Try
+
+private[pekkohttpplayjson] final object ByteStringInputStream {
+  private val byteStringInputStreamMethodTypeOpt = Try {
+    val lookup                = MethodHandles.publicLookup()
+    val inputStreamMethodType = MethodType.methodType(classOf[InputStream])
+    lookup.findVirtual(classOf[ByteString], "asInputStream", inputStreamMethodType)
+  }.toOption
+
+  def apply(bs: ByteString): InputStream = bs match {
+    case cs: ByteString1C =>
+      getInputStreamUnsafe(cs)
+    case _ =>
+      if (byteStringInputStreamMethodTypeOpt.isDefined) {
+        byteStringInputStreamMethodTypeOpt.get.invoke(bs).asInstanceOf[InputStream]
+      } else {
+        getInputStreamUnsafe(bs)
+      }
+  }
+
+  private def getInputStreamUnsafe(bs: ByteString): InputStream =
+    new ByteArrayInputStream(bs.toArrayUnsafe())
+}

pekko-http-play-json/src/main/scala/com/github/pjfanning/pekkohttpplayjson/PlayJsonSupport.scala

@@ -140,7 +140,7 @@ trait PlayJsonSupport {
     *   unmarshaller for any `A` value
     */
   implicit def fromByteStringUnmarshaller[A: Reads]: Unmarshaller[ByteString, A] =
-    Unmarshaller(_ => bs => Future.fromTry(Try(Json.parse(bs.toArrayUnsafe()).as[A])))
+    Unmarshaller(_ => bs => Future.fromTry(Try(Json.parse(ByteStringInputStream(bs)).as[A])))
 
   /**
     * HTTP entity => `Source[A, _]`