re-implement tile saver, without Coroutines or debouncing. It seems to work well.

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

@@ -4,7 +4,7 @@ import dev.adamko.kafkatorio.processor.config.ApplicationProperties
 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.saveMapTiles2
 import dev.adamko.kafkatorio.processor.topology.splitFactorioServerPacketStream
 import java.time.Duration
 import java.util.Properties
@@ -59,7 +59,8 @@ internal class KafkatorioTopology(
 
   private fun saveTiles() {
     val builder = StreamsBuilder()
-    val topology = saveMapTiles(builder)
+//    val topology = saveMapTiles(builder)
+    val topology = saveMapTiles2(builder)
     launchTopology("saveTiles", topology)
   }
 

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

@@ -4,12 +4,17 @@ 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.schema.common.*
 import dev.adamko.kafkatorio.processor.admin.TOPIC_GROUPED_MAP_CHUNKS_STATE
 import dev.adamko.kafkatorio.processor.serdes.kxsBinary
+import dev.adamko.kafkatorio.schema.common.ColourHex
+import dev.adamko.kafkatorio.schema.common.MapChunkPosition
+import dev.adamko.kafkatorio.schema.common.MapTilePosition
+import dev.adamko.kafkatorio.schema.common.SurfaceIndex
+import dev.adamko.kafkatorio.schema.common.toMapChunkPosition
 import dev.adamko.kotka.extensions.consumedAs
 import dev.adamko.kotka.extensions.streams.forEach
 import dev.adamko.kotka.kxs.serde
+import java.awt.Color
 import java.awt.image.BufferedImage
 import java.io.File
 import kotlin.coroutines.CoroutineContext
@@ -36,6 +41,7 @@ import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.flow.runningFold
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.supervisorScope
+import kotlinx.serialization.Serializable
 import org.apache.kafka.streams.StreamsBuilder
 import org.apache.kafka.streams.Topology
 import org.apache.kafka.streams.kstream.KStream
@@ -51,11 +57,12 @@ private val serverMapChunkHandler = ServerMapChunkHandler()
 fun saveMapTiles(
   builder: StreamsBuilder,
 ): Topology {
+  val pid = "saveMapTiles"
 
   val groupedMapChunkTiles: KStream<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> =
     builder.stream(
       TOPIC_GROUPED_MAP_CHUNKS_STATE,
-      consumedAs("consume.grouped-map-chunks", kxsBinary.serde(), kxsBinary.serde())
+      consumedAs("$pid.consume.grouped-map-chunks", kxsBinary.serde(), kxsBinary.serde())
     )
 
   groupedMapChunkTiles
@@ -71,24 +78,24 @@ fun saveMapTiles(
 
 
 @JvmInline
-private value class TilePngFilename(
+value class TilePngFilename(
   val value: String,
 ) {
-  constructor(cmd: Cmd.CreateImage) : this(buildString {
+  constructor(id: ServerMapChunkId) : this(buildString {
     append("src/main/resources/kafkatorio-web-map")
-    append("/s${cmd.surfaceIndex}")
-    append("/z${cmd.chunkSize.zoomLevel}")
-    append("/x${cmd.chunkPosition.x}")
-    append("/y${cmd.chunkPosition.y}")
+    append("/s${id.surfaceIndex}")
+    append("/z${id.chunkSize.zoomLevel}")
+    append("/x${id.chunkPosition.x}")
+    append("/y${id.chunkPosition.y}")
     append(".png")
   })
 }
 
 
-private val TRANSPARENT_AWT = ColourHex.TRANSPARENT.toRgbColor().awt()
+val TRANSPARENT_AWT: Color = ColourHex.TRANSPARENT.toRgbColor().awt()
 
 
-private fun ColourHex.toRgbColor(): RGBColor {
+fun ColourHex.toRgbColor(): RGBColor {
   return RGBColor(
     red.toInt(),
     green.toInt(),
@@ -210,15 +217,13 @@ private class ServerMapChunkHandler : CoroutineScope {
             .asFlow()
         }
         .map { createCmd ->
-          val filename = TilePngFilename(createCmd)
-
           val img = createMapTileImage(
             createCmd.chunkPosition,
             createCmd.tiles,
             createCmd.chunkSize,
           )
 
-          Cmd.SaveImage(filename, img)
+          Cmd.SaveImage(createCmd.tilePngFilename, img)
         }
 
     launch {
@@ -270,17 +275,31 @@ private class ServerMapChunkHandler : CoroutineScope {
 
 private sealed interface Cmd {
 
+  @Serializable
   data class ChunkSubdivide(
     val chunkId: ServerMapChunkId,
     val tiles: Map<MapTilePosition, ColourHex>,
   ) : Cmd
 
+  @Serializable
   data class CreateImage(
     val surfaceIndex: SurfaceIndex,
     val chunkSize: ChunkSize,
     val chunkPosition: MapChunkPosition,
     val tiles: Map<MapTilePosition, ColourHex>,
-  ) : Cmd
+  ) : Cmd {
+    val tilePngFilename: TilePngFilename by lazy {
+      TilePngFilename(buildString {
+        append("src/main/resources/kafkatorio-web-map")
+        append("/s${surfaceIndex}")
+        append("/z${chunkSize.zoomLevel}")
+        append("/x${chunkPosition.x}")
+        append("/y${chunkPosition.y}")
+        append(".png")
+      })
+    }
+  }
+
 
   data class SaveImage(
     val filename: TilePngFilename,

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

@@ -0,0 +1,99 @@
+package dev.adamko.kafkatorio.processor.topology
+
+import com.sksamuel.scrimage.ImmutableImage
+import com.sksamuel.scrimage.ScaleMethod
+import com.sksamuel.scrimage.nio.PngWriter
+import dev.adamko.kafkatorio.processor.admin.TOPIC_GROUPED_MAP_CHUNKS_STATE
+import dev.adamko.kafkatorio.processor.serdes.kxsBinary
+import dev.adamko.kafkatorio.schema.common.ColourHex
+import dev.adamko.kafkatorio.schema.common.toMapChunkPosition
+import dev.adamko.kotka.extensions.consumedAs
+import dev.adamko.kotka.extensions.streams.flatMap
+import dev.adamko.kotka.extensions.streams.forEach
+import dev.adamko.kotka.kxs.serde
+import java.awt.image.BufferedImage
+import java.io.File
+import kotlin.io.path.fileSize
+import kotlin.math.abs
+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
+
+
+fun saveMapTiles2(
+  builder: StreamsBuilder,
+): Topology {
+  val pid = "saveMapTiles"
+
+  val groupedMapChunkTiles: KStream<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> =
+    builder.stream(
+      TOPIC_GROUPED_MAP_CHUNKS_STATE,
+      consumedAs("$pid.consume.grouped-map-chunks", kxsBinary.serde(), kxsBinary.serde())
+    )
+
+  groupedMapChunkTiles.flatMap("$pid.subdivide-chunks") { initialKey, initialChunk ->
+    ChunkSize.entries.flatMap { chunkSize ->
+      initialChunk.map.entries
+        .groupBy(
+          { (tile, _) -> tile.toMapChunkPosition(chunkSize.tilesPerChunk) }
+        ) { (tilePosition, colour) ->
+          tilePosition to colour
+        }
+        .map { (chunkPos, tiles) ->
+          val key = initialKey.copy(
+            chunkSize = chunkSize,
+            chunkPosition = chunkPos
+          )
+
+          val value = ServerMapChunkTiles(key, tiles.toMap())
+
+          key to value
+        }
+    }
+  }.forEach("$pid.save-tiles") { key, value ->
+    val filename = TilePngFilename(key)
+    val chunkColours = value.map
+
+    val chunkImage = ImmutableImage.filled(
+      key.chunkSize.tilesPerChunk,
+      key.chunkSize.tilesPerChunk,
+      TRANSPARENT_AWT,
+      BufferedImage.TYPE_INT_ARGB
+    )
+
+    val chunkOriginX: Int = key.chunkPosition.x * key.chunkSize.tilesPerChunk
+    val chunkOriginY: Int = key.chunkPosition.y * key.chunkSize.tilesPerChunk
+
+    chunkColours.forEach { (tilePosition, colour) ->
+      val rgbColour = colour.toRgbColor()
+
+      val pixelX = abs(abs(tilePosition.x) - abs(chunkOriginX))
+      val pixelY = abs(abs(tilePosition.y) - abs(chunkOriginY))
+
+      chunkImage.setColor(pixelX, pixelY, rgbColour)
+    }
+
+    val img = chunkImage.scaleTo(WEB_MAP_TILE_SIZE_PX, WEB_MAP_TILE_SIZE_PX, ScaleMethod.FastScale)
+
+    val chunkImageFile = File(filename.value)
+    if (chunkImageFile.parentFile.mkdirs()) {
+      println("created new map tile parentFile directory ${chunkImageFile.absolutePath}")
+    }
+
+    val sizeBefore = chunkImageFile.takeIf { it.exists() }?.toPath()?.fileSize()
+
+    val savedTile = img.output(PngWriter.NoCompression, chunkImageFile)
+
+    val sizeAfter = savedTile.takeIf { it.exists() }?.toPath()?.fileSize()
+    if (sizeBefore != sizeAfter) {
+//          println("savedTile $sizeBefore/$sizeAfter: $savedTile")
+    } else {
+      println("savedTile NO-SIZE-CHANGE $sizeBefore/$sizeAfter: $savedTile")
+    }
+  }
+
+  return builder.build()
+}