From bdb2d9577fbd7e3cdade96e4971e85dd301d5b80 Mon Sep 17 00:00:00 2001 From: Adam <897017+aSemy@users.noreply.github.com> Date: Tue, 8 Feb 2022 01:35:52 +0100 Subject: [PATCH] so close to working performantly.... just the tiles are transparent??? - bump some versions - MapTile no longer has objectName field - fix nulls/optional fields in FactorioConfigurationUpdate - use bytes to encode rgba colours - waaaayyyy smaller - increase Kafka record size to ~10mb --- build.gradle.kts | 2 +- gradle/libs.versions.toml | 8 +- .../generated/kafkatorio-schema/schema.d.ts | 1 - .../main/typescript/main/events/converters.ts | 2 - .../main/typescript/main/events/handlers.ts | 3 +- .../src/main/typescript/package-lock.json | 2 +- .../src/main/typescript/package.json | 2 +- .../events/schema/converters/colour.kt | 7 +- .../events/schema/factorioCommonData.kt | 38 ++- .../events/schema/factorioConfiguration.kt | 8 +- .../events/schema/factorioLuaObjects.kt | 11 - .../events/schema/FactorioEventTest.kt | 66 ++++- .../processor/KafkatorioTopology.kt | 18 +- .../processor/topology/factorioGameState.kt | 269 +++++++++++------- .../processor/topology/flowMapTiles.kt | 230 +++++++-------- .../kafkatorio/processor/topology/model.kt | 24 +- .../processor/topology/prototypes.kt | 89 +++--- .../processor/topology/webMapTiles.kt | 164 +++++------ .../kafkatorio/processor/misc/serdeSize.kt | 62 ++++ modules/infra-factorio-server/src/.env | 2 +- modules/infra-kafka-cluster/src/.env | 2 +- .../src/kafka-server.properties | 5 + .../infra-kafka-cluster/src/kafka.Dockerfile | 2 +- modules/web-map/build.gradle.kts | 6 +- 24 files changed, 619 insertions(+), 404 deletions(-) create mode 100644 modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/misc/serdeSize.kt diff --git a/build.gradle.kts b/build.gradle.kts index bfae4e2..3cbf0c0 100644 --- a/build.gradle.kts +++ b/build.gradle.kts @@ -6,7 +6,7 @@ plugins { } group = "dev.adamko.kafkatorio" -version = "0.2.7" +version = "0.2.8" val licenseFile: RegularFile by extra(layout.projectDirectory.file("LICENSE")) diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index 625e7dd..a2ff04b 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -10,17 +10,17 @@ kotlin = "1.6.10" ## libs ## camel-kafkaConnectors = "0.11.0" -http4k = "4.17.6.0" +http4k = "4.19.1.0" kafka = "3.1.0" kotlinx-serialization = "1.3.2" -kvision = "5.7.1" +kvision = "5.8.1" log4j2 = "2.14.1" qosLogback = "1.2.5" slf4j = "1.7.30" ## test ## junit = "5.8.2" -kotest = "5.0.1" +kotest = "5.1.0" ## infra ## factorio = "1.1.53" @@ -99,7 +99,7 @@ kotka = { group = "com.github.blueanvil", name = "kotka", versio ### test ### ############# -mockk = { group = "io.mockk", name = "mockk", version = "1.12.0" } +mockk = { group = "io.mockk", name = "mockk", version = "1.12.2" } ## junit ## junit-bom = { group = "org.junit", name = "junit-bom", version.ref = "junit" } diff --git 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 index da32cea..f1caf5a 100644 --- 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 @@ -43,7 +43,6 @@ interface MapChunkPosition { } interface MapTile { - objectName: string; position: MapTilePosition; prototypeName: string; } diff --git a/modules/events-mod/src/main/typescript/main/events/converters.ts b/modules/events-mod/src/main/typescript/main/events/converters.ts index 26b8e18..6314ba4 100644 --- a/modules/events-mod/src/main/typescript/main/events/converters.ts +++ b/modules/events-mod/src/main/typescript/main/events/converters.ts @@ -98,8 +98,6 @@ export namespace Converters { export function convertTile(tile: LuaTile): MapTile { return { - objectName: tile.object_name, - position: mapTilePosition(tile.position), prototypeName: tile.prototype.name, } diff --git a/modules/events-mod/src/main/typescript/main/events/handlers.ts b/modules/events-mod/src/main/typescript/main/events/handlers.ts index 1599367..569eb27 100644 --- a/modules/events-mod/src/main/typescript/main/events/handlers.ts +++ b/modules/events-mod/src/main/typescript/main/events/handlers.ts @@ -92,8 +92,7 @@ export function handleTilesUpdate( let convertedTiles: MapTile[] = updatedTiles.map((it) => { return { - objectName: "LuaTile", - position: it.position, + position: it.position, // TODO use converters for LuaTile prototypeName: newTileType.name, } }) diff --git a/modules/events-mod/src/main/typescript/package-lock.json b/modules/events-mod/src/main/typescript/package-lock.json index a8216e6..86d4591 100644 --- a/modules/events-mod/src/main/typescript/package-lock.json +++ b/modules/events-mod/src/main/typescript/package-lock.json @@ -1,6 +1,6 @@ { "name": "kafkatorio-events-mod", - "version": "0.2.7", + "version": "0.2.8", "lockfileVersion": 1, "requires": true, "dependencies": { diff --git a/modules/events-mod/src/main/typescript/package.json b/modules/events-mod/src/main/typescript/package.json index c9efe3b..3ce0e46 100644 --- a/modules/events-mod/src/main/typescript/package.json +++ b/modules/events-mod/src/main/typescript/package.json @@ -1,6 +1,6 @@ { "name": "kafkatorio-events-mod", - "version": "0.2.7", + "version": "0.2.8", "private": true, "scripts": { "build": "tstl", diff --git a/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/converters/colour.kt b/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/converters/colour.kt index 88926b3..a4f6fba 100644 --- a/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/converters/colour.kt +++ b/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/converters/colour.kt @@ -23,7 +23,12 @@ fun Colour.toHexadecimal() = if (isPercentage()) { (alpha * 255).coerceIn(0f..255f), ) } else { - this + Colour( + red.coerceIn(0f..255f), + green.coerceIn(0f..255f), + blue.coerceIn(0f..255f), + alpha.coerceIn(0f..255f), + ) } fun Colour.toPercentage() = if (isPercentage()) { diff --git 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 index f519e95..d6ffb50 100644 --- 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 @@ -1,5 +1,6 @@ package dev.adamko.kafkatorio.events.schema +import dev.adamko.kafkatorio.events.schema.converters.toHexadecimal import kotlinx.serialization.EncodeDefault import kotlinx.serialization.Serializable @@ -56,7 +57,7 @@ data class MapBoundingBox( * Red, green, blue and alpha values, all in range `[0, 1]` or all in range `[0, 255]` if any * value is > 1. * - * All values here are optional. Colour channels default to 0, the alpha channel defaults to 1. + * All values here are optional. Colour channels default to `0`, the alpha channel defaults to `1`. */ @Serializable data class Colour( @@ -68,4 +69,39 @@ data class Colour( val blue: Float = 0f, @EncodeDefault val alpha: Float = 1f, +) { + + fun toHex() = toHexadecimal().let { + ColourHex( + it.red.toUInt().toUByte(), + it.green.toUInt().toUByte(), + it.blue.toUInt().toUByte(), + it.alpha.toUInt().toUByte(), + ) + } +} + + +/** Size-efficient version of [Colour] */ +@Serializable +data class ColourHex( + @EncodeDefault + val red: UByte = UByte.MIN_VALUE, + @EncodeDefault + val green: UByte = UByte.MIN_VALUE, + @EncodeDefault + val blue: UByte = UByte.MIN_VALUE, + @EncodeDefault + val alpha: UByte = UByte.MAX_VALUE, +) { + companion object { + val TRANSPARENT = ColourHex(0u, 0u, 0u, 0u) + } +} + + +@Serializable +data class MapTile( + val prototypeName: String, + val position: MapTilePosition, ) diff --git 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 index 994656a..900e28d 100644 --- 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 @@ -16,13 +16,13 @@ data class FactorioConfigurationUpdate( @Serializable data class FactorioGameDataUpdate( - val oldVersion: String?, - val newVersion: String?, + val oldVersion: String? = null, + val newVersion: String? = null, ) @Serializable data class FactorioModInfo( val modName: String, - val currentVersion: String?, - val previousVersion: String?, + val currentVersion: String? = null, + val previousVersion: String? = null, ) diff --git a/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioLuaObjects.kt b/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioLuaObjects.kt index ac89d09..b6e2d09 100644 --- a/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioLuaObjects.kt +++ b/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioLuaObjects.kt @@ -137,17 +137,6 @@ data class MapTiles( override val objectName = ObjectName.LuaTiles } -@Serializable -data class MapTile( - val prototypeName: String, - val position: MapTilePosition, -) -// : FactorioObjectData() -{ - // @EncodeDefault - val objectName: String = "" -} - @Serializable data class ConsoleChatMessage( val authorPlayerIndex: UInt?, diff --git 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 index 65c291c..ce7c8a8 100644 --- 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 @@ -203,7 +203,6 @@ class FactorioEventTest : FunSpec({ "surfaceIndex": 1, "tiles": [ { - "objectName": "LuaTile", "position": { "y": 10, "x": -65 @@ -211,23 +210,20 @@ class FactorioEventTest : FunSpec({ "prototypeName": "refined-concrete" }, { - "objectName": "LuaTile", "position": { "y": 11, "x": -65 }, - "prototypeName": "refined-concrete" + "prototypeName": "red-earth" }, { - "objectName": "LuaTile", "position": { "y": 10, "x": -64 }, - "prototypeName": "refined-concrete" + "prototypeName": "blue-cheese" }, { - "objectName": "LuaTile", "position": { "y": 11, "x": -64 @@ -256,11 +252,11 @@ class FactorioEventTest : FunSpec({ position = MapTilePosition(-65, 10) ), MapTile( - prototypeName = "refined-concrete", + prototypeName = "red-earth", position = MapTilePosition(-65, 11) ), MapTile( - prototypeName = "refined-concrete", + prototypeName = "blue-cheese", position = MapTilePosition(-64, 10) ), MapTile( @@ -279,6 +275,60 @@ class FactorioEventTest : FunSpec({ expected.data.objectName shouldBe FactorioObjectData.ObjectName.LuaTiles } + test("Then: expect encode equals json") { + val encoded = jsonMapperKafkatorio.encodeToString(actual) + encoded.shouldEqualJson(json) + } + } + } + + context("Given: packet FactorioConfigurationUpdate") { + // language=JSON + val json = """ + { + "modVersion": "0.2.8", + "packetType": "CONFIG", + "allMods": [ + { + "modName": "kafkatorio-events", + "currentVersion": "0.2.8", + "previousVersion": "0.2.7" + }, + { + "modName": "base", + "currentVersion": "1.1.53" + } + ], + "factorioData": {} + } + """.trimIndent() + context("When: decoded") { + val actual: FactorioConfigurationUpdate = jsonMapperKafkatorio.decodeFromString(json) + + test("parse") { + + val expected = FactorioConfigurationUpdate( + modVersion = "0.2.8", + factorioData = FactorioGameDataUpdate(null, null), + allMods = listOf( + FactorioModInfo( + modName = "kafkatorio-events", + currentVersion = "0.2.8", + previousVersion = "0.2.7" + ), + FactorioModInfo( + modName = "base", + currentVersion = "1.1.53", + previousVersion = null + ), + ) + ) + + actual shouldBe expected + + expected.packetType shouldBe KafkatorioPacket.PacketType.CONFIG + } + test("Then: expect encode equals json") { val encoded = jsonMapperKafkatorio.encodeToString(actual) encoded.shouldEqualJson(json) 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 9049e61..ec6fbb8 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 @@ -1,16 +1,16 @@ package dev.adamko.kafkatorio.processor -import dev.adamko.kafkatorio.processor.topology.buildFactorioServerMap import dev.adamko.kafkatorio.processor.topology.factorioServerPacketStream +import dev.adamko.kafkatorio.processor.topology.groupTilesIntoChunksWithColours 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 java.time.Duration import org.apache.kafka.streams.KafkaStreams import org.apache.kafka.streams.StreamsBuilder import org.apache.kafka.streams.TopologyDescription - class KafkatorioTopology( private val websocketServer: WebsocketServer, private val builder: StreamsBuilder = StreamsBuilder(), @@ -20,17 +20,27 @@ class KafkatorioTopology( const val sourceTopic = "factorio-server-log" } + fun build() { + val packets = factorioServerPacketStream(builder) + + splitFactorioServerPacketStream(packets) playerUpdatesToWsServer(websocketServer, builder) - val serverMapDataTable = buildFactorioServerMap(packets) +// val serverMapDataTable = buildFactorioServerMap(packets, tilePrototypesTable) - saveMapTiles(serverMapDataTable) + val tileProtoColourDict = tileProtoColourDictionary(packets) + + val tilePrototypesTable = groupTilesIntoChunksWithColours( + packets, + tileProtoColourDict, + ) + saveMapTiles(tilePrototypesTable) // the old way... diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioGameState.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioGameState.kt index 7e9df42..02be069 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioGameState.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioGameState.kt @@ -1,127 +1,196 @@ package dev.adamko.kafkatorio.processor.topology -import dev.adamko.kafkatorio.events.schema.FactorioConfigurationUpdate +import dev.adamko.kafkatorio.events.schema.ColourHex import dev.adamko.kafkatorio.events.schema.FactorioEvent -import dev.adamko.kafkatorio.events.schema.FactorioPrototypes import dev.adamko.kafkatorio.events.schema.KafkatorioPacket import dev.adamko.kafkatorio.events.schema.MapChunk +import dev.adamko.kafkatorio.events.schema.MapChunkPosition 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.processor.serdes.jsonMapper +import dev.adamko.kafkatorio.events.schema.converters.toMapChunkPosition import dev.adamko.kafkatorio.processor.serdes.kxsBinary import dev.adamko.kotka.extensions.groupedAs import dev.adamko.kotka.extensions.materializedAs import dev.adamko.kotka.extensions.materializedWith -import dev.adamko.kotka.extensions.streams.aggregate -import dev.adamko.kotka.extensions.tables.mapValues +import dev.adamko.kotka.extensions.streams.flatMap +import dev.adamko.kotka.extensions.streams.reduce +import dev.adamko.kotka.extensions.tables.join import dev.adamko.kotka.kxs.serde import kotlinx.serialization.Serializable import org.apache.kafka.streams.kstream.KStream import org.apache.kafka.streams.kstream.KTable +//@Serializable +//data class FactorioServerMap( +// val serverId: FactorioServerId, +// val tilePrototypes: Map, +// val surfaces: Map, +//) + + +//@Serializable +//data class FactorioMapSurface( +// val index: SurfaceIndex, +// val tiles: Map, +//) + + @Serializable -data class FactorioServerMap( +data class ServerMapChunkId( val serverId: FactorioServerId, - val tilePrototypes: Map, - val surfaces: Map, + val chunkPosition: MapChunkPosition, + val surfaceIndex: SurfaceIndex, + val zoomLevel: ZoomLevel, ) @Serializable -data class FactorioMapSurface( - val index: SurfaceIndex, - val tiles: Map, -) - -/** Aggregate [KafkatorioPacket]s into a single [FactorioServerMap] (per [serverId][FactorioServerId]). */ -fun buildFactorioServerMap( - factorioServerPacketStream: KStream -): KTable { - - return factorioServerPacketStream - .groupByKey( - groupedAs("factorio-server-map-data-grouping", jsonMapper.serde(), jsonMapper.serde()) - ) - .aggregate( - "factorio-server-map-data-aggregate", - materializedAs("factorio-server-map-aggregate-store", kxsBinary.serde(), kxsBinary.serde()), - { FactorioServerMapAggregator() }, - ) { _: FactorioServerId, packet: KafkatorioPacket, aggregate: FactorioServerMapAggregator -> - aggregate + packet - } - .mapValues( - "finalise-factorio-server-map-aggregate", - materializedWith(kxsBinary.serde(), kxsBinary.serde()) - ) { readOnlyKey, aggregate -> - - val surfaces = aggregate.surfaces.mapValues { (_, surface) -> - FactorioMapSurface(surface.index, surface.tiles) - } - - FactorioServerMap(readOnlyKey, aggregate.tilePrototypes, surfaces) - } - -} - - -@Serializable -private data class FactorioServerMapAggregator( - val tilePrototypes: MutableMap = mutableMapOf(), - val surfaces: MutableMap = mutableMapOf(), +data class ServerMapChunkTiles( + val chunkId: ServerMapChunkId, + val map: Map, ) { - - operator fun plus(packet: KafkatorioPacket): FactorioServerMapAggregator { - return when (packet) { - is FactorioEvent -> { - when (val data = packet.data) { - is MapChunk -> plus(data) - is MapTiles -> plus(data) -// is MapTile -> {} // hmmm... no surface index - else -> this - } - } - is FactorioPrototypes -> plus(packet) - is FactorioConfigurationUpdate -> this - } - } - - operator fun plus(mapTiles: MapTiles): FactorioServerMapAggregator { -// println("FactorioServerMapAggregator: adding ${mapTiles.tiles.size} MapTiles from surface ${mapTiles.surfaceIndex}") - - val surfaceIndex = SurfaceIndex(mapTiles.surfaceIndex) - - val surface = surfaces.getOrPut(surfaceIndex) { FactorioMapSurfaceAggregator(surfaceIndex) } - - val surfaceTiles = mapTiles.tiles.associate { it.position to PrototypeName(it.prototypeName) } - surface.tiles.putAll(surfaceTiles) - - return this - } - - operator fun plus(mapChunk: MapChunk): FactorioServerMapAggregator { -// println("FactorioServerMapAggregator: adding mapChunk ${mapChunk.position}") - return plus(mapChunk.tiles) - } - - operator fun plus(packet: FactorioPrototypes): FactorioServerMapAggregator { -// println("FactorioServerMapAggregator: adding FactorioPrototypes") - - val packetPrototypes = packet.prototypes - .filterIsInstance() - .associateBy { PrototypeName(it.name) } - - tilePrototypes.putAll(packetPrototypes) - - return this - } - + operator fun plus(other: ServerMapChunkTiles) = copy(map = map + other.map) } -@Serializable -private data class FactorioMapSurfaceAggregator( - val index: SurfaceIndex, - val tiles: MutableMap = mutableMapOf(), -) +fun groupTilesIntoChunksWithColours( + factorioServerPacketStream: KStream, + tileProtoColourDict: KTable, +): KTable> { + + // group tiles by server & surface & chunk + val chunkedTilesTable: KTable> = + factorioServerPacketStream + .flatMap("server-map-data.tiles.flatMapByChunk") { key: FactorioServerId, packet: KafkatorioPacket -> + + val mapTileData = when (packet) { + is FactorioEvent -> + when (val data = packet.data) { + is MapChunk -> data.tiles + is MapTiles -> data + else -> null + } + else -> null + } + + val zoomLevel = ZoomLevel.ZOOM_0 + +// +// ZoomLevel +// .values +// .mapNotNull { zoomLevel -> + + mapTileData + ?.tiles + ?.groupBy { tile -> + tile.position.toMapChunkPosition(zoomLevel.chunkSize) + } + ?.map { (chunkPos, tiles) -> + + val chunkId = ServerMapChunkId( + key, + chunkPos, + SurfaceIndex(mapTileData.surfaceIndex), + zoomLevel, + ) + + val tilesMap = tiles.associate { it.position to TileProtoHashCode(it) } + + val chunkTiles = ServerMapChunkTiles(chunkId, tilesMap) + + chunkId to chunkTiles + } ?: listOf() +// }.flatten() + } + .groupByKey( + groupedAs("server-map-data.tiles.group", kxsBinary.serde(), kxsBinary.serde()) + ) + .reduce( + "server-map-data.tiles.reduce", + materializedAs("server-map-data.tiles.store", kxsBinary.serde(), kxsBinary.serde()), + ) { a, b -> a + b } + + + val colourisedChunkTable: KTable> = + chunkedTilesTable + .join( + other = tileProtoColourDict, + name = "server-map-data.join-tiles-with-prototypes", + materialized = materializedWith(kxsBinary.serde(), kxsBinary.serde()), + foreignKeyExtractor = { chunkTiles: ServerMapChunkTiles -> chunkTiles.chunkId.serverId } + ) { tiles: ServerMapChunkTiles, colourDict: TileColourDict -> + + ServerMapChunkTiles( + tiles.chunkId, + tiles.map.mapValues { (_, code) -> + colourDict.map.getOrElse(code) { + println("missing prototype $code") + ColourHex.TRANSPARENT + } + } + ) + } + + return colourisedChunkTable + +} + +//@Serializable +//private data class FactorioServerMapAggregator( +// val tilePrototypes: SortedMap = TreeMap(), +// val surfaces: MutableMap = mutableMapOf(), +//) { +// +// operator fun plus(packet: KafkatorioPacket): FactorioServerMapAggregator { +// return when (packet) { +// is FactorioEvent -> { +// when (val data = packet.data) { +// is MapChunk -> plus(data) +// is MapTiles -> plus(data) +// else -> this +// } +// } +// is FactorioPrototypes -> plus(packet) +// is FactorioConfigurationUpdate -> this +// } +// } +// +// operator fun plus(mapTiles: MapTiles): FactorioServerMapAggregator { +//// println("FactorioServerMapAggregator: adding ${mapTiles.tiles.size} MapTiles from surface ${mapTiles.surfaceIndex}") +// +// val surfaceIndex = SurfaceIndex(mapTiles.surfaceIndex) +// +// val surface = surfaces.getOrPut(surfaceIndex) { FactorioMapSurfaceAggregator(surfaceIndex) } +// +// val surfaceTiles = mapTiles.tiles.associate { it.position to PrototypeName(it.prototypeName) } +// surface.tiles.putAll(surfaceTiles) +// +// return this +// } +// +// operator fun plus(mapChunk: MapChunk): FactorioServerMapAggregator { +//// println("FactorioServerMapAggregator: adding mapChunk ${mapChunk.position}") +// return plus(mapChunk.tiles) +// } +// +// operator fun plus(packet: FactorioPrototypes): FactorioServerMapAggregator { +//// println("FactorioServerMapAggregator: adding FactorioPrototypes") +// +// val packetPrototypes = packet.prototypes +// .filterIsInstance() +// .associate { PrototypeName(it.name) to it.mapColour } +// +// tilePrototypes.putAll(packetPrototypes) +// +// return this +// } +// +//} + +// +//@Serializable +//private data class FactorioMapSurfaceAggregator( +// val index: SurfaceIndex, +// val tiles: MutableMap = mutableMapOf(), +//) diff --git 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 index 2aabdf4..3ee7439 100644 --- 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 @@ -3,10 +3,9 @@ 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.events.schema.Colour +import dev.adamko.kafkatorio.events.schema.ColourHex import dev.adamko.kafkatorio.events.schema.MapChunkPosition import dev.adamko.kafkatorio.events.schema.MapTilePosition -import dev.adamko.kafkatorio.events.schema.converters.toMapChunkPosition import java.awt.image.BufferedImage import java.io.File import kotlin.math.abs @@ -15,17 +14,17 @@ import kotlinx.coroutines.CoroutineName import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.SupervisorJob -import kotlinx.coroutines.delay -import kotlinx.coroutines.flow.MutableStateFlow -import kotlinx.coroutines.flow.asFlow +import kotlinx.coroutines.channels.BufferOverflow +import kotlinx.coroutines.flow.MutableSharedFlow +import kotlinx.coroutines.flow.collect import kotlinx.coroutines.flow.conflate -import kotlinx.coroutines.flow.filterNotNull -import kotlinx.coroutines.flow.flatMapMerge -import kotlinx.coroutines.flow.launchIn -import kotlinx.coroutines.flow.mapLatest -import kotlinx.coroutines.flow.onEach +import kotlinx.coroutines.flow.debounce +import kotlinx.coroutines.flow.distinctUntilChanged +import kotlinx.coroutines.flow.distinctUntilChangedBy +import kotlinx.coroutines.flow.map import kotlinx.coroutines.launch import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.supervisorScope import org.apache.kafka.streams.kstream.KTable @@ -36,106 +35,84 @@ private val cScope = CoroutineScope( ) /** A bridge between Kafka and Kotlin */ -private val serverMapDataFlow: MutableStateFlow = MutableStateFlow(null) +private val chunkedTiles: MutableSharedFlow> = + MutableSharedFlow( + 0, + 1, + BufferOverflow.SUSPEND + ) fun saveMapTiles( - serverMapTable: KTable + serverMapTable: KTable> ) { serverMapTable .toStream() .foreach { _, value -> - runBlocking(Dispatchers.Default) { - serverMapDataFlow.emit(value) + if (value != null) { + runBlocking(Dispatchers.Default) { + println("emitting chunkedTiles ${value.chunkId} tiles: ${value.map.size}") + chunkedTiles.emit(value) + } } } cScope.launch { - println("launching serverMapDataFlow merging") - - - serverMapDataFlow - .filterNotNull() - - serverMapDataFlow - .filterNotNull() - .conflate() -// .transform { serverMapData -> - .flatMapMerge(10) { serverMapData -> - println("flatMapMerge serverMapData ${serverMapData.serverId}, surfaces ${serverMapData.surfaces.keys}") - - val prototypes = serverMapData.tilePrototypes - - serverMapData - .surfaces - .values -// .forEach { surfaceData -> - .asFlow() - .flatMapMerge(10) { surfaceData -> - println("flatMapMerge surfaceData ${surfaceData.index}") - - // get all tile colours for the surface - val surfaceTileColours = surfaceData.tiles - .entries - .mapNotNull { (tilePosition, prototypeName) -> - prototypes[prototypeName]?.let { prototype -> - tilePosition to prototype.mapColour - } - } - println("surfaceTileColours.size ${surfaceTileColours.size}") - - // split a surface into chunks, per zoom-level - ZoomLevel.values -// .forEach { zoom -> - .asFlow() - .flatMapMerge(10) { zoom -> - println("flatMapMerge zoom $zoom") - - surfaceTileColours - .groupBy { (tilePosition, _) -> - tilePosition.toMapChunkPosition(zoom.chunkSize) - } - .entries -// .forEach { (chunkPosition, tileColours) -> - .asFlow() - .mapLatest { (chunkPosition, tileColours) -> - println("mapLatest chunkPosition $chunkPosition") - - // mapLatest + delay = an alternative to debounce-per-key? - delay(30.seconds) - - println("emitting ChunkData $chunkPosition") - - ChunkData( - surfaceData.index, - zoom, - chunkPosition, - tileColours.toMap() - ) - } - } - } - } - .onEach { chunkData -> - - println("starting to save chunkData ${chunkData.chunkPosition}...") - - val chunkImageFile = File(chunkData.filename.value) - if (chunkImageFile.parentFile.mkdirs()) { - println("created new map tile parentFile directory ${chunkImageFile.absolutePath}") + supervisorScope { + chunkedTiles + .conflate() + .distinctUntilChanged() + .debounce(5.seconds) + .map { + println("createMapTileImage ${it.chunkId}") + createMapTileImage(it) } - - val image = - createMapTileImage(chunkData.chunkPosition, chunkData.tileColours, chunkData.zoomLevel) - - println("saving map tile $chunkImageFile") - val savedTile = image.output(PngWriter.NoCompression, chunkImageFile) - println("savedTile: $savedTile") - - } - .launchIn(this) + .collect() + } } + +// cScope.launch { +// println("launching serverMapDataFlow merging") +// +// chunkedTiles +// .runningFold( +// mapOf>>() +// ) { accumulator, value -> +// +// println("folding chunkedTiles ${value.chunkId}, current flows: ${accumulator.keys}") +// +// val chunkFlow = accumulator.getOrElse(value.chunkId) { +// +// println("creating new MutableSharedFlow for ${value.chunkId}") +// +// val input = MutableSharedFlow>( +// replay = 0, +// extraBufferCapacity = 1, +// onBufferOverflow = BufferOverflow.DROP_OLDEST, +// ) +// +// supervisorScope { +// println("launching MutableSharedFlow for ${value.chunkId}") +// input +// .debounce(30.seconds) +// .mapLatest { +// println("createMapTileImage ${it.chunkId}") +// createMapTileImage(it) +// }.launchIn(this) +// } +// +// input +// } +// +// chunkFlow.emit(value) +// +// accumulator + (value.chunkId to chunkFlow) +// } +// .collect() +// +// println("end of cScope.launch") +// } } @@ -145,35 +122,58 @@ private value class TilePngFilename( ) { constructor( - surfaceIndex: SurfaceIndex, - zoom: Int, - chunkX: Int, - chunkY: Int, - ) : this("src/main/resources/kafkatorio-web-map/s${surfaceIndex.index}/z$zoom/x${chunkX}/y${chunkY}.png") - -} - - -private data class ChunkData( - val surfaceIndex: SurfaceIndex, - val zoomLevel: ZoomLevel, - val chunkPosition: MapChunkPosition, - val tileColours: Map, -) { - val filename = TilePngFilename(surfaceIndex, zoomLevel.level, chunkPosition.x, chunkPosition.y) + chunkId: ServerMapChunkId + ) : this(buildString { + append("src/main/resources/kafkatorio-web-map/s") + append(chunkId.surfaceIndex.index) + append("/z") + append(chunkId.zoomLevel.level) + append("/x") + append(chunkId.chunkPosition.x) + append("/y") + append(chunkId.chunkPosition.y) + append(".png") + }) } +private fun createMapTileImage( + chunkTiles: ServerMapChunkTiles +) { + + println("emitting ChunkData ${chunkTiles.chunkId.chunkPosition}") + + println("starting to save chunkData ${chunkTiles.chunkId.chunkPosition}...") + + val filename = TilePngFilename(chunkTiles.chunkId) + + val chunkImageFile = File(filename.value) + if (chunkImageFile.parentFile.mkdirs()) { + println("created new map tile parentFile directory ${chunkImageFile.absolutePath}") + } + + val image = + createMapTileImage( + chunkTiles.chunkId.chunkPosition, + chunkTiles.map, + chunkTiles.chunkId.zoomLevel + ) + + println("saving map tile $chunkImageFile") + val savedTile = image.output(PngWriter.NoCompression, chunkImageFile) + println("savedTile: $savedTile") +} + + private fun createMapTileImage( chunkPosition: MapChunkPosition, - chunkColours: Map, + chunkColours: Map, z: ZoomLevel ): ImmutableImage { - val chunkImage = ImmutableImage.filled( z.chunkSize, z.chunkSize, - COLOUR_TRANSPARENT.awt(), + ColourHex.TRANSPARENT.toRgbColor().awt(), BufferedImage.TYPE_INT_ARGB ) diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/model.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/model.kt index 4f67bc5..3bae3f1 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/model.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/model.kt @@ -2,6 +2,7 @@ package dev.adamko.kafkatorio.processor.topology import com.sksamuel.scrimage.color.RGBColor import dev.adamko.kafkatorio.events.schema.Colour +import dev.adamko.kafkatorio.events.schema.ColourHex import kotlin.math.pow import kotlin.math.roundToInt import kotlinx.serialization.Serializable @@ -23,12 +24,12 @@ value class FactorioServerId(val id: String) { } -fun Colour.toRgbColor(): RGBColor { +fun ColourHex.toRgbColor(): RGBColor { return RGBColor( - red.roundToInt(), - green.roundToInt(), - blue.roundToInt(), - alpha.roundToInt(), + red.toInt(), + green.toInt(), + blue.toInt(), + alpha.toInt(), ) } @@ -36,15 +37,14 @@ fun Colour.toRgbColor(): RGBColor { enum class ZoomLevel( val level: Int ) { - ZOOM_0(0), - ZOOM_1(1), - ZOOM_2(2), - ZOOM_3(3), - ZOOM_4(4), - ZOOM_5(5), + ZOOM_0(0), // 512 + ZOOM_1(1), // 256 + ZOOM_2(2), // 128 + ZOOM_3(3), // 64 + ZOOM_4(4), // 32 ; - val chunkSize: Int = 2f.pow(10 - level).roundToInt() + val chunkSize: Int = 2f.pow(9 - level).roundToInt() init { require(chunkSize > 0) { "chunkSize must be positive" } diff --git 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 index 30b8839..ac19bbd 100644 --- 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 @@ -1,19 +1,17 @@ package dev.adamko.kafkatorio.processor.topology -import com.sksamuel.scrimage.color.RGBColor -import dev.adamko.kafkatorio.events.schema.Colour +import dev.adamko.kafkatorio.events.schema.ColourHex import dev.adamko.kafkatorio.events.schema.FactorioPrototypes import dev.adamko.kafkatorio.events.schema.KafkatorioPacket +import dev.adamko.kafkatorio.events.schema.MapTile import dev.adamko.kafkatorio.events.schema.MapTilePrototype 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.materializedWith -import dev.adamko.kotka.extensions.streams.flatMap +import dev.adamko.kotka.extensions.materializedAs +import dev.adamko.kotka.extensions.streams.mapValues +import dev.adamko.kotka.extensions.streams.toTable import dev.adamko.kotka.kxs.serde import kotlinx.serialization.Serializable -import org.apache.kafka.common.serialization.Serdes -import org.apache.kafka.streams.StreamsBuilder +import org.apache.kafka.streams.kstream.KStream import org.apache.kafka.streams.kstream.KTable @@ -24,50 +22,45 @@ value class PrototypeName(val name: String) { } -//fun prototypesTable(builder: StreamsBuilder): KTable { -// return builder.stream( -// "kafkatorio.${KafkatorioPacket.PacketType.PROTOTYPES}.all", -// consumedAs( -// "consume-all-prototypes-packets", -// jsonMapper.serde(), -// jsonMapper.serde() -// ) -// ) -// .flatMap("map-MapTilePrototype") { _, prototypes: FactorioPrototypes -> -// prototypes.prototypes -// .filterIsInstance() -// .map { PrototypeName(it.name) to it } -// } -// .toTable( -// materializedWith( -// kxsBinary.serde(), -// kxsBinary.serde() -// ) -// ) -//} +@JvmInline +@Serializable +value class TileProtoHashCode(val code: Int) { + constructor(prototype: MapTilePrototype) : this(prototype.name.hashCode()) + constructor(tile: MapTile) : this(tile.prototypeName.hashCode()) +} -/** Get the [Colour] of each [MapTilePrototype] */ -fun tilePrototypeColourTable(builder: StreamsBuilder): KTable { - return builder.stream( - "kafkatorio.${KafkatorioPacket.PacketType.PROTOTYPES}.all", - consumedAs( - "consume-all-prototypes-packets", - jsonMapper.serde(), - jsonMapper.serde() - ) - ) - .flatMap("map-MapTilePrototype") { _, prototypes: FactorioPrototypes -> - prototypes.prototypes - .filterIsInstance() - .map { - PrototypeName(it.name) to it.mapColour - } +@JvmInline +@Serializable +value class TileColourDict(val map: Map) { + operator fun plus(other: TileColourDict) = TileColourDict(this.map + other.map) +} + + +/** Aggregate [KafkatorioPacket]s into a single [FactorioServerMap] (per [serverId][FactorioServerId]). */ +fun tileProtoColourDictionary( + factorioServerPacketStream: KStream +): KTable { + + // get all the prototypes + return factorioServerPacketStream + .mapValues("server-map-data.tile-prototypes.mapValues") { _, packet -> + val map = when (packet) { + is FactorioPrototypes -> packet + .prototypes + .filterIsInstance() + .associate { TileProtoHashCode(it) to it.mapColour.toHex() } + else -> mapOf() + } + TileColourDict(map) } .toTable( - materializedWith( - kxsBinary.serde(), - kxsBinary.serde() + "server-map-data.tile-prototypes", + materializedAs( + "server-map-data.tile-prototypes.store", + jsonMapper.serde(), + jsonMapper.serde() ) ) + } diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/webMapTiles.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/webMapTiles.kt index e5f91c8..2e609a7 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/webMapTiles.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/webMapTiles.kt @@ -13,89 +13,89 @@ import kotlin.math.abs import kotlin.math.roundToInt import org.apache.kafka.streams.kstream.KTable - -val COLOUR_TRANSPARENT = RGBColor.fromAwt(Color(0f, 0f, 0f, 0f)) - -fun saveTileImages( - webMapTiles: KTable -) { - - webMapTiles - .toStream("get-webmap-tile-updates") - .foreach { position: WebMapTileChunkPosition, pixels: WebMapTileChunkPixels -> - runCatching { - saveMapTilesPng(position, pixels.pixels) - }.onFailure { e -> - println("error saving map tile png chunk: $position") - e.printStackTrace() - throw e - } - } -} - -//val saveMapTilesContext = Dispatchers.IO.limitedParallelism(1) - - -private fun saveMapTilesPng( - chunkPosition: WebMapTileChunkPosition, - pixels: Map, -) { - val chunkOriginX: Int = chunkPosition.x * chunkPosition.chunkSize // - 1 - val chunkOriginY: Int = chunkPosition.y * chunkPosition.chunkSize // - 1 - val surfaceIndex: Int = chunkPosition.surfaceIndex.index - - val chunkImage = - ImmutableImage.filled( - chunkPosition.chunkSize, - chunkPosition.chunkSize, - COLOUR_TRANSPARENT.awt(), - BufferedImage.TYPE_INT_ARGB - ) - - pixels.forEach { (tilePosition, colour) -> - - val rgbColour = colour.toRgbColor() - - val pixelX = abs(abs(tilePosition.x) - abs(chunkOriginX)) - val pixelY = abs(abs(tilePosition.y) - abs(chunkOriginY)) - - chunkImage.setColor( - pixelX, - pixelY, - rgbColour - ) - - } - - val zoom = 1u - val file = - File(tileFilename("$surfaceIndex", "$zoom", "${chunkPosition.x}", "${chunkPosition.y}")) - - if (file.parentFile.mkdirs()) { - println("created new map tile parentFile directory ${file.absolutePath}") - } - - println("saving map tile $file") - val savedTile = chunkImage.output(PngWriter.NoCompression, file) - println("savedTile: $savedTile") -} - -fun tileFilename( - surfaceIndex: String, - zoom: String, - chunkX: String, - chunkY: String, -) = "src/main/resources/kafkatorio-web-map/s${surfaceIndex}/z$zoom/x${chunkX}/y${chunkY}.png" - -//fun subdivide(tile: ImmutableImage, size: Int) { // -// val halfSize = size / 2 +//val COLOUR_TRANSPARENT: RGBColor = RGBColor.fromAwt(Color(0f, 0f, 0f, 0f)) // -// val bl = tile.resizeTo(halfSize, halfSize, Position.BottomLeft) -// val savedTile = bl.output(PngWriter.NoCompression, file) -// -// tile.resizeTo(halfSize, halfSize, Position.BottomRight) -// tile.resizeTo(halfSize, halfSize, Position.TopLeft) -// tile.resizeTo(halfSize, halfSize, Position.TopRight) +//fun saveTileImages( +// webMapTiles: KTable +//) { // +// webMapTiles +// .toStream("get-webmap-tile-updates") +// .foreach { position: WebMapTileChunkPosition, pixels: WebMapTileChunkPixels -> +// runCatching { +// saveMapTilesPng(position, pixels.pixels) +// }.onFailure { e -> +// println("error saving map tile png chunk: $position") +// e.printStackTrace() +// throw e +// } +// } //} +// +////val saveMapTilesContext = Dispatchers.IO.limitedParallelism(1) +// +// +//private fun saveMapTilesPng( +// chunkPosition: WebMapTileChunkPosition, +// pixels: Map, +//) { +// val chunkOriginX: Int = chunkPosition.x * chunkPosition.chunkSize // - 1 +// val chunkOriginY: Int = chunkPosition.y * chunkPosition.chunkSize // - 1 +// val surfaceIndex: Int = chunkPosition.surfaceIndex.index +// +// val chunkImage = +// ImmutableImage.filled( +// chunkPosition.chunkSize, +// chunkPosition.chunkSize, +// COLOUR_TRANSPARENT.awt(), +// BufferedImage.TYPE_INT_ARGB +// ) +// +// pixels.forEach { (tilePosition, colour) -> +// +// val rgbColour = colour.toRgbColor() +// +// val pixelX = abs(abs(tilePosition.x) - abs(chunkOriginX)) +// val pixelY = abs(abs(tilePosition.y) - abs(chunkOriginY)) +// +// chunkImage.setColor( +// pixelX, +// pixelY, +// rgbColour +// ) +// +// } +// +// val zoom = 1u +// val file = +// File(tileFilename("$surfaceIndex", "$zoom", "${chunkPosition.x}", "${chunkPosition.y}")) +// +// if (file.parentFile.mkdirs()) { +// println("created new map tile parentFile directory ${file.absolutePath}") +// } +// +// println("saving map tile $file") +// val savedTile = chunkImage.output(PngWriter.NoCompression, file) +// println("savedTile: $savedTile") +//} +// +//fun tileFilename( +// surfaceIndex: String, +// zoom: String, +// chunkX: String, +// chunkY: String, +//) = "src/main/resources/kafkatorio-web-map/s${surfaceIndex}/z$zoom/x${chunkX}/y${chunkY}.png" +// +////fun subdivide(tile: ImmutableImage, size: Int) { +//// +//// val halfSize = size / 2 +//// +//// val bl = tile.resizeTo(halfSize, halfSize, Position.BottomLeft) +//// val savedTile = bl.output(PngWriter.NoCompression, file) +//// +//// tile.resizeTo(halfSize, halfSize, Position.BottomRight) +//// tile.resizeTo(halfSize, halfSize, Position.TopLeft) +//// tile.resizeTo(halfSize, halfSize, Position.TopRight) +//// +////} diff --git a/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/misc/serdeSize.kt b/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/misc/serdeSize.kt new file mode 100644 index 0000000..941cc09 --- /dev/null +++ b/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/misc/serdeSize.kt @@ -0,0 +1,62 @@ +package dev.adamko.kafkatorio.processor.misc + +import dev.adamko.kafkatorio.events.schema.ColourHex +import dev.adamko.kafkatorio.events.schema.MapChunkPosition +import dev.adamko.kafkatorio.events.schema.MapTilePosition +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.SurfaceIndex +import dev.adamko.kafkatorio.processor.topology.ZoomLevel +import java.util.UUID +import kotlinx.serialization.encodeToByteArray +import org.junit.jupiter.api.Test + +class Blah { + + @Test + fun asdad() { + + val tiles = ServerMapChunkTiles( + chunkId = ServerMapChunkId( + serverId = FactorioServerId(UUID.randomUUID().toString()), + chunkPosition = MapChunkPosition(Int.MAX_VALUE, Int.MAX_VALUE), + surfaceIndex = SurfaceIndex(Int.MAX_VALUE), + zoomLevel = ZoomLevel.ZOOM_0 + ), + buildMap { + repeat(512) { x -> + repeat(512) { y -> + put(MapTilePosition(x, y), ColourHex(255u, 255u, 255u, 255u)) + } + } + } + ) + + val encode = kxsBinary.encodeToByteArray(tiles) + + encode.inputStream() + val size = encode.size * Byte.SIZE_BYTES + + println(size) + + + val map2 = buildList { + repeat(512) { x -> + repeat(512) { y -> + val element = (x to y) to Int.MAX_VALUE + add(element) + } + } + } +// .toTypedArray() + val encode2 = kxsBinary.encodeToByteArray(map2) + + encode2.inputStream() + val size2 = encode2.size * Byte.SIZE_BYTES + + println(size2) + + } +} diff --git a/modules/infra-factorio-server/src/.env b/modules/infra-factorio-server/src/.env index aecb9bc..82dbe34 100644 --- a/modules/infra-factorio-server/src/.env +++ b/modules/infra-factorio-server/src/.env @@ -1,4 +1,4 @@ # Do not edit manually. This file was created with task 'dockerEnv' COMPOSE_PROJECT_NAME=kafkatorio FACTORIO_VERSION=1.1.53 -KAFKATORIO_VERSION=0.2.7 +KAFKATORIO_VERSION=0.2.8 diff --git a/modules/infra-kafka-cluster/src/.env b/modules/infra-kafka-cluster/src/.env index 1ebae11..8153f49 100644 --- a/modules/infra-kafka-cluster/src/.env +++ b/modules/infra-kafka-cluster/src/.env @@ -1,3 +1,3 @@ # Do not edit manually. This file was created with task 'dockerEnv' COMPOSE_PROJECT_NAME=kafkatorio -KAFKATORIO_VERSION=0.2.7 +KAFKATORIO_VERSION=0.2.8 diff --git a/modules/infra-kafka-cluster/src/kafka-server.properties b/modules/infra-kafka-cluster/src/kafka-server.properties index 3f7c92c..6c1f2d1 100644 --- a/modules/infra-kafka-cluster/src/kafka-server.properties +++ b/modules/infra-kafka-cluster/src/kafka-server.properties @@ -120,3 +120,8 @@ log.retention.check.interval.ms=300000 ### Compression ### #compression.type=snappy #linger.ms=5 + + +### record size ### +message.max.bytes=10485760 +replica.fetch.max.bytes=10485760 diff --git a/modules/infra-kafka-cluster/src/kafka.Dockerfile b/modules/infra-kafka-cluster/src/kafka.Dockerfile index dca9f09..9dd574d 100644 --- a/modules/infra-kafka-cluster/src/kafka.Dockerfile +++ b/modules/infra-kafka-cluster/src/kafka.Dockerfile @@ -1,4 +1,4 @@ -ARG KAFKA_DL_URL=https://dlcdn.apache.org/kafka/3.0.0/kafka_2.13-3.0.0.tgz +ARG KAFKA_DL_URL=https://dlcdn.apache.org/kafka/3.1.0/kafka_2.13-3.1.0.tgz ## builder ## FROM debian:11-slim as Kafka-Download diff --git a/modules/web-map/build.gradle.kts b/modules/web-map/build.gradle.kts index a7128f7..5c9cbf9 100644 --- a/modules/web-map/build.gradle.kts +++ b/modules/web-map/build.gradle.kts @@ -4,7 +4,7 @@ import org.jetbrains.kotlin.gradle.targets.js.webpack.KotlinWebpackConfig plugins { id("dev.adamko.kafkatorio.lang.kotlin-js") kotlin("plugin.serialization") - id("io.kvision") version "5.7.1" + id("io.kvision") version "5.8.1" } kotlin { @@ -48,8 +48,8 @@ kotlin { implementation(projects.modules.eventsSchema) - implementation("org.jetbrains.kotlinx:kotlinx-nodejs:0.0.7") - implementation("org.jetbrains.kotlinx:kotlinx-html:0.7.3") + implementation(libs.kotlinx.nodejs) + implementation(libs.kotlinx.html) kvision("kvision") kvision("kvision-bootstrap")