kind of slow, but working...

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/admin/CreateTopics.kt

@@ -73,7 +73,7 @@ object CreateTopics {
   }
 
   private fun Admin.createTopics(
-    numPartitions: Int = 10,
+    numPartitions: Int = 3,
     replicationFactor: Short = 1,
     topicNames: () -> Collection<String>,
   ): CreateTopicsResult =

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

@@ -31,6 +31,7 @@ class KafkatorioTopology(
   fun build() {
 
     splitFactorioServerLog(builder)
+
     playerUpdatesToWsServer(websocketServer, builder)
 
     val allMapTilesTable: KTable<TileUpdateRecordKey, MapTile> = allMapTilesTable(builder)

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

@@ -1,6 +1,7 @@
 package dev.adamko.kafkatorio.processor
 
 
+import dev.adamko.kafkatorio.admin.CreateTopics
 import dev.adamko.kafkatorio.processor.config.ApplicationProperties
 import dev.adamko.kafkatorio.processor.tileserver.WebMapTileServer
 import kotlinx.coroutines.runBlocking
@@ -12,6 +13,8 @@ val appProps = ApplicationProperties()
 
 fun main() = runBlocking<Unit> {
 
+  CreateTopics.create()
+
   val wsServer = WebsocketServer()
   val tileServer = WebMapTileServer()
 

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

@@ -8,6 +8,8 @@ import dev.adamko.kafkatorio.events.schema.MapTile
 import dev.adamko.kafkatorio.events.schema.MapTilePosition
 import dev.adamko.kafkatorio.events.schema.MapTiles
 import dev.adamko.kafkatorio.processor.serdes.jsonMapper
+import dev.adamko.kafkatorio.processor.serdes.kxsBinary
+import dev.adamko.kafkatorio.processor.serdes.serde
 import dev.adamko.kotka.extensions.consumedAs
 import dev.adamko.kotka.extensions.materializedWith
 import dev.adamko.kotka.extensions.streams.filter
@@ -71,7 +73,7 @@ fun allMapTilesTable(
     }
     .toTable(
       "all-tiles-keyed-on-position",
-      materializedWith(jsonMapper.serde(), jsonMapper.serde())
+      materializedWith(kxsBinary.serde(), kxsBinary.serde())
     )
     .suppress(
       Suppressed.untilTimeLimit(

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

@@ -12,7 +12,6 @@ import dev.adamko.kafkatorio.events.schema.MapTiles
 import dev.adamko.kafkatorio.events.schema.PlayerData
 import dev.adamko.kafkatorio.events.schema.SurfaceData
 import dev.adamko.kafkatorio.events.schema.converters.toMapChunkPosition
-import dev.adamko.kafkatorio.processor.serdes.KafkatorioPacketSerde
 import dev.adamko.kafkatorio.processor.KafkatorioTopology
 import dev.adamko.kafkatorio.processor.serdes.jsonMapper
 import dev.adamko.kotka.extensions.consumedAs
@@ -56,7 +55,7 @@ fun splitFactorioServerLog(
       producedAs(
         "split-server-log",
         jsonMapper.serde(),
-        KafkatorioPacketSerde
+        jsonMapper.serde(),
       )
     ) { _: FactorioPacketKey, value: KafkatorioPacket, _: RecordContext ->
 //        println("[$key] sending event:${value.eventType} to topic:${value.data.objectName()}")

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

@@ -2,22 +2,21 @@ package dev.adamko.kafkatorio.processor.topology
 
 import dev.adamko.kafkatorio.events.schema.FactorioObjectData
 import dev.adamko.kafkatorio.events.schema.KafkatorioPacket
-import dev.adamko.kafkatorio.processor.serdes.KafkatorioPacketSerde
 import dev.adamko.kafkatorio.processor.WebsocketServer
 import dev.adamko.kafkatorio.processor.serdes.jsonMapper
+import dev.adamko.kotka.extensions.consumedAs
+import dev.adamko.kotka.kxs.serde
 import kotlinx.serialization.encodeToString
-import org.apache.kafka.common.serialization.Serdes
 import org.apache.kafka.streams.StreamsBuilder
-import org.apache.kafka.streams.kstream.Consumed
 
 
 fun playerUpdatesToWsServer(
   websocketServer: WebsocketServer,
   builder: StreamsBuilder = StreamsBuilder(),
 ) {
-  builder.stream(
+  builder.stream<FactorioPacketKey, KafkatorioPacket>(
     "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.LuaPlayer}",
-    Consumed.with(Serdes.String(), KafkatorioPacketSerde)
+    consumedAs("stream-player-updates-for-ws-server", jsonMapper.serde(), jsonMapper.serde())
   )
     .foreach { _, value ->
 //        println("sending ${value.packetType} packet to websocket")

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

@@ -4,6 +4,8 @@ import dev.adamko.kafkatorio.events.schema.FactorioPrototypes
 import dev.adamko.kafkatorio.events.schema.KafkatorioPacket
 import dev.adamko.kafkatorio.events.schema.MapTilePrototype
 import dev.adamko.kafkatorio.processor.serdes.jsonMapper
+import dev.adamko.kafkatorio.processor.serdes.kxsBinary
+import dev.adamko.kafkatorio.processor.serdes.serde
 import dev.adamko.kotka.extensions.consumedAs
 import dev.adamko.kotka.extensions.materializedWith
 import dev.adamko.kotka.extensions.streams.flatMap
@@ -22,12 +24,12 @@ value class PrototypeName(val name: String) {
 
 
 fun prototypesTable(builder: StreamsBuilder): KTable<PrototypeName, MapTilePrototype> {
-  return builder.stream(
+  return builder.stream<FactorioPacketKey, FactorioPrototypes>(
     "kafkatorio.${KafkatorioPacket.PacketType.PROTOTYPES}.all",
     consumedAs(
       "consume-all-prototypes-packets",
-      Serdes.String(),
-      jsonMapper.serde<FactorioPrototypes>()
+      jsonMapper.serde(),
+      jsonMapper.serde()
     )
   )
     .flatMap("map-MapTilePrototype") { _, prototypes: FactorioPrototypes ->
@@ -37,62 +39,8 @@ fun prototypesTable(builder: StreamsBuilder): KTable<PrototypeName, MapTileProto
     }
     .toTable(
       materializedWith(
-        jsonMapper.serde(),
-        jsonMapper.serde()
+        kxsBinary.serde(),
+        kxsBinary.serde()
       )
     )
 }
-
-
-//@Serializable
-//data class FactorioServerPrototypesRecord(
-//  val serverId: FactorioServerId,
-//  val mapTilePrototypes: Map<PrototypeName, MapTilePrototype>,
-//)
-//
-//fun serverPrototypesTable(builder: StreamsBuilder): KTable<FactorioServerId, FactorioServerPrototypesRecord> {
-//
-//  return builder.stream(
-//    "kafkatorio.${KafkatorioPacket.PacketType.PROTOTYPES}.all",
-//    consumedAs(
-//      "consume-all-prototypes-packets",
-//      Serdes.String(),
-//      jsonMapper.serde<FactorioPrototypes>()
-//    )
-//  )
-//    .map("build-FactorioServerPrototypesRecord", ::buildPrototypesRecord)
-//    .peek { _, value ->
-//      println("tile prototypes: ${value.mapTilePrototypes.keys.joinToString()}")
-//    }
-//    .groupByKey(
-//      groupedAs(
-//        "grouping-tile-prototypes-by-surface-index",
-//        jsonMapper.serde(),
-//        jsonMapper.serde()
-//      )
-//    )
-//    .reduce(
-//      "group-surface-prototypes",
-//      materializedWith(
-////          "kafkatorio.surface-prototypes",
-//        jsonMapper.serde(),
-//        jsonMapper.serde()
-//      )
-//    ) { value1: FactorioServerPrototypesRecord, value2: FactorioServerPrototypesRecord ->
-//      val allTiles = value1.mapTilePrototypes + value2.mapTilePrototypes
-//      value1.copy(mapTilePrototypes = allTiles)
-//    }
-//}
-//
-//private fun buildPrototypesRecord(
-//  serverId: String,
-//  prototypes: FactorioPrototypes
-//): Pair<FactorioServerId, FactorioServerPrototypesRecord> {
-//  val tilePrototypes = prototypes.prototypes
-//    .filterIsInstance<MapTilePrototype>()
-//    .associateBy { PrototypeName(it.name) }
-//
-//  val key = FactorioServerId(serverId)
-//  val data = FactorioServerPrototypesRecord(key, tilePrototypes)
-//  return key to data
-//}

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

@@ -5,7 +5,8 @@ import dev.adamko.kafkatorio.events.schema.MapTile
 import dev.adamko.kafkatorio.events.schema.MapTilePosition
 import dev.adamko.kafkatorio.events.schema.MapTilePrototype
 import dev.adamko.kafkatorio.events.schema.converters.toMapChunkPosition
-import dev.adamko.kafkatorio.processor.serdes.jsonMapper
+import dev.adamko.kafkatorio.processor.serdes.kxsBinary
+import dev.adamko.kafkatorio.processor.serdes.serde
 import dev.adamko.kotka.extensions.groupedAs
 import dev.adamko.kotka.extensions.materializedAs
 import dev.adamko.kotka.extensions.namedAs
@@ -13,7 +14,6 @@ import dev.adamko.kotka.extensions.tables.groupBy
 import dev.adamko.kotka.extensions.tables.join
 import dev.adamko.kotka.extensions.tables.mapValues
 import dev.adamko.kotka.extensions.toKeyValue
-import dev.adamko.kotka.kxs.serde
 import java.time.Duration
 import kotlinx.serialization.Serializable
 import org.apache.kafka.streams.kstream.KTable
@@ -60,8 +60,8 @@ fun aggregateWebMapTiles(
         "convert-tiles-to-colours",
         materializedAs(
           "store-web-map-tile-pixels",
-          jsonMapper.serde(),
-          jsonMapper.serde(),
+          kxsBinary.serde(),
+          kxsBinary.serde(),
         ),
         { PrototypeName(it.prototypeName) },
       ) { tile: MapTile, proto: MapTilePrototype ->
@@ -74,8 +74,8 @@ fun aggregateWebMapTiles(
       .groupBy(
         groupedAs(
           "group-webmap-tile-colours-into-chunks",
-          jsonMapper.serde(),
-          jsonMapper.serde(),
+          kxsBinary.serde(),
+          kxsBinary.serde(),
         )
       ) { pos: TileUpdateRecordKey, px: WebMapTilePixel ->
         val (chunkX, chunkY) = pos.tilePosition.toMapChunkPosition(WEB_MAP_IMAGE_TILE_SIZE)
@@ -100,24 +100,24 @@ fun aggregateWebMapTiles(
         namedAs("web-map-tile-colours-into-$WEB_MAP_IMAGE_TILE_SIZE-chunks"),
         materializedAs(
           "web-map-tile-colour-chunks-aggregate",
-          jsonMapper.serde(),
-          jsonMapper.serde()
+          kxsBinary.serde(),
+          kxsBinary.serde()
         )
       )
       .suppress(
         Suppressed.untilTimeLimit<WebMapTileChunkPosition>(
           Duration.ofSeconds(30),
           Suppressed.BufferConfig
-            .maxRecords(2)
+            .maxRecords(30)
 //            .withMaxBytes(31457280  ) // 30MB
-            .emitEarlyWhenFull()
+//            .emitEarlyWhenFull()
         ).withName("web-map-tile-aggregate-debounce")
       )
 
   return webMapTileChunkAggregate
     .mapValues(
       "finalise-web-map-tile-colour-chunk-aggregation",
-      materializedAs("web-map-tile-colour-chunks", jsonMapper.serde(), jsonMapper.serde())
+      materializedAs("web-map-tile-colour-chunks", kxsBinary.serde(), kxsBinary.serde())
     ) { _, v ->
       WebMapTileChunkPixels(v.pixels)
     }

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

@@ -20,10 +20,10 @@ fun saveTileImages(
 
   webMapTiles
     .suppress(
-      Suppressed.untilTimeLimit(
+      Suppressed.untilTimeLimit<WebMapTileChunkPosition>(
         Duration.ofSeconds(30),
         Suppressed.BufferConfig.maxRecords(30)
-      )
+      ).withName("save-tile-images-debounce")
     )
     .toStream("get-webmap-tile-updates")
     .foreach { position: WebMapTileChunkPosition, pixels: WebMapTileChunkPixels ->
@@ -37,13 +37,15 @@ fun saveTileImages(
     }
 }
 
+//val saveMapTilesContext = Dispatchers.IO.limitedParallelism(1)
+
 
 private fun saveMapTilesPng(
   chunkPosition: WebMapTileChunkPosition,
   pixels: Set<WebMapTilePixel>
 ) {
-  val chunkOriginX: Int = chunkPosition.x * chunkPosition.chunkSize
-  val chunkOriginY: Int = chunkPosition.y * chunkPosition.chunkSize
+  val chunkOriginX: Int = chunkPosition.x * chunkPosition.chunkSize // - 1
+  val chunkOriginY: Int = chunkPosition.y * chunkPosition.chunkSize // - 1
   val surfaceIndex: Int = chunkPosition.surfaceIndex.index
 
   val chunkImage =
@@ -76,15 +78,33 @@ private fun saveMapTilesPng(
 
   val zoom = 1u
   val file =
-    File(
-      "src/main/resources/kafkatorio-web-map/s${surfaceIndex}/z$zoom/x${chunkPosition.x}/y${chunkPosition.y}.png"
-    )
+    File(tileFilename("$surfaceIndex", "$zoom", "${chunkPosition.x}", "${chunkPosition.y}"))
 
   if (file.parentFile.mkdirs()) {
     println("created new map tile parentFile directory ${file.absolutePath}")
   }
 
   println("saving map tile $file")
-
-  chunkImage.output(PngWriter.MaxCompression, file)
+  val savedTile = chunkImage.output(PngWriter.NoCompression, file)
+  println("savedTile: $savedTile")
 }
+
+fun tileFilename(
+  surfaceIndex: String,
+  zoom: String,
+  chunkX: String,
+  chunkY: String,
+) = "src/main/resources/kafkatorio-web-map/s${surfaceIndex}/z$zoom/x${chunkX}/y${chunkY}.png"
+
+//fun subdivide(tile: ImmutableImage, size: Int) {
+//
+//  val halfSize = size / 2
+//
+//  val bl = tile.resizeTo(halfSize, halfSize, Position.BottomLeft)
+//  val savedTile = bl.output(PngWriter.NoCompression, file)
+//
+//  tile.resizeTo(halfSize, halfSize, Position.BottomRight)
+//  tile.resizeTo(halfSize, halfSize, Position.TopLeft)
+//  tile.resizeTo(halfSize, halfSize, Position.TopRight)
+//
+//}