tidy up grouping, colouring, saving map chunks

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/KafkatorioTopology.kt

@@ -3,7 +3,6 @@ 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
 import dev.adamko.kafkatorio.processor.topology.saveMapTiles
 import dev.adamko.kafkatorio.processor.topology.splitFactorioServerPacketStream

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/admin/kafkatorioTopics.kt

@@ -17,7 +17,10 @@ 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"
-const val TOPIC_GROUPED_MAP_CHUNKS_STATE_DEBOUNCED = "$TOPIC_GROUPED_MAP_CHUNKS_STATE.debounced"
+
+const val TOPIC_MAP_CHUNK_COLOURED_032_STATE = "$DOMAIN.state.map-chunk-032.colour"
+
+const val TOPIC_MAP_CHUNK_COLOURED_STATE = "$DOMAIN.state.map-chunk.colour"
 
 const val TOPIC_SUBDIVIDED_MAP_TILES = "$DOMAIN.map-tiles.subdivided"
 const val TOPIC_SUBDIVIDED_MAP_TILES_DEBOUNCED = "$TOPIC_SUBDIVIDED_MAP_TILES.debounced"
@@ -25,8 +28,13 @@ const val TOPIC_SUBDIVIDED_MAP_TILES_DEBOUNCED = "$TOPIC_SUBDIVIDED_MAP_TILES.de
 
 fun allTopics(): Set<String> = buildSet {
   add(TOPIC_SRC_SERVER_LOG)
+
   add(TOPIC_GROUPED_MAP_CHUNKS_STATE)
-  add(TOPIC_GROUPED_MAP_CHUNKS_STATE_DEBOUNCED)
+
+  add(TOPIC_MAP_CHUNK_COLOURED_032_STATE)
+
+  add(TOPIC_MAP_CHUNK_COLOURED_STATE)
+
   add(TOPIC_SUBDIVIDED_MAP_TILES)
   add(TOPIC_SUBDIVIDED_MAP_TILES_DEBOUNCED)
 

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/colourMapChunks.kt

@@ -1,6 +1,8 @@
 package dev.adamko.kafkatorio.processor.topology
 
-import dev.adamko.kafkatorio.processor.admin.TOPIC_GROUPED_MAP_CHUNKS_STATE
+import dev.adamko.kafkatorio.processor.admin.TOPIC_MAP_CHUNK_COLOURED_032_STATE
+import dev.adamko.kafkatorio.processor.admin.TOPIC_MAP_CHUNK_COLOURED_STATE
+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
@@ -24,6 +26,7 @@ import dev.adamko.kotka.extensions.tables.toStream
 import dev.adamko.kotka.kxs.serde
 import kotlin.collections.component1
 import kotlin.collections.component2
+import kotlin.time.Duration.Companion.seconds
 import org.apache.kafka.streams.StreamsBuilder
 import org.apache.kafka.streams.Topology
 import org.apache.kafka.streams.kstream.KStream
@@ -48,31 +51,43 @@ fun colourMapChunks(builder: StreamsBuilder): Topology {
       .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<ServerMapChunkId>(),
-      kxsBinary.serde<ServerMapChunkTiles<ColourHex>>(),
-    )
+  chunkTilesColouredTable032.streamMapChunkColouredTo(
+    ChunkSize.CHUNK_032,
+    TOPIC_MAP_CHUNK_COLOURED_032_STATE
   )
 
+  val chunkTilesColoured032Debounced: KStream<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> =
+    builder.stream(
+      TOPIC_MAP_CHUNK_COLOURED_STATE,
+      consumedAs(
+        "$pid.consume.map-chunks-coloured",
+        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
+    chunkTilesColoured032Debounced
       .filter("$pid.combine-chunks.filter-$from-to-$to") { key, _ ->
-      key?.chunkSize == from
-    }.groupByChunkPosition(to)
-      .streamTo(to, TOPIC_GROUPED_MAP_CHUNKS_STATE)
+        key.chunkSize == from
+      }.groupByChunkPosition(to)
+      .streamMapChunkColouredTo(to, TOPIC_MAP_CHUNK_COLOURED_STATE)
   }
 
   return builder.build()
+    .addDebounceProcessor(
+      namePrefix = "groupTilesMapChunks",
+      sourceTopic = TOPIC_MAP_CHUNK_COLOURED_032_STATE,
+      sinkTopic = TOPIC_MAP_CHUNK_COLOURED_STATE,
+      inactivityDuration = 15.seconds,
+      keySerde = kxsBinary.serde<ServerMapChunkId>(),
+      valueSerde = kxsBinary.serde<ServerMapChunkTiles<ColourHex>>(),
+    )
 }
 
 
@@ -182,9 +197,9 @@ private fun KTable<ServerMapChunkId, ServerMapChunkTiles<TileProtoHashCode>>.enr
 }
 
 
-private fun KTable<ServerMapChunkId, ServerMapChunkTiles<ColourHex>>.streamTo(
+fun KTable<ServerMapChunkId, ServerMapChunkTiles<ColourHex>>.streamMapChunkColouredTo(
   chunkSize: ChunkSize,
-  outputTopic: String = TOPIC_GROUPED_MAP_CHUNKS_STATE,
+  outputTopic: String,
 ) {
   val pid = "$pid.output-chunk.${chunkSize.name}"
 
@@ -202,57 +217,13 @@ private fun KTable<ServerMapChunkId, ServerMapChunkTiles<ColourHex>>.streamTo(
       outputTopic,
       producedAs(
         "$pid.grouped-map-chunks",
-        kxsBinary.serde(),
-        kxsBinary.serde(),
+        kxsBinary.serde<ServerMapChunkId>(),
+        kxsBinary.serde<ServerMapChunkTiles<ColourHex>>(),
       )
     )
 }
 
 
-///**
-// * 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<ServerMapChunkId, ServerMapChunkTiles<ColourHex>>.groupByChunkPosition(
-//  chunkSize: ChunkSize = ChunkSize.CHUNK_032
-//): KTable<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> {
-//
-//  val pid = "$pid.groupByChunkPosition.${chunkSize.name}"
-//
-//  return toStream("$pid.intake-stream")
-//    .map("$pid.change-chunk-size") { chunkId: ServerMapChunkId, chunkTiles: ServerMapChunkTiles<ColourHex>? ->
-//
-//      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<ServerMapChunkId>(),
-//        kxsBinary.serde<ServerMapChunkTiles<ColourHex>>(),
-//      )
-//    ).reduce(
-//      "$pid.reduce",
-//      materializedAs(
-//        "$pid.reduce.store",
-//        kxsBinary.serde<ServerMapChunkId>(),
-//        kxsBinary.serde<ServerMapChunkTiles<ColourHex>>(),
-//      )
-//    ) { chunkTiles, otherChunkTiles ->
-//      chunkTiles + otherChunkTiles
-//    }
-//}
-
 /**
  * Group [MapChunkUpdate]s by the Chunk position.
  *

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/groupMapChunks.kt

@@ -1,227 +1 @@
 package dev.adamko.kafkatorio.processor.topology
-
-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
-import dev.adamko.kafkatorio.schema.common.MapTileDictionary.Companion.isNullOrEmpty
-import dev.adamko.kafkatorio.schema.common.MapTilePosition
-import dev.adamko.kafkatorio.schema.common.MapTiles
-import dev.adamko.kafkatorio.schema.common.SurfaceIndex
-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.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
-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 kotlin.time.Duration.Companion.seconds
-import kotlinx.serialization.Serializable
-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
-
-
-@Serializable
-data class ServerMapChunkId(
-  val serverId: FactorioServerId,
-  val chunkPosition: MapChunkPosition,
-  val surfaceIndex: SurfaceIndex,
-  val chunkSize: ChunkSize,
-)
-
-
-@Serializable
-data class ServerMapChunkTiles<Data>(
-  val chunkId: ServerMapChunkId,
-  val map: Map<MapTilePosition, Data>,
-) {
-  operator fun plus(other: ServerMapChunkTiles<Data>) = copy(map = map + other.map)
-}
-
-
-fun groupMapChunks(builder: StreamsBuilder): Topology {
-
-  val mapChunksStream: KStream<FactorioServerId, MapChunkUpdate> = builder.streamPacketData()
-  val protosStream: KStream<FactorioServerId, PrototypesUpdate> = builder.streamPacketData()
-
-  val tileProtoColourDict: KTable<FactorioServerId, TileColourDict> =
-    protosStream.createTilePrototypeTable()
-
-  val groupedMapChunkTiles: KTable<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> =
-    groupTilesIntoChunksWithColours(
-      mapChunksStream,
-      tileProtoColourDict,
-    )
-
-  val pid = "grouped-map-chunk-tiles.output"
-  groupedMapChunkTiles
-    .toStream("$pid.to-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(
-      TOPIC_GROUPED_MAP_CHUNKS_STATE,
-      producedAs(
-        "$pid.grouped-map-chunks",
-        kxsBinary.serde(),
-        kxsBinary.serde(),
-      )
-    )
-
-  return builder.build()
-    .addDebounceProcessor<ServerMapChunkId, ServerMapChunkTiles<ColourHex>>(
-      namePrefix = pid,
-      sourceTopic = TOPIC_GROUPED_MAP_CHUNKS_STATE,
-      sinkTopic = TOPIC_GROUPED_MAP_CHUNKS_STATE_DEBOUNCED,
-      inactivityDuration = 15.seconds,
-      keySerde = kxsBinary.serde(),
-      valueSerde = kxsBinary.serde(),
-    )
-}
-
-
-private fun groupTilesIntoChunksWithColours(
-  mapChunksStream: KStream<FactorioServerId, MapChunkUpdate>,
-  tileProtoColourDict: KTable<FactorioServerId, TileColourDict>,
-): KTable<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> {
-
-  val mapTilesStream: KStream<FactorioServerId, MapTiles> =
-    mapChunksStream.filter { _, value ->
-      !value.tileDictionary.isNullOrEmpty()
-    }.mapValues("map-chunk-update-packets.convert-to-map-tiles") { _, packet: MapChunkUpdate ->
-      val mapTiles = packet.tileDictionary?.toMapTileList() ?: emptyList()
-      MapTiles(packet.key.surfaceIndex, mapTiles)
-    }.filter("map-chunk-update-packets.filter-out.no-tiles") { _, value ->
-      value.tiles.isNotEmpty()
-    }
-
-  val chunkedTilesTable: KTable<ServerMapChunkId, ServerMapChunkTiles<TileProtoHashCode>> =
-    mapTilesStream
-      // group tiles by server & surface & chunk
-      .flatMap("server-map-data.tiles.flatMapByChunk") { serverId: FactorioServerId, mapTiles: MapTiles ->
-
-        val standardChunkSize = ChunkSize.MAX
-
-        val flatMappedChunkTiles: List<Pair<ServerMapChunkId, ServerMapChunkTiles<TileProtoHashCode>>> =
-          mapTiles
-            .tiles
-            .groupBy { tile ->
-              tile.position.toMapChunkPosition(standardChunkSize)
-            }
-            .map { (chunkPos, tiles) ->
-
-              val chunkId = ServerMapChunkId(
-                serverId = serverId,
-                chunkPosition = chunkPos,
-                surfaceIndex = mapTiles.surfaceIndex,
-                chunkSize = standardChunkSize,
-              )
-
-              val tilesMap = tiles.associate { it.position to TileProtoHashCode(it) }
-
-              val chunkTiles = ServerMapChunkTiles(chunkId, tilesMap)
-
-              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()
-      }
-      .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",
-          kxsBinary.serde<ServerMapChunkId>(),
-          kxsBinary.serde<ServerMapChunkTiles<TileProtoHashCode>>(),
-        )
-      )
-      .reduce(
-        "server-map-data.tiles.reduce",
-        materializedAs(
-          "server-map-data.tiles.reduce.store",
-          kxsBinary.serde<ServerMapChunkId>(),
-          kxsBinary.serde<ServerMapChunkTiles<TileProtoHashCode>>(),
-        )
-      ) { 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<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> =
-    chunkedTilesTable
-      .join(
-        other = tileProtoColourDict,
-        tableJoined = tableJoined("server-map-data.join-tiles-with-prototypes"),
-        materialized = materializedAs(
-          "server-map-data.join-tiles-with-prototypes.store",
-          kxsBinary.serde(),
-          kxsBinary.serde(),
-        ),
-        foreignKeyExtractor = { chunkTiles: ServerMapChunkTiles<TileProtoHashCode> ->
-          chunkTiles.chunkId.serverId
-        }
-      ) { tiles: ServerMapChunkTiles<TileProtoHashCode>, colourDict: TileColourDict ->
-//        println("joining tiles:${tiles.map.size} with colourDict:${colourDict.map.size}")
-
-        val missingProtos = mutableSetOf<TileProtoHashCode>()
-
-        val tileColours = tiles.map.mapValues { (_, code) ->
-          colourDict.map.getOrElse(code) {
-            missingProtos.add(code)
-            ColourHex.TRANSPARENT
-          }
-        }
-
-        if (missingProtos.isNotEmpty()) {
-          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
-  get() = MapTilePosition(x, y)

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/model.kt

@@ -1,10 +1,13 @@
 package dev.adamko.kafkatorio.processor.topology
 
 import com.sksamuel.scrimage.color.RGBColor
+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
+import dev.adamko.kafkatorio.schema.common.MapTilePosition
+import dev.adamko.kafkatorio.schema.common.SurfaceIndex
 import java.awt.Color
-import kotlin.math.pow
-import kotlin.math.roundToInt
 import kotlinx.serialization.Serializable
 
 
@@ -13,3 +16,38 @@ import kotlinx.serialization.Serializable
 value class FactorioServerId(private val id: String) {
   override fun toString() = id
 }
+
+
+@Serializable
+data class ServerMapChunkId(
+  val serverId: FactorioServerId,
+  val chunkPosition: MapChunkPosition,
+  val surfaceIndex: SurfaceIndex,
+  val chunkSize: ChunkSize,
+)
+
+
+@Serializable
+data class ServerMapChunkTiles<Data>(
+  val chunkId: ServerMapChunkId,
+  val map: Map<MapTilePosition, Data>,
+) {
+  operator fun plus(other: ServerMapChunkTiles<Data>) = copy(map = map + other.map)
+}
+
+
+val MapTile.position
+  get() = MapTilePosition(x, y)
+
+
+val TRANSPARENT_AWT: Color = ColourHex.TRANSPARENT.toRgbColor().awt()
+
+
+fun ColourHex.toRgbColor(): RGBColor {
+  return RGBColor(
+    red.toInt(),
+    green.toInt(),
+    blue.toInt(),
+    alpha.toInt(),
+  )
+}

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/saveMapTiles.kt

@@ -2,30 +2,17 @@ package dev.adamko.kafkatorio.processor.topology
 
 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.admin.TOPIC_MAP_CHUNK_COLOURED_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.MapTilePosition
-import dev.adamko.kafkatorio.schema.common.toMapChunkPosition
 import dev.adamko.kotka.extensions.consumedAs
-import dev.adamko.kotka.extensions.producedAs
-import dev.adamko.kotka.extensions.streams.flatMap
 import dev.adamko.kotka.extensions.streams.forEach
 import dev.adamko.kotka.kxs.serde
-import java.awt.Color
 import java.awt.image.BufferedImage
 import java.nio.file.Path
 import kotlin.io.path.fileSize
 import kotlin.math.abs
-import kotlin.time.Duration.Companion.seconds
-import kotlin.time.measureTimedValue
 import org.apache.kafka.streams.StreamsBuilder
 import org.apache.kafka.streams.Topology
 import org.apache.kafka.streams.kstream.KStream
@@ -44,26 +31,9 @@ fun saveMapTiles(
   tileDirectory: Path,
 ): Topology {
 
-//  val groupedMapChunkTiles: KStream<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> =
-//    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<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> =
     builder.stream(
-      TOPIC_GROUPED_MAP_CHUNKS_STATE_DEBOUNCED,
+      TOPIC_MAP_CHUNK_COLOURED_STATE,
       consumedAs(
         "$pid.consume.grouped-map-chunks",
         kxsBinary.serde(),
@@ -74,48 +44,9 @@ fun saveMapTiles(
   subdividedMapChunkTilesDebounced.forEachChunkSaveImage(tileDirectory)
 
   return builder.build()
-    .addDebounceProcessor<ServerMapChunkId, ServerMapChunkTiles<ColourHex>>(
-      namePrefix = pid,
-      sourceTopic = TOPIC_GROUPED_MAP_CHUNKS_STATE,
-      sinkTopic = TOPIC_GROUPED_MAP_CHUNKS_STATE_DEBOUNCED,
-      inactivityDuration = 15.seconds,
-      keySerde = kxsBinary.serde(),
-      valueSerde = kxsBinary.serde(),
-    )
 }
 
 
-//private fun KStream<ServerMapChunkId, ServerMapChunkTiles<ColourHex>>.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<MapTilePosition, ColourHex>() }
-//          ) { _, 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<ServerMapChunkId, ServerMapChunkTiles<ColourHex>>.forEachChunkSaveImage(
   tileDirectory: Path,
 ) = forEach(
@@ -186,16 +117,3 @@ value class TilePngFilename(
     append(".png")
   })
 }
-
-
-val TRANSPARENT_AWT: Color = ColourHex.TRANSPARENT.toRgbColor().awt()
-
-
-fun ColourHex.toRgbColor(): RGBColor {
-  return RGBColor(
-    red.toInt(),
-    green.toInt(),
-    blue.toInt(),
-    alpha.toInt(),
-  )
-}

modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/misc/KxsBinaryTests.kt

@@ -1,10 +1,10 @@
 package dev.adamko.kafkatorio.processor.misc
 
 import dev.adamko.kafkatorio.processor.serdes.kxsBinary
-import dev.adamko.kafkatorio.processor.topology.ChunkSize
 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.schema.common.ChunkSize
 import dev.adamko.kafkatorio.schema.common.ColourHex
 import dev.adamko.kafkatorio.schema.common.MapChunkPosition
 import dev.adamko.kafkatorio.schema.common.MapTilePosition

modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/misc/serdeSize.kt

@@ -1,7 +1,6 @@
 package dev.adamko.kafkatorio.processor.misc
 
 import dev.adamko.kafkatorio.processor.serdes.kxsBinary
-import dev.adamko.kafkatorio.processor.topology.ChunkSize
 import dev.adamko.kafkatorio.processor.topology.FactorioServerId
 import dev.adamko.kafkatorio.processor.topology.ServerMapChunkId
 import dev.adamko.kafkatorio.processor.topology.ServerMapChunkTiles