refactoring (I can't remember what... it's untested for sure), and using the latest things from event-schema

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

@@ -1,8 +1,7 @@
 package dev.adamko.kafkatorio.processor
 
 
-import dev.adamko.kafkatorio.admin.KafkatorioKafkaAdmin
-import dev.adamko.kafkatorio.processor.config.ApplicationProperties
+import dev.adamko.kafkatorio.processor.admin.KafkatorioKafkaAdmin
 import dev.adamko.kafkatorio.processor.tileserver.WebMapTileServer
 import kotlinx.coroutines.runBlocking
 import org.http4k.server.PolyHandler

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

@@ -7,6 +7,7 @@ import com.sksamuel.scrimage.nio.PngWriter
 import dev.adamko.kafkatorio.events.schema.ColourHex
 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 java.awt.image.BufferedImage
 import java.io.File
@@ -16,6 +17,8 @@ import kotlinx.coroutines.CoroutineName
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.async
+import kotlinx.coroutines.awaitAll
 import kotlinx.coroutines.channels.BufferOverflow
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.asFlow
@@ -67,8 +70,8 @@ private value class TilePngFilename(
 ) {
   constructor(cmd: Cmd.CreateImage) : this(buildString {
     append("src/main/resources/kafkatorio-web-map")
-    append("/s${cmd.surfaceIndex.index}")
-    append("/z${cmd.zoomLevel.level}")
+    append("/s${cmd.surfaceIndex.value}")
+    append("/z${cmd.chunkSize.zoomLevel}")
     append("/x${cmd.chunkPosition.x}")
     append("/y${cmd.chunkPosition.y}")
     append(".png")
@@ -79,21 +82,157 @@ private value class TilePngFilename(
 private val TRANSPARENT_AWT = ColourHex.TRANSPARENT.toRgbColor().awt()
 
 
+private fun ColourHex.toRgbColor(): RGBColor {
+  return RGBColor(
+    red.toInt(),
+    green.toInt(),
+    blue.toInt(),
+    alpha.toInt(),
+  )
+}
+
+
+private class ServerMapChunkHandler : CoroutineScope {
+
+  override val coroutineContext: CoroutineContext = Dispatchers.Default +
+      SupervisorJob(cScope.coroutineContext.job) +
+      CoroutineName("ServerMapChunkHandler")
+
+
+  /** The inbox for this handler. */
+  private val allChunksFlow = MutableSharedFlow<ServerMapChunkTiles<ColourHex>>(
+    replay = 0,
+    extraBufferCapacity = 10,
+    onBufferOverflow = BufferOverflow.SUSPEND,
+  )
+
+  fun emit(mapChunkTiles: ServerMapChunkTiles<ColourHex>) {
+    launch {
+      allChunksFlow.emit(mapChunkTiles)
+    }
+  }
+
+  /** This flow saves images one-by-one */
+  private val saveImagesFlow = MutableSharedFlow<Cmd.SaveImage>(
+    replay = 0,
+    extraBufferCapacity = 10,
+    onBufferOverflow = BufferOverflow.SUSPEND,
+  )
+
+  init {
+
+    allChunksFlow
+      .runningFold(mapOf<ServerMapChunkId, MutableSharedFlow<Cmd.ChunkSubdivide>>()) { acc, src ->
+
+        val flow = acc.getOrElse(src.chunkId) {
+          println("creating new subdivide-flow for chunkId ${src.chunkId}")
+          createSubdivisionFlow()
+        }
+
+        flow.emit(Cmd.ChunkSubdivide(src.chunkId, src.map))
+
+        acc + (src.chunkId to flow)
+      }
+      .onEach {
+        println("allChunksFlow has ${it.size} subdivide-flows: ${it.keys.joinToString()}")
+      }
+      .launchIn(this)
+
+
+    saveImagesFlow
+      .onEach { (filename, img) ->
+
+        val chunkImageFile = File(filename.value)
+        if (chunkImageFile.parentFile.mkdirs()) {
+          println("created new map tile parentFile directory ${chunkImageFile.absolutePath}")
+        }
+
+        val savedTile = img.output(PngWriter.NoCompression, chunkImageFile)
+        println("savedTile: $savedTile")
+      }
+      .launchIn(this)
+
+  }
+
+
+  /**
+   * Given a chunk, split it into multiple, smaller sub-tiles.
+   *
+   * Send each sub-tile to [saveImagesFlow] to be saved as a PNG
+   */
+  private fun createSubdivisionFlow(): MutableSharedFlow<Cmd.ChunkSubdivide> {
+
+    val subdivisionFlow = MutableSharedFlow<Cmd.ChunkSubdivide>(
+      replay = 0,
+      extraBufferCapacity = 10,
+      onBufferOverflow = BufferOverflow.SUSPEND,
+    )
+
+    val saveImgCommands = subdivisionFlow
+      .flatMapConcat { subdivideCmd ->
+        ChunkSize
+          .values
+          .flatMap { chunkSize ->
+            subdivideCmd.tiles
+              .entries
+              .groupBy(
+                { (tile, _) -> tile.toMapChunkPosition(chunkSize.tilesPerChunk) }
+              ) {
+                it.key to it.value
+              }
+              .map { (chunkPos, tiles) ->
+                async {
+                  Cmd.CreateImage(
+                    surfaceIndex = subdivideCmd.chunkId.surfaceIndex,
+                    chunkSize = chunkSize,
+                    chunkPosition = chunkPos,
+                    tiles = tiles.toMap()
+                  )
+                }
+              }
+          }
+          .awaitAll()
+          .asFlow()
+      }
+      .map { createCmd ->
+        val filename = TilePngFilename(createCmd)
+
+        val img = createMapTileImage(
+          createCmd.chunkPosition,
+          createCmd.tiles,
+          createCmd.chunkSize,
+        )
+
+        Cmd.SaveImage(filename, img)
+      }
+
+    launch {
+      supervisorScope {
+        println("emitting all saveImgCommands to saveImagesFlow")
+        saveImagesFlow.emitAll(saveImgCommands)
+      }
+    }
+
+    return subdivisionFlow
+  }
+
+
+  /** Create (but don't save) a PNG for the given chunk. */
   private fun createMapTileImage(
     chunkPosition: MapChunkPosition,
     chunkColours: Map<MapTilePosition, ColourHex>,
-  z: ZoomLevel
+    chunkSize: ChunkSize
   ): ImmutableImage {
 
     val chunkImage = ImmutableImage.filled(
-    z.chunkSize,
-    z.chunkSize,
+      chunkSize.tilesPerChunk,
+      chunkSize.tilesPerChunk,
       TRANSPARENT_AWT,
       BufferedImage.TYPE_INT_ARGB
     )
 
-  val chunkOriginX: Int = chunkPosition.x * z.chunkSize
-  val chunkOriginY: Int = chunkPosition.y * z.chunkSize
+    val chunkOriginX: Int = chunkPosition.x * chunkSize.tilesPerChunk
+    val chunkOriginY: Int = chunkPosition.y * chunkSize.tilesPerChunk
 
     chunkColours.forEach { (tilePosition, colour) ->
 
@@ -111,132 +250,8 @@ private fun createMapTileImage(
 
     return chunkImage.scaleTo(WEB_MAP_TILE_SIZE_PX, WEB_MAP_TILE_SIZE_PX, ScaleMethod.FastScale)
   }
-
-
-private fun ColourHex.toRgbColor(): RGBColor {
-  return RGBColor(
-    red.toInt(),
-    green.toInt(),
-    blue.toInt(),
-    alpha.toInt(),
-  )
 }
 
-private class ServerMapChunkHandler : CoroutineScope {
-
-  override val coroutineContext: CoroutineContext = Dispatchers.Default +
-      SupervisorJob(cScope.coroutineContext.job) +
-      CoroutineName("ServerMapChunkHandler")
-
-
-  private val allChunksFlow = MutableSharedFlow<ServerMapChunkTiles<ColourHex>>(
-    replay = 0,
-    extraBufferCapacity = 10,
-    onBufferOverflow = BufferOverflow.SUSPEND,
-  )
-
-  private val saveImagesFlow = MutableSharedFlow<Cmd.SaveImage>(
-    replay = 0,
-    extraBufferCapacity = 10,
-    onBufferOverflow = BufferOverflow.SUSPEND,
-  )
-
-
-  init {
-
-    allChunksFlow
-      .runningFold(mapOf<ServerMapChunkId, MutableSharedFlow<Cmd.ChunkSubdivide>>()) { acc, src ->
-
-        val flow = acc.getOrElse(src.chunkId) {
-          println("creating new subdivide-flow for chunkId ${src.chunkId}")
-          subdivisionFlow()
-        }
-
-        flow.emit(Cmd.ChunkSubdivide(src.chunkId, src.map))
-
-        acc + (src.chunkId to flow)
-      }
-      .onEach {
-        println("allChunksFlow has ${it.size} subdivide-flows: ${it.keys.joinToString()}")
-      }
-      .launchIn(this)
-
-    saveImagesFlow
-      .onEach { (filename, img) ->
-
-        val chunkImageFile = File(filename.value)
-        if (chunkImageFile.parentFile.mkdirs()) {
-          println("created new map tile parentFile directory ${chunkImageFile.absolutePath}")
-        }
-
-        val savedTile = img.output(PngWriter.NoCompression, chunkImageFile)
-        println("savedTile: $savedTile")
-      }
-      .launchIn(this)
-
-  }
-
-  fun emit(mapChunkTiles: ServerMapChunkTiles<ColourHex>) {
-    launch {
-      allChunksFlow.emit(mapChunkTiles)
-    }
-  }
-
-  private fun subdivisionFlow(): MutableSharedFlow<Cmd.ChunkSubdivide> {
-
-    val flow = MutableSharedFlow<Cmd.ChunkSubdivide>(
-      replay = 0,
-      extraBufferCapacity = 10,
-      onBufferOverflow = BufferOverflow.SUSPEND,
-    )
-
-    val saveImgCommands = flow
-//      .conflate()
-      .flatMapConcat { subdivideCmd ->
-        ZoomLevel.values
-          .asFlow()
-          .flatMapConcat { zoom ->
-            subdivideCmd.tiles
-              .entries
-              .groupBy({ (tile, _) -> tile.toMapChunkPosition(zoom.chunkSize) }
-              ) {
-                it.key to it.value
-              }
-              .map { (chunkPos, tiles) ->
-                Cmd.CreateImage(
-                  surfaceIndex = subdivideCmd.chunkId.surfaceIndex,
-                  zoomLevel = zoom,
-                  chunkPosition = chunkPos,
-                  tiles = tiles.toMap()
-                )
-              }
-              .asFlow()
-          }
-      }
-      .map { createCmd ->
-        val filename = TilePngFilename(createCmd)
-
-        val img = createMapTileImage(
-          createCmd.chunkPosition,
-          createCmd.tiles,
-          createCmd.zoomLevel,
-        )
-
-        Cmd.SaveImage(filename, img)
-      }
-
-    launch {
-      supervisorScope {
-        println("emitting all saveImgCommands to saveImagesFlow")
-        saveImagesFlow.emitAll(saveImgCommands)
-      }
-    }
-
-    return flow
-  }
-
-
-}
 
 private sealed interface Cmd {
 
@@ -247,7 +262,7 @@ private sealed interface Cmd {
 
   data class CreateImage(
     val surfaceIndex: SurfaceIndex,
-    val zoomLevel: ZoomLevel,
+    val chunkSize: ChunkSize,
     val chunkPosition: MapChunkPosition,
     val tiles: Map<MapTilePosition, ColourHex>,
   ) : Cmd

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

@@ -6,6 +6,7 @@ import dev.adamko.kafkatorio.events.schema.MapChunk
 import dev.adamko.kafkatorio.events.schema.MapChunkPosition
 import dev.adamko.kafkatorio.events.schema.MapTilePosition
 import dev.adamko.kafkatorio.events.schema.MapTiles
+import dev.adamko.kafkatorio.events.schema.SurfaceIndex
 import dev.adamko.kafkatorio.events.schema.converters.toMapChunkPosition
 import dev.adamko.kafkatorio.processor.serdes.kxsBinary
 import dev.adamko.kotka.extensions.groupedAs
@@ -28,7 +29,7 @@ data class ServerMapChunkId(
   val serverId: FactorioServerId,
   val chunkPosition: MapChunkPosition,
   val surfaceIndex: SurfaceIndex,
-  val zoomLevel: ZoomLevel,
+  val chunkSize: ChunkSize,
 )
 
 
@@ -77,20 +78,20 @@ fun groupTilesIntoChunksWithColours(
       .filter { _, value -> value != null && value.tiles.isNotEmpty() }
       .flatMap("server-map-data.tiles.flatMapByChunk") { key: FactorioServerId, mapTiles: MapTiles ->
 
-        val zoomLevel = ZoomLevel.ZOOM_0
+        val standardChunkSize = ChunkSize.MAX
 
         mapTiles
           .tiles
           .groupBy { tile ->
-            tile.position.toMapChunkPosition(zoomLevel.chunkSize)
+            tile.position.toMapChunkPosition(standardChunkSize.tilesPerChunk)
           }
           .map { (chunkPos, tiles) ->
 
             val chunkId = ServerMapChunkId(
               key,
               chunkPos,
-              SurfaceIndex(mapTiles.surfaceIndex),
-              zoomLevel,
+              mapTiles.surfaceIndex,
+              standardChunkSize,
             )
 
             val tilesMap = tiles.associate { it.position to TileProtoHashCode(it) }

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

@@ -5,15 +5,6 @@ import kotlin.math.roundToInt
 import kotlinx.serialization.Serializable
 
 
-@Serializable
-@JvmInline
-value class SurfaceIndex(
-  val index: Int
-) {
-  override fun toString() = "$index"
-}
-
-
 @JvmInline
 @Serializable
 value class FactorioServerId(val id: String) {
@@ -21,23 +12,35 @@ value class FactorioServerId(val id: String) {
 }
 
 
-enum class ZoomLevel(
-  val level: Int
-) {
-  ZOOM_0(0), // 512
-  ZOOM_1(1), // 256
-  ZOOM_2(2), // 128
-  ZOOM_3(3), // 64
-  ZOOM_4(4), // 32
+@Serializable
+enum class ChunkSize(
+  val zoomLevel: Int,
+) : Comparable<ChunkSize> {
+  CHUNK_512(-1), // 512
+  CHUNK_256(0), // 256
+  CHUNK_128(1), // 128
+  CHUNK_064(2), // 64
+  CHUNK_032(3), // 32
   ;
 
-  val chunkSize: Int = 2f.pow(9 - level).roundToInt()
+  val tilesPerChunk: Int = 2f.pow(8 - zoomLevel).roundToInt()
 
   init {
-    require(chunkSize > 0) { "chunkSize must be positive" }
+    require(tilesPerChunk > 0) { "chunkSize must be positive" }
+    // 1000 & 0111 = 0000 =>  pow^2
+    // 1001 & 1000 = 1000 => !pow^2
+    require(tilesPerChunk and (tilesPerChunk - 1) == 0) {
+      "tilesPerChunk must be a power-of-two number"
+    }
   }
 
   companion object {
-    val values: List<ZoomLevel> = values().asList() // better performance, KT-48872
+    // cache values for better performance. KT-48872
+    val values: List<ChunkSize> = values().asList()
+
+    val MAX: ChunkSize = values.maxByOrNull { it.tilesPerChunk }!!
+    val MIN: ChunkSize = values.minByOrNull { it.tilesPerChunk }!!
+    val STANDARD: ChunkSize = values.first { it.zoomLevel == 0 }
+
   }
 }

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

@@ -3,16 +3,17 @@ package dev.adamko.kafkatorio.processor.misc
 import dev.adamko.kafkatorio.events.schema.ColourHex
 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.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.processor.topology.SurfaceIndex
-import dev.adamko.kafkatorio.processor.topology.ZoomLevel
 import java.util.UUID
 import kotlinx.serialization.encodeToByteArray
 import org.junit.jupiter.api.Test
 
+
 class Blah {
 
   @Test
@@ -22,8 +23,8 @@ class Blah {
       chunkId = ServerMapChunkId(
         serverId = FactorioServerId(UUID.randomUUID().toString()),
         chunkPosition = MapChunkPosition(Int.MAX_VALUE, Int.MAX_VALUE),
-        surfaceIndex = SurfaceIndex(Int.MAX_VALUE),
-        zoomLevel = ZoomLevel.ZOOM_0
+        surfaceIndex = SurfaceIndex(UInt.MAX_VALUE),
+        chunkSize = ChunkSize.MAX
       ),
       buildMap {
         repeat(512) { x ->