trigger tile png reload in browser by sending 'saved tile' events over websocket

- create new EventServerPacket for websocket packet data
- move common models to events-schema

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

@@ -3,7 +3,7 @@ package dev.adamko.kafkatorio.processor
 import dev.adamko.kafkatorio.processor.config.ApplicationProperties
 import dev.adamko.kafkatorio.processor.topology.colourMapChunks
 import dev.adamko.kafkatorio.processor.topology.factorioServerPacketStream
-import dev.adamko.kafkatorio.processor.topology.playerUpdatesToWsServer
+import dev.adamko.kafkatorio.processor.topology.broadcastToWebsocket
 import dev.adamko.kafkatorio.processor.topology.saveMapTiles
 import dev.adamko.kafkatorio.processor.topology.splitFactorioServerPacketStream
 import java.time.Duration
@@ -39,7 +39,7 @@ internal class KafkatorioTopology(
     groupTilesMapChunks()
     saveTiles()
 
-    playerUpdates()
+    websocketBroadcaster()
   }
 
   private fun splitPackets() {
@@ -64,13 +64,13 @@ internal class KafkatorioTopology(
     launchTopology("saveTiles", topology)
   }
 
-  private fun playerUpdates() {
+  private fun websocketBroadcaster() {
     val builder = StreamsBuilder()
-    playerUpdatesToWsServer(websocketServer, builder)
+    broadcastToWebsocket(websocketServer, builder)
 
     val topology = builder.build()
 
-    launchTopology("playerUpdates", topology)
+    launchTopology("websocketBroadcaster", topology)
   }
 
   private fun launchTopology(

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

@@ -26,6 +26,9 @@ const val TOPIC_SUBDIVIDED_MAP_TILES = "$DOMAIN.map-tiles.subdivided"
 const val TOPIC_SUBDIVIDED_MAP_TILES_DEBOUNCED = "$TOPIC_SUBDIVIDED_MAP_TILES.debounced"
 
 
+/** Broadcast these events out, via websocket */
+const val TOPIC_BROADCAST_TO_WEBSOCKET = "$DOMAIN.broadcast.websocket"
+
 fun allTopics(): Set<String> = buildSet {
   add(TOPIC_SRC_SERVER_LOG)
 
@@ -38,6 +41,8 @@ fun allTopics(): Set<String> = buildSet {
   add(TOPIC_SUBDIVIDED_MAP_TILES)
   add(TOPIC_SUBDIVIDED_MAP_TILES_DEBOUNCED)
 
+  add(TOPIC_BROADCAST_TO_WEBSOCKET)
+
   addAll(packetTopicNames.values)
 }
 

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

@@ -100,19 +100,19 @@ internal class WebMapTileServer(
           currentETag == null          -> {
 //            println("[${request.etagKey}] currentETag is null (requested: $requestedETag), executing request")
             response
-              .header("Cache-Control", "no-cache")
+              .header("Cache-Control", "no-cache, max-age=0, must-revalidate")
           }
           requestedETag == currentETag -> {
 //            println("[${request.etagKey}] currentETag == requested, returning 304 $requestedETag")
             Response(Status.NOT_MODIFIED)
-              .header("Cache-Control", "no-cache")
+              .header("Cache-Control", "no-cache, max-age=0, must-revalidate")
           }
           else                         -> {
 //            println("[${request.etagKey}] currentETag != requested, executing request and adding currentETag:$currentETag (requested:$requestedETag)")
             response
               .header("ETag", currentETag)
 //              .header("Cache-Control", "no-store")
-              .header("Cache-Control", "no-cache")
+              .header("Cache-Control", "no-cache, max-age=0, must-revalidate")
           }
         }
       }

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

@@ -6,6 +6,8 @@ import dev.adamko.kafkatorio.processor.misc.DebounceProcessor.Companion.addDebou
 import dev.adamko.kafkatorio.processor.serdes.kxsBinary
 import dev.adamko.kafkatorio.schema.common.ChunkSize
 import dev.adamko.kafkatorio.schema.common.ColourHex
+import dev.adamko.kafkatorio.schema.common.FactorioServerId
+import dev.adamko.kafkatorio.schema.common.ServerMapChunkId
 import dev.adamko.kafkatorio.schema.common.leftTopTile
 import dev.adamko.kafkatorio.schema.common.toMapChunkPosition
 import dev.adamko.kafkatorio.schema.packets.MapChunkUpdate

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

@@ -4,6 +4,7 @@ package dev.adamko.kafkatorio.processor.topology
 import dev.adamko.kafkatorio.processor.admin.TOPIC_SRC_SERVER_LOG
 import dev.adamko.kafkatorio.processor.admin.topicName
 import dev.adamko.kafkatorio.processor.serdes.jsonMapper
+import dev.adamko.kafkatorio.schema.common.FactorioServerId
 import dev.adamko.kafkatorio.schema.packets.KafkatorioPacket
 import dev.adamko.kotka.extensions.consumedAs
 import dev.adamko.kotka.extensions.producedAs

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

@@ -1,32 +1,14 @@
 package dev.adamko.kafkatorio.processor.topology
 
 import com.sksamuel.scrimage.color.RGBColor
-import dev.adamko.kafkatorio.schema.common.ChunkSize
 import dev.adamko.kafkatorio.schema.common.ColourHex
-import dev.adamko.kafkatorio.schema.common.MapChunkPosition
 import dev.adamko.kafkatorio.schema.common.MapTile
 import dev.adamko.kafkatorio.schema.common.MapTilePosition
-import dev.adamko.kafkatorio.schema.common.SurfaceIndex
+import dev.adamko.kafkatorio.schema.common.ServerMapChunkId
 import java.awt.Color
 import kotlinx.serialization.Serializable
 
 
-@JvmInline
-@Serializable
-value class FactorioServerId(private val id: String) {
-  override fun toString() = id
-}
-
-
-@Serializable
-data class ServerMapChunkId(
-  val serverId: FactorioServerId,
-  val chunkPosition: MapChunkPosition,
-  val surfaceIndex: SurfaceIndex,
-  val chunkSize: ChunkSize,
-)
-
-
 @Serializable
 data class ServerMapChunkTiles<Data>(
   val chunkId: ServerMapChunkId,

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

@@ -2,6 +2,7 @@ package dev.adamko.kafkatorio.processor.topology
 
 import dev.adamko.kafkatorio.processor.admin.topicName
 import dev.adamko.kafkatorio.processor.serdes.jsonMapper
+import dev.adamko.kafkatorio.schema.common.FactorioServerId
 import dev.adamko.kafkatorio.schema.packets.KafkatorioPacket
 import dev.adamko.kafkatorio.schema.packets.KafkatorioPacketData
 import dev.adamko.kotka.extensions.consumedAs

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

@@ -1,11 +1,17 @@
 package dev.adamko.kafkatorio.processor.topology
 
 import dev.adamko.kafkatorio.processor.WebsocketServer
+import dev.adamko.kafkatorio.processor.admin.TOPIC_BROADCAST_TO_WEBSOCKET
 import dev.adamko.kafkatorio.processor.admin.topicName
+import dev.adamko.kafkatorio.schema.common.FactorioServerId
 import dev.adamko.kafkatorio.schema.jsonMapperKafkatorio
+import dev.adamko.kafkatorio.schema.packets.EventServerPacket
 import dev.adamko.kafkatorio.schema.packets.KafkatorioPacket
 import dev.adamko.kafkatorio.schema.packets.PlayerUpdate
 import dev.adamko.kotka.extensions.consumedAs
+import dev.adamko.kotka.extensions.producedAs
+import dev.adamko.kotka.extensions.streams.forEach
+import dev.adamko.kotka.extensions.streams.map
 import dev.adamko.kotka.kxs.serde
 import kotlinx.serialization.encodeToString
 import kotlinx.serialization.json.Json
@@ -18,15 +24,41 @@ private val jsonMapper: Json = Json {
 }
 
 
-fun playerUpdatesToWsServer(
+private const val pid = "websocketBroadcast"
+
+
+fun broadcastToWebsocket(
   websocketServer: WebsocketServer,
   builder: StreamsBuilder = StreamsBuilder(),
 ) {
-  builder.stream<FactorioServerId, KafkatorioPacket>(
-    PlayerUpdate::class.topicName,
-    consumedAs("EventUpdate.Player-updates-for-ws-server", jsonMapper.serde(), jsonMapper.serde()),
-  ).foreach { _, value ->
-//      println("sending ${value.packetType} packet to websocket")
-    websocketServer.sendMessage(jsonMapper.encodeToString(value))
+
+  builder.preparePlayerUpdatesBroadcast()
+
+  builder.stream<FactorioServerId, EventServerPacket>(
+    TOPIC_BROADCAST_TO_WEBSOCKET,
+    consumedAs("$pid.packets-to-broadcast.consume", jsonMapper.serde(), jsonMapper.serde()),
+  ).forEach(
+    "$pid.websocket-server.send-message"
+  ) { _, esPacket ->
+    websocketServer.sendMessage(jsonMapper.encodeToString(esPacket))
   }
 }
+
+
+private fun StreamsBuilder.preparePlayerUpdatesBroadcast() {
+  stream<FactorioServerId, KafkatorioPacket>(
+    PlayerUpdate::class.topicName,
+    consumedAs("$pid.player-updates", jsonMapper.serde(), jsonMapper.serde()),
+  ).map<FactorioServerId, KafkatorioPacket, FactorioServerId, EventServerPacket>(
+    "$pid.player-updates.convert"
+  ) { serverId, playerPacket ->
+    serverId to EventServerPacket.Kafkatorio(playerPacket)
+  }.to(
+    TOPIC_BROADCAST_TO_WEBSOCKET,
+    producedAs(
+      "$pid.player-updates.stream",
+      jsonMapper.serde(FactorioServerId.serializer()),
+      jsonMapper.serde(EventServerPacket.serializer()),
+    )
+  )
+}

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

@@ -3,6 +3,7 @@ package dev.adamko.kafkatorio.processor.topology
 import dev.adamko.kafkatorio.processor.serdes.kxsBinary
 import dev.adamko.kafkatorio.schema.common.ColourHex
 import dev.adamko.kafkatorio.schema.common.FactorioPrototype
+import dev.adamko.kafkatorio.schema.common.FactorioServerId
 import dev.adamko.kafkatorio.schema.common.MapTile
 import dev.adamko.kafkatorio.schema.common.toHex
 import dev.adamko.kafkatorio.schema.packets.PrototypesUpdate

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

@@ -3,11 +3,19 @@ package dev.adamko.kafkatorio.processor.topology
 import com.sksamuel.scrimage.ImmutableImage
 import com.sksamuel.scrimage.ScaleMethod
 import com.sksamuel.scrimage.nio.PngWriter
+import dev.adamko.kafkatorio.processor.admin.TOPIC_BROADCAST_TO_WEBSOCKET
 import dev.adamko.kafkatorio.processor.admin.TOPIC_MAP_CHUNK_COLOURED_STATE
+import dev.adamko.kafkatorio.processor.serdes.jsonMapper
 import dev.adamko.kafkatorio.processor.serdes.kxsBinary
 import dev.adamko.kafkatorio.schema.common.ColourHex
+import dev.adamko.kafkatorio.schema.common.FactorioServerId
+import dev.adamko.kafkatorio.schema.common.ServerMapChunkId
+import dev.adamko.kafkatorio.schema.common.TilePngFilename
+import dev.adamko.kafkatorio.schema.packets.EventServerPacket
 import dev.adamko.kotka.extensions.consumedAs
-import dev.adamko.kotka.extensions.streams.forEach
+import dev.adamko.kotka.extensions.producedAs
+import dev.adamko.kotka.extensions.streams.map
+import dev.adamko.kotka.extensions.streams.mapValues
 import dev.adamko.kotka.kxs.serde
 import java.awt.image.BufferedImage
 import java.nio.file.Path
@@ -41,15 +49,18 @@ fun saveMapTiles(
       ),
     )
 
-  subdividedMapChunkTilesDebounced.forEachChunkSaveImage(tileDirectory)
+  val savedChunkFilenames: KStream<ServerMapChunkId, TilePngFilename> =
+    subdividedMapChunkTilesDebounced.saveChunkAsImage(tileDirectory)
+
+  savedChunkFilenames.broadcastSavedTileUpdates()
 
   return builder.build()
 }
 
 
-private fun KStream<ServerMapChunkId, ServerMapChunkTiles<ColourHex>>.forEachChunkSaveImage(
+private fun KStream<ServerMapChunkId, ServerMapChunkTiles<ColourHex>>.saveChunkAsImage(
   tileDirectory: Path,
-) = forEach(
+): KStream<ServerMapChunkId, TilePngFilename> = mapValues(
   "$pid.save-tiles"
 ) { chunkId: ServerMapChunkId, tileColours: ServerMapChunkTiles<ColourHex> ->
 
@@ -77,6 +88,7 @@ private fun KStream<ServerMapChunkId, ServerMapChunkTiles<ColourHex>>.forEachChu
 
   val filename = TilePngFilename(chunkId)
   saveTile(filename, tileDirectory, scaledImage)
+  filename
 }
 
 
@@ -105,15 +117,17 @@ private fun saveTile(
 }
 
 
-@JvmInline
-private value class TilePngFilename(
-  val value: String,
-) {
-  constructor(id: ServerMapChunkId) : this(buildString {
-    append("s${id.surfaceIndex}")
-    append("/z${id.chunkSize.zoomLevel}")
-    append("/x${id.chunkPosition.x}")
-    append("/y${id.chunkPosition.y}")
-    append(".png")
-  })
+private fun KStream<ServerMapChunkId, TilePngFilename>.broadcastSavedTileUpdates() {
+  map<ServerMapChunkId, TilePngFilename, FactorioServerId, EventServerPacket>(
+    "$pid.broadcast.filename-update.convert",
+  ) { id, filename ->
+    id.serverId to EventServerPacket.ChunkTileSaved(id, filename)
+  }.to(
+    TOPIC_BROADCAST_TO_WEBSOCKET,
+    producedAs(
+      "$pid.broadcast.filename-update.send",
+      jsonMapper.serde(FactorioServerId.serializer()),
+      jsonMapper.serde(EventServerPacket.serializer()),
+    )
+  )
 }

modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/misc/KxsBinaryTests.kt

@@ -1,11 +1,11 @@
 package dev.adamko.kafkatorio.processor.misc
 
 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.schema.common.ServerMapChunkId
 import dev.adamko.kafkatorio.processor.topology.ServerMapChunkTiles
 import dev.adamko.kafkatorio.schema.common.ChunkSize
 import dev.adamko.kafkatorio.schema.common.ColourHex
+import dev.adamko.kafkatorio.schema.common.FactorioServerId
 import dev.adamko.kafkatorio.schema.common.MapChunkPosition
 import dev.adamko.kafkatorio.schema.common.MapTilePosition
 import dev.adamko.kafkatorio.schema.common.SurfaceIndex
@@ -134,7 +134,8 @@ class KxsBinaryTests : FunSpec({
 
     test("binary round trip") {
       val binaryEncoded = kxsBinary.encodeToByteArray(initial)
-      val binaryDecoded = kxsBinary.decodeFromByteArray<ServerMapChunkTiles<ColourHex>>(binaryEncoded)
+      val binaryDecoded =
+        kxsBinary.decodeFromByteArray<ServerMapChunkTiles<ColourHex>>(binaryEncoded)
 
       binaryDecoded shouldBeEqualToComparingFields initial
     }

modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/misc/serdeSize.kt

@@ -1,10 +1,14 @@
 package dev.adamko.kafkatorio.processor.misc
 
 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.schema.common.ServerMapChunkId
 import dev.adamko.kafkatorio.processor.topology.ServerMapChunkTiles
-import dev.adamko.kafkatorio.schema.common.*
+import dev.adamko.kafkatorio.schema.common.ChunkSize
+import dev.adamko.kafkatorio.schema.common.ColourHex
+import dev.adamko.kafkatorio.schema.common.FactorioServerId
+import dev.adamko.kafkatorio.schema.common.MapChunkPosition
+import dev.adamko.kafkatorio.schema.common.MapTilePosition
+import dev.adamko.kafkatorio.schema.common.SurfaceIndex
 import java.util.UUID
 import kotlinx.serialization.encodeToByteArray
 import org.junit.jupiter.api.Test

modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/topology/SaveMapChunksTest.kt

@@ -3,8 +3,10 @@ package dev.adamko.kafkatorio.processor.topology
 import dev.adamko.kafkatorio.processor.serdes.kxsBinary
 import dev.adamko.kafkatorio.schema.common.ChunkSize
 import dev.adamko.kafkatorio.schema.common.ColourHex
+import dev.adamko.kafkatorio.schema.common.FactorioServerId
 import dev.adamko.kafkatorio.schema.common.MapChunkPosition
 import dev.adamko.kafkatorio.schema.common.MapTilePosition
+import dev.adamko.kafkatorio.schema.common.ServerMapChunkId
 import dev.adamko.kafkatorio.schema.common.SurfaceIndex
 import dev.adamko.kafkatorio.schema.common.iterateTiles
 import dev.adamko.kafkatorio.schema.common.toMapChunkPosition