remove commented out code, try using materializedAs to fix KTable-KTable join not working, make 'saveMapTiles' more encapsulated

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

@@ -1,17 +1,11 @@
 package dev.adamko.kafkatorio.processor
 
-import dev.adamko.kafkatorio.events.schema.ColourHex
 import dev.adamko.kafkatorio.processor.config.ApplicationProperties
-import dev.adamko.kafkatorio.processor.serdes.kxsBinary
-import dev.adamko.kafkatorio.processor.topology.ServerMapChunkId
-import dev.adamko.kafkatorio.processor.topology.ServerMapChunkTiles
 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
-import dev.adamko.kotka.extensions.consumedAs
-import dev.adamko.kotka.kxs.serde
 import java.time.Duration
 import kotlin.coroutines.CoroutineContext
 import kotlin.coroutines.cancellation.CancellationException
@@ -26,7 +20,6 @@ import org.apache.kafka.streams.StreamsBuilder
 import org.apache.kafka.streams.StreamsConfig
 import org.apache.kafka.streams.Topology
 import org.apache.kafka.streams.TopologyDescription
-import org.apache.kafka.streams.kstream.KTable
 
 
 internal class KafkatorioTopology(
@@ -57,16 +50,7 @@ internal class KafkatorioTopology(
 
   private fun saveTiles() {
     val builder = StreamsBuilder()
-
-    val groupedMapChunkTiles: KTable<ServerMapChunkId?, ServerMapChunkTiles<ColourHex>?> =
-      builder.table(
-        TOPIC_GROUPED_MAP_CHUNKS,
-        consumedAs("consume.grouped-map-chunks", kxsBinary.serde(), kxsBinary.serde())
-      )
-
-    saveMapTiles(groupedMapChunkTiles)
-
-    val topology = builder.build()
+    val topology = saveMapTiles(builder)
     launchTopology("saveTiles", topology)
   }
 

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

@@ -9,7 +9,11 @@ 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 dev.adamko.kotka.extensions.streams.*
+import dev.adamko.kafkatorio.processor.KafkatorioTopology
+import dev.adamko.kafkatorio.processor.serdes.kxsBinary
+import dev.adamko.kotka.extensions.consumedAs
+import dev.adamko.kotka.extensions.streams.forEach
+import dev.adamko.kotka.kxs.serde
 import java.awt.image.BufferedImage
 import java.io.File
 import kotlin.coroutines.CoroutineContext
@@ -33,7 +37,9 @@ import kotlinx.coroutines.job
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.supervisorScope
-import org.apache.kafka.streams.kstream.KTable
+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
@@ -48,11 +54,16 @@ private val serverMapChunkHandler = ServerMapChunkHandler()
 
 
 fun saveMapTiles(
-  serverMapTable: KTable<ServerMapChunkId?, ServerMapChunkTiles<ColourHex>?>
-) {
+  builder: StreamsBuilder,
+): Topology {
 
-  serverMapTable
-    .toStream()
+  val groupedMapChunkTiles: KStream<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> =
+    builder.stream(
+      KafkatorioTopology.TOPIC_GROUPED_MAP_CHUNKS,
+      consumedAs("consume.grouped-map-chunks", kxsBinary.serde(), kxsBinary.serde())
+    )
+
+  groupedMapChunkTiles
     .forEach("save-chunked-tiles") { _, value ->
       if (value != null) {
         runBlocking(Dispatchers.Default) {
@@ -62,6 +73,7 @@ fun saveMapTiles(
       }
     }
 
+  return builder.build()
 }
 
 

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

@@ -18,12 +18,20 @@ import dev.adamko.kafkatorio.processor.serdes.jsonMapper
 import dev.adamko.kafkatorio.processor.serdes.kxsBinary
 import dev.adamko.kotka.extensions.consumedAs
 import dev.adamko.kotka.extensions.groupedAs
-import dev.adamko.kotka.extensions.materializedWith
+import dev.adamko.kotka.extensions.materializedAs
 import dev.adamko.kotka.extensions.producedAs
-import dev.adamko.kotka.extensions.streams.*
+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.tables.join
 import dev.adamko.kotka.extensions.tables.toStream
 import dev.adamko.kotka.kxs.serde
+import kotlin.coroutines.CoroutineContext
+import kotlinx.coroutines.CoroutineName
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.SupervisorJob
 import kotlinx.serialization.Serializable
 import org.apache.kafka.streams.StreamsBuilder
 import org.apache.kafka.streams.Topology
@@ -104,12 +112,21 @@ fun groupMapChunks(
 
   groupedMapChunkTiles
     .toStream("stream-grouped-map-tiles")
+    .filter("stream-grouped-map-tiles.filter-not-null") { _, v ->
+      !v?.map.isNullOrEmpty()
+    }
+    .mapValues("stream-grouped-map-tiles.map-not-null") { _, v ->
+      v as ServerMapChunkTiles<ColourHex>
+    }
+    .peek("stream-grouped-map-tiles.peek") { _, v ->
+      println("Grouping map tiles result: ${v.chunkId} / ${v.map.size}")
+    }
     .to(
       KafkatorioTopology.TOPIC_GROUPED_MAP_CHUNKS,
       producedAs(
         "produce.grouped-map-chunks",
-        kxsBinary.serde<ServerMapChunkId?>(),
-        kxsBinary.serde<ServerMapChunkTiles<ColourHex>?>()
+        kxsBinary.serde<ServerMapChunkId>(),
+        kxsBinary.serde<ServerMapChunkTiles<ColourHex>>()
       )
     )
 
@@ -117,38 +134,10 @@ fun groupMapChunks(
 }
 
 private fun groupTilesIntoChunksWithColours(
-//  mapTilePacketsStream: KStream<FactorioServerId, FactorioEvent>,
-//  mapChunksPacketsStream: KStream<FactorioServerId, FactorioEvent>,
   mapChunkUpdatePacketsStream: KStream<FactorioServerId, FactorioEventUpdatePacket>,
   tileProtoColourDict: KTable<FactorioServerId, TileColourDict>,
 ): KTable<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> {
 
-
-//  val mapChunkTilesStream: KStream<FactorioServerId, MapTiles> =
-//    mapChunksPacketsStream
-//      .filter { _, packet: FactorioEvent? ->
-//        when (val data = packet?.data) {
-//          is MapChunk -> data.tiles.tiles.isNotEmpty()
-//          else        -> false
-//        }
-//      }
-//      .mapValues("map-chunk-packets.convert-to-map-tiles") { _, packet: FactorioEvent ->
-//        (packet.data as MapChunk).tiles
-//      }
-//
-//
-//  val mapTilesStream: KStream<FactorioServerId, MapTiles> =
-//    mapTilePacketsStream
-//      .filter { _, packet: FactorioEvent? ->
-//        when (val data = packet?.data) {
-//          is MapTiles -> data.tiles.isNotEmpty()
-//          else        -> false
-//        }
-//      }
-//      .mapValues("map-tiles-packets.convert-to-map-tiles") { _, packet: FactorioEvent ->
-//        packet.data as MapTiles
-//      }
-
   val mapChunkUpdatesStream: KStream<FactorioServerId, MapTiles> =
     mapChunkUpdatePacketsStream
       .filter { _, packet: FactorioEventUpdatePacket? ->
@@ -174,37 +163,38 @@ private fun groupTilesIntoChunksWithColours(
       }
 
 
-  // group tiles by server & surface & chunk
   val chunkedTilesTable: KTable<ServerMapChunkId, ServerMapChunkTiles<TileProtoHashCode>> =
     mapChunkUpdatesStream
-//    mapTilesStream
-//      .merge("merge-mapTilesStream-with-mapChunkTilesStream", mapChunkTilesStream)
-//      .merge("merge-mapTilesStream-with-mapChunkUpdatesStream", mapChunkUpdatesStream)
-//      .filter { _, value -> !value?.tiles.isNullOrEmpty() }
+      // group tiles by server & surface & chunk
       .flatMap("server-map-data.tiles.flatMapByChunk") { serverId: FactorioServerId, mapTiles: MapTiles ->
 
         val standardChunkSize = ChunkSize.MAX
 
-        mapTiles
-          .tiles
-          .groupBy { tile ->
-            tile.position.toMapChunkPosition(standardChunkSize.tilesPerChunk)
-          }
-          .map { (chunkPos, tiles) ->
+        val flatMappedChunkTiles: List<Pair<ServerMapChunkId, ServerMapChunkTiles<TileProtoHashCode>>> =
+          mapTiles
+            .tiles
+            .groupBy { tile ->
+              tile.position.toMapChunkPosition(standardChunkSize.tilesPerChunk)
+            }
+            .map { (chunkPos, tiles) ->
 
-            val chunkId = ServerMapChunkId(
-              serverId = serverId,
-              chunkPosition = chunkPos,
-              surfaceIndex = mapTiles.surfaceIndex,
-              chunkSize = standardChunkSize,
-            )
+              val chunkId = ServerMapChunkId(
+                serverId = serverId,
+                chunkPosition = chunkPos,
+                surfaceIndex = mapTiles.surfaceIndex,
+                chunkSize = standardChunkSize,
+              )
 
-            val tilesMap = tiles.associate { it.position to TileProtoHashCode(it) }
+              val tilesMap = tiles.associate { it.position to TileProtoHashCode(it) }
 
-            val chunkTiles = ServerMapChunkTiles(chunkId, tilesMap)
+              val chunkTiles = ServerMapChunkTiles(chunkId, tilesMap)
 
-            chunkId to chunkTiles
-          }
+              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()
@@ -212,18 +202,18 @@ private fun groupTilesIntoChunksWithColours(
       .groupByKey(
         groupedAs(
           "server-map-data.tiles.group",
-          kxsBinary.serde<ServerMapChunkId?>(),
-          kxsBinary.serde<ServerMapChunkTiles<TileProtoHashCode>?>(),
+          kxsBinary.serde<ServerMapChunkId>(),
+          kxsBinary.serde<ServerMapChunkTiles<TileProtoHashCode>>(),
         )
       )
       .reduce(
-        "server-map-data.tiles.reduce", materializedWith(
-          //          "server-map-data.tiles.store",
+        "server-map-data.tiles.reduce",
+        materializedAs(
+          "server-map-data.tiles.store",
           kxsBinary.serde<ServerMapChunkId>(),
           kxsBinary.serde<ServerMapChunkTiles<TileProtoHashCode>>(),
-        ),
-        ServerMapChunkTiles<TileProtoHashCode>::plus
-      )
+        )
+      ) { chunkTiles, other -> chunkTiles + other }
 
 
   val colourisedChunkTable: KTable<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> =
@@ -231,7 +221,8 @@ private fun groupTilesIntoChunksWithColours(
       .join(
         other = tileProtoColourDict,
         name = "server-map-data.join-tiles-with-prototypes",
-        materialized = materializedWith(
+        materialized = materializedAs(
+          "server-map-data.join-tiles-with-prototypes.store",
           kxsBinary.serde<ServerMapChunkId>(),
           kxsBinary.serde<ServerMapChunkTiles<ColourHex>>(),
         ),
@@ -251,11 +242,12 @@ private fun groupTilesIntoChunksWithColours(
           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

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

@@ -56,8 +56,8 @@ fun tileProtoColourDictionary(
     }
     .toTable(
       "server-map-data.tile-prototypes",
-      materializedWith(
-//        "server-map-data.tile-prototypes.store",
+      materializedAs(
+        "server-map-data.tile-prototypes.store",
         kxsBinary.serde(),
         kxsBinary.serde(),
       )

modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/topology/GroupMapChunksTest.kt

@@ -95,12 +95,12 @@ class GroupMapChunksTest : FunSpec({
       )
     )
 
-    val outputTopic: TestOutputTopic<ServerMapChunkId?, ServerMapChunkTiles<ColourHex>?> =
+    val outputTopic: TestOutputTopic<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> =
       testDriver
         .createOutputTopic(
           KafkatorioTopology.TOPIC_GROUPED_MAP_CHUNKS,
-          kxsBinary.serde<ServerMapChunkId?>().deserializer(),
-          kxsBinary.serde<ServerMapChunkTiles<ColourHex>?>().deserializer(),
+          kxsBinary.serde<ServerMapChunkId>().deserializer(),
+          kxsBinary.serde<ServerMapChunkTiles<ColourHex>>().deserializer(),
         )
 
     val records = outputTopic.readRecordsToList()