From a557c437cc5766d7ab2f32e3c4ea7af87dad79c9 Mon Sep 17 00:00:00 2001 From: Adam <897017+aSemy@users.noreply.github.com> Date: Wed, 19 Jan 2022 22:42:22 +0100 Subject: [PATCH] tiles are visible! the tileserver works! they're the right way up! they have colours! --- README.md | 5 ++ .../processor/KafkatorioTopology.kt | 87 ++++++++++++------- .../processor/tileserver/WebMapTileServer.kt | 11 ++- .../dev/adamko/kafkatorio/webmap/map.kt | 4 +- 4 files changed, 72 insertions(+), 35 deletions(-) diff --git a/README.md b/README.md index 43670cc..f883e79 100644 --- a/README.md +++ b/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 diff --git 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 index 13a5290..f491b6b 100644 --- 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 @@ -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 = 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, + val tilePrototypes: Set, ) -// : TopicRecord - { - // @kotlinx.serialization.Transient -// 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() { + private fun mapChunksStream(tilePrototypesTable: KTable) { val luaTilesUpdatesStream: KStream = 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 { - override val topicKey: TileUpdateRecordKey = - TileUpdateRecordKey(surfaceIndex, tilePosition) + override val topicKey: TileUpdateRecordKey = TileUpdateRecordKey(surfaceIndex, tilePosition) } val allTileUpdatesStream: KStream = @@ -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 = 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 = + 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 - .get(tile.prototypeName) - ?.value() + val prototypeColour = tilePrototypes + .getValue(tile.prototypeName) ?.mapColour ?.toHexadecimal() diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/tileserver/WebMapTileServer.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/tileserver/WebMapTileServer.kt index 5c2ea5e..db5e420 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/tileserver/WebMapTileServer.kt +++ b/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 { diff --git a/modules/web-map/src/main/kotlin/dev/adamko/kafkatorio/webmap/map.kt b/modules/web-map/src/main/kotlin/dev/adamko/kafkatorio/webmap/map.kt index 1e310ee..4016479 100644 --- a/modules/web-map/src/main/kotlin/dev/adamko/kafkatorio/webmap/map.kt +++ b/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) {