reuse the packet-splitter topics for the tile-saver job

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

@@ -1,11 +1,19 @@
 package dev.adamko.kafkatorio.processor
 
+import dev.adamko.kafkatorio.events.schema.FactorioEvent
+import dev.adamko.kafkatorio.events.schema.FactorioObjectData
+import dev.adamko.kafkatorio.events.schema.FactorioPrototypes
+import dev.adamko.kafkatorio.events.schema.KafkatorioPacket
+import dev.adamko.kafkatorio.processor.serdes.jsonMapper
+import dev.adamko.kafkatorio.processor.topology.FactorioServerId
 import dev.adamko.kafkatorio.processor.topology.factorioServerPacketStream
 import dev.adamko.kafkatorio.processor.topology.groupTilesIntoChunksWithColours
 import dev.adamko.kafkatorio.processor.topology.playerUpdatesToWsServer
 import dev.adamko.kafkatorio.processor.topology.saveMapTiles
 import dev.adamko.kafkatorio.processor.topology.splitFactorioServerPacketStream
 import dev.adamko.kafkatorio.processor.topology.tileProtoColourDictionary
+import dev.adamko.kotka.extensions.consumedAs
+import dev.adamko.kotka.kxs.serde
 import java.time.Duration
 import kotlin.coroutines.CoroutineContext
 import kotlinx.coroutines.CoroutineName
@@ -18,6 +26,8 @@ import org.apache.kafka.streams.KafkaStreams
 import org.apache.kafka.streams.StreamsBuilder
 import org.apache.kafka.streams.StreamsConfig
 import org.apache.kafka.streams.Topology
+import org.apache.kafka.streams.kstream.KStream
+
 
 class KafkatorioTopology(
   private val websocketServer: WebsocketServer,
@@ -38,17 +48,37 @@ class KafkatorioTopology(
 
   private fun saveTiles() {
     val builder = StreamsBuilder()
-    val packets = factorioServerPacketStream(builder)
-    val tileProtoColourDict = tileProtoColourDictionary(packets)
 
+
+    val protosStream: KStream<FactorioServerId, FactorioPrototypes> =
+      builder.stream(
+        "kafkatorio.${KafkatorioPacket.PacketType.PROTOTYPES}.all",
+        consumedAs("consume.factorio-protos.all", jsonMapper.serde(), jsonMapper.serde())
+      )
+    val tileProtoColourDict = tileProtoColourDictionary(protosStream)
+
+
+    val mapTilesStream: KStream<FactorioServerId, FactorioEvent> =
+      builder.stream(
+        "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.LuaTiles}",
+        consumedAs("consume.map-tiles", jsonMapper.serde(), jsonMapper.serde())
+      )
+    val mapChunksStream: KStream<FactorioServerId, FactorioEvent> =
+      builder.stream(
+        "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.MapChunk}",
+        consumedAs("consume.map-chunks", jsonMapper.serde(), jsonMapper.serde())
+      )
     val tilePrototypesTable = groupTilesIntoChunksWithColours(
-      packets,
+      mapTilesStream,
+      mapChunksStream,
       tileProtoColourDict,
     )
+
+
     saveMapTiles(tilePrototypesTable)
 
-    val topology = builder.build()
 
+    val topology = builder.build()
     launchTopology("saveTiles", topology)
   }
 
@@ -74,7 +104,7 @@ class KafkatorioTopology(
 
   private fun launchTopology(id: String, topology: Topology) {
 
-    val props =  appProps.kafkaConfig
+    val props = appProps.kafkaConfig
 
     val appId = props.compute(StreamsConfig.APPLICATION_ID_CONFIG) { _, v -> "$v.$id" } as? String
     props.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, appId)

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

@@ -2,7 +2,6 @@ package dev.adamko.kafkatorio.processor.topology
 
 import dev.adamko.kafkatorio.events.schema.ColourHex
 import dev.adamko.kafkatorio.events.schema.FactorioEvent
-import dev.adamko.kafkatorio.events.schema.KafkatorioPacket
 import dev.adamko.kafkatorio.events.schema.MapChunk
 import dev.adamko.kafkatorio.events.schema.MapChunkPosition
 import dev.adamko.kafkatorio.events.schema.MapTilePosition
@@ -12,7 +11,10 @@ import dev.adamko.kafkatorio.processor.serdes.kxsBinary
 import dev.adamko.kotka.extensions.groupedAs
 import dev.adamko.kotka.extensions.materializedAs
 import dev.adamko.kotka.extensions.materializedWith
+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.merge
 import dev.adamko.kotka.extensions.streams.reduce
 import dev.adamko.kotka.extensions.tables.join
 import dev.adamko.kotka.kxs.serde
@@ -20,6 +22,7 @@ import kotlinx.serialization.Serializable
 import org.apache.kafka.streams.kstream.KStream
 import org.apache.kafka.streams.kstream.KTable
 
+
 @Serializable
 data class ServerMapChunkId(
   val serverId: FactorioServerId,
@@ -39,38 +42,54 @@ data class ServerMapChunkTiles<Data>(
 
 
 fun groupTilesIntoChunksWithColours(
-  factorioServerPacketStream: KStream<FactorioServerId, KafkatorioPacket>,
+  mapTilePacketsStream: KStream<FactorioServerId, FactorioEvent>,
+  mapChunksPacketsStream: KStream<FactorioServerId, FactorioEvent>,
   tileProtoColourDict: KTable<FactorioServerId, TileColourDict>,
 ): KTable<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> {
 
+
+  val mapChunkTilesStream: KStream<FactorioServerId, MapTiles> =
+    mapChunksPacketsStream
+      .filter { _, packet: FactorioEvent? ->
+        packet != null
+            && packet.data is MapChunk
+            && !((packet.data as? MapChunk)?.tiles?.tiles.isNullOrEmpty())
+      }
+      .mapValues("map-chunk-packets.convert-to-map-tiles") { _, packet: FactorioEvent ->
+        (packet.data as MapChunk).tiles
+      }
+
+  val mapTilesStream: KStream<FactorioServerId, MapTiles> =
+    mapTilePacketsStream
+      .filter { _, packet: FactorioEvent? ->
+        packet != null
+            && packet.data is MapTiles
+            && !((packet.data as? MapTiles)?.tiles.isNullOrEmpty())
+      }
+      .mapValues("map-tiles-packets.convert-to-map-tiles") { _, packet: FactorioEvent ->
+        packet.data as MapTiles
+      }
+
   // group tiles by server & surface & chunk
   val chunkedTilesTable: KTable<ServerMapChunkId, ServerMapChunkTiles<TileProtoHashCode>> =
-    factorioServerPacketStream
+    mapTilesStream
-      .flatMap("server-map-data.tiles.flatMapByChunk") { key: FactorioServerId, packet: KafkatorioPacket ->
+      .merge("merge-map-chunk-tiles-with-map-tiles", mapChunkTilesStream)
-
+      .filter { _, value -> value != null && value.tiles.isNotEmpty() }
-        val mapTileData = when (packet) {
+      .flatMap("server-map-data.tiles.flatMapByChunk") { key: FactorioServerId, mapTiles: MapTiles ->
-          is FactorioEvent ->
-            when (val data = packet.data) {
-              is MapChunk -> data.tiles
-              is MapTiles -> data
-              else        -> null
-            }
-          else             -> null
-        }
 
         val zoomLevel = ZoomLevel.ZOOM_0
 
-        mapTileData
+        mapTiles
-          ?.tiles
+          .tiles
-          ?.groupBy { tile ->
+          .groupBy { tile ->
             tile.position.toMapChunkPosition(zoomLevel.chunkSize)
           }
-          ?.map { (chunkPos, tiles) ->
+          .map { (chunkPos, tiles) ->
 
             val chunkId = ServerMapChunkId(
               key,
               chunkPos,
-              SurfaceIndex(mapTileData.surfaceIndex),
+              SurfaceIndex(mapTiles.surfaceIndex),
               zoomLevel,
             )
 
@@ -79,8 +98,10 @@ fun groupTilesIntoChunksWithColours(
             val chunkTiles = ServerMapChunkTiles(chunkId, tilesMap)
 
             chunkId to chunkTiles
-          } ?: listOf()
+          }
-
+      }
+      .filter("server-map-data.tiles.filter-not-empty") { _, value ->
+        value.map.isNotEmpty()
       }
       .groupByKey(
         groupedAs("server-map-data.tiles.group", kxsBinary.serde(), kxsBinary.serde())

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

@@ -40,7 +40,7 @@ value class TileColourDict(val map: Map<TileProtoHashCode, ColourHex>)
 
 /** Get the latest map-tile colours per server. */
 fun tileProtoColourDictionary(
-  factorioServerPacketStream: KStream<FactorioServerId, KafkatorioPacket>
+  factorioServerPacketStream: KStream<FactorioServerId, FactorioPrototypes>
 ): KTable<FactorioServerId, TileColourDict> {
 
   // get all the prototypes