From f2ccbbd456cd4a7e22307cac589d86400419235b Mon Sep 17 00:00:00 2001 From: Adam <897017+aSemy@users.noreply.github.com> Date: Fri, 21 Jan 2022 18:31:15 +0100 Subject: [PATCH] - factorio version bump - mod: add a couple of commands for emitting chunks - mod: decrease on_tick event rate - mod: version bump typed-factorio, tstl - kafka admin: decrease replication - kafka admin: filter out existing-topics when creating - events server: add more serdes, hopefully more performant than json - events server: lots of work trying to improve performance... - events server: increase tile size to 256 - events server: fix web-map-tile-image x/y location - kafka: version bump - kafka: increase default partition size - kafka ui: version bump - --- build.gradle.kts | 2 +- gradle/libs.versions.toml | 2 +- .../main/commands/kafkatorioCommands.ts | 58 +++++++- .../main/events/registerHandlers.ts | 2 +- .../src/main/typescript/package-lock.json | 42 +++--- .../src/main/typescript/package.json | 6 +- modules/events-server/build.gradle.kts | 1 + .../adamko/kafkatorio/admin/CreateTopics.kt | 24 +++- .../processor/KafkatorioTopology.kt | 4 +- .../kafkatorio/processor/serdes/kxsBinary.kt | 105 +++++++++++++++ .../{serdes.kt => serdes/kxsJson.kt} | 2 +- .../kafkatorio/processor/serdes/protobuf.kt | 59 +++++++++ .../processor/topology/FactorioServerState.kt | 124 +++++++++--------- .../processor/topology/factorioMapTiles.kt | 66 +++++----- .../topology/factorioServerPackets.kt | 56 +++++++- .../kafkatorio/processor/topology/model.kt | 7 +- .../processor/topology/playerUpdates.kt | 4 +- .../processor/topology/prototypes.kt | 4 +- .../processor/topology/webMapTileAggregate.kt | 39 +++--- .../processor/topology/webMapTiles.kt | 14 +- modules/infra-factorio-server/src/.env | 4 +- modules/infra-kafka-cluster/src/.env | 2 +- .../src/docker-compose.yml | 4 +- .../src/kafka-server.properties | 23 +--- modules/infra-kafka-pipe/src/.env | 2 +- .../dev/adamko/kafkatorio/webmap/map.kt | 5 +- .../adamko/kafkatorio/webmap/websockets.kt | 22 ++-- 27 files changed, 476 insertions(+), 207 deletions(-) create mode 100644 modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes/kxsBinary.kt rename modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/{serdes.kt => serdes/kxsJson.kt} (96%) create mode 100644 modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes/protobuf.kt diff --git a/build.gradle.kts b/build.gradle.kts index c12c822..f23f144 100644 --- a/build.gradle.kts +++ b/build.gradle.kts @@ -6,7 +6,7 @@ plugins { } group = "dev.adamko.kafkatorio" -version = "0.2.4" +version = "0.2.6" val licenseFile: RegularFile by extra(layout.projectDirectory.file("LICENSE")) diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index a9e448b..7dd2fe9 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -23,7 +23,7 @@ junit = "5.8.2" kotest = "5.0.1" ## infra ## -factorio = "1.1.50" +factorio = "1.1.53" confluent-kafkaConnect = "7.0.1" ######### diff --git a/modules/events-mod/src/main/typescript/main/commands/kafkatorioCommands.ts b/modules/events-mod/src/main/typescript/main/commands/kafkatorioCommands.ts index 8087f98..20b692a 100644 --- a/modules/events-mod/src/main/typescript/main/commands/kafkatorioCommands.ts +++ b/modules/events-mod/src/main/typescript/main/commands/kafkatorioCommands.ts @@ -1,4 +1,6 @@ import {emitPrototypes} from "../config-update/prototypes"; +import {handleChunkUpdate} from "../events/handlers"; +import floor = math.floor; commands.add_command( "kafkatorio", @@ -6,7 +8,59 @@ commands.add_command( (e: CustomCommandData) => { if (e.parameter == undefined) { // do nothing - } else if ("PROTOTYPES" == e.parameter.toUpperCase()) { + } else if ("PROTOTYPES" == e.parameter.toUpperCase()) { emitPrototypes() + } else if ("CURRENT_CHUNK" == e.parameter.toUpperCase()) { + + if (e.player_index != undefined) { + + let player = game.players[e.player_index] + + let chunkPosition: ChunkPosition = { + x: floor(player.position.x / 32), + y: floor(player.position.y / 32), + } + + let chunkLeftTop: Position = { + x: chunkPosition.x * 32, + y: chunkPosition.y * 32, + } + let chunkBottomRight: Position = { + x: chunkLeftTop.x + 32 - 1, + y: chunkLeftTop.y + 32 - 1, + } + let chunkArea: BoundingBoxRead = { + left_top: chunkLeftTop, + right_bottom: chunkBottomRight + } + + handleChunkUpdate( + e.tick, + "player-command", + player.surface.index, + chunkPosition, + chunkArea, + ) + } + } else if ("CHUNKS" == e.parameter.toUpperCase()) { + + for (let [, surface] of game.surfaces) { + for (let chunk of surface.get_chunks()) { + if ( + (chunk.x >= -16 || chunk.x <= 16) + && + (chunk.y >= -16 || chunk.y <= 16) + ) { + handleChunkUpdate( + e.tick, + "player-command", + surface.index, + chunk, + chunk.area, + ) + } + } + } } - }) + } +) diff --git a/modules/events-mod/src/main/typescript/main/events/registerHandlers.ts b/modules/events-mod/src/main/typescript/main/events/registerHandlers.ts index 26f2e8f..4468bac 100644 --- a/modules/events-mod/src/main/typescript/main/events/registerHandlers.ts +++ b/modules/events-mod/src/main/typescript/main/events/registerHandlers.ts @@ -48,7 +48,7 @@ script.on_event( script.on_event( defines.events.on_tick, (e: OnTickEvent) => { - if (e.tick % 60 == 0) { + if (e.tick % 1000 == 0) { for (const [, surface] of pairs(game.surfaces)) { handleSurfaceUpdate(e.tick, mapEventIdToName.get(e.name), surface) } diff --git a/modules/events-mod/src/main/typescript/package-lock.json b/modules/events-mod/src/main/typescript/package-lock.json index 29a486e..5d7dc4f 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.4", + "version": "0.2.6", "lockfileVersion": 1, "requires": true, "dependencies": { @@ -24,9 +24,9 @@ "integrity": "sha512-yIovAzMX49sF8Yl58fSCWJ5svSLuaibPxXQJFLmBObTuCr0Mf1KiPopGM9NiFjiYBCbfaa2Fh6breQ6ANVTI0A==" }, "graceful-fs": { - "version": "4.2.8", - "resolved": "https://registry.npmjs.org/graceful-fs/-/graceful-fs-4.2.8.tgz", - "integrity": "sha512-qkIilPUYcNhJpd33n0GBXTB1MMPp14TxEsEs0pTrsSVucApsYzW5V+Q8Qxhik6KU3evy+qkAAowTByymK0avdg==" + "version": "4.2.9", + "resolved": "https://registry.npmjs.org/graceful-fs/-/graceful-fs-4.2.9.tgz", + "integrity": "sha512-NtNxqUcXgpW2iMrfqSfR73Glt39K+BLwWsPs94yR63v45T0Wbej7eRmL5cWfwEgqXnmjQp3zaJTshdRW/qC2ZQ==" }, "has": { "version": "1.0.3", @@ -37,9 +37,9 @@ } }, "is-core-module": { - "version": "2.8.0", - "resolved": "https://registry.npmjs.org/is-core-module/-/is-core-module-2.8.0.tgz", - "integrity": "sha512-vd15qHsaqrRL7dtH6QNuy0ndJmRDrS9HAM1CAiSifNUFv4x1a0CCVsj18hJ1mShxIG6T2i1sO78MkP56r0nYRw==", + "version": "2.8.1", + "resolved": "https://registry.npmjs.org/is-core-module/-/is-core-module-2.8.1.tgz", + "integrity": "sha512-SdNCUs284hr40hFTFP6l0IfZ/RSrMXF3qgoRHd3/79unUTvrFO/JoXwkGm+5J/Oe3E/b5GsnG330uUNgRpu1PA==", "requires": { "has": "^1.0.3" } @@ -55,12 +55,13 @@ "integrity": "sha512-LDJzPVEEEPR+y48z93A0Ed0yXb8pAByGWo/k5YYdYgpY2/2EsOsksJrq7lOHxryrVOn1ejG6oAp8ahvOIQD8sw==" }, "resolve": { - "version": "1.20.0", - "resolved": "https://registry.npmjs.org/resolve/-/resolve-1.20.0.tgz", - "integrity": "sha512-wENBPt4ySzg4ybFQW2TT1zMQucPK95HSh/nq2CFTZVOGut2+pQvSsgtda4d26YrYcr067wjbmzOG8byDPBX63A==", + "version": "1.21.1", + "resolved": "https://registry.npmjs.org/resolve/-/resolve-1.21.1.tgz", + "integrity": "sha512-lfEImVbnolPuaSZuLQ52cAxPBHeI77sPwCOWRdy12UG/CNa8an7oBHH1R+Fp1/mUqSJi4c8TIP6FOIPSZAUrEQ==", "requires": { - "is-core-module": "^2.2.0", - "path-parse": "^1.0.6" + "is-core-module": "^2.8.0", + "path-parse": "^1.0.7", + "supports-preserve-symlinks-flag": "^1.0.0" } }, "source-map": { @@ -68,15 +69,20 @@ "resolved": "https://registry.npmjs.org/source-map/-/source-map-0.7.3.tgz", "integrity": "sha512-CkCj6giN3S+n9qrYiBTX5gystlENnRW5jZeNLHpe6aue+SrHcG5VYwujhW9s4dY31mEGsxBDrHR6oI69fTXsaQ==" }, + "supports-preserve-symlinks-flag": { + "version": "1.0.0", + "resolved": "https://registry.npmjs.org/supports-preserve-symlinks-flag/-/supports-preserve-symlinks-flag-1.0.0.tgz", + "integrity": "sha512-ot0WnXS9fgdkgIcePe6RHNk1WA8+muPa6cSjeR3V8K27q9BB1rTE3R1p7Hv0z1ZyAc8s6Vvv8DIyWf681MAt0w==" + }, "tapable": { "version": "2.2.1", "resolved": "https://registry.npmjs.org/tapable/-/tapable-2.2.1.tgz", "integrity": "sha512-GNzQvQTOIP6RyTfE2Qxb8ZVlNmw0n88vp1szwWRimP02mnTsx3Wtn5qRdqY9w2XduFNUgvOwhNnQsjwCp+kqaQ==" }, "typed-factorio": { - "version": "0.16.0", - "resolved": "https://registry.npmjs.org/typed-factorio/-/typed-factorio-0.16.0.tgz", - "integrity": "sha512-I6W1lhRkeE+mHzLBKBVIDZdquwBjBYaS7jkV0CKiWWkA9ptGqT+MblOkNUdYZ0ykR2Rvn1eF/6pzh8WwQ+mYVQ==" + "version": "0.17.0", + "resolved": "https://registry.npmjs.org/typed-factorio/-/typed-factorio-0.17.0.tgz", + "integrity": "sha512-sFLQn+G/X0IROUBnlPwuIKQlvU561dckdI59WnRaClDRK1o8zyUEJRuJ4R+IJovKUd+I75xuoSdYVIWEwZoeHA==" }, "typescript": { "version": "4.5.4", @@ -84,9 +90,9 @@ "integrity": "sha512-VgYs2A2QIRuGphtzFV7aQJduJ2gyfTljngLzjpfW9FoYZF6xuw1W0vW9ghCKLfcWrCFxK81CSGRAvS1pn4fIUg==" }, "typescript-to-lua": { - "version": "1.3.0", - "resolved": "https://registry.npmjs.org/typescript-to-lua/-/typescript-to-lua-1.3.0.tgz", - "integrity": "sha512-hFjZ7+Cv08WUql2VJChOEb1JH2nqYspyjw+9Og0cDLkP0QaEmM+Eh/slO2exdngDyQye/rrQysLy3zVzJe+PVg==", + "version": "1.3.3", + "resolved": "https://registry.npmjs.org/typescript-to-lua/-/typescript-to-lua-1.3.3.tgz", + "integrity": "sha512-Fkt0+dLufN5CK0G/qpINY7Sb0+SNmWpov+PHhO/dq0HFoNBzR47M1SJd+hQbG92vTVFLBO4f+Np29LuVvoemMA==", "requires": { "enhanced-resolve": "^5.8.2", "resolve": "^1.15.1", diff --git a/modules/events-mod/src/main/typescript/package.json b/modules/events-mod/src/main/typescript/package.json index 7d2945b..b4993c8 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.4", + "version": "0.2.6", "private": true, "scripts": { "build": "tstl", @@ -13,8 +13,8 @@ "dependencies": { "lua-types": "^2.11.0", "@types/node": "^12.0.8", - "typescript-to-lua": "1.3.0", - "typed-factorio": "0.16.0", + "typescript-to-lua": "1.3.3", + "typed-factorio": "0.17.0", "typescript": "4.5.4" }, "license": "UNLICENSED", diff --git a/modules/events-server/build.gradle.kts b/modules/events-server/build.gradle.kts index 2c6cb9f..491230d 100644 --- a/modules/events-server/build.gradle.kts +++ b/modules/events-server/build.gradle.kts @@ -26,6 +26,7 @@ dependencies { implementation(platform(libs.kotlinx.serialization.bom)) implementation(libs.kotlinx.serialization.core) implementation(libs.kotlinx.serialization.json) + implementation("org.jetbrains.kotlinx:kotlinx-serialization-protobuf") implementation(libs.kotlinx.coroutines) diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/admin/CreateTopics.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/admin/CreateTopics.kt index 845a7c3..1152dcc 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/admin/CreateTopics.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/admin/CreateTopics.kt @@ -2,11 +2,13 @@ package dev.adamko.kafkatorio.admin import dev.adamko.kafkatorio.events.schema.FactorioObjectData import dev.adamko.kafkatorio.events.schema.KafkatorioPacket +import dev.adamko.kafkatorio.processor.KafkatorioTopology import java.util.concurrent.TimeUnit import mu.KotlinLogging import org.apache.kafka.clients.admin.Admin import org.apache.kafka.clients.admin.CreateTopicsResult import org.apache.kafka.clients.admin.NewTopic +import org.apache.kafka.clients.admin.TopicListing private val logger = KotlinLogging.logger {} @@ -29,7 +31,7 @@ object CreateTopics { KafkatorioPacket.PacketType.values().forEach { packetType -> val kafkatorioTopics = buildSet { -// add(KafkatorioTopology.sourceTopic) + add(KafkatorioTopology.sourceTopic) when (packetType) { KafkatorioPacket.PacketType.EVENT -> FactorioObjectData.ObjectName.values().forEach { objectName -> @@ -41,7 +43,11 @@ object CreateTopics { add("kafkatorio.${packetType.name}.all") } } + + val currentTopics = currentTopics().map { it.name() } + val result = kafkatorioTopics + .filterNot { it in currentTopics } .run { logger.info("Creating $size topics") kafkaAdmin.createTopics { this } @@ -68,10 +74,18 @@ object CreateTopics { private fun Admin.createTopics( numPartitions: Int = 10, - replicationFactor: Short = 5, + replicationFactor: Short = 1, topicNames: () -> Collection, ): CreateTopicsResult = - createTopics(topicNames().distinct().map { - NewTopic(it, numPartitions, replicationFactor) - }) + createTopics(topicNames() + .distinct() + .map { + NewTopic(it, numPartitions, replicationFactor) + }) + + private fun currentTopics(): MutableCollection { + return kafkaAdmin.listTopics() + .listings() + .get() + } } 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 96cea83..0a68437 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,9 +1,9 @@ package dev.adamko.kafkatorio.processor import dev.adamko.kafkatorio.events.schema.MapTile -import dev.adamko.kafkatorio.events.schema.MapTilePosition import dev.adamko.kafkatorio.events.schema.MapTilePrototype import dev.adamko.kafkatorio.processor.topology.PrototypeName +import dev.adamko.kafkatorio.processor.topology.TileUpdateRecordKey import dev.adamko.kafkatorio.processor.topology.WebMapTileChunkPixels import dev.adamko.kafkatorio.processor.topology.WebMapTileChunkPosition import dev.adamko.kafkatorio.processor.topology.aggregateWebMapTiles @@ -33,7 +33,7 @@ class KafkatorioTopology( splitFactorioServerLog(builder) playerUpdatesToWsServer(websocketServer, builder) - val allMapTilesTable: KTable = allMapTilesTable(builder) + val allMapTilesTable: KTable = allMapTilesTable(builder) val tilePrototypesTable: KTable = prototypesTable(builder) val webMapTiles: KTable = diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes/kxsBinary.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes/kxsBinary.kt new file mode 100644 index 0000000..0c0a66f --- /dev/null +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes/kxsBinary.kt @@ -0,0 +1,105 @@ +package dev.adamko.kafkatorio.processor.serdes + +import java.io.ByteArrayInputStream +import java.io.ByteArrayOutputStream +import java.io.DataInput +import java.io.DataInputStream +import java.io.DataOutput +import java.io.DataOutputStream +import java.util.zip.DeflaterOutputStream +import kotlinx.serialization.BinaryFormat +import kotlinx.serialization.DeserializationStrategy +import kotlinx.serialization.SerializationStrategy +import kotlinx.serialization.descriptors.SerialDescriptor +import kotlinx.serialization.encoding.AbstractDecoder +import kotlinx.serialization.encoding.AbstractEncoder +import kotlinx.serialization.encoding.CompositeDecoder +import kotlinx.serialization.encoding.CompositeEncoder +import kotlinx.serialization.modules.EmptySerializersModule +import kotlinx.serialization.modules.SerializersModule + + +val kxsBinaryMapper: KxsBinary = KxsBinary() + +/** [`https://github.com/Kotlin/kotlinx.serialization/blob/master/docs/formats.md#efficient-binary-format`](https://github.com/Kotlin/kotlinx.serialization/blob/master/docs/formats.md#efficient-binary-format) */ +class KxsBinary( + override val serializersModule: SerializersModule = EmptySerializersModule +) : BinaryFormat { + + override fun encodeToByteArray(serializer: SerializationStrategy, value: T): ByteArray { + val output = ByteArrayOutputStream() + val encoder = KxsDataOutputEncoder(DataOutputStream(output), serializersModule) + encoder.encodeSerializableValue(serializer, value) + return output.toByteArray() + } + + override fun decodeFromByteArray( + deserializer: DeserializationStrategy, + bytes: ByteArray + ): T { + val input = ByteArrayInputStream(bytes) + val decoder = KxsDataInputDecoder(DataInputStream(input), serializersModule) + return decoder.decodeSerializableValue(deserializer) + } +} + +class KxsDataOutputEncoder( + private val output: DataOutput, + override val serializersModule: SerializersModule = EmptySerializersModule, +) : AbstractEncoder() { + + override fun encodeBoolean(value: Boolean) = output.writeByte(if (value) 1 else 0) + override fun encodeByte(value: Byte) = output.writeByte(value.toInt()) + override fun encodeShort(value: Short) = output.writeShort(value.toInt()) + override fun encodeInt(value: Int) = output.writeInt(value) + override fun encodeLong(value: Long) = output.writeLong(value) + override fun encodeFloat(value: Float) = output.writeFloat(value) + override fun encodeDouble(value: Double) = output.writeDouble(value) + override fun encodeChar(value: Char) = output.writeChar(value.code) + override fun encodeString(value: String) = output.writeUTF(value) + override fun encodeEnum(enumDescriptor: SerialDescriptor, index: Int) = output.writeInt(index) + + override fun beginCollection( + descriptor: SerialDescriptor, + collectionSize: Int + ): CompositeEncoder { + encodeInt(collectionSize) + return this + } + + override fun encodeNull() = encodeBoolean(false) + override fun encodeNotNullMark() = encodeBoolean(true) +} + +class KxsDataInputDecoder( + private val input: DataInput, + override val serializersModule: SerializersModule = EmptySerializersModule, + var elementsCount: Int = 0, +) : AbstractDecoder() { + private var elementIndex = 0 + override fun decodeBoolean(): Boolean = input.readByte().toInt() != 0 + override fun decodeByte(): Byte = input.readByte() + override fun decodeShort(): Short = input.readShort() + override fun decodeInt(): Int = input.readInt() + override fun decodeLong(): Long = input.readLong() + override fun decodeFloat(): Float = input.readFloat() + override fun decodeDouble(): Double = input.readDouble() + override fun decodeChar(): Char = input.readChar() + override fun decodeString(): String = input.readUTF() + override fun decodeEnum(enumDescriptor: SerialDescriptor): Int = input.readInt() + + override fun decodeElementIndex(descriptor: SerialDescriptor): Int { + if (elementIndex == elementsCount) return CompositeDecoder.DECODE_DONE + return elementIndex++ + } + + override fun beginStructure(descriptor: SerialDescriptor): CompositeDecoder = + KxsDataInputDecoder(input, serializersModule, descriptor.elementsCount) + + override fun decodeSequentially(): Boolean = true + + override fun decodeCollectionSize(descriptor: SerialDescriptor): Int = + decodeInt().also { elementsCount = it } + + override fun decodeNotNullMark(): Boolean = decodeBoolean() +} diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes/kxsJson.kt similarity index 96% rename from modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes.kt rename to modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes/kxsJson.kt index e178d78..4cb82b9 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes/kxsJson.kt @@ -1,4 +1,4 @@ -package dev.adamko.kafkatorio.processor +package dev.adamko.kafkatorio.processor.serdes import dev.adamko.kafkatorio.events.schema.KafkatorioPacket import dev.adamko.kafkatorio.events.schema.jsonMapperKafkatorio diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes/protobuf.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes/protobuf.kt new file mode 100644 index 0000000..527121f --- /dev/null +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes/protobuf.kt @@ -0,0 +1,59 @@ +package dev.adamko.kafkatorio.processor.serdes + +import dev.adamko.kafkatorio.events.schema.jsonMapperKafkatorio +import kotlinx.serialization.BinaryFormat +import kotlinx.serialization.decodeFromByteArray +import kotlinx.serialization.encodeToByteArray +import kotlinx.serialization.protobuf.ProtoBuf +import org.apache.kafka.common.serialization.Deserializer +import org.apache.kafka.common.serialization.Serde +import org.apache.kafka.common.serialization.Serializer + +val protoMapper: ProtoBuf = ProtoBuf { + serializersModule = jsonMapperKafkatorio.serializersModule +} + + +inline fun BinaryFormat.kafkaSerializer() = + Serializer { topic: String, data: T -> + runCatching { + encodeToByteArray(data) + }.getOrElse { e -> + println( + """ + Exception on encodeToString + Topic: $topic + topicData: $data + topicData as T: $data + """.trimIndent() + ) + e.printStackTrace() + throw e + } + } + +inline fun BinaryFormat.kafkaDeserializer() = + Deserializer { topic: String, data: ByteArray -> + runCatching { + decodeFromByteArray(data) + }.getOrElse { e -> + println( + """ + Exception on decodeFromString, + Topic: $topic + topicData: $data + topicData as T: ${data as? T} + """.trimIndent() + ) + e.printStackTrace() + throw e + } + } + +inline fun ProtoBuf.serde() = object : Serde { + + override fun serializer(): Serializer = kafkaSerializer() + + override fun deserializer(): Deserializer = kafkaDeserializer() + +} diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/FactorioServerState.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/FactorioServerState.kt index 48a80b0..a2f6045 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/FactorioServerState.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/FactorioServerState.kt @@ -1,16 +1,16 @@ package dev.adamko.kafkatorio.processor.topology -import dev.adamko.kafkatorio.events.schema.KafkatorioPacket -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.processor.jsonMapper +//import dev.adamko.kafkatorio.events.schema.KafkatorioPacket +//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.processor.serdes.jsonMapper import kotlinx.serialization.Serializable -import kotlinx.serialization.decodeFromString -import org.apache.kafka.common.serialization.Serdes -import org.apache.kafka.streams.StreamsBuilder -import org.apache.kafka.streams.kstream.Consumed -import org.apache.kafka.streams.kstream.KGroupedStream +//import kotlinx.serialization.decodeFromString +//import org.apache.kafka.common.serialization.Serdes +//import org.apache.kafka.streams.StreamsBuilder +//import org.apache.kafka.streams.kstream.Consumed +//import org.apache.kafka.streams.kstream.KGroupedStream private const val sourceTopic = "factorio-server-log" @@ -20,55 +20,55 @@ value class FactorioServerId(val id: String) { override fun toString() = id } -@Serializable -data class FactorioServerState( - val serverId: FactorioServerId, - val mapTilePrototypes: Set, - val surfaceIndexes: Set, -) - - -private fun gatherGameState(builder: StreamsBuilder = StreamsBuilder()) { - - val serverPackets: KGroupedStream = - builder.stream( - sourceTopic, - Consumed.with(Serdes.String(), Serdes.String()) - ) - .mapValues { readOnlyKey, value -> - println("Mapping $readOnlyKey:$value") - jsonMapper.decodeFromString(value) - } - .groupByKey() - - @Serializable - data class TileSurfaceData( - val surfaceIndex: Int, - val position: MapTilePosition, - val prototype: MapTilePrototype, - ) - - @Serializable - data class MapChunkData( - val position: MapChunkPosition, - val tiles: List, - ) - - @Serializable - data class FactorioSurfaceData( - val surfaceIndex: Int, - val chunks: List, - ) - - @Serializable - data class FactorioServerState( - val surfaces: Map, - val tilePrototypes: List, - ) - - -// serverPackets -// .aggregate( -// Initializer { FactorioServerState("", emptyMap(), emptyList()) } -// ) -} +//@Serializable +//data class FactorioServerState( +// val serverId: FactorioServerId, +// val mapTilePrototypes: Set, +// val surfaceIndexes: Set, +//) +// +// +//private fun gatherGameState(builder: StreamsBuilder = StreamsBuilder()) { +// +// val serverPackets: KGroupedStream = +// builder.stream( +// sourceTopic, +// Consumed.with(Serdes.String(), Serdes.String()) +// ) +// .mapValues { readOnlyKey, value -> +// println("Mapping $readOnlyKey:$value") +// jsonMapper.decodeFromString(value) +// } +// .groupByKey() +// +// @Serializable +// data class TileSurfaceData( +// val surfaceIndex: Int, +// val position: MapTilePosition, +// val prototype: MapTilePrototype, +// ) +// +// @Serializable +// data class MapChunkData( +// val position: MapChunkPosition, +// val tiles: List, +// ) +// +// @Serializable +// data class FactorioSurfaceData( +// val surfaceIndex: Int, +// val chunks: List, +// ) +// +// @Serializable +// data class FactorioServerState( +// val surfaces: Map, +// val tilePrototypes: List, +// ) +// +// +//// serverPackets +//// .aggregate( +//// Initializer { FactorioServerState("", emptyMap(), emptyList()) } +//// ) +//} diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioMapTiles.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioMapTiles.kt index de01bd0..30ed815 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioMapTiles.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioMapTiles.kt @@ -7,7 +7,7 @@ import dev.adamko.kafkatorio.events.schema.MapChunk import dev.adamko.kafkatorio.events.schema.MapTile import dev.adamko.kafkatorio.events.schema.MapTilePosition import dev.adamko.kafkatorio.events.schema.MapTiles -import dev.adamko.kafkatorio.processor.jsonMapper +import dev.adamko.kafkatorio.processor.serdes.jsonMapper import dev.adamko.kotka.extensions.consumedAs import dev.adamko.kotka.extensions.materializedWith import dev.adamko.kotka.extensions.streams.filter @@ -17,12 +17,12 @@ import dev.adamko.kotka.extensions.streams.toTable import dev.adamko.kotka.kxs.serde import java.time.Duration 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 import org.apache.kafka.streams.kstream.Suppressed + @Serializable data class TileUpdateRecordKey( val surfaceIndex: SurfaceIndex, @@ -32,26 +32,42 @@ data class TileUpdateRecordKey( fun allMapTilesTable( builder: StreamsBuilder -): KTable { +): 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 luaTilesUpdatesStream: KStream = + builder.stream( + "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.LuaTiles}", + consumedAs("consume-map-tiles-packets", jsonMapper.serde(), jsonMapper.serde()) + ) + .filter("events.filter.map-tiles") { _: FactorioPacketKey, 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 } + val chunkTilesUpdateStream: KStream = + builder.stream( + "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.MapChunk}", + consumedAs("consume-map-chunk-packets", jsonMapper.serde(), jsonMapper.serde()) + ) + .filter("events.filter.map-chunks") { _: FactorioPacketKey, event: FactorioEvent -> + event.data is MapChunk + } + .mapValues("events.extract-map-chunks") { _, event: FactorioEvent -> + (event.data as? MapChunk)!!.tiles + } return luaTilesUpdatesStream .merge(chunkTilesUpdateStream) .flatMap("flat-map-all-tiles") { _, value -> - value.tiles.map { it.position to it } + value.tiles.map { tile -> + val key = TileUpdateRecordKey( + SurfaceIndex(value.surfaceIndex), + tile.position + ) + key to tile + } } .toTable( "all-tiles-keyed-on-position", @@ -66,21 +82,3 @@ fun allMapTilesTable( ) ) } - -//fun buildTilesTable( -// builder: StreamsBuilder -//): KTable { -// -// return luaTilesUpdatesStream -// .merge("luaTilesUpdatesStream-and-chunkTilesUpdateStream", chunkTilesUpdateStream) -// .flatMapTopicRecords("all-tiles-convert-to-TileUpdateRecord") { _, mapTiles: MapTiles -> -// mapTiles.tiles.map { tile -> -// TileUpdateRecord( -// SurfaceIndex(mapTiles.surfaceIndex), -// tile.position, -// tile, -// ) -// } -// } -// .toTable(materializedWith(jsonMapper.serde(), jsonMapper.serde())) -//} diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioServerPackets.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioServerPackets.kt index d071ad2..17fbcf1 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioServerPackets.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioServerPackets.kt @@ -1,22 +1,39 @@ package dev.adamko.kafkatorio.processor.topology +import dev.adamko.kafkatorio.events.schema.ConsoleChatMessage +import dev.adamko.kafkatorio.events.schema.EntityData import dev.adamko.kafkatorio.events.schema.FactorioConfigurationUpdate 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.processor.KafkatorioPacketSerde +import dev.adamko.kafkatorio.events.schema.MapChunk +import dev.adamko.kafkatorio.events.schema.MapTile +import dev.adamko.kafkatorio.events.schema.MapTiles +import dev.adamko.kafkatorio.events.schema.PlayerData +import dev.adamko.kafkatorio.events.schema.SurfaceData +import dev.adamko.kafkatorio.events.schema.converters.toMapChunkPosition +import dev.adamko.kafkatorio.processor.serdes.KafkatorioPacketSerde import dev.adamko.kafkatorio.processor.KafkatorioTopology -import dev.adamko.kafkatorio.processor.jsonMapper +import dev.adamko.kafkatorio.processor.serdes.jsonMapper import dev.adamko.kotka.extensions.consumedAs import dev.adamko.kotka.extensions.producedAs -import dev.adamko.kotka.extensions.streams.mapValues +import dev.adamko.kotka.extensions.streams.map import dev.adamko.kotka.extensions.streams.to +import dev.adamko.kotka.kxs.serde +import kotlinx.serialization.Serializable import kotlinx.serialization.decodeFromString import org.apache.kafka.common.serialization.Serdes import org.apache.kafka.streams.StreamsBuilder import org.apache.kafka.streams.processor.RecordContext +@Serializable +data class FactorioPacketKey( + val serverId: FactorioServerId, + /** arbitrary distinctive key */ + val key: String, +) + fun splitFactorioServerLog( builder: StreamsBuilder = StreamsBuilder(), ) { @@ -29,17 +46,19 @@ fun splitFactorioServerLog( Serdes.String(), ) ) - .mapValues("decode-packets") { _: String, value: String -> + .map("decode-packets") { serverId: String, value: String -> // println("Mapping $readOnlyKey:$value") - jsonMapper.decodeFromString(value) + val packet = jsonMapper.decodeFromString(value) + val key: String = createPacketKey(packet) + FactorioPacketKey(FactorioServerId(serverId), key) to packet } .to( producedAs( "split-server-log", - Serdes.String(), + jsonMapper.serde(), KafkatorioPacketSerde ) - ) { _: String, value: KafkatorioPacket, _: RecordContext -> + ) { _: FactorioPacketKey, value: KafkatorioPacket, _: RecordContext -> // println("[$key] sending event:${value.eventType} to topic:${value.data.objectName()}") when (value) { is FactorioEvent -> @@ -52,3 +71,26 @@ fun splitFactorioServerLog( } } + +private fun createPacketKey(packet: KafkatorioPacket): String { + + return when (packet) { + is FactorioEvent -> factorioEventKey(packet) + is FactorioConfigurationUpdate, + is FactorioPrototypes -> packet.hashCode().toString() + } +} + +private fun factorioEventKey(event: FactorioEvent): String { + val id: String = when (val data = event.data) { + is ConsoleChatMessage -> data.authorPlayerIndex?.toString() + is EntityData -> data.type + is MapChunk -> data.position.toString() + is MapTile -> data.position.toString() + is MapTiles -> data.tiles.firstOrNull()?.position?.toMapChunkPosition()?.toString() + is PlayerData -> (data.characterUnitNumber?.toString() ?: data.name) + is SurfaceData -> data.index.toString() + } ?: event.tick.toString() + + return id +} 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 1ba0d5c..58a3a8f 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 @@ -3,8 +3,9 @@ package dev.adamko.kafkatorio.processor.topology import kotlinx.serialization.Serializable @Serializable -data class SurfaceIndex( - val surfaceIndex: Int +@JvmInline +value class SurfaceIndex( + val index: Int ) { - override fun toString() = "$surfaceIndex" + override fun toString() = "$index" } diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/playerUpdates.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/playerUpdates.kt index da1838c..86f24d7 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/playerUpdates.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/playerUpdates.kt @@ -2,9 +2,9 @@ package dev.adamko.kafkatorio.processor.topology import dev.adamko.kafkatorio.events.schema.FactorioObjectData import dev.adamko.kafkatorio.events.schema.KafkatorioPacket -import dev.adamko.kafkatorio.processor.KafkatorioPacketSerde +import dev.adamko.kafkatorio.processor.serdes.KafkatorioPacketSerde import dev.adamko.kafkatorio.processor.WebsocketServer -import dev.adamko.kafkatorio.processor.jsonMapper +import dev.adamko.kafkatorio.processor.serdes.jsonMapper import kotlinx.serialization.encodeToString import org.apache.kafka.common.serialization.Serdes import org.apache.kafka.streams.StreamsBuilder 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 a09c5b7..3f44473 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 @@ -3,12 +3,10 @@ package dev.adamko.kafkatorio.processor.topology import dev.adamko.kafkatorio.events.schema.FactorioPrototypes import dev.adamko.kafkatorio.events.schema.KafkatorioPacket import dev.adamko.kafkatorio.events.schema.MapTilePrototype -import dev.adamko.kafkatorio.processor.jsonMapper +import dev.adamko.kafkatorio.processor.serdes.jsonMapper import dev.adamko.kotka.extensions.consumedAs -//import dev.adamko.kotka.extensions.groupedAs import dev.adamko.kotka.extensions.materializedWith import dev.adamko.kotka.extensions.streams.flatMap -//import dev.adamko.kotka.extensions.streams.* import dev.adamko.kotka.kxs.serde import kotlinx.serialization.Serializable import org.apache.kafka.common.serialization.Serdes diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/webMapTileAggregate.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/webMapTileAggregate.kt index 954a46c..c1f04fd 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/webMapTileAggregate.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/webMapTileAggregate.kt @@ -5,7 +5,7 @@ import dev.adamko.kafkatorio.events.schema.MapTile import dev.adamko.kafkatorio.events.schema.MapTilePosition import dev.adamko.kafkatorio.events.schema.MapTilePrototype import dev.adamko.kafkatorio.events.schema.converters.toMapChunkPosition -import dev.adamko.kafkatorio.processor.jsonMapper +import dev.adamko.kafkatorio.processor.serdes.jsonMapper import dev.adamko.kotka.extensions.groupedAs import dev.adamko.kotka.extensions.materializedAs import dev.adamko.kotka.extensions.namedAs @@ -19,26 +19,16 @@ import kotlinx.serialization.Serializable import org.apache.kafka.streams.kstream.KTable import org.apache.kafka.streams.kstream.Suppressed -/** The tiles of the webmap will be this many pixels high/wide */ -const val WEB_MAP_IMAGE_TILE_SIZE = 32 -//@Serializable -//data class MapChunkDataPosition( -// val position: MapChunkPosition, -// val surfaceIndex: SurfaceIndex, -//) -// -//@Serializable -//data class MapChunkData( -// val chunkPosition: MapChunkDataPosition, -// val tiles: Set, -//) +/** The tiles of the webmap will be this many pixels high/wide */ +const val WEB_MAP_IMAGE_TILE_SIZE = 256 @Serializable data class WebMapTileChunkPosition( val chunkSize: Int, val x: Int, val y: Int, + val surfaceIndex: SurfaceIndex, ) @Serializable @@ -59,11 +49,11 @@ private data class WebMapTileChunkPixelsAcc( ) fun aggregateWebMapTiles( - allMapTilesTable: KTable, + allMapTilesTable: KTable, tilePrototypesTable: KTable, ): KTable { - val webmapTileColours: KTable = + val webmapTileColours: KTable = allMapTilesTable .join( tilePrototypesTable, @@ -87,9 +77,14 @@ fun aggregateWebMapTiles( jsonMapper.serde(), jsonMapper.serde(), ) - ) { pos: MapTilePosition, px: WebMapTilePixel -> - val (chunkX, chunkY) = pos.toMapChunkPosition(WEB_MAP_IMAGE_TILE_SIZE) - val chunkPos = WebMapTileChunkPosition(WEB_MAP_IMAGE_TILE_SIZE, chunkX, chunkY) + ) { pos: TileUpdateRecordKey, px: WebMapTilePixel -> + val (chunkX, chunkY) = pos.tilePosition.toMapChunkPosition(WEB_MAP_IMAGE_TILE_SIZE) + val chunkPos = WebMapTileChunkPosition( + WEB_MAP_IMAGE_TILE_SIZE, + chunkX, + chunkY, + pos.surfaceIndex + ) (chunkPos to px).toKeyValue() } .aggregate( @@ -103,7 +98,11 @@ fun aggregateWebMapTiles( acc.apply { pixels.minus(oldValue) } }, namedAs("web-map-tile-colours-into-$WEB_MAP_IMAGE_TILE_SIZE-chunks"), - materializedAs("web-map-tile-colour-chunks-aggregate", jsonMapper.serde(), jsonMapper.serde()) + materializedAs( + "web-map-tile-colour-chunks-aggregate", + jsonMapper.serde(), + jsonMapper.serde() + ) ) .suppress( Suppressed.untilTimeLimit( 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 5d8b231..0ffc2b0 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 @@ -3,7 +3,6 @@ package dev.adamko.kafkatorio.processor.topology import com.sksamuel.scrimage.ImmutableImage import com.sksamuel.scrimage.color.RGBColor import com.sksamuel.scrimage.nio.PngWriter -import dev.adamko.kafkatorio.events.schema.MAP_CHUNK_SIZE import dev.adamko.kotka.extensions.tables.toStream import java.awt.Color import java.awt.image.BufferedImage @@ -43,15 +42,14 @@ private fun saveMapTilesPng( chunkPosition: WebMapTileChunkPosition, pixels: Set ) { - val surfaceIndex = 1 // TODO get surface index - - val chunkOriginX = chunkPosition.x * chunkPosition.chunkSize - val chunkOriginY = chunkPosition.y * chunkPosition.chunkSize + val chunkOriginX: Int = chunkPosition.x * chunkPosition.chunkSize + val chunkOriginY: Int = chunkPosition.y * chunkPosition.chunkSize + val surfaceIndex: Int = chunkPosition.surfaceIndex.index val chunkImage = ImmutableImage.filled( - MAP_CHUNK_SIZE, - MAP_CHUNK_SIZE, + chunkPosition.chunkSize, + chunkPosition.chunkSize, Color.BLACK, BufferedImage.TYPE_INT_ARGB ) @@ -79,7 +77,7 @@ private fun saveMapTilesPng( val zoom = 1u val file = File( - "src/main/resources/kafkatorio-web-map/s${surfaceIndex}/z$zoom/x${chunkOriginX}/y${chunkOriginY}.png" + "src/main/resources/kafkatorio-web-map/s${surfaceIndex}/z$zoom/x${chunkPosition.x}/y${chunkPosition.y}.png" ) if (file.parentFile.mkdirs()) { diff --git a/modules/infra-factorio-server/src/.env b/modules/infra-factorio-server/src/.env index c8f48e4..129db2a 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.50 -KAFKATORIO_VERSION=0.2.4 +FACTORIO_VERSION=1.1.53 +KAFKATORIO_VERSION=0.2.6 diff --git a/modules/infra-kafka-cluster/src/.env b/modules/infra-kafka-cluster/src/.env index 363e10c..afa5814 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.4 +KAFKATORIO_VERSION=0.2.6 diff --git a/modules/infra-kafka-cluster/src/docker-compose.yml b/modules/infra-kafka-cluster/src/docker-compose.yml index c95fa8e..e05f8c9 100644 --- a/modules/infra-kafka-cluster/src/docker-compose.yml +++ b/modules/infra-kafka-cluster/src/docker-compose.yml @@ -11,7 +11,7 @@ services: context: "." target: "kafka-kraft" args: - KAFKA_DL_URL: "https://dlcdn.apache.org/kafka/3.0.0/kafka_2.13-3.0.0.tgz" + KAFKA_DL_URL: "https://dlcdn.apache.org/kafka/3.1.0/kafka_2.13-3.1.0.tgz" networks: - "kafka-nw" - "kafkatorio-nw" @@ -20,7 +20,7 @@ services: - "9093:9093" kafka-ui: - image: provectuslabs/kafka-ui:0.3.1 + image: provectuslabs/kafka-ui:0.3.2 container_name: "kafka-ui" restart: "unless-stopped" networks: diff --git a/modules/infra-kafka-cluster/src/kafka-server.properties b/modules/infra-kafka-cluster/src/kafka-server.properties index 70ebc99..3f7c92c 100644 --- a/modules/infra-kafka-cluster/src/kafka-server.properties +++ b/modules/infra-kafka-cluster/src/kafka-server.properties @@ -1,22 +1,5 @@ -# Licensed to the Apache Software Foundation (ASF) under one or more -# contributor license agreements. See the NOTICE file distributed with -# this work for additional information regarding copyright ownership. -# The ASF licenses this file to You under the Apache License, Version 2.0 -# (the "License"); you may not use this file except in compliance with -# the License. You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -# # This configuration file is intended for use in KRaft mode, where # Apache ZooKeeper is not present. See config/kraft/README.md for details. -# ############################# Server Basics ############################# @@ -83,7 +66,7 @@ log.dirs=/tmp/kraft-combined-logs # The default number of log partitions per topic. More partitions allow greater # parallelism for consumption, but this will also result in more files across # the brokers. -num.partitions=1 +num.partitions=3 # The number of threads per data directory to be used for log recovery at startup and flushing at shutdown. # This value is recommended to be increased for installations with data dirs located in RAID array. @@ -133,3 +116,7 @@ log.segment.bytes=1073741824 # The interval at which log segments are checked to see if they can be deleted according # to the retention policies log.retention.check.interval.ms=300000 + +### Compression ### +#compression.type=snappy +#linger.ms=5 diff --git a/modules/infra-kafka-pipe/src/.env b/modules/infra-kafka-pipe/src/.env index 363e10c..afa5814 100644 --- a/modules/infra-kafka-pipe/src/.env +++ b/modules/infra-kafka-pipe/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.4 +KAFKATORIO_VERSION=0.2.6 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 f29c4a2..ff73df7 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 @@ -33,9 +33,10 @@ fun createFactorioMap() = Maps { """http://localhost:9073/tiles/s1/z{z}/x{x}/y{y}.png""" ) { attribution = "kafkatorio" +// tileSize = 256 tms = true - minZoom = -2 - maxZoom = 1 +// minZoom = -2 +// maxZoom = 1 maxNativeZoom = 1 minNativeZoom = 1 }.addTo(this) diff --git a/modules/web-map/src/main/kotlin/dev/adamko/kafkatorio/webmap/websockets.kt b/modules/web-map/src/main/kotlin/dev/adamko/kafkatorio/webmap/websockets.kt index 36c8a7c..2030fe5 100644 --- a/modules/web-map/src/main/kotlin/dev/adamko/kafkatorio/webmap/websockets.kt +++ b/modules/web-map/src/main/kotlin/dev/adamko/kafkatorio/webmap/websockets.kt @@ -4,9 +4,11 @@ import dev.adamko.kafkatorio.events.schema.ConsoleChatMessage import dev.adamko.kafkatorio.events.schema.EntityData import dev.adamko.kafkatorio.events.schema.FactorioConfigurationUpdate 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.MapTile +import dev.adamko.kafkatorio.events.schema.MapTiles import dev.adamko.kafkatorio.events.schema.PlayerData import dev.adamko.kafkatorio.events.schema.SurfaceData import io.kvision.redux.ReduxStore @@ -37,18 +39,22 @@ class WebsocketService( println(data.replace('\n', ' ')) when (val event: KafkatorioPacket = jsonMapper.decodeFromString(data)) { - is FactorioEvent -> { + is FactorioEvent -> { when (val eventData = event.data) { - is PlayerData -> + is PlayerData -> reduxStore.dispatch(FactorioUpdate.PlayerUpdate(event, eventData)) - is ConsoleChatMessage -> {} - is EntityData -> {} - is MapChunk -> {} - is MapTile -> {} - is SurfaceData -> {} + is ConsoleChatMessage, + is EntityData, + is MapChunk, + is MapTile, + is SurfaceData, + is MapTiles -> { + } } } - is FactorioConfigurationUpdate -> {} + is FactorioConfigurationUpdate, + is FactorioPrototypes -> { + } } } else {