Add Bytes Serde (#100)

conduktor · Oct 31, 2022 · 883649e · 883649e
1 parent 4ce24be
commit 883649e
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 4 deletions.
diff --git a/src/main/scala/zio/kafka/serde/Serdes.scala b/src/main/scala/zio/kafka/serde/Serdes.scala
@@ -1,12 +1,13 @@
 package zio.kafka.serde
 
-import java.nio.ByteBuffer
-import java.util.UUID
-
-import org.apache.kafka.common.serialization.{ Serde => KafkaSerde, Serdes => KafkaSerdes }
 import org.apache.kafka.common.header.Headers
+import org.apache.kafka.common.serialization.{ Serde => KafkaSerde, Serdes => KafkaSerdes }
+import org.apache.kafka.common.utils.Bytes
 import zio.{ RIO, Task }
 
+import java.nio.ByteBuffer
+import java.util.UUID
+
 private[zio] trait Serdes {
   lazy val long: Serde[Any, Long]     = convertPrimitiveSerde(KafkaSerdes.Long()).inmap(Long2long)(long2Long)
   lazy val int: Serde[Any, Int]       = convertPrimitiveSerde(KafkaSerdes.Integer()).inmap(Integer2int)(int2Integer)
@@ -15,6 +16,7 @@ private[zio] trait Serdes {
   lazy val double: Serde[Any, Double] = convertPrimitiveSerde(KafkaSerdes.Double()).inmap(Double2double)(double2Double)
   lazy val string: Serde[Any, String] = convertPrimitiveSerde(KafkaSerdes.String())
   lazy val byteArray: Serde[Any, Array[Byte]] = convertPrimitiveSerde(KafkaSerdes.ByteArray())
+  lazy val bytes: Serde[Any, Bytes]           = convertPrimitiveSerde(KafkaSerdes.Bytes())
   lazy val byteBuffer: Serde[Any, ByteBuffer] = convertPrimitiveSerde(KafkaSerdes.ByteBuffer())
   lazy val uuid: Serde[Any, UUID]             = convertPrimitiveSerde(KafkaSerdes.UUID())
 

diff --git a/src/test/scala/zio/kafka/serde/SerdeSpec.scala b/src/test/scala/zio/kafka/serde/SerdeSpec.scala
@@ -7,6 +7,9 @@ import zio.test._
 import scala.reflect.ClassTag
 
 object SerdeSpec extends DefaultRunnableSpec {
+
+  private val anyBytes = Gen.listOf(Gen.anyByte).map(bytes => new org.apache.kafka.common.utils.Bytes(bytes.toArray))
+
   override def spec = suite("Serde")(
     testSerde(Serde.string, Gen.anyString),
     testSerde(Serde.int, Gen.anyInt),
@@ -15,6 +18,7 @@ object SerdeSpec extends DefaultRunnableSpec {
     testSerde(Serde.double, Gen.anyDouble),
     testSerde(Serde.long, Gen.anyLong),
     testSerde(Serde.uuid, Gen.anyUUID),
+    testSerde(Serde.bytes, anyBytes),
     testSerde(Serde.byteArray, Gen.listOf(Gen.anyByte).map(_.toArray)),
     suite("asOption")(
       testM("serialize and deserialize None values to null and visa versa") {