diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes/binaryFormat.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes/binaryFormat.kt new file mode 100644 index 0000000..a1587f6 --- /dev/null +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes/binaryFormat.kt @@ -0,0 +1,50 @@ +package dev.adamko.kafkatorio.processor.serdes + +import kotlinx.serialization.BinaryFormat +import kotlinx.serialization.decodeFromByteArray +import kotlinx.serialization.encodeToByteArray +import org.apache.kafka.common.serialization.Deserializer +import org.apache.kafka.common.serialization.Serde +import org.apache.kafka.common.serialization.Serializer + + +inline fun BinaryFormat.kafkaSerializer() = + Serializer { topic: String, data: T -> + runCatching { + encodeToByteArray(data) + }.getOrElse { e -> + println( + """ + Exception on encodeToString + Topic: $topic + topicData: $data + topicData as T: $data + """.trimIndent() + ) + e.printStackTrace() + throw e + } + } + +inline fun BinaryFormat.kafkaDeserializer() = + Deserializer { topic: String, data: ByteArray -> + runCatching { + decodeFromByteArray(data) + }.getOrElse { e -> + println( + """ + Exception on decodeFromString, + Topic: $topic + topicData: $data + topicData as T: ${data as? T} + """.trimIndent() + ) + e.printStackTrace() + throw e + } + } + +inline fun BinaryFormat.serde() = object : Serde { + override fun serializer(): Serializer = kafkaSerializer() + override fun deserializer(): Deserializer = kafkaDeserializer() +} diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes/kxsJson.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes/kxsJson.kt index 4cb82b9..0096037 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes/kxsJson.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes/kxsJson.kt @@ -12,6 +12,7 @@ import org.http4k.format.ConfigurableKotlinxSerialization import org.http4k.format.asConfigurable import org.http4k.format.withStandardMappings + object KafkatorioPacketSerde : Serde { override fun serializer() = Serializer { _, message -> jsonMapper.encodeToString(message).encodeToByteArray() diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes/protobuf.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes/protobuf.kt index 527121f..8491485 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes/protobuf.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes/protobuf.kt @@ -1,59 +1,6 @@ package dev.adamko.kafkatorio.processor.serdes -import dev.adamko.kafkatorio.events.schema.jsonMapperKafkatorio -import kotlinx.serialization.BinaryFormat -import kotlinx.serialization.decodeFromByteArray -import kotlinx.serialization.encodeToByteArray import kotlinx.serialization.protobuf.ProtoBuf -import org.apache.kafka.common.serialization.Deserializer -import org.apache.kafka.common.serialization.Serde -import org.apache.kafka.common.serialization.Serializer - -val protoMapper: ProtoBuf = ProtoBuf { - serializersModule = jsonMapperKafkatorio.serializersModule -} -inline fun BinaryFormat.kafkaSerializer() = - Serializer { topic: String, data: T -> - runCatching { - encodeToByteArray(data) - }.getOrElse { e -> - println( - """ - Exception on encodeToString - Topic: $topic - topicData: $data - topicData as T: $data - """.trimIndent() - ) - e.printStackTrace() - throw e - } - } - -inline fun BinaryFormat.kafkaDeserializer() = - Deserializer { topic: String, data: ByteArray -> - runCatching { - decodeFromByteArray(data) - }.getOrElse { e -> - println( - """ - Exception on decodeFromString, - Topic: $topic - topicData: $data - topicData as T: ${data as? T} - """.trimIndent() - ) - e.printStackTrace() - throw e - } - } - -inline fun ProtoBuf.serde() = object : Serde { - - override fun serializer(): Serializer = kafkaSerializer() - - override fun deserializer(): Deserializer = kafkaDeserializer() - -} +val protoMapper: ProtoBuf = ProtoBuf { }