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 f491b6b..969f136 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 @@ -16,7 +16,6 @@ import dev.adamko.kafkatorio.events.schema.MapTilePosition import dev.adamko.kafkatorio.events.schema.MapTilePrototype import dev.adamko.kafkatorio.events.schema.MapTiles import dev.adamko.kafkatorio.events.schema.converters.leftTopTile -import dev.adamko.kafkatorio.events.schema.converters.rightBottomTile import dev.adamko.kafkatorio.events.schema.converters.toHexadecimal import dev.adamko.kafkatorio.events.schema.converters.toMapChunkPosition import dev.adamko.kotka.extensions.consumedAs @@ -24,14 +23,16 @@ 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.aggregate import dev.adamko.kotka.extensions.streams.filter -import dev.adamko.kotka.extensions.streams.flatMap +import dev.adamko.kotka.extensions.streams.groupBy 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.reduce 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.extensions.tables.mapValues import dev.adamko.kotka.kxs.serde import dev.adamko.kotka.topicdata.TopicRecord import dev.adamko.kotka.topicdata.flatMapTopicRecords @@ -42,6 +43,7 @@ import java.time.Duration import kotlin.math.abs import kotlin.math.roundToInt import kotlinx.serialization.Serializable +import kotlinx.serialization.Transient import kotlinx.serialization.decodeFromString import kotlinx.serialization.encodeToString import org.apache.kafka.common.serialization.Serdes @@ -68,37 +70,41 @@ class KafkatorioTopology( const val sourceTopic = "factorio-server-log" } - private lateinit var tilePrototypesStore: ReadOnlyKeyValueStore> +// private lateinit var tilePrototypesStore: ReadOnlyKeyValueStore> fun build() { parseAndSplitServerLog() sendPlayerUpdatesToWebSocket() // createMapTileImages() - val tilePrototypesTable: KTable = tilePrototypesTable() - mapChunksStream(tilePrototypesTable) + val tilePrototypesTable: KTable = tilePrototypesTable() + + val tilesTable: KTable = buildTilesTable() + + val tilesGroupedByChunk: KTable = + buildChunkedTilesTable(tilesTable) + + createChunkTileImage(tilePrototypesTable, tilesGroupedByChunk) val topology = builder.build() val streams = KafkaStreams(topology, appProps.kafkaConfig) + // streams.cleanUp() - streams.setUncaughtExceptionHandler(StreamsExceptionHandler()) - - Runtime.getRuntime().addShutdownHook(Thread { streams.close(Duration.ofSeconds(10)) }) - + Runtime.getRuntime().addShutdownHook(Thread { streams.close(Duration.ofSeconds(1)) }) val description: TopologyDescription = topology.describe() println(description) streams.start() - tilePrototypesStore = streams.store( - StoreQueryParameters.fromNameAndType( - tilePrototypesTable.queryableStoreName(), - QueryableStoreTypes.timestampedKeyValueStore() - ) - ) +// tilePrototypesStore = streams.store( +// StoreQueryParameters.fromNameAndType( +// tilePrototypesTable.queryableStoreName(), +// QueryableStoreTypes.timestampedKeyValueStore() +// ) +// ) } private fun parseAndSplitServerLog() { @@ -124,11 +130,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" } } @@ -146,30 +152,67 @@ class KafkatorioTopology( } } - private fun tilePrototypesTable(): KTable { + @Serializable + data class SurfaceIndex( + val surfaceIndex: Int + ) { + override fun toString() = "$surfaceIndex" + } + + @Serializable + data class SurfacePrototypesData( + val surfaceIndex: SurfaceIndex, + val mapTilePrototypes: Set, + ) + + private fun tilePrototypesTable(): KTable { // val tilePrototypesStreamName = "kafkatorio.stream.prototypes-tiles" // val tilePrototypesStoreName = "kafkatorio.store.prototypes-tiles" return builder.stream( "kafkatorio.${KafkatorioPacket.PacketType.PROTOTYPES}.all", - Consumed.with(Serdes.String(), jsonMapper.serde()) + consumedAs( + "consume-all-prototypes-packets", + Serdes.String(), + jsonMapper.serde() + ) ) - .flatMap("flatten-map-tile-prototypes") { _, prototypes: FactorioPrototypes -> - prototypes.prototypes + .map("flatten-map-tile-prototypes") { _, prototypes: FactorioPrototypes -> + val tilePrototypes = prototypes.prototypes .filterIsInstance() - .map { proto -> proto.name to proto } + .toSet() + val key = SurfaceIndex(1) // TODO make surfaceIndex come from Factorio, not hard coded + val data = SurfacePrototypesData(key, tilePrototypes) + key to data } .peek { _, value -> - println("tile prototype: ${value.name}") + println("tile prototypes: ${value.mapTilePrototypes.joinToString { it.name }}") } - .toTable( - materializedAs( - "kafkatorio.prototypes.map-tiles", + .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: SurfacePrototypesData, value2: SurfacePrototypesData -> + value1.copy(mapTilePrototypes = value1.mapTilePrototypes + value2.mapTilePrototypes) + } +// .toTable( +// materializedAs( +// "kafkatorio.prototypes.map-tiles", +// jsonMapper.serde(), +// jsonMapper.serde() +// ) +// ) // .to( // tilePrototypesStreamName, // ) @@ -190,137 +233,166 @@ class KafkatorioTopology( } @Serializable - data class MapTileWithPrototype( - val position: MapTilePosition, - val prototype: MapTilePrototype, + data class TileUpdateRecordKey( + val surfaceIndex: SurfaceIndex, + val tilePosition: MapTilePosition, ) + @Serializable + data class TileUpdateRecord( + val surfaceIndex: SurfaceIndex, + val tilePosition: MapTilePosition, + val tile: MapTile, + ) : TopicRecord { + @Transient + override val topicKey: TileUpdateRecordKey = TileUpdateRecordKey(surfaceIndex, tilePosition) + } + + private fun buildTilesTable(): KTable { + + val luaTilesUpdatesStream: KStream = builder.stream( + "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.LuaTiles}", + consumedAs("consume-map-tiles-packets", Serdes.String(), jsonMapper.serde()) + ) + .filter("events.filter.map-tiles") { _: String, event: FactorioEvent -> event.data is MapTiles } + .mapValues("events.extract-map-tiles") { _, event: FactorioEvent -> (event.data as? MapTiles)!! } + + val chunkTilesUpdateStream: KStream = builder.stream( + "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.MapChunk}", + consumedAs("consume-map-chunk-packets", Serdes.String(), jsonMapper.serde()) + ) + .filter("events.filter.map-chunks") { _: String, event: FactorioEvent -> event.data is MapChunk } + .mapValues("events.extract-map-chunks") { _, event: FactorioEvent -> (event.data as? MapChunk)!!.tiles } + + return luaTilesUpdatesStream + .merge("luaTilesUpdatesStream-and-chunkTilesUpdateStream", chunkTilesUpdateStream) + .flatMapTopicRecords("all-tiles-convert-to-TileUpdateRecord") { _, mapTiles: MapTiles -> +// println("all map tiles update ${mapTiles.tiles.size}") + mapTiles.tiles.map { tile -> + TileUpdateRecord( + SurfaceIndex(mapTiles.surfaceIndex), + tile.position, + tile, + ) + } + } + .toTable(materializedWith(jsonMapper.serde(), jsonMapper.serde())) +// .suppress( +// Suppressed.untilTimeLimit( +// Duration.ofSeconds(30), +// Suppressed.BufferConfig +// .maxRecords(100) +// .emitEarlyWhenFull() +// ) +// ) + } + @Serializable data class MapChunkDataPosition( val position: MapChunkPosition, - val surfaceIndex: Int, + val surfaceIndex: SurfaceIndex, ) @Serializable data class MapChunkData( val chunkPosition: MapChunkDataPosition, val tiles: Set, - val tilePrototypes: Set, ) - private fun mapChunksStream(tilePrototypesTable: KTable) { - - val luaTilesUpdatesStream: KStream = builder.stream( - "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.LuaTiles}", - Consumed.with(Serdes.String(), jsonMapper.serde()) - ) -// .peek { k, v -> -// println("trying to handle MapChunk update... $k: ${v.eventType}") -// } - .filter("events.filter.map-tiles") { _: String, event: FactorioEvent -> event.data is MapTiles } - .mapValues("events.extract-map-tiles") { _, event: FactorioEvent -> (event.data as? MapTiles)!! } -// .peek { key, value -> -// println("MapChunk update $key, tiles count: ${value.tiles.size}") -// } - - - val chunkTilesUpdateStream: KStream = builder.stream( - "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.MapChunk}", - Consumed.with(Serdes.String(), jsonMapper.serde()) - ) - .peek { k, v -> - println("trying to handle LuaTiles update... $k: ${v.eventType} / ${v.data is MapChunk}") - } - .filter("events.filter.map-chunks") { _: String, event: FactorioEvent -> event.data is MapChunk } - .mapValues("events.extract-map-chunks") { _, event: FactorioEvent -> (event.data as? MapChunk)!!.tiles } - .peek { key, value -> - println("LuaTiles update $key, tiles count: ${value.tiles.size}") - } + private fun buildChunkedTilesTable( + tilesTable: KTable + ): KTable { @Serializable - data class TileUpdateRecordKey( - val surfaceIndex: Int, - val tilePosition: MapTilePosition, + data class MapChunkDataAggregator( + val tiles: MutableSet = mutableSetOf(), ) - @Serializable - data class TileUpdateRecord( - val surfaceIndex: Int, - val tilePosition: MapTilePosition, - val tile: MapTile, - val tilePrototype: MapTilePrototype?, - ) : TopicRecord { - override val topicKey: TileUpdateRecordKey = TileUpdateRecordKey(surfaceIndex, tilePosition) - } - - val allTileUpdatesStream: KStream = - luaTilesUpdatesStream - .merge("luaTilesUpdatesStream-and-chunkTilesUpdateStream", chunkTilesUpdateStream) - .flatMapTopicRecords("all-tiles-convert-to-TileUpdateRecord") { _, mapTiles: MapTiles -> -// println("all map tiles update ${mapTiles.tiles.size}") - mapTiles.tiles.map { tile -> - TileUpdateRecord( - mapTiles.surfaceIndex, - tile.position, - 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 - .map("map-tile-updates-to-MapChunkData") { _: TileUpdateRecordKey, value: TileUpdateRecord -> - val key = MapChunkDataPosition( - value.tilePosition.toMapChunkPosition(), - value.surfaceIndex, - ) - 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, - tilePrototypes = tileA.tilePrototypes + tileB.tilePrototypes - ) - } - .suppress( - Suppressed.untilTimeLimit( - Duration.ofSeconds(30), - Suppressed.BufferConfig.maxRecords(32*32*2) - ) - ) - - chunksTable + return tilesTable .toStream() - .foreach { key: MapChunkDataPosition, value: MapChunkData -> +// .repartition( +// repartitionedAs( +// "repartition-all-tiles-before-chunk-grouping", +// jsonMapper.serde(), +// jsonMapper.serde() +// ) +// ) + .groupBy( + groupedAs("grouping-tiles-by-chunk-pos", jsonMapper.serde(), jsonMapper.serde()), + ) { key: TileUpdateRecordKey, value: TileUpdateRecord -> + MapChunkDataPosition( + value.tilePosition.toMapChunkPosition(), + key.surfaceIndex + ) + } + .aggregate( + "reducing-tiles-grouped-by-chunk-position", + materializedWith(jsonMapper.serde(), jsonMapper.serde()), + { MapChunkDataAggregator() }, + ) { _: MapChunkDataPosition, value: TileUpdateRecord, aggregate: MapChunkDataAggregator -> + aggregate.tiles.add(value.tile) + aggregate + } + // TODO maybe set serdes here? + .mapValues( + "tiles-grouped-by-chunk-position-finalise-aggregation", + materializedWith(jsonMapper.serde(), jsonMapper.serde()), + ) { k, v -> + MapChunkData(k, v.tiles) + } + + } + + private fun createChunkTileImage( + tilePrototypesTable: KTable, + tilesGroupedByChunk: KTable, + ) { + + @Serializable + data class ChunkTilesAndProtos( + val chunkTiles: MapChunkData, + val protos: SurfacePrototypesData + ) + +// tilesGroupedByChunk +// .join( +// tilePrototypesTable, +// { cd: MapChunkData -> cd.chunkPosition.surfaceIndex }, +// { chunkTiles: MapChunkData, protos: SurfacePrototypesData -> +// ChunkTilesAndProtos(chunkTiles, protos) +// }, +// namedAs("joining-chunks-tiles-prototypes"), +// materializedAs( +// "joined-chunked-tiles-with-prototypes", +// jsonMapper.serde(), +// jsonMapper.serde() +// ), +// ) + tilesGroupedByChunk + .join( + tilePrototypesTable, + "enrich-tiles-with-prototype-colour", + materializedAs( + "joined-chunked-tiles-with-prototypes", + jsonMapper.serde(), + jsonMapper.serde() + ), + { it.chunkPosition.surfaceIndex }, + ) { chunkTiles: MapChunkData, protos: SurfacePrototypesData -> + ChunkTilesAndProtos(chunkTiles, protos) + } + .suppress( + Suppressed.untilTimeLimit( + Duration.ofSeconds(30), + Suppressed.BufferConfig.maxRecords(30) + ) + ) + .toStream() + .foreach { _, (chunkTiles: MapChunkData, protos: SurfacePrototypesData) -> runCatching { - saveMapTilesPng(key, value) + saveMapTilesPng(chunkTiles, protos) }.onFailure { e -> - println("error saving map tile png chunk:${value.chunkPosition}") + println("error saving map tile png chunk:${chunkTiles.chunkPosition}") e.printStackTrace() throw e } @@ -328,14 +400,16 @@ class KafkatorioTopology( } - private fun saveMapTilesPng(key: MapChunkDataPosition, chunk: MapChunkData) { + private fun saveMapTilesPng(chunkTiles: MapChunkData, protos: SurfacePrototypesData) { + + val chunkPos = chunkTiles.chunkPosition val tilePrototypes: Map = - chunk.tilePrototypes.associateBy { it.name }.withDefault { null } + protos.mapTilePrototypes.associateBy { it.name }.withDefault { null } - val chunkOriginX = chunk.chunkPosition.position.leftTopTile.x + val chunkOriginX = chunkTiles.chunkPosition.position.leftTopTile.x // val chunkOriginY = chunk.chunkPosition.position.rightBottomTile.y - val chunkOriginY = chunk.chunkPosition.position.leftTopTile.y + val chunkOriginY = chunkTiles.chunkPosition.position.leftTopTile.y val chunkImage = ImmutableImage.filled( @@ -345,7 +419,7 @@ class KafkatorioTopology( BufferedImage.TYPE_INT_ARGB ) - chunk.tiles.forEach { tile -> + chunkTiles.tiles.forEach { tile -> val prototypeColour = tilePrototypes .getValue(tile.prototypeName) @@ -378,7 +452,9 @@ class KafkatorioTopology( val zoom = 1u val file = - File("src/main/resources/kafkatorio-web-map/s${key.surfaceIndex}/z$zoom/x${key.position.x}/y${key.position.y}.png") + File( + "src/main/resources/kafkatorio-web-map/s${chunkPos.surfaceIndex}/z$zoom/x${chunkOriginX}/y${chunkOriginY}.png" + ) if (file.parentFile.mkdirs()) { println("created new map tile parentFile directory ${file.absolutePath}") @@ -389,100 +465,4 @@ class KafkatorioTopology( chunkImage.output(PngWriter.MaxCompression, file) } -// -// private fun createMapTileImages() { -// -// val mapTilesByPrototypeName: KStream = -// builder.stream( -// "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.LuaTiles}", -// Consumed.with(Serdes.String(), KafkatorioPacketSerde) -// ) -// .filter { _, packet -> packet is FactorioEvent && packet.data is MapTiles } -// .mapValues { packet -> ((packet as? FactorioEvent)?.data as? MapTiles)!! } -// .flatMap { _, mapTiles -> -// mapTiles -// .tiles -// .map { tile -> tile.prototypeName to tile } -// .map { it.toKeyValue() } -// } -//// .groupByKey() -// -// @Serializable -// data class TileSurfaceCoord( -// val surfaceIndex: Int, -// val position: MapTilePosition, -// ) -// -// @Serializable -// data class TileSurfaceState( -// val surfaceIndex: Int, -// val position: MapTilePosition, -// val prototypeName: String, -// ) -// -// val mapTilesTable: KTable = -// builder.stream( -// "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.LuaTiles}", -// Consumed.with(Serdes.String(), KafkatorioPacketSerde) -// ) -// .filter { _, packet -> packet is FactorioEvent && packet.data is MapTiles } -// .mapValues { packet -> ((packet as? FactorioEvent)?.data as? MapTiles)!! } -// .flatMap({ _, mapTiles: MapTiles -> -// mapTiles.tiles.map { tile: MapTile -> -// val key = TileSurfaceCoord(mapTiles.surfaceIndex, tile.position) -// val value = TileSurfaceState(mapTiles.surfaceIndex, tile.position, tile.prototypeName) -// (key to value).toKeyValue() -// } -// } -// ) -// .toTable( -// materializedAs( -// "kafkatorio.table.map-tile-status", -// jsonMapper.serde(), -// jsonMapper.serde(), -// ) -// ) -// -// -// val tilePrototypesTable: KTable = -// builder.stream( -// "kafkatorio.${KafkatorioPacket.PacketType.CONFIG}.FactorioConfigurationUpdate", -// Consumed.with(Serdes.String(), KafkatorioPacketSerde) -// ) -// .flatMap(KeyValueMapper>> { _, packet -> -// when (packet) { -// is FactorioConfigurationUpdate -> -// packet.prototypes -// .filterIsInstance() -// .map { proto -> proto.name to proto } -// .map { it.toKeyValue() } -// is FactorioEvent -> -// emptyList() -// } -// }) -// .toTable( -// materializedAs("kafkatorio.table.tile-prototypes", Serdes.String(), jsonMapper.serde()) -// ) -// -// -// @Serializable -// data class TileSurfaceData( -// val surfaceIndex: Int, -// val position: MapTilePosition, -// val prototype: MapTilePrototype, -// ) -// -// val enrichedTiles: KTable = -// mapTilesTable.join( -// tilePrototypesTable, -// { tile: TileSurfaceState -> -// tile.prototypeName -// }, -// { tile: TileSurfaceState, proto: MapTilePrototype -> -// TileSurfaceData(tile.surfaceIndex, tile.position, proto) -// }, -// ) -// -// } - } diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/eventProcessor.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/eventProcessor.kt index bc3ab88..faaf92b 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/eventProcessor.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/eventProcessor.kt @@ -3,6 +3,7 @@ package dev.adamko.kafkatorio.processor import dev.adamko.kafkatorio.processor.config.ApplicationProperties import dev.adamko.kafkatorio.processor.tileserver.WebMapTileServer +import kotlinx.coroutines.launch import kotlinx.coroutines.runBlocking import org.http4k.server.PolyHandler import org.http4k.server.Undertow @@ -10,7 +11,7 @@ import org.http4k.server.asServer val appProps = ApplicationProperties() -fun main() = runBlocking { +fun main() = runBlocking { val wsServer = WebsocketServer() val tileServer = WebMapTileServer() @@ -18,11 +19,16 @@ fun main() = runBlocking { val topology = KafkatorioTopology(wsServer) topology.build() - PolyHandler( + val webServer = PolyHandler( tileServer.build(), wsServer.build(), ).asServer(Undertow(9073)) - .start() - .block() + +// Runtime.getRuntime().addShutdownHook(Thread { +// webServer.stop() +// }) +// launch { + webServer.start().block() +// } } 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 4016479..f29c4a2 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 @@ -34,11 +34,16 @@ fun createFactorioMap() = Maps { ) { attribution = "kafkatorio" tms = true - minZoom = 1 -// maxZoom = 1 + minZoom = -2 + maxZoom = 1 maxNativeZoom = 1 + minNativeZoom = 1 }.addTo(this) + L.scale { + + } + with(options) { crs = CRS.Simple }