Speedup BinaryCodec::streamDecoder implementations

Author: guizmaii

build.sbt

@@ -1,7 +1,8 @@
-import sbtcrossproject.CrossPlugin.autoImport._
-import BuildHelper.{ crossProjectSettings, _ }
-import org.scalajs.sbtplugin.ScalaJSPlugin.autoImport._
+import sbtcrossproject.CrossPlugin.autoImport.*
+import BuildHelper.{crossProjectSettings, *}
+import org.scalajs.sbtplugin.ScalaJSPlugin.autoImport.*
 import com.typesafe.tools.mima.plugin.MimaKeys.mimaPreviousArtifacts
+import sbt.{Resolver, ThisBuild}
 
 Global / onChangedBuildSource := ReloadOnSourceChanges
 
@@ -35,7 +36,9 @@ inThisBuild(
     licenses := Seq("Apache-2.0" -> url(s"${scmInfo.value.map(_.browseUrl).get}/blob/v${version.value}/LICENSE")),
     pgpPassphrase := sys.env.get("PGP_PASSWORD").map(_.toArray),
     pgpPublicRing := file("/tmp/public.asc"),
-    pgpSecretRing := file("/tmp/secret.asc")
+    pgpSecretRing := file("/tmp/secret.asc"),
+    resolvers ++= Resolver.sonatypeOssRepos("snapshots"),
+    resolvers ++= Resolver.sonatypeOssRepos("staging")
   )
 )
 

project/BuildHelper.scala

@@ -33,7 +33,7 @@ object BuildHelper {
   val Scala213: String = versions("2.13")
   val Scala3: String   = versions("3.3")
 
-  val zioVersion                   = "2.1.15"
+  val zioVersion                   = "2.1.17+28-a8d9c23d-SNAPSHOT"
   val zioJsonVersion               = "0.7.25"
   val zioPreludeVersion            = "1.0.0-RC39"
   val zioOpticsVersion             = "0.2.2"

zio-schema-avro/src/main/scala/zio/schema/codec/AvroCodec.scala

@@ -60,11 +60,8 @@ object AvroCodec {
         decodeValue(decoded, schema)
       }
 
-      override def streamDecoder: ZPipeline[Any, DecodeError, Byte, A] = ZPipeline.mapChunksZIO { chunk =>
-        ZIO.fromEither(
-          decode(chunk).map(Chunk(_))
-        )
-      }
+      override def streamDecoder: ZPipeline[Any, DecodeError, Byte, A] =
+        ZPipeline.mapChunksEither(bytes => decode(bytes).map(Chunk(_)))
 
       override def encodeGenericRecord(value: A)(implicit schema: Schema[A]): GenericData.Record =
         encodeValue(value, schema).asInstanceOf[GenericData.Record]

zio-schema-msg-pack/src/main/scala/zio/schema/codec/MessagePackCodec.scala

@@ -28,13 +28,9 @@ object MessagePackCodec {
       }
 
       override def streamDecoder: ZPipeline[Any, DecodeError, Byte, A] =
-        ZPipeline.mapChunksZIO { chunk =>
-          ZIO.fromEither(
-            decodeChunk(chunk).map(Chunk(_))
-          )
-        }
+        ZPipeline.mapChunksEither(bytes => decodeChunk(bytes).map(Chunk.single))
 
-      private def decodeChunk(chunk: Chunk[Byte]) =
+      private def decodeChunk(chunk: Chunk[Byte]): Either[DecodeError, A] =
         new MessagePackDecoder(chunk)
           .decode(schema)
           .left

zio-schema-protobuf/shared/src/main/scala/zio/schema/codec/ProtobufCodec.scala

@@ -25,15 +25,13 @@ object ProtobufCodec {
         new Decoder(whole).decode(schema)
 
       override def streamDecoder: ZPipeline[Any, DecodeError, Byte, A] =
-        ZPipeline.mapChunksZIO(chunk => ZIO.fromEither(new Decoder(chunk).decode(schema).map(Chunk(_))))
+        ZPipeline.mapChunksEither(bytes => new Decoder(bytes).decode(schema).map(Chunk.single))
 
       override def encode(value: A): Chunk[Byte] =
         Encoder.process(schema, value)
 
       override def streamEncoder: ZPipeline[Any, Nothing, A, Byte] =
-        ZPipeline.mapChunks(
-          _.flatMap(encode)
-        )
+        ZPipeline.mapChunks(_.flatMap(encode))
     }
 
   object Protobuf {

zio-schema-thrift/src/main/scala/zio/schema/codec/ThriftCodec.scala

@@ -27,11 +27,7 @@ object ThriftCodec {
           decodeChunk(whole)
 
       override def streamDecoder: ZPipeline[Any, DecodeError, Byte, A] =
-        ZPipeline.mapChunksZIO { chunk =>
-          ZIO.fromEither(
-            decodeChunk(chunk).map(Chunk(_))
-          )
-        }
+        ZPipeline.mapChunksEither(bytes => decodeChunk(bytes).map(Chunk.single))
 
       override def encode(value: A): Chunk[Byte] =
         new Encoder().encode(schema, value)