handle logical types

api/avro4k-core.api

@@ -73,6 +73,7 @@ public abstract interface class com/github/avrokotlin/avro4k/AvroDecoder : kotli
 	public abstract fun decodeBytes ()[B
 	public abstract fun decodeFixed ()Lorg/apache/avro/generic/GenericFixed;
 	public abstract fun decodeValue ()Ljava/lang/Object;
+	public abstract fun getAvro ()Lcom/github/avrokotlin/avro4k/Avro;
 	public abstract fun getCurrentWriterSchema ()Lorg/apache/avro/Schema;
 }
 
@@ -117,6 +118,7 @@ public abstract interface class com/github/avrokotlin/avro4k/AvroEncoder : kotli
 	public abstract fun encodeBytes ([B)V
 	public abstract fun encodeFixed (Lorg/apache/avro/generic/GenericFixed;)V
 	public abstract fun encodeFixed ([B)V
+	public abstract fun getAvro ()Lcom/github/avrokotlin/avro4k/Avro;
 	public abstract fun getCurrentWriterSchema ()Lorg/apache/avro/Schema;
 }
 

src/main/kotlin/com/github/avrokotlin/avro4k/Avro.kt

@@ -1,6 +1,7 @@
 package com.github.avrokotlin.avro4k
 
 import com.github.avrokotlin.avro4k.internal.EnumResolver
+import com.github.avrokotlin.avro4k.internal.LogicalTypesSerializersCollector
 import com.github.avrokotlin.avro4k.internal.PolymorphicResolver
 import com.github.avrokotlin.avro4k.internal.RecordResolver
 import com.github.avrokotlin.avro4k.internal.schema.ValueVisitor
@@ -39,6 +40,10 @@ public sealed class Avro(
     internal val recordResolver = RecordResolver(this)
     internal val polymorphicResolver = PolymorphicResolver(serializersModule)
     internal val enumResolver = EnumResolver()
+    internal val logicalTypeSerializers: Map<String, KSerializer<Any>> =
+        LogicalTypesSerializersCollector(configuration)
+            .apply { serializersModule.dumpTo(this) }
+            .serializers
 
     public companion object Default : Avro(
         AvroConfiguration(),

src/main/kotlin/com/github/avrokotlin/avro4k/AvroDecoder.kt

@@ -45,6 +45,8 @@ public interface AvroDecoder : Decoder {
     @ExperimentalSerializationApi
     public val currentWriterSchema: Schema
 
+    public val avro: Avro
+
     /**
      * Decode a [Schema.Type.BYTES] value.
      *

src/main/kotlin/com/github/avrokotlin/avro4k/AvroEncoder.kt

@@ -33,6 +33,8 @@ public interface AvroEncoder : Encoder {
     @ExperimentalSerializationApi
     public val currentWriterSchema: Schema
 
+    public val avro: Avro
+
     /**
      * Encodes a [Schema.Type.BYTES] value from a [ByteBuffer].
      */

src/main/kotlin/com/github/avrokotlin/avro4k/internal/LogicalTypesSerializersCollector.kt

@@ -0,0 +1,45 @@
+package com.github.avrokotlin.avro4k.internal
+
+import com.github.avrokotlin.avro4k.AvroConfiguration
+import com.github.avrokotlin.avro4k.serializer.AvroSchemaSupplier
+import com.github.avrokotlin.avro4k.serializer.ElementLocation
+import com.github.avrokotlin.avro4k.serializer.SchemaSupplierContext
+import kotlinx.serialization.DeserializationStrategy
+import kotlinx.serialization.KSerializer
+import kotlinx.serialization.SerializationStrategy
+import kotlinx.serialization.modules.SerializersModuleCollector
+import kotlin.reflect.KClass
+
+internal class LogicalTypesSerializersCollector(
+    private val configuration: AvroConfiguration,
+) : SerializersModuleCollector {
+    internal var serializers: MutableMap<String, KSerializer<Any>> = mutableMapOf()
+
+    override fun <T : Any> contextual(kClass: KClass<T>, serializer: KSerializer<T>) {
+        if (serializer !is AvroSchemaSupplier || serializer.descriptor.isNullable) {
+            return
+        }
+        val schema = runCatching { serializer.getSchema(SimpleContext(configuration)) }.getOrNull()
+        if (schema?.logicalType != null) {
+            @Suppress("UNCHECKED_CAST")
+            serializers[schema.logicalType.name] = serializer as KSerializer<Any>
+        }
+    }
+
+    override fun <T : Any> contextual(kClass: KClass<T>, provider: (typeArgumentsSerializers: List<KSerializer<*>>) -> KSerializer<*>) {
+    }
+
+    override fun <Base : Any, Sub : Base> polymorphic(baseClass: KClass<Base>, actualClass: KClass<Sub>, actualSerializer: KSerializer<Sub>) {
+    }
+
+    override fun <Base : Any> polymorphicDefaultDeserializer(baseClass: KClass<Base>, defaultDeserializerProvider: (className: String?) -> DeserializationStrategy<Base>?) {
+    }
+
+    override fun <Base : Any> polymorphicDefaultSerializer(baseClass: KClass<Base>, defaultSerializerProvider: (value: Base) -> SerializationStrategy<Base>?) {
+    }
+}
+
+private class SimpleContext(
+    override val configuration: AvroConfiguration,
+    override val inlinedElements: List<ElementLocation> = emptyList(),
+) : SchemaSupplierContext

src/main/kotlin/com/github/avrokotlin/avro4k/internal/decoder/direct/AbstractAvroDirectDecoder.kt

@@ -39,7 +39,7 @@ import org.apache.avro.generic.GenericData
 import org.apache.avro.generic.GenericFixed
 
 internal abstract class AbstractAvroDirectDecoder(
-    protected val avro: Avro,
+    override val avro: Avro,
     protected val binaryDecoder: org.apache.avro.io.Decoder,
 ) : AbstractInterceptingDecoder(), UnionDecoder {
     abstract override var currentWriterSchema: Schema

src/main/kotlin/com/github/avrokotlin/avro4k/internal/decoder/generic/AbstractAvroGenericDecoder.kt

@@ -1,6 +1,5 @@
 package com.github.avrokotlin.avro4k.internal.decoder.generic
 
-import com.github.avrokotlin.avro4k.Avro
 import com.github.avrokotlin.avro4k.AvroDecoder
 import com.github.avrokotlin.avro4k.internal.BadDecodedValueError
 import com.github.avrokotlin.avro4k.internal.SerializerLocatorMiddleware
@@ -29,8 +28,6 @@ import java.math.BigDecimal
 import java.nio.ByteBuffer
 
 internal abstract class AbstractAvroGenericDecoder : AbstractDecoder(), AvroDecoder {
-    internal abstract val avro: Avro
-
     abstract override fun decodeNotNullMark(): Boolean
 
     abstract override fun decodeValue(): Any

src/main/kotlin/com/github/avrokotlin/avro4k/internal/encoder/direct/AbstractAvroDirectEncoder.kt

@@ -27,7 +27,7 @@ internal class AvroValueDirectEncoder(
 ) : AbstractAvroDirectEncoder(avro, binaryEncoder)
 
 internal sealed class AbstractAvroDirectEncoder(
-    protected val avro: Avro,
+    override val avro: Avro,
     protected val binaryEncoder: org.apache.avro.io.Encoder,
 ) : AbstractEncoder(), AvroEncoder, UnionEncoder {
     private var selectedUnionIndex: Int = -1

src/main/kotlin/com/github/avrokotlin/avro4k/internal/encoder/direct/RecordDirectEncoder.kt

@@ -79,7 +79,7 @@ private class RecordSequentialDirectEncoder(
 private class RecordBadOrderDirectEncoder(
     private val classDescriptor: ClassDescriptorForWriterSchema,
     private val schema: Schema,
-    private val avro: Avro,
+    override val avro: Avro,
     private val binaryEncoder: org.apache.avro.io.Encoder,
 ) : AbstractEncoder(), AvroEncoder {
     // Each time we encode a field, if the next expected schema field index is not the good one, it is buffered until it's the time to encode it

src/main/kotlin/com/github/avrokotlin/avro4k/internal/encoder/generic/AbstractAvroGenericEncoder.kt

@@ -1,6 +1,5 @@
 package com.github.avrokotlin.avro4k.internal.encoder.generic
 
-import com.github.avrokotlin.avro4k.Avro
 import com.github.avrokotlin.avro4k.AvroEncoder
 import com.github.avrokotlin.avro4k.UnionEncoder
 import com.github.avrokotlin.avro4k.encodeResolving
@@ -22,8 +21,6 @@ import org.apache.avro.generic.GenericFixed
 import java.nio.ByteBuffer
 
 internal abstract class AbstractAvroGenericEncoder : AbstractEncoder(), AvroEncoder, UnionEncoder {
-    abstract val avro: Avro
-
     abstract override var currentWriterSchema: Schema
 
     abstract override fun encodeValue(value: Any)

src/main/kotlin/com/github/avrokotlin/avro4k/serializer/GenericDataSerializers.kt

@@ -96,7 +96,10 @@ public object GenericDataSerializer : AvroSerializer<Any>("GenericData") {
     override fun deserializeAvro(decoder: AvroDecoder): Any {
         (decoder as UnionDecoder).decodeAndResolveUnion()
 
-        // TODO: logical types
+        decoder.currentWriterSchema.logicalType
+            ?.let { decoder.avro.logicalTypeSerializers[it.name] }
+            ?.let { return it.deserialize(decoder) }
+
         @Suppress("WHEN_ENUM_CAN_BE_NULL_IN_JAVA")
         return when (decoder.currentWriterSchema.type) {
             Schema.Type.DOUBLE -> decoder.decodeDouble()