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

improve event-schema polymorphism with enums

Browse source
This commit is contained in:
Adam 2022-01-08 23:09:27 +01:00
parent 3875b9c309
commit 51e0aff823
18 changed files with 120 additions and 67 deletions
Download patch file Download diff file Expand all files Collapse all files

14
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/FactorioEventsTopology.kt
View file

@ -1,7 +1,8 @@
package dev.adamko.kafkatorio.processor
import dev.adamko.kafkatorio.events.schema.FactorioConfigurationUpdate
import dev.adamko.kafkatorio.events.schema.FactorioEvent
import dev.adamko.kafkatorio.events.schema.FactorioObjectData
import dev.adamko.kafkatorio.events.schema.KafkatorioPacket
import java.time.Duration
import kotlinx.serialization.decodeFromString
import kotlinx.serialization.encodeToString
@ -27,7 +28,7 @@ class FactorioEventsTopology(
)
.mapValues { readOnlyKey, value ->
println("Mapping $readOnlyKey:$value")
jsonMapper.decodeFromString<FactorioEvent<FactorioObjectData>>(value)
jsonMapper.decodeFromString<KafkatorioPacket>(value)
}
.peek { _, value ->
websocketServer.sendMessage(jsonMapper.encodeToString(value))
@ -35,9 +36,14 @@ class FactorioEventsTopology(
.to(
{ _, value, _ ->
// println("[$key] sending event:${value.eventType} to topic:${value.data.objectName()}")
value.data.objectName
when (value) {
is FactorioEvent<*> ->
"kafkatorio.${value.packetType.name}.${value.data.objectName.name}"
is FactorioConfigurationUpdate ->
"kafkatorio.${value.packetType.name}"
}
},
Produced.with(Serdes.String(), JsonSerdes)
Produced.with(Serdes.String(), KafkatorioPacketSerde)
)
val topology = builder.build()

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

@ -16,4 +16,5 @@ fun main() = runBlocking {
topology.build()
wsServer.build().asServer(Undertow(9073)).start().block()
}

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

@ -1,7 +1,6 @@
package dev.adamko.kafkatorio.processor
import dev.adamko.kafkatorio.events.schema.FactorioEvent
import dev.adamko.kafkatorio.events.schema.FactorioObjectData
import dev.adamko.kafkatorio.events.schema.KafkatorioPacket
import kotlinx.serialization.decodeFromString
import kotlinx.serialization.encodeToString
import kotlinx.serialization.json.Json
@ -14,12 +13,12 @@ import org.http4k.format.asConfigurable
import org.http4k.format.withStandardMappings
object JsonSerdes : Serde<FactorioEvent<FactorioObjectData>> {
override fun serializer() = Serializer<FactorioEvent<FactorioObjectData>> { _, message ->
jsonMapper.encodeToString<FactorioEvent<FactorioObjectData>>(message).encodeToByteArray()
object KafkatorioPacketSerde : Serde<KafkatorioPacket> {
override fun serializer() = Serializer<KafkatorioPacket> { _, message ->
jsonMapper.encodeToString<KafkatorioPacket>(message).encodeToByteArray()
}
override fun deserializer() = Deserializer<FactorioEvent<FactorioObjectData>> { _, bytes ->
override fun deserializer() = Deserializer<KafkatorioPacket> { _, bytes ->
jsonMapper.decodeFromString(bytes.decodeToString())
}
}