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

fixing 'when' statements, tidying up some code,

Browse source
This commit is contained in:
Adam 2022-02-20 22:30:56 +01:00
parent d14d0cd991
commit 7481c8162a
6 changed files with 17 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

15
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/admin/KafkatorioKafkaAdmin.kt
View file

@ -33,17 +33,19 @@ object KafkatorioKafkaAdmin {
add(KafkatorioTopology.TOPIC_GROUPED_MAP_CHUNKS)
add(KafkatorioTopology.TOPIC_SRC_SERVER_LOG)
val base = topicNameBase(packetType)
when (packetType) {
KafkatorioPacket.PacketType.EVENT ->
FactorioObjectData.ObjectName.values.forEach { objectName ->
add("kafkatorio.${packetType.name}.${objectName.name}")
add("$base.${objectName.name}")
}
KafkatorioPacket.PacketType.CONFIG ->
add("kafkatorio.${packetType.name}.FactorioConfigurationUpdate")
KafkatorioPacket.PacketType.PROTOTYPES ->
add("kafkatorio.${packetType.name}.all")
KafkatorioPacket.PacketType.CONFIG -> add("$base.FactorioConfigurationUpdate")
KafkatorioPacket.PacketType.PROTOTYPES -> add("$base.all")
KafkatorioPacket.PacketType.UPDATE -> add("$base.all")
}
}
}.minus(currentTopics.toSet())
@ -87,4 +89,7 @@ object KafkatorioKafkaAdmin {
.listings()
.get()
}
fun topicNameBase(packetType: KafkatorioPacket.PacketType): String =
"kafkatorio.${packetType.name}"
}

3
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioServerPackets.kt
View file

@ -6,6 +6,7 @@ import dev.adamko.kafkatorio.events.schema.FactorioEventUpdatePacket
import dev.adamko.kafkatorio.events.schema.FactorioPrototypes
import dev.adamko.kafkatorio.events.schema.KafkatorioPacket
import dev.adamko.kafkatorio.processor.KafkatorioTopology
import dev.adamko.kafkatorio.processor.admin.KafkatorioKafkaAdmin
import dev.adamko.kafkatorio.processor.serdes.jsonMapper
import dev.adamko.kotka.extensions.consumedAs
import dev.adamko.kotka.extensions.producedAs
@ -51,7 +52,7 @@ fun splitFactorioServerPacketStream(
) { _: FactorioServerId, value: KafkatorioPacket, _: RecordContext ->
// println("[$key] sending event:${value.eventType} to topic:${value.data.objectName()}")
val base = "kafkatorio.${value.packetType.name}"
val base = KafkatorioKafkaAdmin.topicNameBase(value.packetType)
val type: String = when (value) {
is FactorioEvent -> value.data.objectName.name