tidy up topology

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.saveMapTiles2
+import dev.adamko.kafkatorio.processor.topology.saveMapTiles
 import dev.adamko.kafkatorio.processor.topology.splitFactorioServerPacketStream
 import java.time.Duration
 import java.util.Properties
@@ -60,8 +60,7 @@ internal class KafkatorioTopology(
 
   private fun saveTiles() {
     val builder = StreamsBuilder()
-//    val topology = saveMapTiles(builder)
+    val topology = saveMapTiles(builder, appProps.webmapTileDir)
-    val topology = saveMapTiles2(builder, appProps.webmapTileDir)
     launchTopology("saveTiles", topology)
   }
 
@@ -87,8 +86,6 @@ internal class KafkatorioTopology(
     val appId = props.compute(StreamsConfig.APPLICATION_ID_CONFIG) { _, v -> "$v.$id" } as? String
     props.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, appId)
 
-//    props.setProperty(StreamsConfig.MAX_TASK_IDLE_MS_CONFIG, "${30.minutes.inWholeMilliseconds}")
-
     val streams = KafkaStreams(topology, props)
     streams.setUncaughtExceptionHandler(StreamsExceptionHandler {
       coroutineContext.job.cancel(CancellationException(cause = it))

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

@@ -1,309 +0,0 @@
-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.KafkatorioTopology.Companion.rootJob
-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
-import kotlin.io.path.fileSize
-import kotlin.math.abs
-import kotlin.time.Duration.Companion.seconds
-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.Flow
-import kotlinx.coroutines.flow.MutableSharedFlow
-import kotlinx.coroutines.flow.asFlow
-import kotlinx.coroutines.flow.debounce
-import kotlinx.coroutines.flow.distinctUntilChangedBy
-import kotlinx.coroutines.flow.emitAll
-import kotlinx.coroutines.flow.flatMapConcat
-import kotlinx.coroutines.flow.launchIn
-import kotlinx.coroutines.flow.map
-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
-
-
-private const val WEB_MAP_TILE_SIZE_PX = 256
-
-
-/** A bridge between Kafka and Kotlin */
-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("$pid.consume.grouped-map-chunks", kxsBinary.serde(), kxsBinary.serde())
-    )
-
-  groupedMapChunkTiles
-    .forEach("save-chunked-tiles") { _, value ->
-      if (value != null) {
-        println("emitting chunkedTiles ${value.chunkId} tiles: ${value.map.size}")
-        serverMapChunkHandler.emit(value)
-      }
-    }
-
-  return builder.build()
-}
-
-
-@JvmInline
-value class TilePngFilename(
-  val value: String,
-) {
-  constructor(id: ServerMapChunkId) : this(buildString {
-    append("s${id.surfaceIndex}")
-    append("/z${id.chunkSize.zoomLevel}")
-    append("/x${id.chunkPosition.x}")
-    append("/y${id.chunkPosition.y}")
-    append(".png")
-  })
-}
-
-
-val TRANSPARENT_AWT: Color = ColourHex.TRANSPARENT.toRgbColor().awt()
-
-
-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(rootJob) + 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(mutableMapOf<ServerMapChunkId, MutableSharedFlow<Cmd.ChunkSubdivide>>()) { acc, src ->
-
-        val flow = acc.getOrPut(src.chunkId) {
-          println("creating new subdivide-flow for chunkId ${src.chunkId}")
-          createSubdivisionFlow()
-        }
-
-        flow.emit(Cmd.ChunkSubdivide(src.chunkId, src.map))
-
-        acc
-      }
-      .distinctUntilChangedBy { it.size }
-      .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 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")
-        }
-      }
-      .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: Flow<Cmd.SaveImage> =
-      subdivisionFlow
-        .debounce(30.seconds)
-        .flatMapConcat { subdivideCmd ->
-          ChunkSize
-            .entries
-            .flatMap { chunkSize ->
-              subdivideCmd
-                .tiles
-                .entries
-                .groupBy(
-                  { (tile, _) -> tile.toMapChunkPosition(chunkSize.lengthInTiles) }
-                ) { (tilePosition, colour) ->
-                  tilePosition to colour
-                }
-                .map { (chunkPos, tiles) ->
-                  async {
-                    Cmd.CreateImage(
-                      surfaceIndex = subdivideCmd.chunkId.surfaceIndex,
-                      chunkSize = chunkSize,
-                      chunkPosition = chunkPos,
-                      tiles = tiles.toMap()
-                    )
-                  }
-                }
-            }
-            .awaitAll()
-            .asFlow()
-        }
-        .map { createCmd ->
-          val img = createMapTileImage(
-            createCmd.chunkPosition,
-            createCmd.tiles,
-            createCmd.chunkSize,
-          )
-
-          Cmd.SaveImage(createCmd.tilePngFilename, 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>,
-    chunkSize: ChunkSize
-  ): ImmutableImage {
-
-    val chunkImage = ImmutableImage.filled(
-      chunkSize.lengthInTiles,
-      chunkSize.lengthInTiles,
-      TRANSPARENT_AWT,
-      BufferedImage.TYPE_INT_ARGB
-    )
-
-    val chunkOriginX: Int = chunkPosition.x * chunkSize.lengthInTiles
-    val chunkOriginY: Int = chunkPosition.y * chunkSize.lengthInTiles
-
-    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
-      )
-    }
-
-    return chunkImage.scaleTo(WEB_MAP_TILE_SIZE_PX, WEB_MAP_TILE_SIZE_PX, ScaleMethod.FastScale)
-  }
-}
-
-
-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 {
-    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,
-    val image: ImmutableImage,
-  ) : Cmd
-
-}

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

@@ -107,7 +107,6 @@ private fun groupTilesIntoChunksWithColours(
   tileProtoColourDict: KTable<FactorioServerId, TileColourDict>,
 ): KTable<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> {
 
-
   val mapTilesStream: KStream<FactorioServerId, MapTiles> =
     mapChunksStream.filter { _, value ->
       !value.tileDictionary.isNullOrEmpty()
@@ -117,10 +116,6 @@ private fun groupTilesIntoChunksWithColours(
     }.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<ServerMapChunkId, ServerMapChunkTiles<TileProtoHashCode>> =
     mapTilesStream

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

@@ -1,5 +1,8 @@
 package dev.adamko.kafkatorio.processor.topology
 
+import com.sksamuel.scrimage.color.RGBColor
+import dev.adamko.kafkatorio.schema.common.ColourHex
+import java.awt.Color
 import kotlin.math.pow
 import kotlin.math.roundToInt
 import kotlinx.serialization.Serializable

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

@@ -2,6 +2,7 @@ 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_DEBOUNCED
 import dev.adamko.kafkatorio.processor.admin.TOPIC_SUBDIVIDED_MAP_TILES
@@ -16,6 +17,7 @@ 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
@@ -27,16 +29,18 @@ import org.apache.kafka.streams.Topology
 import org.apache.kafka.streams.kstream.KStream
 
 
+private const val pid = "saveMapTiles"
+
 private const val WEB_MAP_TILE_SIZE_PX = 256
 
+
 /**
  * @param[tileDirectory] [dev.adamko.kafkatorio.processor.config.ApplicationProperties.webmapTileDir]
  */
-fun saveMapTiles2(
+fun saveMapTiles(
   builder: StreamsBuilder,
   tileDirectory: Path,
 ): Topology {
-  val pid = "saveMapTiles"
 
   val groupedMapChunkTiles: KStream<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> =
     builder.stream(
@@ -44,37 +48,9 @@ fun saveMapTiles2(
       consumedAs("$pid.consume.grouped-map-chunks", kxsBinary.serde(), kxsBinary.serde()),
     )
 
-  groupedMapChunkTiles.flatMap("$pid.subdivide-chunks") { initialKey, initialChunk ->
+  groupedMapChunkTiles
-
+    .flatMapSubdividedChunk()
-    val (chunks, time) = measureTimedValue {
+    .to(
-      ChunkSize.entries.flatMap { chunkSize ->
-        initialChunk.map
-          .entries
-          .groupingBy { (tilePosition, _) ->
-            val chunkPosition = tilePosition.toMapChunkPosition(chunkSize.lengthInTiles)
-//            println("$tilePosition groupingBy $chunkPosition")
-            initialKey.copy(
-              chunkSize = chunkSize,
-              chunkPosition = chunkPosition,
-            )
-          }.fold(
-            initialValueSelector = { _, _ -> mutableMapOf<MapTilePosition, ColourHex>() }
-          ) { _, accumulator, (tilePosition, colour) ->
-//            println("fold $tilePosition $colour")
-            accumulator[tilePosition] = colour
-            accumulator
-          }
-          .map { (chunkId, tiles) ->
-//            println("map $chunkId ${tiles.size}")
-            chunkId to ServerMapChunkTiles(chunkId, tiles)
-          }
-      }
-
-    }
-    println("subdivided ${initialKey.chunkPosition} to ${chunks.size} chunks in $time")
-
-    chunks
-  }.to(
       TOPIC_SUBDIVIDED_MAP_TILES,
       producedAs(
         "$pid.produce.grouped-map-chunks",
@@ -93,8 +69,56 @@ fun saveMapTiles2(
       ),
     )
 
-  subdividedMapChunkTilesDebounced
+  subdividedMapChunkTilesDebounced.forEachChunkSaveImage(tileDirectory)
-    .forEach("$pid.save-tiles") { chunkId: ServerMapChunkId, tileColours: ServerMapChunkTiles<ColourHex> ->
+
+  return builder.build()
+    .addDebounceProcessor<ServerMapChunkId, ServerMapChunkTiles<ColourHex>>(
+      namePrefix = pid,
+      sourceTopic = TOPIC_SUBDIVIDED_MAP_TILES,
+      sinkTopic = TOPIC_SUBDIVIDED_MAP_TILES_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.lengthInTiles)
+            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(
+  "$pid.save-tiles"
+) { chunkId: ServerMapChunkId, tileColours: ServerMapChunkTiles<ColourHex> ->
 
   val chunkImage = ImmutableImage.filled(
     chunkId.chunkSize.lengthInTiles,
@@ -122,16 +146,6 @@ fun saveMapTiles2(
   saveTile(filename, tileDirectory, scaledImage)
 }
 
-  return builder.build()
-    .addDebounceProcessor<ServerMapChunkId, ServerMapChunkTiles<ColourHex>>(
-      namePrefix = pid,
-      sourceTopic = TOPIC_SUBDIVIDED_MAP_TILES,
-      sinkTopic = TOPIC_SUBDIVIDED_MAP_TILES_DEBOUNCED,
-      inactivityDuration = 15.seconds,
-      keySerde = kxsBinary.serde(),
-      valueSerde = kxsBinary.serde(),
-    )
-}
 
 @Synchronized
 fun saveTile(
@@ -156,3 +170,30 @@ fun saveTile(
   }
   println("savedTile $sizeDescription: $savedTile")
 }
+
+
+@JvmInline
+value class TilePngFilename(
+  val value: String,
+) {
+  constructor(id: ServerMapChunkId) : this(buildString {
+    append("s${id.surfaceIndex}")
+    append("/z${id.chunkSize.zoomLevel}")
+    append("/x${id.chunkPosition.x}")
+    append("/y${id.chunkPosition.y}")
+    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/topology/SaveMapChunksTest.kt

@@ -78,7 +78,7 @@ class SaveMapChunksTest : FunSpec({
       val now = LocalTime.now().format(DateTimeFormatter.ISO_TIME).filter { it.isLetterOrDigit() }
       Path("build/test/save-map-chunks-${now}")
     }
-    private val topology = saveMapTiles2(streamsBuilder, outputDir)
+    private val topology = saveMapTiles(streamsBuilder, outputDir)
     val testDriver: TopologyTestDriver = TopologyTestDriver(topology)
 
     val groupedMapChunksInputTopic: TestInputTopic<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> =