From 66fa1fb53a2ee0a2c953ba5e6c39207eda917365 Mon Sep 17 00:00:00 2001 From: Adam <897017+aSemy@users.noreply.github.com> Date: Wed, 9 Feb 2022 23:02:00 +0100 Subject: [PATCH] reuse the packet-splitter topics for the tile-saver job --- .../processor/KafkatorioTopology.kt | 40 ++++++++++-- .../processor/topology/factorioGameState.kt | 63 ++++++++++++------- .../processor/topology/prototypes.kt | 2 +- 3 files changed, 78 insertions(+), 27 deletions(-) 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 33d9a16..c0d86bb 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,19 @@ 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.FactorioPrototypes +import dev.adamko.kafkatorio.events.schema.KafkatorioPacket +import dev.adamko.kafkatorio.processor.serdes.jsonMapper +import dev.adamko.kafkatorio.processor.topology.FactorioServerId import dev.adamko.kafkatorio.processor.topology.factorioServerPacketStream import dev.adamko.kafkatorio.processor.topology.groupTilesIntoChunksWithColours import dev.adamko.kafkatorio.processor.topology.playerUpdatesToWsServer 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.kxs.serde import java.time.Duration import kotlin.coroutines.CoroutineContext import kotlinx.coroutines.CoroutineName @@ -18,6 +26,8 @@ import org.apache.kafka.streams.KafkaStreams 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 + class KafkatorioTopology( private val websocketServer: WebsocketServer, @@ -38,17 +48,37 @@ class KafkatorioTopology( private fun saveTiles() { val builder = StreamsBuilder() - val packets = factorioServerPacketStream(builder) - val tileProtoColourDict = tileProtoColourDictionary(packets) + + val protosStream: KStream = + builder.stream( + "kafkatorio.${KafkatorioPacket.PacketType.PROTOTYPES}.all", + consumedAs("consume.factorio-protos.all", jsonMapper.serde(), jsonMapper.serde()) + ) + val tileProtoColourDict = tileProtoColourDictionary(protosStream) + + + val mapTilesStream: KStream = + builder.stream( + "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.LuaTiles}", + consumedAs("consume.map-tiles", jsonMapper.serde(), jsonMapper.serde()) + ) + val mapChunksStream: KStream = + builder.stream( + "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.MapChunk}", + consumedAs("consume.map-chunks", jsonMapper.serde(), jsonMapper.serde()) + ) val tilePrototypesTable = groupTilesIntoChunksWithColours( - packets, + mapTilesStream, + mapChunksStream, tileProtoColourDict, ) + + saveMapTiles(tilePrototypesTable) - val topology = builder.build() + val topology = builder.build() launchTopology("saveTiles", topology) } @@ -74,7 +104,7 @@ class KafkatorioTopology( private fun launchTopology(id: String, topology: Topology) { - val props = appProps.kafkaConfig + val props = appProps.kafkaConfig val appId = props.compute(StreamsConfig.APPLICATION_ID_CONFIG) { _, v -> "$v.$id" } as? String props.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, appId) 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/factorioGameState.kt index c49d6fb..2130f77 100644 --- 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/factorioGameState.kt @@ -2,7 +2,6 @@ package dev.adamko.kafkatorio.processor.topology import dev.adamko.kafkatorio.events.schema.ColourHex import dev.adamko.kafkatorio.events.schema.FactorioEvent -import dev.adamko.kafkatorio.events.schema.KafkatorioPacket import dev.adamko.kafkatorio.events.schema.MapChunk import dev.adamko.kafkatorio.events.schema.MapChunkPosition import dev.adamko.kafkatorio.events.schema.MapTilePosition @@ -12,7 +11,10 @@ import dev.adamko.kafkatorio.processor.serdes.kxsBinary import dev.adamko.kotka.extensions.groupedAs import dev.adamko.kotka.extensions.materializedAs import dev.adamko.kotka.extensions.materializedWith +import dev.adamko.kotka.extensions.streams.filter import dev.adamko.kotka.extensions.streams.flatMap +import dev.adamko.kotka.extensions.streams.mapValues +import dev.adamko.kotka.extensions.streams.merge import dev.adamko.kotka.extensions.streams.reduce import dev.adamko.kotka.extensions.tables.join import dev.adamko.kotka.kxs.serde @@ -20,6 +22,7 @@ import kotlinx.serialization.Serializable import org.apache.kafka.streams.kstream.KStream import org.apache.kafka.streams.kstream.KTable + @Serializable data class ServerMapChunkId( val serverId: FactorioServerId, @@ -39,38 +42,54 @@ data class ServerMapChunkTiles( fun groupTilesIntoChunksWithColours( - factorioServerPacketStream: KStream, + mapTilePacketsStream: KStream, + mapChunksPacketsStream: KStream, tileProtoColourDict: KTable, ): KTable> { + + val mapChunkTilesStream: KStream = + mapChunksPacketsStream + .filter { _, packet: FactorioEvent? -> + packet != null + && packet.data is MapChunk + && !((packet.data as? MapChunk)?.tiles?.tiles.isNullOrEmpty()) + } + .mapValues("map-chunk-packets.convert-to-map-tiles") { _, packet: FactorioEvent -> + (packet.data as MapChunk).tiles + } + + val mapTilesStream: KStream = + mapTilePacketsStream + .filter { _, packet: FactorioEvent? -> + packet != null + && packet.data is MapTiles + && !((packet.data as? MapTiles)?.tiles.isNullOrEmpty()) + } + .mapValues("map-tiles-packets.convert-to-map-tiles") { _, packet: FactorioEvent -> + packet.data as MapTiles + } + // group tiles by server & surface & chunk val chunkedTilesTable: KTable> = - factorioServerPacketStream - .flatMap("server-map-data.tiles.flatMapByChunk") { key: FactorioServerId, packet: KafkatorioPacket -> - - val mapTileData = when (packet) { - is FactorioEvent -> - when (val data = packet.data) { - is MapChunk -> data.tiles - is MapTiles -> data - else -> null - } - else -> null - } + mapTilesStream + .merge("merge-map-chunk-tiles-with-map-tiles", mapChunkTilesStream) + .filter { _, value -> value != null && value.tiles.isNotEmpty() } + .flatMap("server-map-data.tiles.flatMapByChunk") { key: FactorioServerId, mapTiles: MapTiles -> val zoomLevel = ZoomLevel.ZOOM_0 - mapTileData - ?.tiles - ?.groupBy { tile -> + mapTiles + .tiles + .groupBy { tile -> tile.position.toMapChunkPosition(zoomLevel.chunkSize) } - ?.map { (chunkPos, tiles) -> + .map { (chunkPos, tiles) -> val chunkId = ServerMapChunkId( key, chunkPos, - SurfaceIndex(mapTileData.surfaceIndex), + SurfaceIndex(mapTiles.surfaceIndex), zoomLevel, ) @@ -79,8 +98,10 @@ fun groupTilesIntoChunksWithColours( val chunkTiles = ServerMapChunkTiles(chunkId, tilesMap) chunkId to chunkTiles - } ?: listOf() - + } + } + .filter("server-map-data.tiles.filter-not-empty") { _, value -> + value.map.isNotEmpty() } .groupByKey( groupedAs("server-map-data.tiles.group", kxsBinary.serde(), kxsBinary.serde()) diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/prototypes.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/prototypes.kt index 63ce083..b061bf7 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/prototypes.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/prototypes.kt @@ -40,7 +40,7 @@ value class TileColourDict(val map: Map) /** Get the latest map-tile colours per server. */ fun tileProtoColourDictionary( - factorioServerPacketStream: KStream + factorioServerPacketStream: KStream ): KTable { // get all the prototypes