From f0590b9df4d08114cd08130771ac17f49a848c88 Mon Sep 17 00:00:00 2001 From: Adam <897017+aSemy@users.noreply.github.com> Date: Sat, 14 May 2022 12:50:13 +0200 Subject: [PATCH] mad wicked sick tile grouping optimisations --- .../kafkatorio/schema/common/positions.kt | 103 +++++- .../processor/KafkatorioTopology.kt | 3 +- .../processor/topology/colourMapChunks.kt | 313 ++++++++++++++++++ .../processor/topology/groupMapChunks.kt | 5 +- .../kafkatorio/processor/topology/model.kt | 35 -- .../processor/topology/playerUpdates.kt | 9 +- .../processor/topology/prototypes.kt | 10 +- .../processor/topology/saveMapTiles.kt | 100 +++--- .../processor/topology/SaveMapChunksTest.kt | 5 +- 9 files changed, 474 insertions(+), 109 deletions(-) create mode 100644 modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/colourMapChunks.kt diff --git a/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/schema/common/positions.kt b/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/schema/common/positions.kt index 16a0739..1d63eee 100644 --- a/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/schema/common/positions.kt +++ b/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/schema/common/positions.kt @@ -3,6 +3,8 @@ package dev.adamko.kafkatorio.schema.common import dev.adamko.kxstsgen.core.* import dev.adamko.kxstsgen.core.experiments.TupleSerializer import kotlin.math.floor +import kotlin.math.pow +import kotlin.math.roundToInt import kotlinx.serialization.Serializable /** @@ -58,6 +60,33 @@ data class MapChunkPosition( ) } } + + private val bounds: MapChunkPositionBounds by lazy { + MapChunkPositionBounds(this, ChunkSize.CHUNK_032) + } + + operator fun contains(tilePosition: MapTilePosition): Boolean { + return tilePosition in bounds + } +} + + +@Serializable +data class MapChunkPositionBounds( + val leftTop: MapTilePosition, + val rightBottom: MapTilePosition, +) : ClosedRange { + + constructor(chunkPosition: MapChunkPosition, chunkSize: ChunkSize) : this( + chunkPosition.leftTopTile(chunkSize), + chunkPosition.rightBottomTile(chunkSize), + ) + + override val start: MapTilePosition + get() = leftTop + + override val endInclusive: MapTilePosition + get() = rightBottom } @@ -71,7 +100,7 @@ data class MapChunkPosition( data class MapTilePosition( val x: Int, val y: Int, -) { +) : Comparable { object Serializer : TupleSerializer( "MapTilePosition", { @@ -85,6 +114,12 @@ data class MapTilePosition( return MapTilePosition(x, y) } } + + override operator fun compareTo(other: MapTilePosition): Int = + when { + other.x != x -> x.compareTo(other.x) + else -> y.compareTo(other.y) + } } @@ -100,21 +135,28 @@ operator fun MapTilePosition.div(divisor: Int) = operator fun MapTilePosition.plus(addend: Int) = MapTilePosition(x + addend, y + addend) +operator fun MapTilePosition.plus(addend: MapTilePosition) = + MapTilePosition(x + addend.x, y + addend.y) -fun MapTilePosition.toMapChunkPosition(chunkSize: Int = MAP_CHUNK_SIZE) = + +operator fun MapTilePosition.minus(subtrahend: MapTilePosition) = + MapTilePosition(x - subtrahend.x, y - subtrahend.y) + + +fun MapTilePosition.toMapChunkPosition(chunkSize: ChunkSize) = MapChunkPosition( - floor(x.toDouble() / chunkSize.toDouble()).toInt(), - floor(y.toDouble() / chunkSize.toDouble()).toInt(), + floor(x.toDouble() / chunkSize.lengthInTiles.toDouble()).toInt(), + floor(y.toDouble() / chunkSize.lengthInTiles.toDouble()).toInt(), ) fun MapEntityPosition.toMapTilePosition() = MapChunkPosition(floor(x).toInt(), floor(y).toInt()) -fun MapEntityPosition.toMapChunkPosition(chunkSize: Int = MAP_CHUNK_SIZE) = +fun MapEntityPosition.toMapChunkPosition(chunkSize: ChunkSize) = MapChunkPosition( - floor(x / chunkSize.toDouble()).toInt(), - floor(y / chunkSize.toDouble()).toInt(), + floor(x / chunkSize.lengthInTiles.toDouble()).toInt(), + floor(y / chunkSize.lengthInTiles.toDouble()).toInt(), ) @@ -128,18 +170,17 @@ operator fun MapChunkPosition.minus(subtrahend: Int) = MapChunkPosition(x - subtrahend, y - subtrahend) -fun MapChunkPosition.leftTopTile(chunkSize: Int = MAP_CHUNK_SIZE): MapTilePosition = +fun MapChunkPosition.leftTopTile(chunkSize: ChunkSize): MapTilePosition = MapTilePosition( - x * chunkSize, - y * chunkSize, + x * chunkSize.lengthInTiles, + y * chunkSize.lengthInTiles, ) - -fun MapChunkPosition.rightBottomTile(chunkSize: Int = MAP_CHUNK_SIZE): MapTilePosition = - leftTopTile(chunkSize) + (chunkSize - 1) +fun MapChunkPosition.rightBottomTile(chunkSize: ChunkSize): MapTilePosition = + leftTopTile(chunkSize) + (chunkSize.lengthInTiles - 1) -fun MapChunkPosition.iterateTiles(chunkSize: Int = MAP_CHUNK_SIZE): Iterator { +fun MapChunkPosition.iterateTiles(chunkSize: ChunkSize): Iterator { val leftTop = leftTopTile(chunkSize) val rightBottom = rightBottomTile(chunkSize) return iterator { @@ -158,3 +199,37 @@ fun MapChunkPosition.iterateTiles(chunkSize: Int = MAP_CHUNK_SIZE): Iterator { + CHUNK_512(-1), + CHUNK_256(0), + CHUNK_128(1), + CHUNK_064(2), + CHUNK_032(3), + ; + + /** 32, 64, 128, 256, or 512 */ + val lengthInTiles: Int = 2f.pow(8 - zoomLevel).roundToInt() + + init { + require(lengthInTiles > 0) { "tilesPerChunk $lengthInTiles must be positive" } + // 1000 & 0111 = 0000 => pow^2 + // 1001 & 1000 = 1000 => !pow^2 + require(lengthInTiles and (lengthInTiles - 1) == 0) { + "tilesPerChunk $lengthInTiles must be a power-of-two number" + } + } + + companion object { + // cache values for better performance. KT-48872 + val entries: Set = values().toSet() + + val MAX: ChunkSize = entries.maxByOrNull { it.lengthInTiles }!! + val MIN: ChunkSize = entries.minByOrNull { it.lengthInTiles }!! + val STANDARD: ChunkSize = entries.first { it.zoomLevel == 0 } + } +} 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 5bea0c6..359ee48 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,6 +1,7 @@ package dev.adamko.kafkatorio.processor import dev.adamko.kafkatorio.processor.config.ApplicationProperties +import dev.adamko.kafkatorio.processor.topology.colourMapChunks import dev.adamko.kafkatorio.processor.topology.factorioServerPacketStream import dev.adamko.kafkatorio.processor.topology.groupMapChunks import dev.adamko.kafkatorio.processor.topology.playerUpdatesToWsServer @@ -54,7 +55,7 @@ internal class KafkatorioTopology( private fun groupTilesMapChunks() { val builder = StreamsBuilder() - val topology = groupMapChunks(builder) + val topology = colourMapChunks(builder) launchTopology("groupTilesMapChunks", topology) } diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/colourMapChunks.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/colourMapChunks.kt new file mode 100644 index 0000000..050d3a7 --- /dev/null +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/colourMapChunks.kt @@ -0,0 +1,313 @@ +package dev.adamko.kafkatorio.processor.topology + +import dev.adamko.kafkatorio.processor.admin.TOPIC_GROUPED_MAP_CHUNKS_STATE +import dev.adamko.kafkatorio.processor.serdes.kxsBinary +import dev.adamko.kafkatorio.schema.common.ChunkSize +import dev.adamko.kafkatorio.schema.common.ColourHex +import dev.adamko.kafkatorio.schema.common.leftTopTile +import dev.adamko.kafkatorio.schema.common.toMapChunkPosition +import dev.adamko.kafkatorio.schema.packets.MapChunkUpdate +import dev.adamko.kafkatorio.schema.packets.PrototypesUpdate +import dev.adamko.kotka.extensions.consumedAs +import dev.adamko.kotka.extensions.groupedAs +import dev.adamko.kotka.extensions.materializedAs +import dev.adamko.kotka.extensions.producedAs +import dev.adamko.kotka.extensions.repartitionedAs +import dev.adamko.kotka.extensions.streams.filter +import dev.adamko.kotka.extensions.streams.map +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.tableJoined +import dev.adamko.kotka.extensions.tables.join +import dev.adamko.kotka.extensions.tables.toStream +import dev.adamko.kotka.kxs.serde +import kotlin.collections.component1 +import kotlin.collections.component2 +import org.apache.kafka.streams.StreamsBuilder +import org.apache.kafka.streams.Topology +import org.apache.kafka.streams.kstream.KStream +import org.apache.kafka.streams.kstream.KTable + + +private const val pid: String = "colourMapChunks" + + +/** Add colour to each [MapChunkUpdate] */ +fun colourMapChunks(builder: StreamsBuilder): Topology { + + val mapChunksStream: KStream = builder.streamPacketData() + val protosStream: KStream = builder.streamPacketData() + + val protosTable: KTable = + protosStream.createTilePrototypeTable() + + val chunkTilesColouredTable032: KTable> = + mapChunksStream + .convertToServerMapChunkTiles() + .reduceServerMapChunkTilesToTable() + .enrichWithColourData(protosTable) + + chunkTilesColouredTable032.streamTo(ChunkSize.CHUNK_032, TOPIC_GROUPED_MAP_CHUNKS_STATE) + + val groupedMapChunks = builder.stream( + TOPIC_GROUPED_MAP_CHUNKS_STATE, + consumedAs( + "$pid.consume.grouped-map-chunks", + kxsBinary.serde(), + kxsBinary.serde>(), + ) + ) + + mapOf( + ChunkSize.CHUNK_032 to ChunkSize.CHUNK_064, + ChunkSize.CHUNK_064 to ChunkSize.CHUNK_128, + ChunkSize.CHUNK_128 to ChunkSize.CHUNK_256, + ChunkSize.CHUNK_256 to ChunkSize.CHUNK_512, + ).forEach { (from, to) -> + groupedMapChunks + .filter("$pid.combine-chunks.filter-$from-to-$to") { key, _ -> + key?.chunkSize == from + }.groupByChunkPosition(to) + .streamTo(to, TOPIC_GROUPED_MAP_CHUNKS_STATE) + } + +// val chunkTilesColouredTable064 = +// chunkTilesColouredTable032.groupByChunkPosition(ChunkSize.CHUNK_064) +// chunkTilesColouredTable064.streamTo(ChunkSize.CHUNK_064, TOPIC_GROUPED_MAP_CHUNKS_STATE) +// +// val chunkTilesColouredTable128 = +// chunkTilesColouredTable064.groupByChunkPosition(ChunkSize.CHUNK_128) +// chunkTilesColouredTable128.streamTo(ChunkSize.CHUNK_128, TOPIC_GROUPED_MAP_CHUNKS_STATE) +// +// val chunkTilesColouredTable256 = +// chunkTilesColouredTable128.groupByChunkPosition(ChunkSize.CHUNK_256) +// chunkTilesColouredTable256.streamTo(ChunkSize.CHUNK_256, TOPIC_GROUPED_MAP_CHUNKS_STATE) +// +// val chunkTilesColouredTable512 = +// chunkTilesColouredTable256.groupByChunkPosition(ChunkSize.CHUNK_512) +// chunkTilesColouredTable512.streamTo(ChunkSize.CHUNK_512, TOPIC_GROUPED_MAP_CHUNKS_STATE) + + return builder.build() +} + + +private fun KStream.convertToServerMapChunkTiles() + : KStream> { + + val pid = "$pid.convertToMapTiles" + + return map("$pid.map") { serverId: FactorioServerId, update: MapChunkUpdate -> + + val mapTileList = update.tileDictionary?.toMapTileList() ?: emptyList() + + val (validMapTilesList, invalidMapTilesList) = mapTileList.partition { + it.position in update.key.chunkPosition + } + + if (invalidMapTilesList.isNotEmpty()) { + println("WARNING [reduceMapTilesToTable] MapChunkUpdate contained out-of-bounds tiles $invalidMapTilesList") + } + + val validMapTilesToProtoHashCode = + validMapTilesList.associate { it.position to TileProtoHashCode(it) } + + val chunkId = ServerMapChunkId( + serverId = serverId, + chunkPosition = update.key.chunkPosition, + surfaceIndex = update.key.surfaceIndex, + chunkSize = ChunkSize.CHUNK_032, + ) + + val chunkTiles = ServerMapChunkTiles(chunkId, validMapTilesToProtoHashCode) + + chunkId to chunkTiles + } +} + + +private fun KStream>.reduceServerMapChunkTilesToTable() + : KTable> { + + val pid = "$pid.reduceServerMapChunkTilesToTable" + + return repartition( + repartitionedAs( + "$pid.pre-table-repartition", + kxsBinary.serde(), + kxsBinary.serde>(), + // force, otherwise KTable-KTable FK join doesn't work + numberOfPartitions = 1, + ) + ).groupByKey( + groupedAs( + "$pid.group-by-key", + kxsBinary.serde(), + kxsBinary.serde>(), + ) + ).reduce( + "$pid.reduce", + materializedAs( + "$pid.reduce.store", + kxsBinary.serde(), + kxsBinary.serde>(), + ) + ) { chunkTiles, otherChunkTiles -> + chunkTiles + otherChunkTiles + } +} + + +private fun KTable>.enrichWithColourData( + protosTable: KTable, +): KTable> { + + val pid = "$pid.enrichWithColourData" + + return this.join( + other = protosTable, + tableJoined = tableJoined("$pid.join-tiles-with-prototypes"), + materialized = materializedAs( + "$pid.join-tiles-with-prototypes.store", + kxsBinary.serde(), + kxsBinary.serde(), + ), + foreignKeyExtractor = { chunkTiles: ServerMapChunkTiles -> + chunkTiles.chunkId.serverId + } + ) { tiles: ServerMapChunkTiles, colourDict: TileColourDict -> +// println("joining tiles:${tiles.map.size} with colourDict:${colourDict.map.size}") + + val missingProtos = mutableSetOf() + + val tileColours = tiles.map.mapValues { (_, code) -> + colourDict.map.getOrElse(code) { + missingProtos.add(code) + ColourHex.TRANSPARENT + } + } + + if (missingProtos.isNotEmpty()) { + println("[$pid] missing ${missingProtos.size} tile prototypes: ${missingProtos.joinToString { "${it.code}" }}") + } + + println("[$pid] Set tile colours for chunk ${tiles.chunkId}") + + ServerMapChunkTiles(tiles.chunkId, tileColours) + } +} + + +private fun KTable>.streamTo( + chunkSize: ChunkSize, + outputTopic: String = TOPIC_GROUPED_MAP_CHUNKS_STATE, +) { + val pid = "$pid.output-chunk.${chunkSize.name}" + + toStream("$pid.stream") + .filter("$pid.filter-tiles-not-empty") { _, v -> + !v?.map.isNullOrEmpty() + } + .mapValues("$pid.map-not-null") { _, v -> + requireNotNull(v) + } + .peek("$pid.print-group-result") { _, v -> + println("Grouping map tiles result: ${v.chunkId} / ${v.map.size}") + } + .to( + outputTopic, + producedAs( + "$pid.grouped-map-chunks", + kxsBinary.serde(), + kxsBinary.serde(), + ) + ) +} + + +///** +// * Group [MapChunkUpdate]s by the Chunk position. +// * +// * They should already be grouped, but do it again to make sure, and to filter out empty updates. +// */ +//private fun KTable>.groupByChunkPosition( +// chunkSize: ChunkSize = ChunkSize.CHUNK_032 +//): KTable> { +// +// val pid = "$pid.groupByChunkPosition.${chunkSize.name}" +// +// return toStream("$pid.intake-stream") +// .map("$pid.change-chunk-size") { chunkId: ServerMapChunkId, chunkTiles: ServerMapChunkTiles? -> +// +// val newChunkPosition = chunkId.chunkPosition +// .leftTopTile(chunkId.chunkSize) +// .toMapChunkPosition(chunkSize) +// +// val newId = chunkId.copy( +// chunkSize = chunkSize, +// chunkPosition = newChunkPosition, +// ) +// +// val newChunkTiles = chunkTiles?.map ?: emptyMap() +// +// newId to ServerMapChunkTiles(chunkId = newId, map = newChunkTiles) +// }.groupByKey( +// groupedAs( +// "$pid.group-by-key", +// kxsBinary.serde(), +// kxsBinary.serde>(), +// ) +// ).reduce( +// "$pid.reduce", +// materializedAs( +// "$pid.reduce.store", +// kxsBinary.serde(), +// kxsBinary.serde>(), +// ) +// ) { chunkTiles, otherChunkTiles -> +// chunkTiles + otherChunkTiles +// } +//} + +/** + * Group [MapChunkUpdate]s by the Chunk position. + * + * They should already be grouped, but do it again to make sure, and to filter out empty updates. + */ +private fun KStream>.groupByChunkPosition( + chunkSize: ChunkSize = ChunkSize.CHUNK_032 +): KTable> { + + val pid = "$pid.groupByChunkPosition.${chunkSize.name}" + + return map("$pid.change-chunk-size") { chunkId: ServerMapChunkId, chunkTiles: ServerMapChunkTiles? -> + + val newChunkPosition = chunkId.chunkPosition + .leftTopTile(chunkId.chunkSize) + .toMapChunkPosition(chunkSize) + + val newId = chunkId.copy( + chunkSize = chunkSize, + chunkPosition = newChunkPosition, + ) + + val newChunkTiles = chunkTiles?.map ?: emptyMap() + + newId to ServerMapChunkTiles(chunkId = newId, map = newChunkTiles) + }.groupByKey( + groupedAs( + "$pid.group-by-key", + kxsBinary.serde(), + kxsBinary.serde>(), + ) + ).reduce( + "$pid.reduce", + materializedAs( + "$pid.reduce.store", + kxsBinary.serde(), + kxsBinary.serde>(), + ) + ) { chunkTiles, otherChunkTiles -> + chunkTiles + otherChunkTiles + } +} 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 eb87d07..f0c9e09 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 @@ -4,6 +4,7 @@ import dev.adamko.kafkatorio.processor.admin.TOPIC_GROUPED_MAP_CHUNKS_STATE import dev.adamko.kafkatorio.processor.admin.TOPIC_GROUPED_MAP_CHUNKS_STATE_DEBOUNCED import dev.adamko.kafkatorio.processor.misc.DebounceProcessor.Companion.addDebounceProcessor import dev.adamko.kafkatorio.processor.serdes.kxsBinary +import dev.adamko.kafkatorio.schema.common.ChunkSize import dev.adamko.kafkatorio.schema.common.ColourHex import dev.adamko.kafkatorio.schema.common.MapChunkPosition import dev.adamko.kafkatorio.schema.common.MapTile @@ -61,7 +62,7 @@ fun groupMapChunks(builder: StreamsBuilder): Topology { val protosStream: KStream = builder.streamPacketData() val tileProtoColourDict: KTable = - tileProtoColourDictionary(protosStream) + protosStream.createTilePrototypeTable() val groupedMapChunkTiles: KTable> = groupTilesIntoChunksWithColours( @@ -128,7 +129,7 @@ private fun groupTilesIntoChunksWithColours( mapTiles .tiles .groupBy { tile -> - tile.position.toMapChunkPosition(standardChunkSize.lengthInTiles) + tile.position.toMapChunkPosition(standardChunkSize) } .map { (chunkPos, tiles) -> diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/model.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/model.kt index 44fb966..0a70dda 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/model.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/model.kt @@ -13,38 +13,3 @@ import kotlinx.serialization.Serializable value class FactorioServerId(private val id: String) { override fun toString() = id } - - -@Serializable -enum class ChunkSize( - val zoomLevel: Int, -) : Comparable { - CHUNK_512(-1), - CHUNK_256(0), - CHUNK_128(1), - CHUNK_064(2), - CHUNK_032(3), - ; - - /** 32, 64, 128, 256, or 512 */ - val lengthInTiles: Int = 2f.pow(8 - zoomLevel).roundToInt() - - init { - require(lengthInTiles > 0) { "tilesPerChunk $lengthInTiles must be positive" } - // 1000 & 0111 = 0000 => pow^2 - // 1001 & 1000 = 1000 => !pow^2 - require(lengthInTiles and (lengthInTiles - 1) == 0) { - "tilesPerChunk $lengthInTiles must be a power-of-two number" - } - } - - companion object { - // cache values for better performance. KT-48872 - val entries: Set = values().toSet() - - val MAX: ChunkSize = entries.maxByOrNull { it.lengthInTiles }!! - val MIN: ChunkSize = entries.minByOrNull { it.lengthInTiles }!! - val STANDARD: ChunkSize = entries.first { it.zoomLevel == 0 } - - } -} diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/playerUpdates.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/playerUpdates.kt index ef10585..48a910a 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/playerUpdates.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/playerUpdates.kt @@ -2,15 +2,22 @@ package dev.adamko.kafkatorio.processor.topology import dev.adamko.kafkatorio.processor.WebsocketServer import dev.adamko.kafkatorio.processor.admin.topicName -import dev.adamko.kafkatorio.processor.serdes.jsonMapper +import dev.adamko.kafkatorio.schema.jsonMapperKafkatorio import dev.adamko.kafkatorio.schema.packets.KafkatorioPacket import dev.adamko.kafkatorio.schema.packets.PlayerUpdate import dev.adamko.kotka.extensions.consumedAs import dev.adamko.kotka.kxs.serde import kotlinx.serialization.encodeToString +import kotlinx.serialization.json.Json import org.apache.kafka.streams.StreamsBuilder +private val jsonMapper: Json = Json { + prettyPrint = false + serializersModule = jsonMapperKafkatorio.serializersModule +} + + fun playerUpdatesToWsServer( websocketServer: WebsocketServer, builder: StreamsBuilder = StreamsBuilder(), 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 19d0fba..7716819 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 @@ -32,12 +32,12 @@ value class TileColourDict(val map: Map) /** Get the latest map-tile colours per server. */ -fun tileProtoColourDictionary( - factorioServerPacketStream: KStream -): KTable { - val pid = "tileProtoColourDictionary" +fun KStream.createTilePrototypeTable() + : KTable { - return factorioServerPacketStream.mapValues( + val pid = "createTilePrototypeTable" + + return mapValues( "$pid.map-values" ) { _, protoPacket: PrototypesUpdate -> val map = protoPacket diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/saveMapTiles.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/saveMapTiles.kt index 71f062b..52ca06d 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/saveMapTiles.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/saveMapTiles.kt @@ -4,11 +4,13 @@ import com.sksamuel.scrimage.ImmutableImage import com.sksamuel.scrimage.ScaleMethod import com.sksamuel.scrimage.color.RGBColor import com.sksamuel.scrimage.nio.PngWriter +import dev.adamko.kafkatorio.processor.admin.TOPIC_GROUPED_MAP_CHUNKS_STATE import dev.adamko.kafkatorio.processor.admin.TOPIC_GROUPED_MAP_CHUNKS_STATE_DEBOUNCED import dev.adamko.kafkatorio.processor.admin.TOPIC_SUBDIVIDED_MAP_TILES import dev.adamko.kafkatorio.processor.admin.TOPIC_SUBDIVIDED_MAP_TILES_DEBOUNCED import dev.adamko.kafkatorio.processor.misc.DebounceProcessor.Companion.addDebounceProcessor import dev.adamko.kafkatorio.processor.serdes.kxsBinary +import dev.adamko.kafkatorio.schema.common.ChunkSize import dev.adamko.kafkatorio.schema.common.ColourHex import dev.adamko.kafkatorio.schema.common.MapTilePosition import dev.adamko.kafkatorio.schema.common.toMapChunkPosition @@ -42,28 +44,28 @@ fun saveMapTiles( tileDirectory: Path, ): Topology { - val groupedMapChunkTiles: KStream> = - builder.stream( - TOPIC_GROUPED_MAP_CHUNKS_STATE_DEBOUNCED, - consumedAs("$pid.consume.grouped-map-chunks", kxsBinary.serde(), kxsBinary.serde()), - ) - - groupedMapChunkTiles - .flatMapSubdividedChunk() - .to( - TOPIC_SUBDIVIDED_MAP_TILES, - producedAs( - "$pid.produce.grouped-map-chunks", - kxsBinary.serde(), - kxsBinary.serde(), - ) - ) +// val groupedMapChunkTiles: KStream> = +// builder.stream( +// TOPIC_GROUPED_MAP_CHUNKS_STATE_DEBOUNCED, +// consumedAs("$pid.consume.grouped-map-chunks", kxsBinary.serde(), kxsBinary.serde()), +// ) +// +// groupedMapChunkTiles +// .flatMapSubdividedChunk() +// .to( +// TOPIC_SUBDIVIDED_MAP_TILES, +// producedAs( +// "$pid.produce.grouped-map-chunks", +// kxsBinary.serde(), +// kxsBinary.serde(), +// ) +// ) val subdividedMapChunkTilesDebounced: KStream> = builder.stream( - TOPIC_SUBDIVIDED_MAP_TILES_DEBOUNCED, + TOPIC_GROUPED_MAP_CHUNKS_STATE_DEBOUNCED, consumedAs( - "$pid.consume.debounced-subdivided-chunks", + "$pid.consume.grouped-map-chunks", kxsBinary.serde(), kxsBinary.serde(), ), @@ -74,8 +76,8 @@ fun saveMapTiles( return builder.build() .addDebounceProcessor>( namePrefix = pid, - sourceTopic = TOPIC_SUBDIVIDED_MAP_TILES, - sinkTopic = TOPIC_SUBDIVIDED_MAP_TILES_DEBOUNCED, + sourceTopic = TOPIC_GROUPED_MAP_CHUNKS_STATE, + sinkTopic = TOPIC_GROUPED_MAP_CHUNKS_STATE_DEBOUNCED, inactivityDuration = 15.seconds, keySerde = kxsBinary.serde(), valueSerde = kxsBinary.serde(), @@ -83,35 +85,35 @@ fun saveMapTiles( } -private fun KStream>.flatMapSubdividedChunk() = - flatMap("$pid.subdivide-chunks") { initialKey, initialChunk -> - - val (chunks, time) = measureTimedValue { - ChunkSize.entries.flatMap { chunkSize -> - initialChunk.map - .entries - .groupingBy { (tilePosition, _) -> - val chunkPosition = tilePosition.toMapChunkPosition(chunkSize.lengthInTiles) - initialKey.copy( - chunkSize = chunkSize, - chunkPosition = chunkPosition, - ) - }.fold( - initialValueSelector = { _, _ -> mutableMapOf() } - ) { _, accumulator, (tilePosition, colour) -> - accumulator[tilePosition] = colour - accumulator - } - .map { (chunkId, tiles) -> - chunkId to ServerMapChunkTiles(chunkId, tiles) - } - } - } - - println("subdivided ${initialKey.chunkPosition} to ${chunks.size} chunks in $time") - - chunks - } +//private fun KStream>.flatMapSubdividedChunk() = +// flatMap("$pid.subdivide-chunks") { initialKey, initialChunk -> +// +// val (chunks, time) = measureTimedValue { +// ChunkSize.entries.flatMap { chunkSize -> +// initialChunk.map +// .entries +// .groupingBy { (tilePosition, _) -> +// val chunkPosition = tilePosition.toMapChunkPosition(chunkSize) +// initialKey.copy( +// chunkSize = chunkSize, +// chunkPosition = chunkPosition, +// ) +// }.fold( +// initialValueSelector = { _, _ -> mutableMapOf() } +// ) { _, accumulator, (tilePosition, colour) -> +// accumulator[tilePosition] = colour +// accumulator +// } +// .map { (chunkId, tiles) -> +// chunkId to ServerMapChunkTiles(chunkId, tiles) +// } +// } +// } +// +// println("subdivided ${initialKey.chunkPosition} to ${chunks.size} chunks in $time") +// +// chunks +// } private fun KStream>.forEachChunkSaveImage( diff --git a/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/topology/SaveMapChunksTest.kt b/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/topology/SaveMapChunksTest.kt index 94ebdc0..22dd44b 100644 --- a/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/topology/SaveMapChunksTest.kt +++ b/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/topology/SaveMapChunksTest.kt @@ -1,6 +1,7 @@ package dev.adamko.kafkatorio.processor.topology import dev.adamko.kafkatorio.processor.serdes.kxsBinary +import dev.adamko.kafkatorio.schema.common.ChunkSize import dev.adamko.kafkatorio.schema.common.ColourHex import dev.adamko.kafkatorio.schema.common.MapChunkPosition import dev.adamko.kafkatorio.schema.common.MapTilePosition @@ -142,11 +143,11 @@ class SaveMapChunksTest : FunSpec({ println("chunk size: ${chunkId.chunkSize} ${chunkId.chunkSize.lengthInTiles}") val map = chunkId.chunkPosition - .iterateTiles(chunkId.chunkSize.lengthInTiles) + .iterateTiles(chunkId.chunkSize) .asSequence() .map { tilePos -> val colour = - tileColours.getOrPut(tilePos.toMapChunkPosition(ChunkSize.MIN.lengthInTiles)) { + tileColours.getOrPut(tilePos.toMapChunkPosition(ChunkSize.MIN)) { colourHexArb().sample(RandomSource.default()).value } tilePos to colour