diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/admin/kafkatorioTopics.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/admin/kafkatorioTopics.kt index 5a7781e..18e483a 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/admin/kafkatorioTopics.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/admin/kafkatorioTopics.kt @@ -14,7 +14,6 @@ import kotlin.reflect.KClass private const val DOMAIN = "kafkatorio" - const val TOPIC_SRC_SERVER_LOG = "$DOMAIN.src.server-log" const val TOPIC_GROUPED_MAP_CHUNKS_STATE = "$DOMAIN.state.map-chunks.grouped" 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 0cb62d3..aae4413 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 @@ -13,9 +13,9 @@ 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.groupedAs -import dev.adamko.kotka.extensions.materializedAs import dev.adamko.kotka.extensions.materializedWith 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.flatMap import dev.adamko.kotka.extensions.streams.mapValues @@ -66,21 +66,22 @@ fun groupMapChunks(builder: StreamsBuilder): Topology { tileProtoColourDict, ) + val pid = "grouped-map-chunk-tiles.output" groupedMapChunkTiles - .toStream("stream-grouped-map-tiles") - .filter("stream-grouped-map-tiles.filter-not-null") { _, v -> + .toStream("$pid.to-stream") + .filter("$pid.filter-tiles-not-empty") { _, v -> !v?.map.isNullOrEmpty() } - .mapValues("stream-grouped-map-tiles.map-not-null") { _, v -> + .mapValues("$pid.map-not-null") { _, v -> requireNotNull(v) } - .peek("stream-grouped-map-tiles.peek") { _, v -> + .peek("$pid.print-group-result") { _, v -> println("Grouping map tiles result: ${v.chunkId} / ${v.map.size}") } .to( TOPIC_GROUPED_MAP_CHUNKS_STATE, producedAs( - "produce.grouped-map-chunks", + "$pid.grouped-map-chunks", kxsBinary.serde(), kxsBinary.serde>() ) @@ -102,9 +103,12 @@ private fun groupTilesIntoChunksWithColours( }.mapValues("map-chunk-update-packets.convert-to-map-tiles") { _, packet: MapChunkUpdate -> val mapTiles = packet.tileDictionary?.toMapTileList() ?: emptyList() MapTiles(packet.key.surfaceIndex, mapTiles) - }.filter { _, value -> + }.filter("map-chunk-update-packets.filter-out.no-tiles") { _, value -> value.tiles.isNotEmpty() } +// .peek { _, value -> +// println("[${System.currentTimeMillis()}] mapTilesStream: surface:${value.surfaceIndex}, tile count: ${value.tiles.size}") +// } val chunkedTilesTable: KTable> = @@ -115,8 +119,6 @@ private fun groupTilesIntoChunksWithColours( val standardChunkSize = ChunkSize.MAX val flatMappedChunkTiles: List>> = - // println("[${System.currentTimeMillis()}] Flat-mapped ${flatMappedChunkTiles.size} chunk, with ${flatMappedChunkTiles.joinToString { it.second.map.size.toString() }} tiles") - mapTiles .tiles .groupBy { tile -> @@ -145,6 +147,15 @@ private fun groupTilesIntoChunksWithColours( .filter("server-map-data.tiles.filter-not-empty") { _, value -> value.map.isNotEmpty() } + .repartition( + repartitionedAs( + "server-map-data.tiles.pre-table-repartition", + kxsBinary.serde(), + kxsBinary.serde(), + // force, otherwise KTable-KTable FK join doesn't work + numberOfPartitions = 1, + ) + ) .groupByKey( groupedAs( "server-map-data.tiles.group", @@ -155,24 +166,35 @@ private fun groupTilesIntoChunksWithColours( .reduce( "server-map-data.tiles.reduce", materializedWith( +// materializedAs( +// "server-map-data.tiles.reduce.store", kxsBinary.serde(), kxsBinary.serde>(), ) - ) { chunkTiles, other -> chunkTiles + other } - + ) { chunkTiles, other -> +// println( +// "[${System.currentTimeMillis()}] reducing tiles " + +// "chunkTiles:${chunkTiles.chunkId.chunkPosition}, ${chunkTiles.chunkId.chunkSize}, ${chunkTiles.map.size} " + +// "other:${other.chunkId.chunkPosition}, ${other.chunkId.chunkSize}, ${other.map.size}" +// ) + chunkTiles + other + } val colourisedChunkTable: KTable> = chunkedTilesTable .join( other = tileProtoColourDict, tableJoined = tableJoined("server-map-data.join-tiles-with-prototypes"), - materialized = materializedAs( - "server-map-data.join-tiles-with-prototypes.store", + materialized = materializedWith( +// "server-map-data.join-tiles-with-prototypes.store", kxsBinary.serde(), kxsBinary.serde>(), ), - foreignKeyExtractor = { chunkTiles: ServerMapChunkTiles -> chunkTiles.chunkId.serverId } + foreignKeyExtractor = { chunkTiles: ServerMapChunkTiles -> + chunkTiles.chunkId.serverId + } ) { tiles: ServerMapChunkTiles, colourDict: TileColourDict -> +// println("joining tiles:${tiles.map.size} with colourDict:${colourDict.map.size}") val missingProtos = mutableSetOf() diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/packetDataStreams.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/packetDataStreams.kt index 31a1bde..81f7e43 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/packetDataStreams.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/packetDataStreams.kt @@ -5,6 +5,7 @@ import dev.adamko.kafkatorio.processor.serdes.jsonMapper import dev.adamko.kafkatorio.schema.packets.KafkatorioPacket import dev.adamko.kafkatorio.schema.packets.KafkatorioPacketData import dev.adamko.kotka.extensions.consumedAs +import dev.adamko.kotka.extensions.streams.filter import dev.adamko.kotka.extensions.streams.mapValues import dev.adamko.kotka.kxs.serde import org.apache.kafka.streams.StreamsBuilder @@ -16,12 +17,14 @@ inline fun StreamsBuilder.streamPacketData(): val simpleName = T::class.simpleName!! val topicName = T::class.topicName + val taskName = "consume.${simpleName}" + return stream( topicName, - consumedAs("consume.${simpleName}", jsonMapper.serde(), jsonMapper.serde()) - ).filter { key: FactorioServerId?, value: KafkatorioPacket? -> + consumedAs("$taskName.input", jsonMapper.serde(), jsonMapper.serde()) + ).filter("$taskName.filter-is-instance") { key: FactorioServerId?, value: KafkatorioPacket? -> key != null && value?.data is T - }.mapValues("consume.map-values.${simpleName}") { _: FactorioServerId, packet: KafkatorioPacket -> + }.mapValues("$taskName.map-values") { _: FactorioServerId, packet: KafkatorioPacket -> packet.data as T } } 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 4661507..7514753 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 @@ -4,9 +4,10 @@ import dev.adamko.kafkatorio.processor.serdes.kxsBinary import dev.adamko.kafkatorio.schema.common.ColourHex import dev.adamko.kafkatorio.schema.common.MapTile import dev.adamko.kafkatorio.schema.common.toHex -import dev.adamko.kafkatorio.schema.prototypes.FactorioPrototype import dev.adamko.kafkatorio.schema.packets.PrototypesUpdate +import dev.adamko.kafkatorio.schema.prototypes.FactorioPrototype import dev.adamko.kotka.extensions.materializedAs +import dev.adamko.kotka.extensions.repartitionedAs import dev.adamko.kotka.extensions.streams.filter import dev.adamko.kotka.extensions.streams.mapValues import dev.adamko.kotka.extensions.streams.toTable @@ -34,51 +35,35 @@ value class TileColourDict(val map: Map) fun tileProtoColourDictionary( factorioServerPacketStream: KStream ): KTable { + val pid = "tileProtoColourDictionary" - return factorioServerPacketStream.mapValues("server-map-data.tile-prototypes.map-values") { _, protoPacket: PrototypesUpdate -> + return factorioServerPacketStream.mapValues( + "$pid.map-values" + ) { _, protoPacket: PrototypesUpdate -> val map = protoPacket .prototypes .filterIsInstance() .associate { TileProtoHashCode(it) to it.mapColour.toHex() } TileColourDict(map) - }.filter("server-map-data.tile-prototypes.filterMapTileProtos") { _, dict: TileColourDict -> + }.filter("$pid.filterMapTileProtos") { _, dict: TileColourDict -> dict.map.isNotEmpty() }.peek { serverId, dict: TileColourDict -> - println("server $serverId has TileColourDict[${dict.map.size}]: ${dict.map.entries.joinToString()}") - }.toTable( - "server-map-data.tile-prototypes", + println("$pid: server $serverId has TileColourDict[${dict.map.size}]: ${dict.map.entries.joinToString()}") + }.repartition( + repartitionedAs( + "$pid.pre-table-repartition", + kxsBinary.serde(), + kxsBinary.serde(), + // force, otherwise KTable-KTable FK join doesn't work + numberOfPartitions = 1, + ) + ).toTable( + "$pid.create-table", materializedAs( - "server-map-data.tile-prototypes.store", + "$pid.output-store", kxsBinary.serde(), kxsBinary.serde(), ) ) } - -// get all the prototypes -// return factorioServerPacketStream -// .mapValues("server-map-data.tile-prototypes.mapValues") { _, packet: KafkatorioPacket -> -// val map = when (packet) { -// is FactorioPrototype -> packet -// .prototypes -// .filterIsInstance() -// .associate { TileProtoHashCode(it) to it.mapColour.toHex() } -// else -> mapOf() -// } -// TileColourDict(map) -// }.filter("server-map-data.tile-prototypes.filterMapTileProtos") { _, dict: TileColourDict -> -// dict.map.isNotEmpty() -// }.peek { serverId, dict: TileColourDict -> -// println("server $serverId has TileColourDict[${dict.map.size}]: ${dict.map.entries.joinToString()}") -// } -// .toTable( -// "server-map-data.tile-prototypes", -// materializedAs( -// "server-map-data.tile-prototypes.store", -// kxsBinary.serde(), -// kxsBinary.serde(), -// ) -// ) -// -//}