tiles are visible! the tileserver works! they're the right way up! they have colours!

README.md

@@ -11,6 +11,11 @@
 While this project is under development, [no license](https://choosealicense.com/no-permission/)
 is provided.
 
+## cmds
+
+`./bin/kafka-streams-application-reset.sh --application-id kafkatorio-events-processor --input-topics factorio-server-log`
+
+
 ## todo
 
 * add settings

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

@@ -22,6 +22,7 @@ import dev.adamko.kafkatorio.events.schema.converters.toMapChunkPosition
 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.producedAs
 import dev.adamko.kotka.extensions.streams.filter
 import dev.adamko.kotka.extensions.streams.flatMap
@@ -29,6 +30,8 @@ import dev.adamko.kotka.extensions.streams.map
 import dev.adamko.kotka.extensions.streams.mapValues
 import dev.adamko.kotka.extensions.streams.merge
 import dev.adamko.kotka.extensions.streams.to
+import dev.adamko.kotka.extensions.streams.toTable
+import dev.adamko.kotka.extensions.tables.join
 import dev.adamko.kotka.kxs.serde
 import dev.adamko.kotka.topicdata.TopicRecord
 import dev.adamko.kotka.topicdata.flatMapTopicRecords
@@ -73,17 +76,17 @@ class KafkatorioTopology(
     sendPlayerUpdatesToWebSocket()
 //    createMapTileImages()
     val tilePrototypesTable: KTable<String, MapTilePrototype> = tilePrototypesTable()
-    mapChunksStream()
+    mapChunksStream(tilePrototypesTable)
 
     val topology = builder.build()
 
     val streams = KafkaStreams(topology, appProps.kafkaConfig)
+//    streams.cleanUp()
 
     streams.setUncaughtExceptionHandler(StreamsExceptionHandler())
 
     Runtime.getRuntime().addShutdownHook(Thread { streams.close(Duration.ofSeconds(10)) })
 
-//  streams.cleanUp()
 
     val description: TopologyDescription = topology.describe()
     println(description)
@@ -121,11 +124,11 @@ class KafkatorioTopology(
       ) { _: String, value: KafkatorioPacket, _: RecordContext ->
 //        println("[$key] sending event:${value.eventType} to topic:${value.data.objectName()}")
         when (value) {
-          is FactorioEvent ->
+          is FactorioEvent               ->
             "kafkatorio.${value.packetType.name}.${value.data.objectName.name}"
           is FactorioConfigurationUpdate ->
             "kafkatorio.${value.packetType.name}.FactorioConfigurationUpdate"
-          is FactorioPrototypes ->
+          is FactorioPrototypes          ->
             "kafkatorio.${value.packetType.name}.all"
         }
       }
@@ -167,7 +170,6 @@ class KafkatorioTopology(
           jsonMapper.serde()
         )
       )
-
 //      .to(
 //        tilePrototypesStreamName,
 //      )
@@ -187,6 +189,12 @@ class KafkatorioTopology(
 //    )
   }
 
+  @Serializable
+  data class MapTileWithPrototype(
+    val position: MapTilePosition,
+    val prototype: MapTilePrototype,
+  )
+
   @Serializable
   data class MapChunkDataPosition(
     val position: MapChunkPosition,
@@ -197,24 +205,10 @@ class KafkatorioTopology(
   data class MapChunkData(
     val chunkPosition: MapChunkDataPosition,
     val tiles: Set<MapTile>,
+    val tilePrototypes: Set<MapTilePrototype>,
   )
-//    : TopicRecord<MapChunkDataPosition>
-  {
 
-    //    @kotlinx.serialization.Transient
+  private fun mapChunksStream(tilePrototypesTable: KTable<String, MapTilePrototype>) {
-//    override val topicKey by ::chunkPosition
-//    @kotlinx.serialization.Transient
-//    val surfaceIndex: Int by chunkPosition::surfaceIndex
-
-//    @kotlinx.serialization.Transient
-//    val leftTopTile = chunkPosition.position.leftTopTile
-//    @kotlinx.serialization.Transient
-//    val rightBottom = chunkPosition.position.rightBottomTile
-//    @kotlinx.serialization.Transient
-//    val leftBottom = MapTilePosition(leftTopTile.x, rightBottom.x)
-  }
-
-  private fun mapChunksStream() {
 
     val luaTilesUpdatesStream: KStream<String, MapTiles> = builder.stream(
       "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.LuaTiles}",
@@ -253,10 +247,10 @@ class KafkatorioTopology(
     data class TileUpdateRecord(
       val surfaceIndex: Int,
       val tilePosition: MapTilePosition,
-      val tile: MapTile
+      val tile: MapTile,
+      val tilePrototype: MapTilePrototype?,
     ) : TopicRecord<TileUpdateRecordKey> {
-      override val topicKey: TileUpdateRecordKey =
+      override val topicKey: TileUpdateRecordKey = TileUpdateRecordKey(surfaceIndex, tilePosition)
-        TileUpdateRecordKey(surfaceIndex, tilePosition)
     }
 
     val allTileUpdatesStream: KStream<TileUpdateRecordKey, TileUpdateRecord> =
@@ -268,10 +262,27 @@ class KafkatorioTopology(
             TileUpdateRecord(
               mapTiles.surfaceIndex,
               tile.position,
-              tile
+              tile,
+              null
             )
           }
         }
+        .toTable(
+          "all-tile-updates-table",
+          materializedWith(jsonMapper.serde(), jsonMapper.serde())
+        )
+        .join(
+          tilePrototypesTable,
+          "enrich-tiles-with-prototypes",
+          materializedWith(jsonMapper.serde(), jsonMapper.serde()),
+          { it.tile.prototypeName },
+        ) { v1: TileUpdateRecord, v2: MapTilePrototype ->
+          v1.copy(tilePrototype = v2)
+        }
+        .toStream()
+        .filter { _, value ->
+          value.tilePrototype != null
+        }
 
     val chunksTable: KTable<MapChunkDataPosition, MapChunkData> =
       allTileUpdatesStream
@@ -280,19 +291,26 @@ class KafkatorioTopology(
             value.tilePosition.toMapChunkPosition(),
             value.surfaceIndex,
           )
-          key to MapChunkData(key, setOf(value.tile))
+          key to MapChunkData(
+            key,
+            setOf(value.tile),
+            setOfNotNull(value.tilePrototype)
+          )
         }
         // group all tiles by chunk position
         .groupByKey(
           groupedAs("group-tile-updates-per-chunk-position", jsonMapper.serde(), jsonMapper.serde())
         )
         .reduce { tileA, tileB ->
-          tileA.copy(tiles = tileA.tiles + tileB.tiles)
+          tileA.copy(
+            tiles = tileA.tiles + tileB.tiles,
+            tilePrototypes = tileA.tilePrototypes + tileB.tilePrototypes
+          )
         }
         .suppress(
           Suppressed.untilTimeLimit(
             Duration.ofSeconds(30),
-            Suppressed.BufferConfig.maxRecords(30)
+            Suppressed.BufferConfig.maxRecords(32*32*2)
           )
         )
 
@@ -309,11 +327,15 @@ class KafkatorioTopology(
       }
   }
 
-//  @Suppress("KotlinConstantConditions")
+
   private fun saveMapTilesPng(key: MapChunkDataPosition, chunk: MapChunkData) {
 
+    val tilePrototypes: Map<String, MapTilePrototype?> =
+      chunk.tilePrototypes.associateBy { it.name }.withDefault { null }
+
     val chunkOriginX = chunk.chunkPosition.position.leftTopTile.x
-    val chunkOriginY = chunk.chunkPosition.position.rightBottomTile.y
+//    val chunkOriginY = chunk.chunkPosition.position.rightBottomTile.y
+    val chunkOriginY = chunk.chunkPosition.position.leftTopTile.y
 
     val chunkImage =
       ImmutableImage.filled(
@@ -325,9 +347,8 @@ class KafkatorioTopology(
 
     chunk.tiles.forEach { tile ->
 
-      val prototypeColour = tilePrototypesStore
+      val prototypeColour = tilePrototypes
-        .get(tile.prototypeName)
+        .getValue(tile.prototypeName)
-        ?.value()
         ?.mapColour
         ?.toHexadecimal()
 

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

@@ -1,6 +1,8 @@
 package dev.adamko.kafkatorio.processor.tileserver
 
 import org.http4k.core.HttpHandler
+import org.http4k.core.then
+import org.http4k.filter.CachingFilters
 import org.http4k.routing.ResourceLoader
 import org.http4k.routing.RoutingHttpHandler
 import org.http4k.routing.bind
@@ -10,7 +12,14 @@ import org.http4k.routing.static
 class WebMapTileServer {
 
   private val routes: RoutingHttpHandler = routes(
-    "/tiles" bind static(ResourceLoader.Classpath("/kafkatorio-web-map"))
+    "/tiles" bind CachingFilters.Response.NoCache()
+      .then(
+        static(
+          ResourceLoader.Directory(
+            """D:\Users\Adam\Projects\games\kafkatorio\modules\events-server\src\main\resources\kafkatorio-web-map\"""
+          )
+        )
+      )
   )
 
   fun build(): HttpHandler {

modules/web-map/src/main/kotlin/dev/adamko/kafkatorio/webmap/map.kt

@@ -30,11 +30,13 @@ fun createFactorioMap() = Maps {
 
 //    DefaultTileLayers.OpenStreetMap.addTo(this)
     L.tileLayer(
-      """http://localhost:9073/tiles/s1/z1/x{x}/y{y}.png"""
+      """http://localhost:9073/tiles/s1/z{z}/x{x}/y{y}.png"""
     ) {
       attribution = "kafkatorio"
       tms = true
       minZoom = 1
+//      maxZoom = 1
+      maxNativeZoom = 1
     }.addTo(this)
 
     with(options) {