- tidy up groupMapChunks.kt

- add ticks to all Kafkatorio Packets - add Kafka Streams test for groupMapChunks.kt
2022-03-12 16:42:24 +01:00 · 2022-03-12 16:42:24 +01:00 · b8ba8b7c9e
commit b8ba8b7c9e
parent bfdee32d61
22 changed files with 271 additions and 98 deletions
--- a/gradle/libs.versions.toml
+++ b/gradle/libs.versions.toml
@ -91,8 +91,9 @@ gcp-speech = { group = "com.google.cloud",  name = "google-cloud-speech"

 ## kafka ##
 kafka-streams          = { group = "org.apache.kafka", name = "kafka-streams",            version.ref = "kafka" }
+kafka-streamsTestUtils = { group = "org.apache.kafka", name = "kafka-streams-test-utils", version.ref = "kafka" }
 # kafka helpers
-kotka         = { group = "com.github.blueanvil", name = "kotka",         version =     "1.1.1" }
+#kotka         = { group = "com.github.blueanvil", name = "kotka",         version =     "1.1.1" }

 jsoup = { group = "org.jsoup", name = "jsoup", version = "1.14.3" }
 skrapeit = { group = "it.skrape", name = "skrapeit", version = "1.2.0" }
--- a/modules/events-mod/src/main/typescript/generated/kafkatorio-schema/config-update.d.ts
+++ b/modules/events-mod/src/main/typescript/generated/kafkatorio-schema/config-update.d.ts
@ -16,4 +16,5 @@ interface FactorioConfigurationUpdate extends KafkatorioPacket {
  factorioData: FactorioGameDataUpdate;
  modVersion: string;
  packetType: PacketType;
+  tick: uint;
 }
--- a/modules/events-mod/src/main/typescript/generated/kafkatorio-schema/prototype.d.ts
+++ b/modules/events-mod/src/main/typescript/generated/kafkatorio-schema/prototype.d.ts
@ -10,6 +10,7 @@ interface FactorioPrototypes extends KafkatorioPacket {
  modVersion: string;
  packetType: PacketType;
  prototypes: FactorioPrototype[];
+  tick: uint;
 }

 interface MapTilePrototype extends FactorioPrototype {
--- a/modules/events-mod/src/main/typescript/generated/kafkatorio-schema/schema.d.ts
+++ b/modules/events-mod/src/main/typescript/generated/kafkatorio-schema/schema.d.ts
@ -5,6 +5,7 @@ type PacketType = "EVENT" | "CONFIG" | "PROTOTYPES" | "UPDATE";
 interface KafkatorioPacket {
  modVersion: string;
  packetType: PacketType;
+  tick: uint;
 }

 interface FactorioEvent extends KafkatorioPacket {
--- a/modules/events-mod/src/main/typescript/main/config-update/handleConfigurationUpdate.ts
+++ b/modules/events-mod/src/main/typescript/main/config-update/handleConfigurationUpdate.ts
@ -6,7 +6,8 @@ export function emitConfigurationUpdate(changeData: ConfigurationChangedData) {
    modVersion: global.MOD_VERSION,
    packetType: "CONFIG",
    allMods: allMods(changeData),
-    factorioData: factorioData(changeData)
+    factorioData: factorioData(changeData),
+    tick: game.tick,
  })

 }
--- a/modules/events-mod/src/main/typescript/main/config-update/prototypes.ts
+++ b/modules/events-mod/src/main/typescript/main/config-update/prototypes.ts
@ -6,7 +6,8 @@ export function emitPrototypes() {
  emitPacket<FactorioPrototypes>({
    modVersion: global.MOD_VERSION,
    packetType: "PROTOTYPES",
-    prototypes: prototypes()
+    prototypes: prototypes(),
+    tick: game.tick,
  })
 }

--- a/modules/events-mod/src/main/typescript/main/events/emitEvent.ts
+++ b/modules/events-mod/src/main/typescript/main/events/emitEvent.ts
@ -7,6 +7,6 @@ export function emitEvent<T extends FactorioObjectData>(eventData: T, tick: uint
    packetType: "EVENT",
    eventType: eventType,
    modVersion: global.MOD_VERSION,
-    tick: tick
+    tick: tick,
  })
 }
--- a/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioCommonData.kt
+++ b/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioCommonData.kt
@ -112,8 +112,9 @@ interface EntityIdentifiers {
 data class MapTileDictionary(
  /** Map an X,Y coordinate a prototype name */
  val tilesXY: Map<String, Map<String, PrototypeKey>>,
-  val protos: Map<PrototypeKey, PrototypeName>
+  val protos: Map<PrototypeKey, PrototypeName>,
 ) {
+
  fun toMapTileList(): List<MapTile> = buildList {
    tilesXY.forEach { (xString, row) ->
      row.forEach { (yString, protoIndex) ->
--- a/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioConfiguration.kt
+++ b/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioConfiguration.kt
@ -7,6 +7,7 @@ import kotlinx.serialization.Serializable
@Serializable
 data class FactorioConfigurationUpdate(
  override val modVersion: String,
+  override val tick: Tick,
  val factorioData: FactorioGameDataUpdate,
  val allMods: List<FactorioModInfo>,
 ) : KafkatorioPacket() {
--- a/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioEventUpdates.kt
+++ b/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioEventUpdates.kt
@ -15,8 +15,7 @@ import kotlinx.serialization.json.jsonPrimitive
@Serializable
 data class FactorioEventUpdatePacket(
  override val modVersion: String,
-  /** game time */
-  val tick: Tick,
+  override val tick: Tick,
  val update: FactorioEventUpdate,
 ) : KafkatorioPacket() {
  @EncodeDefault
--- a/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioEvents.kt
+++ b/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioEvents.kt
@ -7,10 +7,9 @@ import kotlinx.serialization.Serializable
@Serializable
 data class FactorioEvent(
  override val modVersion: String,
+  override val tick: Tick,
  /** the initial Factorio event (`defines.events`) trigger */
  val eventType: String,
-  /** game time */
-  val tick: Tick,
  val data: FactorioObjectData,
 ) : KafkatorioPacket() {
  @EncodeDefault
--- a/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioPrototypes.kt
+++ b/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioPrototypes.kt
@ -15,6 +15,7 @@ import kotlinx.serialization.json.jsonPrimitive
@Serializable
 data class FactorioPrototypes(
  override val modVersion: String,
+  override val tick: Tick,
  val prototypes: List<FactorioPrototype>,
 ) : KafkatorioPacket() {
  @EncodeDefault
--- a/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/kafkatorioPacket.kt
+++ b/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/kafkatorioPacket.kt
@ -14,6 +14,7 @@ import kotlinx.serialization.json.jsonPrimitive
 sealed class KafkatorioPacket {
  /** Schema versioning */
  abstract val modVersion: String
+  abstract val tick: Tick
  @EncodeDefault
  abstract val packetType: PacketType

--- a/modules/events-schema/src/commonTest/kotlin/dev/adamko/kafkatorio/events/schema/FactorioEventTest.kt
+++ b/modules/events-schema/src/commonTest/kotlin/dev/adamko/kafkatorio/events/schema/FactorioEventTest.kt
@ -289,6 +289,7 @@ class FactorioEventTest : FunSpec({
               {
                 "modVersion": "0.2.8",
                 "packetType": "CONFIG",
+                 "tick": 534,
                 "allMods": [
                   {
                     "modName": "kafkatorio-events",
@ -310,6 +311,7 @@ class FactorioEventTest : FunSpec({

        val expected = FactorioConfigurationUpdate(
          modVersion = "0.2.8",
+          tick = Tick(534u),
          factorioData = FactorioGameDataUpdate(null, null),
          allMods = listOf(
            FactorioModInfo(
--- a/modules/events-schema/src/commonTest/kotlin/dev/adamko/kafkatorio/events/schema/PrototypesTest.kt
+++ b/modules/events-schema/src/commonTest/kotlin/dev/adamko/kafkatorio/events/schema/PrototypesTest.kt
@ -12,6 +12,7 @@ class PrototypesTest : BehaviorSpec({
      {
         "modVersion":"0.2.4",
         "packetType":"PROTOTYPES",
+         "tick": 534,
         "prototypes":[ 
            {
               "prototypeObjectName":"LuaTilePrototype",
@ -46,6 +47,7 @@ class PrototypesTest : BehaviorSpec({
      {
         "modVersion":"0.3.2",
         "packetType":"PROTOTYPES",
+         "tick": 534,
         "prototypes":[
            {
               "prototypeObjectName":"LuaTilePrototype",
--- a/modules/events-server/build.gradle.kts
+++ b/modules/events-server/build.gradle.kts
@ -20,6 +20,7 @@ val kafkaStateDir = layout.projectDirectory.dir("kafka-state")
 dependencies {

  implementation(libs.kafka.streams)
+  testImplementation(libs.kafka.streamsTestUtils)

  implementation(projects.modules.eventsSchema)

@ -40,7 +41,7 @@ dependencies {

  implementation("com.sksamuel.scrimage:scrimage-core:4.0.24")

-  val kotkaVersion = "v0.0.14"
+  val kotkaVersion = "v0.0.15"
  implementation("com.github.adamko-dev.kotka-streams:kotka-streams:$kotkaVersion") {
    isChanging = true
  }
--- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/KafkatorioTopology.kt
+++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/KafkatorioTopology.kt
@ -1,30 +1,20 @@
 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
 import dev.adamko.kafkatorio.processor.config.ApplicationProperties
-import dev.adamko.kafkatorio.processor.serdes.jsonMapper
 import dev.adamko.kafkatorio.processor.serdes.kxsBinary
-import dev.adamko.kafkatorio.processor.topology.FactorioServerId
 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.groupTilesIntoChunksWithColours
+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.kafkatorio.processor.topology.tileProtoColourDictionary
 import dev.adamko.kotka.extensions.consumedAs
-import dev.adamko.kotka.extensions.producedAs
-import dev.adamko.kotka.extensions.tables.toStream
 import dev.adamko.kotka.kxs.serde
 import java.time.Duration
 import kotlin.coroutines.CoroutineContext
+import kotlin.coroutines.cancellation.CancellationException
 import kotlinx.coroutines.CoroutineName
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
@ -36,7 +26,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.KStream
 import org.apache.kafka.streams.kstream.KTable


@ -62,54 +51,7 @@ internal class KafkatorioTopology(

  private fun groupTilesMapChunks() {
    val builder = StreamsBuilder()
-
-
-    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 mapChunksUpdatesStream: KStream<FactorioServerId, FactorioEventUpdatePacket> =
-      builder.stream(
-        "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,
-        mapChunksUpdatesStream,
-        tileProtoColourDict,
-      )
-
-    groupedMapChunkTiles
-      .toStream("stream-grouped-map-tiles")
-      .to(
-        TOPIC_GROUPED_MAP_CHUNKS,
-        producedAs(
-          "produce.grouped-map-chunks",
-          kxsBinary.serde<ServerMapChunkId?>(),
-          kxsBinary.serde<ServerMapChunkTiles<ColourHex>?>()
-        )
-      )
-
-    val topology = builder.build()
+    val topology = groupMapChunks(builder)
    launchTopology("groupTilesMapChunks", topology)
  }

--- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes/kxsBinary.kt
+++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes/kxsBinary.kt
@ -60,12 +60,14 @@ class KxsDataOutputEncoder(
  override fun encodeLong(value: Long) = output.writeLong(value)
  override fun encodeShort(value: Short) = output.writeShort(value.toInt())
  override fun encodeString(value: String) = output.writeUTF(value)
+//  override fun encodeString(value: String) = encodeByteArray(value.toByteArray(Charsets.UTF_8))

  override fun beginCollection(
    descriptor: SerialDescriptor,
    collectionSize: Int
  ): CompositeEncoder {
-    encodeCompactSize(collectionSize)
+    encodeInt(collectionSize)
+//    encodeCompactSize(collectionSize)
    return this
  }

@ -113,6 +115,7 @@ class KxsDataInputDecoder(
  override fun decodeLong(): Long = input.readLong()
  override fun decodeShort(): Short = input.readShort()
  override fun decodeString(): String = input.readUTF()
+//  override fun decodeString(): String = decodeByteArray().toString(Charsets.UTF_8)

  override fun decodeElementIndex(descriptor: SerialDescriptor): Int {
    if (elementIndex == elementsCount) return CompositeDecoder.DECODE_DONE
@ -125,7 +128,8 @@ class KxsDataInputDecoder(
  override fun decodeSequentially(): Boolean = true

  override fun decodeCollectionSize(descriptor: SerialDescriptor): Int =
-    decodeCompactSize().also { elementsCount = it }
+    decodeInt().also { elementsCount = it }
+//    decodeCompactSize().also { elementsCount = it }

  override fun decodeNotNullMark(): Boolean = decodeBoolean()

--- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/flowMapTiles.kt
+++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/flowMapTiles.kt
@ -9,6 +9,7 @@ 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 java.awt.image.BufferedImage
 import java.io.File
 import kotlin.coroutines.CoroutineContext
@ -52,7 +53,7 @@ fun saveMapTiles(

  serverMapTable
    .toStream()
-    .foreach { _, value ->
+    .forEach("save-chunked-tiles") { _, value ->
      if (value != null) {
        runBlocking(Dispatchers.Default) {
          println("emitting chunkedTiles ${value.chunkId} tiles: ${value.map.size}")
--- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/groupMapChunks.kt
+++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/groupMapChunks.kt
@ -1,7 +1,10 @@
 package dev.adamko.kafkatorio.processor.topology

 import dev.adamko.kafkatorio.events.schema.ColourHex
+import dev.adamko.kafkatorio.events.schema.FactorioEventUpdate
 import dev.adamko.kafkatorio.events.schema.FactorioEventUpdatePacket
+import dev.adamko.kafkatorio.events.schema.FactorioPrototypes
+import dev.adamko.kafkatorio.events.schema.KafkatorioPacket
 import dev.adamko.kafkatorio.events.schema.MapChunkPosition
 import dev.adamko.kafkatorio.events.schema.MapChunkUpdate
 import dev.adamko.kafkatorio.events.schema.MapTile
@ -10,16 +13,20 @@ import dev.adamko.kafkatorio.events.schema.MapTilePosition
 import dev.adamko.kafkatorio.events.schema.MapTiles
 import dev.adamko.kafkatorio.events.schema.SurfaceIndex
 import dev.adamko.kafkatorio.events.schema.converters.toMapChunkPosition
+import dev.adamko.kafkatorio.processor.KafkatorioTopology
+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.materializedAs
 import dev.adamko.kotka.extensions.materializedWith
-import dev.adamko.kotka.extensions.streams.flatMap
-import dev.adamko.kotka.extensions.streams.mapValues
-import dev.adamko.kotka.extensions.streams.reduce
+import dev.adamko.kotka.extensions.producedAs
+import dev.adamko.kotka.extensions.streams.*
 import dev.adamko.kotka.extensions.tables.join
+import dev.adamko.kotka.extensions.tables.toStream
 import dev.adamko.kotka.kxs.serde
 import kotlinx.serialization.Serializable
+import org.apache.kafka.streams.StreamsBuilder
+import org.apache.kafka.streams.Topology
 import org.apache.kafka.streams.kstream.KStream
 import org.apache.kafka.streams.kstream.KTable

@ -42,7 +49,74 @@ data class ServerMapChunkTiles<Data>(
 }


-fun groupTilesIntoChunksWithColours(
+fun groupMapChunks(
+  builder: StreamsBuilder,
+  prototypesTopicName: String = "kafkatorio.${KafkatorioPacket.PacketType.PROTOTYPES}.all",
+  mapChunkUpdatesTopicName: String = "kafkatorio.${KafkatorioPacket.PacketType.UPDATE}.${FactorioEventUpdate.FactorioEventUpdateType.MAP_CHUNK}",
+): Topology {
+
+  val protosStream: KStream<FactorioServerId, FactorioPrototypes> =
+    builder.stream(
+      prototypesTopicName,
+      consumedAs("consume.factorio-protos.all", jsonMapper.serde(), jsonMapper.serde())
+    )
+  val tileProtoColourDict: KTable<FactorioServerId, TileColourDict> =
+    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 mapChunksUpdatesStream: KStream<FactorioServerId, FactorioEventUpdatePacket> =
+    builder.stream(
+      mapChunkUpdatesTopicName,
+      consumedAs(
+        "consume.map-chunk-updates",
+        jsonMapper.serde<FactorioServerId>(),
+        jsonMapper.serde<FactorioEventUpdatePacket>(),
+      )
+    )
+//        .repartition(
+//        repartitionedAs(
+//          "map-chunk-updates.repartition.change-serde",
+//          kxsBinary.serde<FactorioServerId>(),
+//          kxsBinary.serde<FactorioEventUpdatePacket>(),
+//          numberOfPartitions = 3,
+//        )
+//      )
+
+  val groupedMapChunkTiles: KTable<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> =
+    groupTilesIntoChunksWithColours(
+//        mapTilesStream,
+//        mapChunksStream,
+      mapChunksUpdatesStream,
+      tileProtoColourDict,
+    )
+
+  groupedMapChunkTiles
+    .toStream("stream-grouped-map-tiles")
+    .to(
+      KafkatorioTopology.TOPIC_GROUPED_MAP_CHUNKS,
+      producedAs(
+        "produce.grouped-map-chunks",
+        kxsBinary.serde<ServerMapChunkId?>(),
+        kxsBinary.serde<ServerMapChunkTiles<ColourHex>?>()
+      )
+    )
+
+  return builder.build()
+}
+
+private fun groupTilesIntoChunksWithColours(
 //  mapTilePacketsStream: KStream<FactorioServerId, FactorioEvent>,
 //  mapChunksPacketsStream: KStream<FactorioServerId, FactorioEvent>,
  mapChunkUpdatePacketsStream: KStream<FactorioServerId, FactorioEventUpdatePacket>,
@ -107,7 +181,7 @@ fun groupTilesIntoChunksWithColours(
 //      .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 ->
+      .flatMap("server-map-data.tiles.flatMapByChunk") { serverId: FactorioServerId, mapTiles: MapTiles ->

        val standardChunkSize = ChunkSize.MAX

@ -119,10 +193,10 @@ fun groupTilesIntoChunksWithColours(
          .map { (chunkPos, tiles) ->

            val chunkId = ServerMapChunkId(
-              key,
-              chunkPos,
-              mapTiles.surfaceIndex,
-              standardChunkSize,
+              serverId = serverId,
+              chunkPosition = chunkPos,
+              surfaceIndex = mapTiles.surfaceIndex,
+              chunkSize = standardChunkSize,
            )

            val tilesMap = tiles.associate { it.position to TileProtoHashCode(it) }
@ -132,20 +206,24 @@ 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<ServerMapChunkId>(),
-          kxsBinary.serde<ServerMapChunkTiles<TileProtoHashCode>>()
+          kxsBinary.serde<ServerMapChunkId?>(),
+          kxsBinary.serde<ServerMapChunkTiles<TileProtoHashCode>?>(),
        )
      )
      .reduce(
-        "server-map-data.tiles.reduce",
-        materializedAs("server-map-data.tiles.store", kxsBinary.serde(), kxsBinary.serde()),
-      ) { a, b -> a + b }
+        "server-map-data.tiles.reduce", materializedWith(
+          //          "server-map-data.tiles.store",
+          kxsBinary.serde<ServerMapChunkId>(),
+          kxsBinary.serde<ServerMapChunkTiles<TileProtoHashCode>>(),
+        ),
+        ServerMapChunkTiles<TileProtoHashCode>::plus
+      )


  val colourisedChunkTable: KTable<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> =
@ -153,7 +231,10 @@ fun groupTilesIntoChunksWithColours(
      .join(
        other = tileProtoColourDict,
        name = "server-map-data.join-tiles-with-prototypes",
-        materialized = materializedWith(kxsBinary.serde(), kxsBinary.serde()),
+        materialized = materializedWith(
+          kxsBinary.serde<ServerMapChunkId>(),
+          kxsBinary.serde<ServerMapChunkTiles<ColourHex>>(),
+        ),
        foreignKeyExtractor = { chunkTiles: ServerMapChunkTiles<TileProtoHashCode> -> chunkTiles.chunkId.serverId }
      ) { tiles: ServerMapChunkTiles<TileProtoHashCode>, colourDict: TileColourDict ->

--- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/prototypes.kt
+++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/prototypes.kt
@ -7,7 +7,9 @@ import dev.adamko.kafkatorio.events.schema.MapTile
 import dev.adamko.kafkatorio.events.schema.MapTilePrototype
 import dev.adamko.kafkatorio.events.schema.converters.toHex
 import dev.adamko.kafkatorio.processor.serdes.jsonMapper
+import dev.adamko.kafkatorio.processor.serdes.kxsBinary
 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.mapValues
 import dev.adamko.kotka.extensions.streams.toTable
@ -54,10 +56,10 @@ fun tileProtoColourDictionary(
    }
    .toTable(
      "server-map-data.tile-prototypes",
-      materializedAs(
-        "server-map-data.tile-prototypes.store",
-        jsonMapper.serde(),
-        jsonMapper.serde(),
+      materializedWith(
+//        "server-map-data.tile-prototypes.store",
+        kxsBinary.serde(),
+        kxsBinary.serde(),
      )
    )

--- a/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/topology/GroupMapChunksTest.kt
+++ b/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/topology/GroupMapChunksTest.kt
@ -0,0 +1,130 @@
+package dev.adamko.kafkatorio.processor.topology
+
+import dev.adamko.kafkatorio.events.schema.Colour
+import dev.adamko.kafkatorio.events.schema.ColourHex
+import dev.adamko.kafkatorio.events.schema.FactorioEventUpdatePacket
+import dev.adamko.kafkatorio.events.schema.FactorioPrototypes
+import dev.adamko.kafkatorio.events.schema.MapChunkPosition
+import dev.adamko.kafkatorio.events.schema.MapChunkUpdate
+import dev.adamko.kafkatorio.events.schema.MapTileDictionary
+import dev.adamko.kafkatorio.events.schema.MapTilePosition
+import dev.adamko.kafkatorio.events.schema.MapTilePrototype
+import dev.adamko.kafkatorio.events.schema.PrototypeKey
+import dev.adamko.kafkatorio.events.schema.PrototypeName
+import dev.adamko.kafkatorio.events.schema.SurfaceIndex
+import dev.adamko.kafkatorio.events.schema.Tick
+import dev.adamko.kafkatorio.processor.KafkatorioTopology
+import dev.adamko.kafkatorio.processor.serdes.jsonMapper
+import dev.adamko.kafkatorio.processor.serdes.kxsBinary
+import dev.adamko.kotka.kxs.serde
+import io.kotest.assertions.withClue
+import io.kotest.core.spec.style.FunSpec
+import io.kotest.matchers.collections.shouldHaveSize
+import io.kotest.matchers.equality.shouldBeEqualToComparingFields
+import io.kotest.matchers.nulls.shouldNotBeNull
+import io.kotest.matchers.should
+import org.apache.kafka.streams.StreamsBuilder
+import org.apache.kafka.streams.TestInputTopic
+import org.apache.kafka.streams.TestOutputTopic
+import org.apache.kafka.streams.TopologyTestDriver
+
+
+class GroupMapChunksTest : FunSpec({
+
+  test("basic test") {
+
+    val prototypesTopicName = "prototypes-topic"
+    val mapChunkUpdatesTopicName = "map-chunk-updates-topic"
+
+    val topology = groupMapChunks(
+      StreamsBuilder(),
+      prototypesTopicName,
+      mapChunkUpdatesTopicName,
+    )
+
+    val testDriver = TopologyTestDriver(topology)
+
+    val prototypesTestInputTopic: TestInputTopic<FactorioServerId, FactorioPrototypes> =
+      testDriver.createInputTopic(
+        prototypesTopicName,
+        jsonMapper.serde<FactorioServerId>().serializer(),
+        jsonMapper.serde<FactorioPrototypes>().serializer(),
+      )
+
+    val serverId = FactorioServerId("test-server-id")
+
+    prototypesTestInputTopic.pipeInput(
+      serverId,
+      FactorioPrototypes(
+        modVersion = "1.2.3",
+        tick = Tick(456u),
+        listOf(
+          MapTilePrototype(
+            PrototypeName("fake-proto"),
+            layer = 1u,
+            mapColour = Colour(0.3f, 0.2f, 0.1f, 1.0f),
+            collisionMasks = listOf("fake-mask"),
+            order = "fake-order",
+            canBeMined = false,
+          )
+        )
+      )
+    )
+
+    val mapChunkUpdatesInputTopic: TestInputTopic<FactorioServerId, FactorioEventUpdatePacket> =
+      testDriver.createInputTopic(
+        mapChunkUpdatesTopicName,
+        jsonMapper.serde<FactorioServerId>().serializer(),
+        jsonMapper.serde<FactorioEventUpdatePacket>().serializer(),
+      )
+    mapChunkUpdatesInputTopic.pipeInput(
+      serverId,
+      FactorioEventUpdatePacket(
+        modVersion = "1.2.3",
+        tick = Tick(854u),
+        update = MapChunkUpdate(
+          MapChunkPosition(11, 22),
+          SurfaceIndex(1u),
+          tileDictionary = MapTileDictionary(
+            tilesXY = mapOf(
+              "1" to mapOf("2" to PrototypeKey("33"))
+            ),
+            protos = mapOf(PrototypeKey("33") to PrototypeName("fake-proto"))
+          )
+        )
+      )
+    )
+
+    val outputTopic: TestOutputTopic<ServerMapChunkId?, ServerMapChunkTiles<ColourHex>?> =
+      testDriver
+        .createOutputTopic(
+          KafkatorioTopology.TOPIC_GROUPED_MAP_CHUNKS,
+          kxsBinary.serde<ServerMapChunkId?>().deserializer(),
+          kxsBinary.serde<ServerMapChunkTiles<ColourHex>?>().deserializer(),
+        )
+
+    val records = outputTopic.readRecordsToList()
+
+    withClue(records) {
+      records shouldHaveSize 1
+
+      records.map { it.key to it.value }.first() should { (key, value) ->
+        key.shouldNotBeNull()
+        key shouldBeEqualToComparingFields ServerMapChunkId(
+          serverId = serverId,
+          chunkPosition = MapChunkPosition(0, 0),
+          surfaceIndex = SurfaceIndex(1u),
+          chunkSize = ChunkSize.CHUNK_512,
+        )
+        value.shouldNotBeNull()
+        value shouldBeEqualToComparingFields ServerMapChunkTiles(
+          chunkId = key,
+          map = mapOf(
+            MapTilePosition(1, 2) to ColourHex(76u, 51u, 25u, 255u)
+          )
+        )
+      }
+    }
+  }
+
+})