added API to allow encoding and decoding to/from Generic record, to a… (

#573) * added API to allow encoding and decoding to/from Generic record, to allow integration with Schema Registry (AWS Glue) * fixed imports * added API to allow encoding and decoding to/from Generic record, to allow integration with Schema Registry (AWS Glue) * fixed imports * fix: fixed README --------- Co-authored-by: John A. De Goes <[email protected]>
zio · Aug 22, 2023 · 620d57d · 620d57d
1 parent 9420122
commit 620d57d
Show file tree

Hide file tree

Showing 2 changed files with 26 additions and 3 deletions.
diff --git a/zio-schema-avro/shared/src/main/scala/zio/schema/codec/AvroCodec.scala b/zio-schema-avro/shared/src/main/scala/zio/schema/codec/AvroCodec.scala
@@ -25,8 +25,13 @@ import zio.{ Chunk, Unsafe, ZIO }
 
 object AvroCodec {
 
-  implicit def schemaBasedBinaryCodec[A](implicit schema: Schema[A]): BinaryCodec[A] =
-    new BinaryCodec[A] {
+  trait ExtendedBinaryCodec[A] extends BinaryCodec[A] {
+    def encodeGenericRecord(value: A)(implicit schema: Schema[A]): GenericData.Record
+    def decodeGenericRecord(value: GenericRecord)(implicit schema: Schema[A]): Either[DecodeError, A]
+  }
+
+  implicit def schemaBasedBinaryCodec[A](implicit schema: Schema[A]): ExtendedBinaryCodec[A] =
+    new ExtendedBinaryCodec[A] {
 
       val avroSchema: SchemaAvro =
         AvroSchemaCodec.encodeToApacheAvro(schema).getOrElse(throw new Exception("Avro schema could not be generated."))
@@ -59,6 +64,12 @@ object AvroCodec {
           decode(chunk).map(Chunk(_))
         )
       }
+
+      override def encodeGenericRecord(value: A)(implicit schema: Schema[A]): GenericData.Record =
+        encodeValue(value, schema).asInstanceOf[GenericData.Record]
+
+      override def decodeGenericRecord(value: GenericRecord)(implicit schema: Schema[A]): Either[DecodeError, A] =
+        decodeValue(value, schema)
     }
 
   private def decodeValue[A](raw: Any, schema: Schema[A]): Either[DecodeError, A] = schema match {

diff --git a/zio-schema-avro/shared/src/test/scala-2/zio/schema/codec/AvroCodecSpec.scala b/zio-schema-avro/shared/src/test/scala-2/zio/schema/codec/AvroCodecSpec.scala
@@ -20,6 +20,8 @@ import java.time.{
 }
 import java.util.UUID
 
+import org.apache.avro.generic.GenericData
+
 import zio._
 import zio.schema.codec.AvroAnnotations.avroEnum
 import zio.schema.{ DeriveSchema, Schema }
@@ -140,7 +142,8 @@ object AvroCodecSpec extends ZIOSpecDefault {
     sequenceDecoderSpec,
     genericRecordDecoderSpec,
     enumDecoderSpec,
-    streamEncodingDecodingSpec
+    streamEncodingDecodingSpec,
+    genericRecordEncodeDecodeSpec
   )
 
   private val primitiveEncoderSpec = suite("Avro Codec - Encoder primitive spec")(
@@ -695,4 +698,13 @@ object AvroCodecSpec extends ZIOSpecDefault {
 
     })
 
+  private val genericRecordEncodeDecodeSpec = suite("AvroCodec - encode/decode Generic Record")(
+    test("Encode/Decode") {
+      val codec                       = AvroCodec.schemaBasedBinaryCodec[Record]
+      val generic: GenericData.Record = codec.encodeGenericRecord(Record("John", 42))
+      val result                      = codec.decodeGenericRecord(generic)
+      assertTrue(result == Right(Record("John", 42)))
+    }
+  )
+
 }