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

fetch serialName with kxs, minor formatting/refactoring

Browse source
This commit is contained in:
Adam 2021-12-03 22:18:17 +01:00
parent 42b60ac0c3
commit 29d190b8f4
2 changed files with 15 additions and 12 deletions
Download patch file Download diff file Expand all files Collapse all files

7
modules/events-processor/src/main/kotlin/dev/adamko/kafkatorio/processor/main.kt
View file

@ -21,7 +21,8 @@ import org.http4k.format.ConfigurableKotlinxSerialization
import org.http4k.format.asConfigurable
import org.http4k.format.withStandardMappings
object KSX : ConfigurableKotlinxSerialization({
object KXS : ConfigurableKotlinxSerialization({
ignoreUnknownKeys = true
prettyPrint = true
prettyPrintIndent = " "
@ -37,7 +38,6 @@ object JsonSerdes : Serde<FactorioEvent<FactorioObjectData>> {
override fun deserializer() = Deserializer<FactorioEvent<FactorioObjectData>> { _, bytes ->
jsonMapper.decodeFromString(bytes.decodeToString())
}
}
fun main() {
@ -53,7 +53,7 @@ fun main() {
.mapValues { readOnlyKey, value ->
println("Mapping $readOnlyKey:$value")
jsonMapper.decodeFromString<FactorioEvent<FactorioObjectData>>(value)
// KSX.asA<FactorioEvent<FactorioObjectData>>(value)
// KXS.asA<FactorioEvent<FactorioObjectData>>(value)
}
.to(
{ key, value, recordContext -> value.data.objectName },
@ -63,7 +63,6 @@ fun main() {
val topology = builder.build()
val properties = appProps.kafkaConfig
val streams = KafkaStreams(topology, properties)
streams.setUncaughtExceptionHandler(StreamsExceptionHandler())