From b8ba8b7c9ed63c0d142c551e880029cb3d823216 Mon Sep 17 00:00:00 2001 From: Adam <897017+aSemy@users.noreply.github.com> Date: Sat, 12 Mar 2022 16:42:24 +0100 Subject: [PATCH] - tidy up groupMapChunks.kt - add ticks to all Kafkatorio Packets - add Kafka Streams test for groupMapChunks.kt --- gradle/libs.versions.toml | 5 +- .../kafkatorio-schema/config-update.d.ts | 1 + .../kafkatorio-schema/prototype.d.ts | 1 + .../generated/kafkatorio-schema/schema.d.ts | 1 + .../handleConfigurationUpdate.ts | 3 +- .../main/config-update/prototypes.ts | 3 +- .../main/typescript/main/events/emitEvent.ts | 2 +- .../events/schema/factorioCommonData.kt | 3 +- .../events/schema/factorioConfiguration.kt | 1 + .../events/schema/factorioEventUpdates.kt | 3 +- .../events/schema/factorioEvents.kt | 3 +- .../events/schema/factorioPrototypes.kt | 1 + .../events/schema/kafkatorioPacket.kt | 1 + .../events/schema/FactorioEventTest.kt | 2 + .../events/schema/PrototypesTest.kt | 2 + modules/events-server/build.gradle.kts | 3 +- .../processor/KafkatorioTopology.kt | 64 +-------- .../kafkatorio/processor/serdes/kxsBinary.kt | 8 +- .../processor/topology/flowMapTiles.kt | 3 +- .../processor/topology/groupMapChunks.kt | 119 +++++++++++++--- .../processor/topology/prototypes.kt | 10 +- .../processor/topology/GroupMapChunksTest.kt | 130 ++++++++++++++++++ 22 files changed, 271 insertions(+), 98 deletions(-) create mode 100644 modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/topology/GroupMapChunksTest.kt diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index 1295f7c..0f0eae2 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -90,9 +90,10 @@ gcp-speech = { group = "com.google.cloud", name = "google-cloud-speech" ## kafka ## -kafka-streams = { group = "org.apache.kafka", name = "kafka-streams", version.ref = "kafka" } +kafka-streams = { group = "org.apache.kafka", name = "kafka-streams", version.ref = "kafka" } +kafka-streamsTestUtils = { group = "org.apache.kafka", name = "kafka-streams-test-utils", version.ref = "kafka" } # kafka helpers -kotka = { group = "com.github.blueanvil", name = "kotka", version = "1.1.1" } +#kotka = { group = "com.github.blueanvil", name = "kotka", version = "1.1.1" } jsoup = { group = "org.jsoup", name = "jsoup", version = "1.14.3" } skrapeit = { group = "it.skrape", name = "skrapeit", version = "1.2.0" } diff --git a/modules/events-mod/src/main/typescript/generated/kafkatorio-schema/config-update.d.ts b/modules/events-mod/src/main/typescript/generated/kafkatorio-schema/config-update.d.ts index 3145904..6fa06d0 100644 --- a/modules/events-mod/src/main/typescript/generated/kafkatorio-schema/config-update.d.ts +++ b/modules/events-mod/src/main/typescript/generated/kafkatorio-schema/config-update.d.ts @@ -16,4 +16,5 @@ interface FactorioConfigurationUpdate extends KafkatorioPacket { factorioData: FactorioGameDataUpdate; modVersion: string; packetType: PacketType; + tick: uint; } diff --git a/modules/events-mod/src/main/typescript/generated/kafkatorio-schema/prototype.d.ts b/modules/events-mod/src/main/typescript/generated/kafkatorio-schema/prototype.d.ts index 01d4958..5dbd1c0 100644 --- a/modules/events-mod/src/main/typescript/generated/kafkatorio-schema/prototype.d.ts +++ b/modules/events-mod/src/main/typescript/generated/kafkatorio-schema/prototype.d.ts @@ -10,6 +10,7 @@ interface FactorioPrototypes extends KafkatorioPacket { modVersion: string; packetType: PacketType; prototypes: FactorioPrototype[]; + tick: uint; } interface MapTilePrototype extends FactorioPrototype { 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 7091073..3c42507 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 @@ -5,6 +5,7 @@ type PacketType = "EVENT" | "CONFIG" | "PROTOTYPES" | "UPDATE"; interface KafkatorioPacket { modVersion: string; packetType: PacketType; + tick: uint; } interface FactorioEvent extends KafkatorioPacket { diff --git a/modules/events-mod/src/main/typescript/main/config-update/handleConfigurationUpdate.ts b/modules/events-mod/src/main/typescript/main/config-update/handleConfigurationUpdate.ts index dab8aef..783f611 100644 --- a/modules/events-mod/src/main/typescript/main/config-update/handleConfigurationUpdate.ts +++ b/modules/events-mod/src/main/typescript/main/config-update/handleConfigurationUpdate.ts @@ -6,7 +6,8 @@ export function emitConfigurationUpdate(changeData: ConfigurationChangedData) { modVersion: global.MOD_VERSION, packetType: "CONFIG", allMods: allMods(changeData), - factorioData: factorioData(changeData) + factorioData: factorioData(changeData), + tick: game.tick, }) } diff --git a/modules/events-mod/src/main/typescript/main/config-update/prototypes.ts b/modules/events-mod/src/main/typescript/main/config-update/prototypes.ts index d4d2bfe..bd96832 100644 --- a/modules/events-mod/src/main/typescript/main/config-update/prototypes.ts +++ b/modules/events-mod/src/main/typescript/main/config-update/prototypes.ts @@ -6,7 +6,8 @@ export function emitPrototypes() { emitPacket({ modVersion: global.MOD_VERSION, packetType: "PROTOTYPES", - prototypes: prototypes() + prototypes: prototypes(), + tick: game.tick, }) } diff --git a/modules/events-mod/src/main/typescript/main/events/emitEvent.ts b/modules/events-mod/src/main/typescript/main/events/emitEvent.ts index ab98b7f..d095fbc 100644 --- a/modules/events-mod/src/main/typescript/main/events/emitEvent.ts +++ b/modules/events-mod/src/main/typescript/main/events/emitEvent.ts @@ -7,6 +7,6 @@ export function emitEvent(eventData: T, tick: uint packetType: "EVENT", eventType: eventType, modVersion: global.MOD_VERSION, - tick: tick + tick: tick, }) } 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 16d4b55..65b7e36 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 @@ -112,8 +112,9 @@ interface EntityIdentifiers { data class MapTileDictionary( /** Map an X,Y coordinate a prototype name */ val tilesXY: Map>, - val protos: Map + val protos: Map, ) { + fun toMapTileList(): List = buildList { tilesXY.forEach { (xString, row) -> row.forEach { (yString, protoIndex) -> 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 900e28d..37aa57f 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 @@ -7,6 +7,7 @@ import kotlinx.serialization.Serializable @Serializable data class FactorioConfigurationUpdate( override val modVersion: String, + override val tick: Tick, val factorioData: FactorioGameDataUpdate, val allMods: List, ) : KafkatorioPacket() { diff --git a/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioEventUpdates.kt b/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioEventUpdates.kt index e66f16f..e930361 100644 --- a/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioEventUpdates.kt +++ b/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioEventUpdates.kt @@ -15,8 +15,7 @@ import kotlinx.serialization.json.jsonPrimitive @Serializable data class FactorioEventUpdatePacket( override val modVersion: String, - /** game time */ - val tick: Tick, + override val tick: Tick, val update: FactorioEventUpdate, ) : KafkatorioPacket() { @EncodeDefault diff --git a/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioEvents.kt b/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioEvents.kt index 0cb882e..177bf21 100644 --- a/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioEvents.kt +++ b/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioEvents.kt @@ -7,10 +7,9 @@ import kotlinx.serialization.Serializable @Serializable data class FactorioEvent( override val modVersion: String, + override val tick: Tick, /** the initial Factorio event (`defines.events`) trigger */ val eventType: String, - /** game time */ - val tick: Tick, val data: FactorioObjectData, ) : KafkatorioPacket() { @EncodeDefault diff --git a/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioPrototypes.kt b/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioPrototypes.kt index 0fb80b2..c5c3f3e 100644 --- a/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioPrototypes.kt +++ b/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioPrototypes.kt @@ -15,6 +15,7 @@ import kotlinx.serialization.json.jsonPrimitive @Serializable data class FactorioPrototypes( override val modVersion: String, + override val tick: Tick, val prototypes: List, ) : KafkatorioPacket() { @EncodeDefault diff --git a/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/kafkatorioPacket.kt b/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/kafkatorioPacket.kt index e5e947e..e562dfa 100644 --- a/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/kafkatorioPacket.kt +++ b/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/kafkatorioPacket.kt @@ -14,6 +14,7 @@ import kotlinx.serialization.json.jsonPrimitive sealed class KafkatorioPacket { /** Schema versioning */ abstract val modVersion: String + abstract val tick: Tick @EncodeDefault abstract val packetType: PacketType 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 248e22b..bd93da6 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 @@ -289,6 +289,7 @@ class FactorioEventTest : FunSpec({ { "modVersion": "0.2.8", "packetType": "CONFIG", + "tick": 534, "allMods": [ { "modName": "kafkatorio-events", @@ -310,6 +311,7 @@ class FactorioEventTest : FunSpec({ val expected = FactorioConfigurationUpdate( modVersion = "0.2.8", + tick = Tick(534u), factorioData = FactorioGameDataUpdate(null, null), allMods = listOf( FactorioModInfo( diff --git a/modules/events-schema/src/commonTest/kotlin/dev/adamko/kafkatorio/events/schema/PrototypesTest.kt b/modules/events-schema/src/commonTest/kotlin/dev/adamko/kafkatorio/events/schema/PrototypesTest.kt index 5356ad2..974af26 100644 --- a/modules/events-schema/src/commonTest/kotlin/dev/adamko/kafkatorio/events/schema/PrototypesTest.kt +++ b/modules/events-schema/src/commonTest/kotlin/dev/adamko/kafkatorio/events/schema/PrototypesTest.kt @@ -12,6 +12,7 @@ class PrototypesTest : BehaviorSpec({ { "modVersion":"0.2.4", "packetType":"PROTOTYPES", + "tick": 534, "prototypes":[ { "prototypeObjectName":"LuaTilePrototype", @@ -46,6 +47,7 @@ class PrototypesTest : BehaviorSpec({ { "modVersion":"0.3.2", "packetType":"PROTOTYPES", + "tick": 534, "prototypes":[ { "prototypeObjectName":"LuaTilePrototype", diff --git a/modules/events-server/build.gradle.kts b/modules/events-server/build.gradle.kts index a5f5c14..e9c353c 100644 --- a/modules/events-server/build.gradle.kts +++ b/modules/events-server/build.gradle.kts @@ -20,6 +20,7 @@ val kafkaStateDir = layout.projectDirectory.dir("kafka-state") dependencies { implementation(libs.kafka.streams) + testImplementation(libs.kafka.streamsTestUtils) implementation(projects.modules.eventsSchema) @@ -40,7 +41,7 @@ dependencies { implementation("com.sksamuel.scrimage:scrimage-core:4.0.24") - val kotkaVersion = "v0.0.14" + val kotkaVersion = "v0.0.15" implementation("com.github.adamko-dev.kotka-streams:kotka-streams:$kotkaVersion") { isChanging = true } 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 3d35424..076db30 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,30 +1,20 @@ package dev.adamko.kafkatorio.processor import dev.adamko.kafkatorio.events.schema.ColourHex -import dev.adamko.kafkatorio.events.schema.FactorioEvent -import dev.adamko.kafkatorio.events.schema.FactorioEventUpdate -import dev.adamko.kafkatorio.events.schema.FactorioEventUpdatePacket -import dev.adamko.kafkatorio.events.schema.FactorioObjectData -import dev.adamko.kafkatorio.events.schema.FactorioPrototypes -import dev.adamko.kafkatorio.events.schema.KafkatorioPacket import dev.adamko.kafkatorio.processor.config.ApplicationProperties -import dev.adamko.kafkatorio.processor.serdes.jsonMapper 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.factorioServerPacketStream -import dev.adamko.kafkatorio.processor.topology.groupTilesIntoChunksWithColours +import dev.adamko.kafkatorio.processor.topology.groupMapChunks 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 dev.adamko.kotka.extensions.consumedAs -import dev.adamko.kotka.extensions.producedAs -import dev.adamko.kotka.extensions.tables.toStream import dev.adamko.kotka.kxs.serde import java.time.Duration import kotlin.coroutines.CoroutineContext +import kotlin.coroutines.cancellation.CancellationException import kotlinx.coroutines.CoroutineName import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers @@ -36,7 +26,6 @@ import org.apache.kafka.streams.StreamsBuilder import org.apache.kafka.streams.StreamsConfig import org.apache.kafka.streams.Topology import org.apache.kafka.streams.TopologyDescription -import org.apache.kafka.streams.kstream.KStream import org.apache.kafka.streams.kstream.KTable @@ -62,54 +51,7 @@ internal class KafkatorioTopology( private fun groupTilesMapChunks() { val builder = StreamsBuilder() - - - val protosStream: KStream = - builder.stream( - "kafkatorio.${KafkatorioPacket.PacketType.PROTOTYPES}.all", - consumedAs("consume.factorio-protos.all", jsonMapper.serde(), jsonMapper.serde()) - ) - val tileProtoColourDict = tileProtoColourDictionary(protosStream) - - -// val mapTilesStream: KStream = -// builder.stream( -// "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.LuaTiles}", -// consumedAs("consume.map-tiles", jsonMapper.serde(), jsonMapper.serde()) -// ) -// -// val mapChunksStream: KStream = -// builder.stream( -// "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.MapChunk}", -// consumedAs("consume.map-chunks", jsonMapper.serde(), jsonMapper.serde()) -// ) - - val mapChunksUpdatesStream: KStream = - builder.stream( - "kafkatorio.${KafkatorioPacket.PacketType.UPDATE}.${FactorioEventUpdate.FactorioEventUpdateType.MAP_CHUNK}", - consumedAs("consume.map-chunk-updates", jsonMapper.serde(), jsonMapper.serde()) - ) - - val groupedMapChunkTiles: KTable> = - groupTilesIntoChunksWithColours( -// mapTilesStream, -// mapChunksStream, - mapChunksUpdatesStream, - tileProtoColourDict, - ) - - groupedMapChunkTiles - .toStream("stream-grouped-map-tiles") - .to( - TOPIC_GROUPED_MAP_CHUNKS, - producedAs( - "produce.grouped-map-chunks", - kxsBinary.serde(), - kxsBinary.serde?>() - ) - ) - - val topology = builder.build() + val topology = groupMapChunks(builder) launchTopology("groupTilesMapChunks", topology) } 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 index b1be03c..4e1a99c 100644 --- 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 @@ -60,12 +60,14 @@ class KxsDataOutputEncoder( override fun encodeLong(value: Long) = output.writeLong(value) override fun encodeShort(value: Short) = output.writeShort(value.toInt()) override fun encodeString(value: String) = output.writeUTF(value) +// override fun encodeString(value: String) = encodeByteArray(value.toByteArray(Charsets.UTF_8)) override fun beginCollection( descriptor: SerialDescriptor, collectionSize: Int ): CompositeEncoder { - encodeCompactSize(collectionSize) + encodeInt(collectionSize) +// encodeCompactSize(collectionSize) return this } @@ -113,6 +115,7 @@ class KxsDataInputDecoder( override fun decodeLong(): Long = input.readLong() override fun decodeShort(): Short = input.readShort() override fun decodeString(): String = input.readUTF() +// override fun decodeString(): String = decodeByteArray().toString(Charsets.UTF_8) override fun decodeElementIndex(descriptor: SerialDescriptor): Int { if (elementIndex == elementsCount) return CompositeDecoder.DECODE_DONE @@ -125,7 +128,8 @@ class KxsDataInputDecoder( override fun decodeSequentially(): Boolean = true override fun decodeCollectionSize(descriptor: SerialDescriptor): Int = - decodeCompactSize().also { elementsCount = it } + decodeInt().also { elementsCount = it } +// decodeCompactSize().also { elementsCount = it } override fun decodeNotNullMark(): Boolean = decodeBoolean() 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 f0b189e..69c5304 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 @@ -9,6 +9,7 @@ import dev.adamko.kafkatorio.events.schema.MapChunkPosition import dev.adamko.kafkatorio.events.schema.MapTilePosition import dev.adamko.kafkatorio.events.schema.SurfaceIndex import dev.adamko.kafkatorio.events.schema.converters.toMapChunkPosition +import dev.adamko.kotka.extensions.streams.* import java.awt.image.BufferedImage import java.io.File import kotlin.coroutines.CoroutineContext @@ -52,7 +53,7 @@ fun saveMapTiles( serverMapTable .toStream() - .foreach { _, value -> + .forEach("save-chunked-tiles") { _, value -> if (value != null) { runBlocking(Dispatchers.Default) { println("emitting chunkedTiles ${value.chunkId} tiles: ${value.map.size}") diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/groupMapChunks.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/groupMapChunks.kt index 28eaab6..1858986 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/groupMapChunks.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/groupMapChunks.kt @@ -1,7 +1,10 @@ package dev.adamko.kafkatorio.processor.topology import dev.adamko.kafkatorio.events.schema.ColourHex +import dev.adamko.kafkatorio.events.schema.FactorioEventUpdate import dev.adamko.kafkatorio.events.schema.FactorioEventUpdatePacket +import dev.adamko.kafkatorio.events.schema.FactorioPrototypes +import dev.adamko.kafkatorio.events.schema.KafkatorioPacket import dev.adamko.kafkatorio.events.schema.MapChunkPosition import dev.adamko.kafkatorio.events.schema.MapChunkUpdate import dev.adamko.kafkatorio.events.schema.MapTile @@ -10,16 +13,20 @@ import dev.adamko.kafkatorio.events.schema.MapTilePosition import dev.adamko.kafkatorio.events.schema.MapTiles import dev.adamko.kafkatorio.events.schema.SurfaceIndex import dev.adamko.kafkatorio.events.schema.converters.toMapChunkPosition +import dev.adamko.kafkatorio.processor.KafkatorioTopology +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.groupedAs -import dev.adamko.kotka.extensions.materializedAs import dev.adamko.kotka.extensions.materializedWith -import dev.adamko.kotka.extensions.streams.flatMap -import dev.adamko.kotka.extensions.streams.mapValues -import dev.adamko.kotka.extensions.streams.reduce +import dev.adamko.kotka.extensions.producedAs +import dev.adamko.kotka.extensions.streams.* import dev.adamko.kotka.extensions.tables.join +import dev.adamko.kotka.extensions.tables.toStream import dev.adamko.kotka.kxs.serde import kotlinx.serialization.Serializable +import org.apache.kafka.streams.StreamsBuilder +import org.apache.kafka.streams.Topology import org.apache.kafka.streams.kstream.KStream import org.apache.kafka.streams.kstream.KTable @@ -42,7 +49,74 @@ data class ServerMapChunkTiles( } -fun groupTilesIntoChunksWithColours( +fun groupMapChunks( + builder: StreamsBuilder, + prototypesTopicName: String = "kafkatorio.${KafkatorioPacket.PacketType.PROTOTYPES}.all", + mapChunkUpdatesTopicName: String = "kafkatorio.${KafkatorioPacket.PacketType.UPDATE}.${FactorioEventUpdate.FactorioEventUpdateType.MAP_CHUNK}", +): Topology { + + val protosStream: KStream = + builder.stream( + prototypesTopicName, + consumedAs("consume.factorio-protos.all", jsonMapper.serde(), jsonMapper.serde()) + ) + val tileProtoColourDict: KTable = + tileProtoColourDictionary(protosStream) + + +// val mapTilesStream: KStream = +// builder.stream( +// "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.LuaTiles}", +// consumedAs("consume.map-tiles", jsonMapper.serde(), jsonMapper.serde()) +// ) +// +// val mapChunksStream: KStream = +// builder.stream( +// "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.MapChunk}", +// consumedAs("consume.map-chunks", jsonMapper.serde(), jsonMapper.serde()) +// ) + + val mapChunksUpdatesStream: KStream = + builder.stream( + mapChunkUpdatesTopicName, + consumedAs( + "consume.map-chunk-updates", + jsonMapper.serde(), + jsonMapper.serde(), + ) + ) +// .repartition( +// repartitionedAs( +// "map-chunk-updates.repartition.change-serde", +// kxsBinary.serde(), +// kxsBinary.serde(), +// numberOfPartitions = 3, +// ) +// ) + + val groupedMapChunkTiles: KTable> = + groupTilesIntoChunksWithColours( +// mapTilesStream, +// mapChunksStream, + mapChunksUpdatesStream, + tileProtoColourDict, + ) + + groupedMapChunkTiles + .toStream("stream-grouped-map-tiles") + .to( + KafkatorioTopology.TOPIC_GROUPED_MAP_CHUNKS, + producedAs( + "produce.grouped-map-chunks", + kxsBinary.serde(), + kxsBinary.serde?>() + ) + ) + + return builder.build() +} + +private fun groupTilesIntoChunksWithColours( // mapTilePacketsStream: KStream, // mapChunksPacketsStream: KStream, mapChunkUpdatePacketsStream: KStream, @@ -107,7 +181,7 @@ fun groupTilesIntoChunksWithColours( // .merge("merge-mapTilesStream-with-mapChunkTilesStream", mapChunkTilesStream) // .merge("merge-mapTilesStream-with-mapChunkUpdatesStream", mapChunkUpdatesStream) // .filter { _, value -> !value?.tiles.isNullOrEmpty() } - .flatMap("server-map-data.tiles.flatMapByChunk") { key: FactorioServerId, mapTiles: MapTiles -> + .flatMap("server-map-data.tiles.flatMapByChunk") { serverId: FactorioServerId, mapTiles: MapTiles -> val standardChunkSize = ChunkSize.MAX @@ -119,10 +193,10 @@ fun groupTilesIntoChunksWithColours( .map { (chunkPos, tiles) -> val chunkId = ServerMapChunkId( - key, - chunkPos, - mapTiles.surfaceIndex, - standardChunkSize, + serverId = serverId, + chunkPosition = chunkPos, + surfaceIndex = mapTiles.surfaceIndex, + chunkSize = standardChunkSize, ) val tilesMap = tiles.associate { it.position to TileProtoHashCode(it) } @@ -132,20 +206,24 @@ fun groupTilesIntoChunksWithColours( chunkId to chunkTiles } } -// .filter("server-map-data.tiles.filter-not-empty") { _, value -> -// value.map.isNotEmpty() -// } + .filter("server-map-data.tiles.filter-not-empty") { _, value -> + value.map.isNotEmpty() + } .groupByKey( groupedAs( "server-map-data.tiles.group", - kxsBinary.serde(), - kxsBinary.serde>() + kxsBinary.serde(), + kxsBinary.serde?>(), ) ) .reduce( - "server-map-data.tiles.reduce", - materializedAs("server-map-data.tiles.store", kxsBinary.serde(), kxsBinary.serde()), - ) { a, b -> a + b } + "server-map-data.tiles.reduce", materializedWith( + // "server-map-data.tiles.store", + kxsBinary.serde(), + kxsBinary.serde>(), + ), + ServerMapChunkTiles::plus + ) val colourisedChunkTable: KTable> = @@ -153,7 +231,10 @@ fun groupTilesIntoChunksWithColours( .join( other = tileProtoColourDict, name = "server-map-data.join-tiles-with-prototypes", - materialized = materializedWith(kxsBinary.serde(), kxsBinary.serde()), + materialized = materializedWith( + kxsBinary.serde(), + kxsBinary.serde>(), + ), foreignKeyExtractor = { chunkTiles: ServerMapChunkTiles -> chunkTiles.chunkId.serverId } ) { tiles: ServerMapChunkTiles, colourDict: TileColourDict -> 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 b63d67e..ad25cfb 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 @@ -7,7 +7,9 @@ import dev.adamko.kafkatorio.events.schema.MapTile import dev.adamko.kafkatorio.events.schema.MapTilePrototype import dev.adamko.kafkatorio.events.schema.converters.toHex import dev.adamko.kafkatorio.processor.serdes.jsonMapper +import dev.adamko.kafkatorio.processor.serdes.kxsBinary import dev.adamko.kotka.extensions.materializedAs +import dev.adamko.kotka.extensions.materializedWith import dev.adamko.kotka.extensions.streams.filter import dev.adamko.kotka.extensions.streams.mapValues import dev.adamko.kotka.extensions.streams.toTable @@ -54,10 +56,10 @@ fun tileProtoColourDictionary( } .toTable( "server-map-data.tile-prototypes", - materializedAs( - "server-map-data.tile-prototypes.store", - jsonMapper.serde(), - jsonMapper.serde(), + materializedWith( +// "server-map-data.tile-prototypes.store", + kxsBinary.serde(), + kxsBinary.serde(), ) ) diff --git a/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/topology/GroupMapChunksTest.kt b/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/topology/GroupMapChunksTest.kt new file mode 100644 index 0000000..bed9571 --- /dev/null +++ b/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/topology/GroupMapChunksTest.kt @@ -0,0 +1,130 @@ +package dev.adamko.kafkatorio.processor.topology + +import dev.adamko.kafkatorio.events.schema.Colour +import dev.adamko.kafkatorio.events.schema.ColourHex +import dev.adamko.kafkatorio.events.schema.FactorioEventUpdatePacket +import dev.adamko.kafkatorio.events.schema.FactorioPrototypes +import dev.adamko.kafkatorio.events.schema.MapChunkPosition +import dev.adamko.kafkatorio.events.schema.MapChunkUpdate +import dev.adamko.kafkatorio.events.schema.MapTileDictionary +import dev.adamko.kafkatorio.events.schema.MapTilePosition +import dev.adamko.kafkatorio.events.schema.MapTilePrototype +import dev.adamko.kafkatorio.events.schema.PrototypeKey +import dev.adamko.kafkatorio.events.schema.PrototypeName +import dev.adamko.kafkatorio.events.schema.SurfaceIndex +import dev.adamko.kafkatorio.events.schema.Tick +import dev.adamko.kafkatorio.processor.KafkatorioTopology +import dev.adamko.kafkatorio.processor.serdes.jsonMapper +import dev.adamko.kafkatorio.processor.serdes.kxsBinary +import dev.adamko.kotka.kxs.serde +import io.kotest.assertions.withClue +import io.kotest.core.spec.style.FunSpec +import io.kotest.matchers.collections.shouldHaveSize +import io.kotest.matchers.equality.shouldBeEqualToComparingFields +import io.kotest.matchers.nulls.shouldNotBeNull +import io.kotest.matchers.should +import org.apache.kafka.streams.StreamsBuilder +import org.apache.kafka.streams.TestInputTopic +import org.apache.kafka.streams.TestOutputTopic +import org.apache.kafka.streams.TopologyTestDriver + + +class GroupMapChunksTest : FunSpec({ + + test("basic test") { + + val prototypesTopicName = "prototypes-topic" + val mapChunkUpdatesTopicName = "map-chunk-updates-topic" + + val topology = groupMapChunks( + StreamsBuilder(), + prototypesTopicName, + mapChunkUpdatesTopicName, + ) + + val testDriver = TopologyTestDriver(topology) + + val prototypesTestInputTopic: TestInputTopic = + testDriver.createInputTopic( + prototypesTopicName, + jsonMapper.serde().serializer(), + jsonMapper.serde().serializer(), + ) + + val serverId = FactorioServerId("test-server-id") + + prototypesTestInputTopic.pipeInput( + serverId, + FactorioPrototypes( + modVersion = "1.2.3", + tick = Tick(456u), + listOf( + MapTilePrototype( + PrototypeName("fake-proto"), + layer = 1u, + mapColour = Colour(0.3f, 0.2f, 0.1f, 1.0f), + collisionMasks = listOf("fake-mask"), + order = "fake-order", + canBeMined = false, + ) + ) + ) + ) + + val mapChunkUpdatesInputTopic: TestInputTopic = + testDriver.createInputTopic( + mapChunkUpdatesTopicName, + jsonMapper.serde().serializer(), + jsonMapper.serde().serializer(), + ) + mapChunkUpdatesInputTopic.pipeInput( + serverId, + FactorioEventUpdatePacket( + modVersion = "1.2.3", + tick = Tick(854u), + update = MapChunkUpdate( + MapChunkPosition(11, 22), + SurfaceIndex(1u), + tileDictionary = MapTileDictionary( + tilesXY = mapOf( + "1" to mapOf("2" to PrototypeKey("33")) + ), + protos = mapOf(PrototypeKey("33") to PrototypeName("fake-proto")) + ) + ) + ) + ) + + val outputTopic: TestOutputTopic?> = + testDriver + .createOutputTopic( + KafkatorioTopology.TOPIC_GROUPED_MAP_CHUNKS, + kxsBinary.serde().deserializer(), + kxsBinary.serde?>().deserializer(), + ) + + val records = outputTopic.readRecordsToList() + + withClue(records) { + records shouldHaveSize 1 + + records.map { it.key to it.value }.first() should { (key, value) -> + key.shouldNotBeNull() + key shouldBeEqualToComparingFields ServerMapChunkId( + serverId = serverId, + chunkPosition = MapChunkPosition(0, 0), + surfaceIndex = SurfaceIndex(1u), + chunkSize = ChunkSize.CHUNK_512, + ) + value.shouldNotBeNull() + value shouldBeEqualToComparingFields ServerMapChunkTiles( + chunkId = key, + map = mapOf( + MapTilePosition(1, 2) to ColourHex(76u, 51u, 25u, 255u) + ) + ) + } + } + } + +})