From 8bb73b8291285ab7bbf942522c27159500e49a94 Mon Sep 17 00:00:00 2001 From: Adam <897017+aSemy@users.noreply.github.com> Date: Sun, 13 Mar 2022 11:06:30 +0100 Subject: [PATCH] remove commented out code, try using materializedAs to fix KTable-KTable join not working, make 'saveMapTiles' more encapsulated --- .../processor/KafkatorioTopology.kt | 18 +-- .../processor/topology/flowMapTiles.kt | 24 +++- .../processor/topology/groupMapChunks.kt | 116 ++++++++---------- .../processor/topology/prototypes.kt | 4 +- .../processor/topology/GroupMapChunksTest.kt | 6 +- 5 files changed, 78 insertions(+), 90 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 076db30..1b22f49 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,17 +1,11 @@ package dev.adamko.kafkatorio.processor -import dev.adamko.kafkatorio.events.schema.ColourHex import dev.adamko.kafkatorio.processor.config.ApplicationProperties -import dev.adamko.kafkatorio.processor.serdes.kxsBinary -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.groupMapChunks import dev.adamko.kafkatorio.processor.topology.playerUpdatesToWsServer import dev.adamko.kafkatorio.processor.topology.saveMapTiles import dev.adamko.kafkatorio.processor.topology.splitFactorioServerPacketStream -import dev.adamko.kotka.extensions.consumedAs -import dev.adamko.kotka.kxs.serde import java.time.Duration import kotlin.coroutines.CoroutineContext import kotlin.coroutines.cancellation.CancellationException @@ -26,7 +20,6 @@ import org.apache.kafka.streams.StreamsBuilder import org.apache.kafka.streams.StreamsConfig import org.apache.kafka.streams.Topology import org.apache.kafka.streams.TopologyDescription -import org.apache.kafka.streams.kstream.KTable internal class KafkatorioTopology( @@ -57,16 +50,7 @@ internal class KafkatorioTopology( 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() + val topology = saveMapTiles(builder) launchTopology("saveTiles", topology) } diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/flowMapTiles.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/flowMapTiles.kt index 69c5304..cab13f7 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/flowMapTiles.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/flowMapTiles.kt @@ -9,7 +9,11 @@ import dev.adamko.kafkatorio.events.schema.MapChunkPosition import dev.adamko.kafkatorio.events.schema.MapTilePosition import dev.adamko.kafkatorio.events.schema.SurfaceIndex import dev.adamko.kafkatorio.events.schema.converters.toMapChunkPosition -import dev.adamko.kotka.extensions.streams.* +import dev.adamko.kafkatorio.processor.KafkatorioTopology +import dev.adamko.kafkatorio.processor.serdes.kxsBinary +import dev.adamko.kotka.extensions.consumedAs +import dev.adamko.kotka.extensions.streams.forEach +import dev.adamko.kotka.kxs.serde import java.awt.image.BufferedImage import java.io.File import kotlin.coroutines.CoroutineContext @@ -33,7 +37,9 @@ import kotlinx.coroutines.job import kotlinx.coroutines.launch import kotlinx.coroutines.runBlocking import kotlinx.coroutines.supervisorScope -import org.apache.kafka.streams.kstream.KTable +import org.apache.kafka.streams.StreamsBuilder +import org.apache.kafka.streams.Topology +import org.apache.kafka.streams.kstream.KStream private const val WEB_MAP_TILE_SIZE_PX = 256 @@ -48,11 +54,16 @@ private val serverMapChunkHandler = ServerMapChunkHandler() fun saveMapTiles( - serverMapTable: KTable?> -) { + builder: StreamsBuilder, +): Topology { - serverMapTable - .toStream() + val groupedMapChunkTiles: KStream> = + builder.stream( + KafkatorioTopology.TOPIC_GROUPED_MAP_CHUNKS, + consumedAs("consume.grouped-map-chunks", kxsBinary.serde(), kxsBinary.serde()) + ) + + groupedMapChunkTiles .forEach("save-chunked-tiles") { _, value -> if (value != null) { runBlocking(Dispatchers.Default) { @@ -62,6 +73,7 @@ fun saveMapTiles( } } + return builder.build() } diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/groupMapChunks.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/groupMapChunks.kt index 1858986..d9b0bbd 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/groupMapChunks.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/groupMapChunks.kt @@ -18,12 +18,20 @@ import dev.adamko.kafkatorio.processor.serdes.jsonMapper import dev.adamko.kafkatorio.processor.serdes.kxsBinary import dev.adamko.kotka.extensions.consumedAs import dev.adamko.kotka.extensions.groupedAs -import dev.adamko.kotka.extensions.materializedWith +import dev.adamko.kotka.extensions.materializedAs import dev.adamko.kotka.extensions.producedAs -import dev.adamko.kotka.extensions.streams.* +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.peek +import dev.adamko.kotka.extensions.streams.reduce import dev.adamko.kotka.extensions.tables.join import dev.adamko.kotka.extensions.tables.toStream import dev.adamko.kotka.kxs.serde +import kotlin.coroutines.CoroutineContext +import kotlinx.coroutines.CoroutineName +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.SupervisorJob import kotlinx.serialization.Serializable import org.apache.kafka.streams.StreamsBuilder import org.apache.kafka.streams.Topology @@ -104,12 +112,21 @@ fun groupMapChunks( groupedMapChunkTiles .toStream("stream-grouped-map-tiles") + .filter("stream-grouped-map-tiles.filter-not-null") { _, v -> + !v?.map.isNullOrEmpty() + } + .mapValues("stream-grouped-map-tiles.map-not-null") { _, v -> + v as ServerMapChunkTiles + } + .peek("stream-grouped-map-tiles.peek") { _, v -> + println("Grouping map tiles result: ${v.chunkId} / ${v.map.size}") + } .to( KafkatorioTopology.TOPIC_GROUPED_MAP_CHUNKS, producedAs( "produce.grouped-map-chunks", - kxsBinary.serde(), - kxsBinary.serde?>() + kxsBinary.serde(), + kxsBinary.serde>() ) ) @@ -117,38 +134,10 @@ fun groupMapChunks( } private fun groupTilesIntoChunksWithColours( -// mapTilePacketsStream: KStream, -// mapChunksPacketsStream: KStream, mapChunkUpdatePacketsStream: KStream, tileProtoColourDict: KTable, ): KTable> { - -// val mapChunkTilesStream: KStream = -// mapChunksPacketsStream -// .filter { _, packet: FactorioEvent? -> -// when (val data = packet?.data) { -// is MapChunk -> data.tiles.tiles.isNotEmpty() -// else -> false -// } -// } -// .mapValues("map-chunk-packets.convert-to-map-tiles") { _, packet: FactorioEvent -> -// (packet.data as MapChunk).tiles -// } -// -// -// val mapTilesStream: KStream = -// mapTilePacketsStream -// .filter { _, packet: FactorioEvent? -> -// when (val data = packet?.data) { -// is MapTiles -> data.tiles.isNotEmpty() -// else -> false -// } -// } -// .mapValues("map-tiles-packets.convert-to-map-tiles") { _, packet: FactorioEvent -> -// packet.data as MapTiles -// } - val mapChunkUpdatesStream: KStream = mapChunkUpdatePacketsStream .filter { _, packet: FactorioEventUpdatePacket? -> @@ -174,37 +163,38 @@ private fun groupTilesIntoChunksWithColours( } - // group tiles by server & surface & chunk val chunkedTilesTable: KTable> = mapChunkUpdatesStream -// mapTilesStream -// .merge("merge-mapTilesStream-with-mapChunkTilesStream", mapChunkTilesStream) -// .merge("merge-mapTilesStream-with-mapChunkUpdatesStream", mapChunkUpdatesStream) -// .filter { _, value -> !value?.tiles.isNullOrEmpty() } + // group tiles by server & surface & chunk .flatMap("server-map-data.tiles.flatMapByChunk") { serverId: FactorioServerId, mapTiles: MapTiles -> val standardChunkSize = ChunkSize.MAX - mapTiles - .tiles - .groupBy { tile -> - tile.position.toMapChunkPosition(standardChunkSize.tilesPerChunk) - } - .map { (chunkPos, tiles) -> + val flatMappedChunkTiles: List>> = + mapTiles + .tiles + .groupBy { tile -> + tile.position.toMapChunkPosition(standardChunkSize.tilesPerChunk) + } + .map { (chunkPos, tiles) -> - val chunkId = ServerMapChunkId( - serverId = serverId, - chunkPosition = chunkPos, - surfaceIndex = mapTiles.surfaceIndex, - chunkSize = standardChunkSize, - ) + val chunkId = ServerMapChunkId( + serverId = serverId, + chunkPosition = chunkPos, + surfaceIndex = mapTiles.surfaceIndex, + chunkSize = standardChunkSize, + ) - val tilesMap = tiles.associate { it.position to TileProtoHashCode(it) } + val tilesMap = tiles.associate { it.position to TileProtoHashCode(it) } - val chunkTiles = ServerMapChunkTiles(chunkId, tilesMap) + val chunkTiles = ServerMapChunkTiles(chunkId, tilesMap) - chunkId to chunkTiles - } + chunkId to chunkTiles + } + +// println("[${System.currentTimeMillis()}] Flat-mapped ${flatMappedChunkTiles.size} chunk, with ${flatMappedChunkTiles.joinToString { it.second.map.size.toString() }} tiles") + + flatMappedChunkTiles } .filter("server-map-data.tiles.filter-not-empty") { _, value -> value.map.isNotEmpty() @@ -212,18 +202,18 @@ private fun groupTilesIntoChunksWithColours( .groupByKey( groupedAs( "server-map-data.tiles.group", - kxsBinary.serde(), - kxsBinary.serde?>(), + kxsBinary.serde(), + kxsBinary.serde>(), ) ) .reduce( - "server-map-data.tiles.reduce", materializedWith( - // "server-map-data.tiles.store", + "server-map-data.tiles.reduce", + materializedAs( + "server-map-data.tiles.store", kxsBinary.serde(), kxsBinary.serde>(), - ), - ServerMapChunkTiles::plus - ) + ) + ) { chunkTiles, other -> chunkTiles + other } val colourisedChunkTable: KTable> = @@ -231,7 +221,8 @@ private fun groupTilesIntoChunksWithColours( .join( other = tileProtoColourDict, name = "server-map-data.join-tiles-with-prototypes", - materialized = materializedWith( + materialized = materializedAs( + "server-map-data.join-tiles-with-prototypes.store", kxsBinary.serde(), kxsBinary.serde>(), ), @@ -251,11 +242,12 @@ private fun groupTilesIntoChunksWithColours( println("missing ${missingProtos.size} tile prototypes: ${missingProtos.joinToString { "${it.code}" }}") } + println("Set tile colours for chunk ${tiles.chunkId}") + ServerMapChunkTiles(tiles.chunkId, tileColours) } return colourisedChunkTable - } val MapTile.position 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 ad25cfb..9f307eb 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 @@ -56,8 +56,8 @@ fun tileProtoColourDictionary( } .toTable( "server-map-data.tile-prototypes", - materializedWith( -// "server-map-data.tile-prototypes.store", + materializedAs( + "server-map-data.tile-prototypes.store", kxsBinary.serde(), kxsBinary.serde(), ) diff --git a/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/topology/GroupMapChunksTest.kt b/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/topology/GroupMapChunksTest.kt index bed9571..4d46be7 100644 --- a/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/topology/GroupMapChunksTest.kt +++ b/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/topology/GroupMapChunksTest.kt @@ -95,12 +95,12 @@ class GroupMapChunksTest : FunSpec({ ) ) - val outputTopic: TestOutputTopic?> = + val outputTopic: TestOutputTopic> = testDriver .createOutputTopic( KafkatorioTopology.TOPIC_GROUPED_MAP_CHUNKS, - kxsBinary.serde().deserializer(), - kxsBinary.serde?>().deserializer(), + kxsBinary.serde().deserializer(), + kxsBinary.serde>().deserializer(), ) val records = outputTopic.readRecordsToList()