lazzexus/Kafkatorio
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

tweaking the kxs serde stuff

Browse source
This commit is contained in:
Adam 2022-01-23 14:00:51 +01:00
parent 960782dc90
commit 28f985bdca
3 changed files with 52 additions and 54 deletions
Download patch file Download diff file Expand all files Collapse all files

50
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes/binaryFormat.kt Normal file
View file

@ -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 <reified T> 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 <reified T> BinaryFormat.kafkaDeserializer() =
Deserializer { topic: String, data: ByteArray ->
runCatching {
decodeFromByteArray<T>(data)
}.getOrElse { e ->
println(
"""
Exception on decodeFromString,
Topic: $topic
topicData: $data
topicData as T: ${data as? T}
""".trimIndent()
)
e.printStackTrace()
throw e
}
}
inline fun <reified T> BinaryFormat.serde() = object : Serde<T> {
override fun serializer(): Serializer<T> = kafkaSerializer()
override fun deserializer(): Deserializer<T> = kafkaDeserializer()
}

1
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes/kxsJson.kt
View file

@ -12,6 +12,7 @@ import org.http4k.format.ConfigurableKotlinxSerialization
import org.http4k.format.asConfigurable
import org.http4k.format.withStandardMappings
object KafkatorioPacketSerde : Serde<KafkatorioPacket> {
override fun serializer() = Serializer<KafkatorioPacket> { _, message ->
jsonMapper.encodeToString<KafkatorioPacket>(message).encodeToByteArray()

55
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes/protobuf.kt
View file

@ -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 <reified T> 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 <reified T> BinaryFormat.kafkaDeserializer() =
Deserializer { topic: String, data: ByteArray ->
runCatching {
decodeFromByteArray<T>(data)
}.getOrElse { e ->
println(
"""
Exception on decodeFromString,
Topic: $topic
topicData: $data
topicData as T: ${data as? T}
""".trimIndent()
)
e.printStackTrace()
throw e
}
}
inline fun <reified T> ProtoBuf.serde() = object : Serde<T> {
override fun serializer(): Serializer<T> = kafkaSerializer()
override fun deserializer(): Deserializer<T> = kafkaDeserializer()
}
val protoMapper: ProtoBuf = ProtoBuf { }