From b37da373aee6048ce26848527b2cc288574d6ae3 Mon Sep 17 00:00:00 2001 From: Adam <897017+aSemy@users.noreply.github.com> Date: Wed, 9 Feb 2022 23:38:12 +0100 Subject: [PATCH] split up the tile-chunk-grouping and save-tile applications --- modules/events-server/build.gradle.kts | 5 +++ .../adamko/kafkatorio/admin/CreateTopics.kt | 6 ++- .../processor/KafkatorioTopology.kt | 45 +++++++++++++++---- .../topology/factorioServerPackets.kt | 2 +- ...factorioGameState.kt => groupMapChunks.kt} | 0 5 files changed, 48 insertions(+), 10 deletions(-) rename modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/{factorioGameState.kt => groupMapChunks.kt} (100%) diff --git a/modules/events-server/build.gradle.kts b/modules/events-server/build.gradle.kts index 5a35bec..8e140b2 100644 --- a/modules/events-server/build.gradle.kts +++ b/modules/events-server/build.gradle.kts @@ -90,6 +90,11 @@ val kafkatorioEventsServerKafkaForceReset by tasks.registering { logging.captureStandardOutput(LogLevel.LIFECYCLE) commandLine = parseSpaceSeparatedArgs(""" docker exec -d kafka $cmd """) } + exec { + val cmd = reset("kafkatorio-events-processor.groupTilesMapChunks") + logging.captureStandardOutput(LogLevel.LIFECYCLE) + commandLine = parseSpaceSeparatedArgs(""" docker exec -d kafka $cmd """) + } } } diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/admin/CreateTopics.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/admin/CreateTopics.kt index dd9b89b..c1784f6 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/admin/CreateTopics.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/admin/CreateTopics.kt @@ -32,7 +32,10 @@ object KafkaAdmin { val kafkatorioTopics = buildSet { KafkatorioPacket.PacketType.values().forEach { packetType -> - add(KafkatorioTopology.sourceTopic) + + add(KafkatorioTopology.TOPIC_GROUPED_MAP_CHUNKS) + add(KafkatorioTopology.TOPIC_SRC_SERVER_LOG) + when (packetType) { KafkatorioPacket.PacketType.EVENT -> FactorioObjectData.ObjectName.values().forEach { objectName -> @@ -43,6 +46,7 @@ object KafkaAdmin { KafkatorioPacket.PacketType.PROTOTYPES -> add("kafkatorio.${packetType.name}.all") } + } }.minus(currentTopics.toSet()) diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/KafkatorioTopology.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/KafkatorioTopology.kt index c0d86bb..efba5aa 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/KafkatorioTopology.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/KafkatorioTopology.kt @@ -1,11 +1,15 @@ package dev.adamko.kafkatorio.processor +import dev.adamko.kafkatorio.events.schema.ColourHex import dev.adamko.kafkatorio.events.schema.FactorioEvent import dev.adamko.kafkatorio.events.schema.FactorioObjectData import dev.adamko.kafkatorio.events.schema.FactorioPrototypes import dev.adamko.kafkatorio.events.schema.KafkatorioPacket import dev.adamko.kafkatorio.processor.serdes.jsonMapper +import dev.adamko.kafkatorio.processor.serdes.kxsBinary import dev.adamko.kafkatorio.processor.topology.FactorioServerId +import dev.adamko.kafkatorio.processor.topology.ServerMapChunkId +import dev.adamko.kafkatorio.processor.topology.ServerMapChunkTiles import dev.adamko.kafkatorio.processor.topology.factorioServerPacketStream import dev.adamko.kafkatorio.processor.topology.groupTilesIntoChunksWithColours import dev.adamko.kafkatorio.processor.topology.playerUpdatesToWsServer @@ -13,6 +17,8 @@ import dev.adamko.kafkatorio.processor.topology.saveMapTiles import dev.adamko.kafkatorio.processor.topology.splitFactorioServerPacketStream import dev.adamko.kafkatorio.processor.topology.tileProtoColourDictionary import dev.adamko.kotka.extensions.consumedAs +import dev.adamko.kotka.extensions.producedAs +import dev.adamko.kotka.extensions.tables.toStream import dev.adamko.kotka.kxs.serde import java.time.Duration import kotlin.coroutines.CoroutineContext @@ -27,6 +33,7 @@ import org.apache.kafka.streams.StreamsBuilder import org.apache.kafka.streams.StreamsConfig import org.apache.kafka.streams.Topology import org.apache.kafka.streams.kstream.KStream +import org.apache.kafka.streams.kstream.KTable class KafkatorioTopology( @@ -37,16 +44,18 @@ class KafkatorioTopology( Dispatchers.Default + SupervisorJob() + CoroutineName("KafkatorioTopology") companion object { - const val sourceTopic = "factorio-server-log" + const val TOPIC_SRC_SERVER_LOG = "factorio-server-log" + const val TOPIC_GROUPED_MAP_CHUNKS = "kafkatorio.MAP.grouped-map-chunks" } fun start() { + groupTilesMapChunks() saveTiles() splitPackets() playerUpdates() } - private fun saveTiles() { + private fun groupTilesMapChunks() { val builder = StreamsBuilder() @@ -68,15 +77,35 @@ class KafkatorioTopology( "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.MapChunk}", consumedAs("consume.map-chunks", jsonMapper.serde(), jsonMapper.serde()) ) - val tilePrototypesTable = groupTilesIntoChunksWithColours( - mapTilesStream, - mapChunksStream, - tileProtoColourDict, - ) + val groupedMapChunkTiles: KTable> = + groupTilesIntoChunksWithColours( + mapTilesStream, + mapChunksStream, + tileProtoColourDict, + ) + + groupedMapChunkTiles + .toStream("stream-grouped-map-tiles") + .to( + TOPIC_GROUPED_MAP_CHUNKS, + producedAs("produce.grouped-map-chunks", kxsBinary.serde(), kxsBinary.serde()) + ) - saveMapTiles(tilePrototypesTable) + val topology = builder.build() + launchTopology("groupTilesMapChunks", topology) + } + private fun saveTiles() { + val builder = StreamsBuilder() + + val groupedMapChunkTiles: KTable> = + builder.table( + TOPIC_GROUPED_MAP_CHUNKS, + consumedAs("consume.grouped-map-chunks", kxsBinary.serde(), kxsBinary.serde()) + ) + + saveMapTiles(groupedMapChunkTiles) val topology = builder.build() launchTopology("saveTiles", topology) diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioServerPackets.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioServerPackets.kt index 7731256..d1869d3 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioServerPackets.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioServerPackets.kt @@ -23,7 +23,7 @@ fun factorioServerPacketStream( ): KStream { return builder.stream( - KafkatorioTopology.sourceTopic, + KafkatorioTopology.TOPIC_SRC_SERVER_LOG, consumedAs( "read-raw-packets-from-server", Serdes.String(), diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioGameState.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/groupMapChunks.kt similarity index 100% rename from modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioGameState.kt rename to modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/groupMapChunks.kt