Add MessagePack item serializer and validator

benchmarks/src/main/scala/fs2/data/benchmarks/MsgPackItemParserBenchmarks.scala

@@ -22,7 +22,6 @@ import org.openjdk.jmh.annotations._
 
 import cats.effect.SyncIO
 
-import scodec.bits._
 import fs2._
 
 @OutputTimeUnit(TimeUnit.MICROSECONDS)
@@ -32,23 +31,19 @@ import fs2._
 @Warmup(iterations = 3, time = 2)
 @Measurement(iterations = 10, time = 2)
 class MsgPackItemParserBenchmarks {
-
-  // The file contains hex representation of the values so we have to convert it
-  val msgpackBytes: ByteVector = ByteVector
-    .fromHex(
-      fs2.io
-        .readClassLoaderResource[SyncIO]("twitter_msgpack.txt", 4096)
-        .through(fs2.text.utf8.decode)
-        .compile
-        .string
-        .unsafeRunSync()
-    )
-    .get
+  val msgpackBytes: Stream[SyncIO, Byte] =
+    fs2.io
+      .readClassLoaderResource[SyncIO]("twitter_msgpack.mp", 4096)
+      .chunks
+      .compile
+      .toList
+      .unsafeRunSync()
+      .map(Stream.chunk)
+      .fold(Stream.empty)(_ ++ _)
 
   @Benchmark
   def parseMsgpackItems() =
-    Stream
-      .chunk(Chunk.byteVector(msgpackBytes))
+    msgpackBytes
       .through(fs2.data.msgpack.low.items[SyncIO])
       .compile
       .drain

benchmarks/src/main/scala/fs2/data/benchmarks/MsgPackItemSerializerBenchmarks.scala

@@ -0,0 +1,58 @@
+/*
+ * Copyright 2024 fs2-data Project
+ *
+ * Licensed 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 fs2
+package data.benchmarks
+
+import java.util.concurrent.TimeUnit
+import org.openjdk.jmh.annotations._
+
+import cats.effect.SyncIO
+
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+@BenchmarkMode(Array(Mode.AverageTime))
+@State(org.openjdk.jmh.annotations.Scope.Benchmark)
+@Fork(value = 1)
+@Warmup(iterations = 3, time = 2)
+@Measurement(iterations = 10, time = 2)
+class MsgPackItemSerializerBenchmarks {
+  val msgpackItems: Stream[SyncIO, fs2.data.msgpack.low.MsgpackItem] =
+    fs2.io
+      .readClassLoaderResource[SyncIO]("twitter_msgpack.mp", 4096)
+      .through(fs2.data.msgpack.low.items[SyncIO])
+      .chunks
+      .compile
+      .toList
+      .unsafeRunSync()
+      .map(Stream.chunk)
+      .fold(Stream.empty)(_ ++ _)
+
+  @Benchmark
+  def serialize() =
+    msgpackItems
+      .through(fs2.data.msgpack.low.toNonValidatedBinary[SyncIO])
+      .compile
+      .drain
+      .unsafeRunSync()
+
+  @Benchmark
+  def withValidation() =
+    msgpackItems
+      .through(fs2.data.msgpack.low.toBinary[SyncIO])
+      .compile
+      .drain
+      .unsafeRunSync()
+}

msgpack/src/main/scala/fs2/data/msgpack/exceptions.scala

@@ -0,0 +1,32 @@
+/*
+ * Copyright 2024 fs2-data Project
+ *
+ * Licensed 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 fs2
+package data
+package msgpack
+
+abstract class MsgpackException(msg: String, cause: Throwable = null) extends Exception(msg, cause)
+
+case class MsgpackMalformedItemException(msg: String, position: Option[Long] = None, inner: Throwable = null)
+    extends MsgpackException(position.fold(msg)(pos => s"at position $pos"), inner)
+
+case class MsgpackUnexpectedEndOfStreamException(position: Option[Long] = None, inner: Throwable = null)
+    extends MsgpackException(
+      position.fold("Unexpected end of stream")(pos => s"Unexpected end of stream starting at position $pos"),
+      inner)
+
+case class MsgpackMalformedByteStreamException(msg: String, inner: Throwable = null)
+    extends MsgpackException(msg, inner)

msgpack/src/main/scala/fs2/data/msgpack/low/internal/FormatParsers.scala

@@ -34,14 +34,14 @@ private[internal] object FormatParsers {
   def parseArray[F[_]](length: Int, ctx: ParserContext[F])(implicit
       F: RaiseThrowable[F]): Pull[F, MsgpackItem, ParserContext[F]] = {
     requireBytes(length, ctx).map { res =>
-      res.accumulate(v => MsgpackItem.Array(v.toInt(false, ByteOrdering.BigEndian)))
+      res.accumulate(v => MsgpackItem.Array(v.toLong(false)))
     }
   }
 
   def parseMap[F[_]](length: Int, ctx: ParserContext[F])(implicit
       F: RaiseThrowable[F]): Pull[F, MsgpackItem, ParserContext[F]] = {
     requireBytes(length, ctx).map { res =>
-      res.accumulate(v => MsgpackItem.Map(v.toInt(false, ByteOrdering.BigEndian)))
+      res.accumulate(v => MsgpackItem.Map(v.toLong(false)))
     }
   }
 
@@ -63,7 +63,7 @@ private[internal] object FormatParsers {
           res <- requireBytes(8, res.toContext)
           seconds = res.result.toLong(false)
         } yield res.toContext.prepend(MsgpackItem.Timestamp96(nanosec, seconds))
-      case _ => Pull.raiseError(new MsgpackParsingException(s"Invalid timestamp length: ${length}"))
+      case _ => Pull.raiseError(MsgpackMalformedByteStreamException(s"Invalid timestamp length: ${length}"))
     }
   }
 

msgpack/src/main/scala/fs2/data/msgpack/low/internal/Helpers.scala

@@ -23,7 +23,6 @@ package internal
 import scodec.bits.ByteVector
 
 private[internal] object Helpers {
-  case class MsgpackParsingException(str: String) extends Exception
 
   /** @param chunk Current chunk
     * @param idx Index of the current [[Byte]] in `chunk`
@@ -67,7 +66,7 @@ private[internal] object Helpers {
       // Inbounds chunk access is guaranteed by `ensureChunk`
       Pull.pure(ctx.next.toResult(ctx.chunk(ctx.idx)))
     } {
-      Pull.raiseError(new MsgpackParsingException("Unexpected end of input"))
+      Pull.raiseError(MsgpackUnexpectedEndOfStreamException())
     }
   }
 
@@ -93,7 +92,7 @@ private[internal] object Helpers {
           go(count - available, ParserContext(chunk, slice.size, rest, acc), newBytes)
         }
       } {
-        Pull.raiseError(new MsgpackParsingException("Unexpected end of input"))
+        Pull.raiseError(MsgpackUnexpectedEndOfStreamException())
       }
     }
 

msgpack/src/main/scala/fs2/data/msgpack/low/internal/ItemParser.scala

@@ -37,7 +37,7 @@ private[low] object ItemParser {
 
       ((byte & 0xff): @switch) match {
         case Headers.Nil       => Pull.pure(ctx.prepend(MsgpackItem.Nil))
-        case Headers.NeverUsed => Pull.raiseError(new MsgpackParsingException("Reserved value 0xc1 used"))
+        case Headers.NeverUsed => Pull.raiseError(MsgpackMalformedByteStreamException("Reserved value 0xc1 used"))
         case Headers.False     => Pull.pure(ctx.prepend(MsgpackItem.False))
         case Headers.True      => Pull.pure(ctx.prepend(MsgpackItem.True))
         case Headers.Bin8      => parseBin(1, ctx)
@@ -77,13 +77,13 @@ private[low] object ItemParser {
           // fixmap
           else if ((byte & 0xf0) == 0x80) {
             val length = byte & 0x0f // 0x8f- 0x80
-            Pull.pure(ctx.prepend(MsgpackItem.Map(length)))
+            Pull.pure(ctx.prepend(MsgpackItem.Map(length.toLong)))
           }
 
           // fixarray
           else if ((byte & 0xf0) == 0x90) {
             val length = byte & 0x0f // 0x9f- 0x90
-            Pull.pure(ctx.prepend(MsgpackItem.Array(length)))
+            Pull.pure(ctx.prepend(MsgpackItem.Array(length.toLong)))
           }
 
           // fixstr
@@ -98,7 +98,7 @@ private[low] object ItemParser {
           else if ((byte & 0xe0) == 0xe0) {
             Pull.pure(ctx.prepend(MsgpackItem.SignedInt(ByteVector(byte))))
           } else {
-            Pull.raiseError(new MsgpackParsingException(s"Invalid type ${byte}"))
+            Pull.raiseError(MsgpackMalformedByteStreamException(s"Invalid type ${byte}"))
           }
         }
       }