- wicked sick size optimisation of map-tiles update packet

- fix ts-generator recursive value class unwrapping
- move kafka state dir
- try fixing kafka streams for tile generation (simplifying it, and fixing some nullability problems)

modules/events-server/.gitignore

@@ -0,0 +1 @@
+kafka-state/

modules/events-server/build.gradle.kts

@@ -66,7 +66,7 @@ tasks.withType<KotlinCompile> {
 val kafkaStateDirDelete by tasks.registering(Delete::class) {
   group = project.name
   mustRunAfter(kafkaConsumersDelete)
-  delete(layout.buildDirectory.dir("kafka-state"))
+  delete(layout.projectDirectory.dir("kafka-state"))
 }
 
 val kafkaConsumersDelete by tasks.registering(KafkaConsumerGroupsTask.DeleteAll::class) {

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

@@ -2,6 +2,8 @@ package dev.adamko.kafkatorio.processor
 
 import dev.adamko.kafkatorio.events.schema.ColourHex
 import dev.adamko.kafkatorio.events.schema.FactorioEvent
+import dev.adamko.kafkatorio.events.schema.FactorioEventUpdate
+import dev.adamko.kafkatorio.events.schema.FactorioEventUpdatePacket
 import dev.adamko.kafkatorio.events.schema.FactorioObjectData
 import dev.adamko.kafkatorio.events.schema.FactorioPrototypes
 import dev.adamko.kafkatorio.events.schema.KafkatorioPacket
@@ -33,6 +35,7 @@ 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.TopologyDescription
 import org.apache.kafka.streams.kstream.KStream
 import org.apache.kafka.streams.kstream.KTable
 
@@ -69,20 +72,29 @@ internal class KafkatorioTopology(
     val tileProtoColourDict = tileProtoColourDictionary(protosStream)
 
 
-    val mapTilesStream: KStream<FactorioServerId, FactorioEvent> =
+//    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 mapChunksUpdatesStream: KStream<FactorioServerId, FactorioEventUpdatePacket> =
       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())
+        "kafkatorio.${KafkatorioPacket.PacketType.UPDATE}.${FactorioEventUpdate.FactorioEventUpdateType.MAP_CHUNK}",
+        consumedAs("consume.map-chunk-updates", jsonMapper.serde(), jsonMapper.serde())
       )
+
     val groupedMapChunkTiles: KTable<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> =
       groupTilesIntoChunksWithColours(
-        mapTilesStream,
-        mapChunksStream,
+//        mapTilesStream,
+//        mapChunksStream,
+        mapChunksUpdatesStream,
         tileProtoColourDict,
       )
 
@@ -90,7 +102,11 @@ internal class KafkatorioTopology(
       .toStream("stream-grouped-map-tiles")
       .to(
         TOPIC_GROUPED_MAP_CHUNKS,
-        producedAs("produce.grouped-map-chunks", kxsBinary.serde(), kxsBinary.serde())
+        producedAs(
+          "produce.grouped-map-chunks",
+          kxsBinary.serde<ServerMapChunkId?>(),
+          kxsBinary.serde<ServerMapChunkTiles<ColourHex>?>()
+        )
       )
 
 
@@ -140,10 +156,19 @@ internal class KafkatorioTopology(
     val appId = props.compute(StreamsConfig.APPLICATION_ID_CONFIG) { _, v -> "$v.$id" } as? String
     props.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, appId)
 
-
     val streams = KafkaStreams(topology, props)
     streams.setUncaughtExceptionHandler(StreamsExceptionHandler())
 
+    val description: TopologyDescription = topology.describe()
+    println(
+      """
+        |----------------
+        |$appId
+        |$description
+        |----------------
+      """.trimMargin()
+    )
+
     coroutineContext.job.invokeOnCompletion {
       println("closing $id KafkaStreams")
       streams.close(Duration.ofSeconds(1))

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

@@ -1,10 +1,11 @@
 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.MapChunk
+import dev.adamko.kafkatorio.events.schema.FactorioEventUpdatePacket
 import dev.adamko.kafkatorio.events.schema.MapChunkPosition
+import dev.adamko.kafkatorio.events.schema.MapChunkUpdate
 import dev.adamko.kafkatorio.events.schema.MapTile
+import dev.adamko.kafkatorio.events.schema.MapTileDictionary.Companion.isNullOrEmpty
 import dev.adamko.kafkatorio.events.schema.MapTilePosition
 import dev.adamko.kafkatorio.events.schema.MapTiles
 import dev.adamko.kafkatorio.events.schema.SurfaceIndex
@@ -13,10 +14,8 @@ 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
@@ -44,39 +43,70 @@ data class ServerMapChunkTiles<Data>(
 
 
 fun groupTilesIntoChunksWithColours(
-  mapTilePacketsStream: KStream<FactorioServerId, FactorioEvent>,
-  mapChunksPacketsStream: KStream<FactorioServerId, FactorioEvent>,
+//  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? ->
-        packet != null
-            && packet.data is MapChunk
-            && !((packet.data as? MapChunk)?.tiles?.tiles.isNullOrEmpty())
+//  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? ->
+        when (val data = packet?.update) {
+          is MapChunkUpdate -> !data.tileDictionary.isNullOrEmpty()
+          else              -> false
+        }
       }
-      .mapValues("map-chunk-packets.convert-to-map-tiles") { _, packet: FactorioEvent ->
-        (packet.data as MapChunk).tiles
+      .mapValues("map-chunk-update-packets.convert-to-map-tiles") { _, packet: FactorioEventUpdatePacket ->
+
+        val mapChunkUpdate: MapChunkUpdate = requireNotNull(packet.update as? MapChunkUpdate) {
+          "invalid UpdatePacket type $packet"
+        }
+
+        val tiles: List<MapTile> = requireNotNull(mapChunkUpdate.tileDictionary?.toMapTileList()) {
+          "invalid MapChunkUpdate packet, tiles must not be null"
+        }
+        require(tiles.isNotEmpty()) {
+          "invalid MapChunkUpdate packet, tiles must not be empty"
+        }
+
+        MapTiles(mapChunkUpdate.surfaceIndex, 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>> =
-    mapTilesStream
-      .merge("merge-map-chunk-tiles-with-map-tiles", mapChunkTilesStream)
-      .filter { _, value -> value != null && value.tiles.isNotEmpty() }
+    mapChunkUpdatesStream
+//    mapTilesStream
+//      .merge("merge-mapTilesStream-with-mapChunkTilesStream", mapChunkTilesStream)
+//      .merge("merge-mapTilesStream-with-mapChunkUpdatesStream", mapChunkUpdatesStream)
+//      .filter { _, value -> !value?.tiles.isNullOrEmpty() }
       .flatMap("server-map-data.tiles.flatMapByChunk") { key: FactorioServerId, mapTiles: MapTiles ->
 
         val standardChunkSize = ChunkSize.MAX
@@ -102,11 +132,15 @@ fun groupTilesIntoChunksWithColours(
             chunkId to chunkTiles
           }
       }
-      .filter("server-map-data.tiles.filter-not-empty") { _, value ->
-        value.map.isNotEmpty()
-      }
+//      .filter("server-map-data.tiles.filter-not-empty") { _, value ->
+//        value.map.isNotEmpty()
+//      }
       .groupByKey(
-        groupedAs("server-map-data.tiles.group", kxsBinary.serde(), kxsBinary.serde())
+        groupedAs(
+          "server-map-data.tiles.group",
+          kxsBinary.serde<ServerMapChunkId>(),
+          kxsBinary.serde<ServerMapChunkTiles<TileProtoHashCode>>()
+        )
       )
       .reduce(
         "server-map-data.tiles.reduce",

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

@@ -1,6 +1,6 @@
 package dev.adamko.kafkatorio.processor.topology
 
-import dev.adamko.kafkatorio.events.schema.FactorioObjectData
+import dev.adamko.kafkatorio.events.schema.FactorioEventUpdate
 import dev.adamko.kafkatorio.events.schema.KafkatorioPacket
 import dev.adamko.kafkatorio.processor.WebsocketServer
 import dev.adamko.kafkatorio.processor.serdes.jsonMapper
@@ -15,8 +15,8 @@ fun playerUpdatesToWsServer(
   builder: StreamsBuilder = StreamsBuilder(),
 ) {
   builder.stream<FactorioServerId, KafkatorioPacket>(
-    "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.LuaPlayer}",
-    consumedAs("stream-player-updates-for-ws-server", jsonMapper.serde(), jsonMapper.serde())
+    "kafkatorio.${KafkatorioPacket.PacketType.UPDATE}.${FactorioEventUpdate.FactorioEventUpdateType.PLAYER}",
+    consumedAs("EventUpdate.Player-updates-for-ws-server", jsonMapper.serde(), jsonMapper.serde()),
   )
     .foreach { _, value ->
 //      println("sending ${value.packetType} packet to websocket")