From b055397d25d721cfbaebc6aecaf687c165fa70a0 Mon Sep 17 00:00:00 2001 From: Adam <897017+aSemy@users.noreply.github.com> Date: Wed, 2 Mar 2022 18:38:51 +0100 Subject: [PATCH] - wicked sick size optimisation of map-tiles update packet - fix ts-generator recursive value class unwrapping - move kafka state dir - try fixing kafka streams for tile generation (simplifying it, and fixing some nullability problems) --- .../kafkatorio-schema/prototype.d.ts | 2 +- .../generated/kafkatorio-schema/schema.d.ts | 5 + .../generated/kafkatorio-schema/update.d.ts | 14 +-- .../typescript/main/cache/EventDataCache.ts | 2 +- .../main/events/onMapChunkEventListeners.ts | 41 ++++--- .../events/schema/factorioCommonData.kt | 39 ++++++- .../events/schema/factorioEventUpdates.kt | 9 +- .../events/schema/ValueClassTransformer.kt | 101 +++++++++++------- .../adamko/kafkatorio/events/schema/kt2ts.kt | 4 - modules/events-server/.gitignore | 1 + modules/events-server/build.gradle.kts | 2 +- .../processor/KafkatorioTopology.kt | 49 ++++++--- .../processor/topology/groupMapChunks.kt | 96 +++++++++++------ .../processor/topology/playerUpdates.kt | 6 +- 14 files changed, 253 insertions(+), 118 deletions(-) create mode 100644 modules/events-server/.gitignore 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 9021c35..01d4958 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 @@ -15,7 +15,7 @@ interface FactorioPrototypes extends KafkatorioPacket { interface MapTilePrototype extends FactorioPrototype { canBeMined: boolean; collisionMasks: string[]; - layer: int; + layer: uint; mapColour: Colour; name: string; order: string; 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 ca76f05..7091073 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 @@ -59,3 +59,8 @@ interface EntityIdentifiers { type: string; unitNumber: uint | null; } + +interface MapTileDictionary { + protos: { [key: string]: string }; + tilesXY: { [key: string]: { [key: string]: string } }; +} diff --git a/modules/events-mod/src/main/typescript/generated/kafkatorio-schema/update.d.ts b/modules/events-mod/src/main/typescript/generated/kafkatorio-schema/update.d.ts index a9598fc..ed61598 100644 --- a/modules/events-mod/src/main/typescript/generated/kafkatorio-schema/update.d.ts +++ b/modules/events-mod/src/main/typescript/generated/kafkatorio-schema/update.d.ts @@ -7,7 +7,7 @@ interface FactorioEventUpdateKey { } interface FactorioEventUpdate extends FactorioEventUpdateKey { - eventCounts: { [key: string]: int } | null; + eventCounts: { [key: string]: uint } | null; updateType: FactorioEventUpdateType; } @@ -30,7 +30,7 @@ interface PlayerUpdate extends FactorioEventUpdate, PlayerUpdateKey { colour: Colour | null; diedCause: EntityIdentifiers | null; disconnectReason: string | null; - eventCounts: { [key: string]: int } | null; + eventCounts: { [key: string]: uint } | null; forceIndex: uint | null; index: uint; isAdmin: boolean | null; @@ -57,12 +57,12 @@ interface EntityUpdateKey extends EntityIdentifiers, FactorioEventUpdateKey { interface EntityUpdate extends FactorioEventUpdate, EntityIdentifiers, EntityUpdateKey { chunkPosition: MapChunkPosition | null; - eventCounts: { [key: string]: int } | null; - graphicsVariation: number | null; + eventCounts: { [key: string]: uint } | null; + graphicsVariation: uint8 | null; health: float | null; isActive: boolean | null; isRotatable: boolean | null; - lastUser: int | null; + lastUser: uint | null; localisedDescription: string | null; localisedName: string | null; name: string; @@ -79,12 +79,12 @@ interface MapChunkUpdateKey extends FactorioEventUpdateKey { interface MapChunkUpdate extends FactorioEventUpdate, MapChunkUpdateKey { chunkPosition: MapChunkPosition; - eventCounts: { [key: string]: int } | null; + eventCounts: { [key: string]: uint } | null; force: uint | null; isDeleted: boolean | null; player: uint | null; robot: EntityIdentifiers | null; surfaceIndex: uint; - tiles: MapTile[] | null; + tileDictionary: MapTileDictionary | null; updateType: FactorioEventUpdateType; } diff --git a/modules/events-mod/src/main/typescript/main/cache/EventDataCache.ts b/modules/events-mod/src/main/typescript/main/cache/EventDataCache.ts index 7920507..ed59119 100644 --- a/modules/events-mod/src/main/typescript/main/cache/EventDataCache.ts +++ b/modules/events-mod/src/main/typescript/main/cache/EventDataCache.ts @@ -20,7 +20,7 @@ export namespace EventDataCache { log("Initialising global.DEFAULT_CACHE_DURATION_TICKS") global.defaultCacheDurationTicks = { - "PLAYER": 60, // 1 second + "PLAYER": 15, // 0.25 seconds "MAP_CHUNK": 60 * 30, // 30 seconds "ENTITY": 60 * 2, // 2 seconds "default": 60 * 5, // 5 seconds diff --git a/modules/events-mod/src/main/typescript/main/events/onMapChunkEventListeners.ts b/modules/events-mod/src/main/typescript/main/events/onMapChunkEventListeners.ts index f2e6665..7d6851b 100644 --- a/modules/events-mod/src/main/typescript/main/events/onMapChunkEventListeners.ts +++ b/modules/events-mod/src/main/typescript/main/events/onMapChunkEventListeners.ts @@ -8,6 +8,17 @@ import CacheData = EventDataCache.CacheData; type MapChunkUpdater = (data: CacheData<"MAP_CHUNK">) => void +const mapProtoNameToKey: { [key: string]: string } = {} +let protoIndex: uint = 0 + +function getProtoKey(protoName: string): string { + if (mapProtoNameToKey[protoName] == undefined) { + mapProtoNameToKey[protoName] = `${protoIndex++}` + } + return mapProtoNameToKey[protoName] +} + + function mapTilesUpdateDebounce( surface: LuaSurface | undefined, chunkPosition: ChunkPositionTable, @@ -27,24 +38,26 @@ function mapTilesUpdateDebounce( updateType: "MAP_CHUNK", }, data => { - if (data.tiles == undefined) { - data.tiles = [] + data.tileDictionary ??= { + tilesXY: {}, + protos: {}, } - for (const tile of tiles) { - data.tiles.push( - { - x: tile.position.x, - y: tile.position.y, - proto: tile.name, - } - ) + for (let tile of tiles) { + + const xString = `${tile.position.x}` + const yString = `${tile.position.y}` + const protoKey = getProtoKey(tile.name) + + data.tileDictionary.protos[protoKey] = tile.name + + data.tileDictionary.tilesXY[xString] ??= {} + data.tileDictionary.tilesXY[xString][yString] = protoKey } - if (data.eventCounts == undefined) { - data.eventCounts = {} - } - data.eventCounts[eventName] = ((data.eventCounts ?? {}) [eventName] ?? 0) + 1 + data.eventCounts ??= {} + data.eventCounts[eventName] ??= 0 + data.eventCounts[eventName]++ if (updater != undefined) { updater(data) 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 993e00d..16d4b55 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,7 +1,6 @@ package dev.adamko.kafkatorio.events.schema import kotlin.jvm.JvmInline -import kotlinx.serialization.EncodeDefault import kotlinx.serialization.Serializable @@ -101,3 +100,41 @@ interface EntityIdentifiers { val name: String val type: String } + + +/** + * Size-optimised 2D map of x/y tile positions and prototype-name. + * + * Each X/Y coordinate maps to an arbitrary [PrototypeKey] (which is only valid for any specific + * instance). [protos] is specific to each instance of [MapTileDictionary], can be used to convert from a + */ +@Serializable +data class MapTileDictionary( + /** Map an X,Y coordinate a prototype name */ + val tilesXY: Map>, + val protos: Map +) { + fun toMapTileList(): List = buildList { + tilesXY.forEach { (xString, row) -> + row.forEach { (yString, protoIndex) -> + val x = xString.toIntOrNull() + val y = yString.toIntOrNull() + val protoName = protos[protoIndex] + if (x != null && y != null && protoName != null) { + add(MapTile(x, y, protoName)) + } + } + } + } + + companion object { + fun MapTileDictionary?.isNullOrEmpty(): Boolean { + return this == null || (tilesXY.isEmpty() && tilesXY.all { it.value.isEmpty() }) + } + } +} + + +@Serializable +@JvmInline +value class PrototypeKey(val index: String) 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 fbcd3a1..e66f16f 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 @@ -147,7 +147,7 @@ data class EntityUpdate( override val eventCounts: Map? = null, val chunkPosition: MapChunkPosition? = null, - val graphicsVariation: UShort? = null, + val graphicsVariation: UByte? = null, val health: Float? = null, val isActive: Boolean? = null, val isRotatable: Boolean? = null, @@ -180,11 +180,8 @@ data class MapChunkUpdate( val player: PlayerIndex? = null, val robot: EntityIdentifiers? = null, val force: ForceIndex? = null, - /** updated tiles - might be partial */ - // note: this must be a List, not a Map - // because JSON can't have non-string keys. - @Contextual - val tiles: List? = null, + /** updated tiles - might be partial and not all tiles in the chunk */ + val tileDictionary: MapTileDictionary? = null, val isDeleted: Boolean? = null, ) : FactorioEventUpdate(), MapChunkUpdateKey { @EncodeDefault diff --git a/modules/events-schema/src/jvmMain/kotlin/dev/adamko/kafkatorio/events/schema/ValueClassTransformer.kt b/modules/events-schema/src/jvmMain/kotlin/dev/adamko/kafkatorio/events/schema/ValueClassTransformer.kt index bdc9747..daedc28 100644 --- a/modules/events-schema/src/jvmMain/kotlin/dev/adamko/kafkatorio/events/schema/ValueClassTransformer.kt +++ b/modules/events-schema/src/jvmMain/kotlin/dev/adamko/kafkatorio/events/schema/ValueClassTransformer.kt @@ -1,11 +1,15 @@ package dev.adamko.kafkatorio.events.schema +import kotlin.contracts.ExperimentalContracts +import kotlin.contracts.contract import kotlin.reflect.KClass +import kotlin.reflect.KClassifier import kotlin.reflect.KProperty import kotlin.reflect.KType import kotlin.reflect.KTypeProjection import kotlin.reflect.full.createType import kotlin.reflect.full.declaredMemberProperties +import kotlin.reflect.full.starProjectedType import kotlin.reflect.full.withNullability import me.ntrrgc.tsGenerator.ClassTransformer @@ -14,54 +18,77 @@ import me.ntrrgc.tsGenerator.ClassTransformer */ object ValueClassTransformer : ClassTransformer { + private val mappedValueClasses: MutableMap = mutableMapOf( + // don't unwrap unsigned numbers - they're mapped to Factorio types + UInt::class.starProjectedType to UInt::class.starProjectedType, + UByte::class.starProjectedType to UByte::class.starProjectedType, + UShort::class.starProjectedType to UShort::class.starProjectedType, + ULong::class.starProjectedType to ULong::class.starProjectedType, + ) + override fun transformPropertyType(type: KType, property: KProperty<*>, klass: KClass<*>): KType { - - val classifier = type.classifier ?: return super.transformPropertyType(type, property, klass) - - val valueClassType = unwrapValueClassType(type) - - if (valueClassType != null) { - println("$klass has a value class $type -> $valueClassType") - return valueClassType + val unwrappedType = + unwrapValueClassType(type, property, klass) + ?: unwrapArgs(type, property, klass) + return when (unwrappedType) { + null -> super.transformPropertyType(type, property, klass) + else -> super.transformPropertyType(unwrappedType, property, klass) } + } - val hasValueClassTypeParameters = type.arguments - .mapNotNull { it.type?.classifier } - .filterIsInstance>() - .any { it.isValue } + private fun unwrapArgs(type: KType, property: KProperty<*>, klass: KClass<*>): KType? { - if (hasValueClassTypeParameters) { + val classifier: KClassifier = type.classifier ?: return type - val args = type.arguments.map { arg -> - when (val vct = unwrapValueClassType(arg.type)) { - null -> arg - else -> KTypeProjection(arg.variance, vct.withNullability(arg.type!!.isMarkedNullable)) - } + return if (type.arguments.isNotEmpty()) { + println("unwrapping args for $type - ${type.arguments}") + + val args = type.arguments.map { arg: KTypeProjection -> + // recursively map each argument + val argType = arg.type!! + val converted = transformPropertyType(argType, property, klass) + KTypeProjection(arg.variance, converted.withNullability(argType.isMarkedNullable)) } - println("$klass has a prop with value class type parameters ${type.arguments} -> $args") - - return classifier.createType( + classifier.createType( arguments = args, nullable = type.isMarkedNullable, annotations = type.annotations, ) - } - - return super.transformPropertyType(type, property, klass) - } - - private fun unwrapValueClassType(type: KType?): KType? { - val klass = type?.classifier as? KClass<*> ?: return null - - return when { - klass.isValue -> klass - .declaredMemberProperties - .first() - .returnType - .withNullability(type.isMarkedNullable) - - else -> null + } else { + null } } + + private fun unwrapValueClassType(type: KType, property: KProperty<*>, klass: KClass<*>): KType? { + + val valueKlass: KClass<*> = type.classifier as? KClass<*> ?: return null + + return if (isValueClass(type)) { + val mappedType = mappedValueClasses.getOrPut(type.withNullability(false)) { + + val result = valueKlass + .declaredMemberProperties + .first() + .returnType + .withNullability(type.isMarkedNullable) + + println("unwrapping value class $type -> $result") + + // recursively unwrap... + val transformed = transformPropertyType(result, property, klass) + transformed.withNullability(false) + } + mappedType.withNullability(type.isMarkedNullable) + } else { + null + } + } + + @OptIn(ExperimentalContracts::class) + private fun isValueClass(type: KType?): Boolean { + contract { returnsNotNull() implies (type != null) } + return (type?.classifier as? KClass<*>)?.isValue == true + } + } diff --git a/modules/events-schema/src/jvmMain/kotlin/dev/adamko/kafkatorio/events/schema/kt2ts.kt b/modules/events-schema/src/jvmMain/kotlin/dev/adamko/kafkatorio/events/schema/kt2ts.kt index e9c51ef..8483604 100644 --- a/modules/events-schema/src/jvmMain/kotlin/dev/adamko/kafkatorio/events/schema/kt2ts.kt +++ b/modules/events-schema/src/jvmMain/kotlin/dev/adamko/kafkatorio/events/schema/kt2ts.kt @@ -21,12 +21,9 @@ fun main(args: Array) { add(FactorioPrototypes::class) add(FactorioConfigurationUpdate::class) - add(FactorioEventUpdatePacket::class) addAll(FactorioEventUpdate::class.sealedSubclasses) addAll(FactorioEventUpdateKey::class.sealedSubclasses) -// addAll(FactorioEventUpdateData::class.sealedSubclasses) - }, mappings = mapOf( // builtin Factorio numeric types > `typed-factorio/generated/builtin-types.d.ts` @@ -74,7 +71,6 @@ private fun splitDefinitions(definitions: String): Map { .split("\n\n") .groupBy { def -> - when { "interface FactorioObjectData " in def || diff --git a/modules/events-server/.gitignore b/modules/events-server/.gitignore new file mode 100644 index 0000000..e7abcfb --- /dev/null +++ b/modules/events-server/.gitignore @@ -0,0 +1 @@ +kafka-state/ diff --git a/modules/events-server/build.gradle.kts b/modules/events-server/build.gradle.kts index 4007c64..f0a19ef 100644 --- a/modules/events-server/build.gradle.kts +++ b/modules/events-server/build.gradle.kts @@ -66,7 +66,7 @@ tasks.withType { val kafkaStateDirDelete by tasks.registering(Delete::class) { group = project.name mustRunAfter(kafkaConsumersDelete) - delete(layout.buildDirectory.dir("kafka-state")) + delete(layout.projectDirectory.dir("kafka-state")) } val kafkaConsumersDelete by tasks.registering(KafkaConsumerGroupsTask.DeleteAll::class) { 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 26269b1..5fe672b 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 @@ -2,6 +2,8 @@ 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 @@ -33,6 +35,7 @@ import org.apache.kafka.streams.KafkaStreams 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 @@ -69,20 +72,29 @@ internal class KafkatorioTopology( val tileProtoColourDict = tileProtoColourDictionary(protosStream) - val mapTilesStream: KStream = +// 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.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()) + "kafkatorio.${KafkatorioPacket.PacketType.UPDATE}.${FactorioEventUpdate.FactorioEventUpdateType.MAP_CHUNK}", + consumedAs("consume.map-chunk-updates", jsonMapper.serde(), jsonMapper.serde()) ) + val groupedMapChunkTiles: KTable> = groupTilesIntoChunksWithColours( - mapTilesStream, - mapChunksStream, +// mapTilesStream, +// mapChunksStream, + mapChunksUpdatesStream, tileProtoColourDict, ) @@ -90,7 +102,11 @@ internal class KafkatorioTopology( .toStream("stream-grouped-map-tiles") .to( TOPIC_GROUPED_MAP_CHUNKS, - producedAs("produce.grouped-map-chunks", kxsBinary.serde(), kxsBinary.serde()) + producedAs( + "produce.grouped-map-chunks", + kxsBinary.serde(), + kxsBinary.serde?>() + ) ) @@ -140,10 +156,19 @@ internal class KafkatorioTopology( val appId = props.compute(StreamsConfig.APPLICATION_ID_CONFIG) { _, v -> "$v.$id" } as? String props.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, appId) - val streams = KafkaStreams(topology, props) streams.setUncaughtExceptionHandler(StreamsExceptionHandler()) + val description: TopologyDescription = topology.describe() + println( + """ + |---------------- + |$appId + |$description + |---------------- + """.trimMargin() + ) + coroutineContext.job.invokeOnCompletion { println("closing $id KafkaStreams") streams.close(Duration.ofSeconds(1)) 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 cb78bd9..28eaab6 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,10 +1,11 @@ package dev.adamko.kafkatorio.processor.topology import dev.adamko.kafkatorio.events.schema.ColourHex -import dev.adamko.kafkatorio.events.schema.FactorioEvent -import dev.adamko.kafkatorio.events.schema.MapChunk +import dev.adamko.kafkatorio.events.schema.FactorioEventUpdatePacket import dev.adamko.kafkatorio.events.schema.MapChunkPosition +import dev.adamko.kafkatorio.events.schema.MapChunkUpdate import dev.adamko.kafkatorio.events.schema.MapTile +import dev.adamko.kafkatorio.events.schema.MapTileDictionary.Companion.isNullOrEmpty import dev.adamko.kafkatorio.events.schema.MapTilePosition import dev.adamko.kafkatorio.events.schema.MapTiles import dev.adamko.kafkatorio.events.schema.SurfaceIndex @@ -13,10 +14,8 @@ 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.filter import dev.adamko.kotka.extensions.streams.flatMap import dev.adamko.kotka.extensions.streams.mapValues -import dev.adamko.kotka.extensions.streams.merge import dev.adamko.kotka.extensions.streams.reduce import dev.adamko.kotka.extensions.tables.join import dev.adamko.kotka.kxs.serde @@ -44,39 +43,70 @@ data class ServerMapChunkTiles( fun groupTilesIntoChunksWithColours( - mapTilePacketsStream: KStream, - mapChunksPacketsStream: KStream, +// mapTilePacketsStream: KStream, +// mapChunksPacketsStream: KStream, + mapChunkUpdatePacketsStream: KStream, tileProtoColourDict: KTable, ): KTable> { - val mapChunkTilesStream: KStream = - mapChunksPacketsStream - .filter { _, packet: FactorioEvent? -> - packet != null - && packet.data is MapChunk - && !((packet.data as? MapChunk)?.tiles?.tiles.isNullOrEmpty()) +// val mapChunkTilesStream: KStream = +// mapChunksPacketsStream +// .filter { _, packet: FactorioEvent? -> +// when (val data = packet?.data) { +// is MapChunk -> data.tiles.tiles.isNotEmpty() +// else -> false +// } +// } +// .mapValues("map-chunk-packets.convert-to-map-tiles") { _, packet: FactorioEvent -> +// (packet.data as MapChunk).tiles +// } +// +// +// val mapTilesStream: KStream = +// mapTilePacketsStream +// .filter { _, packet: FactorioEvent? -> +// when (val data = packet?.data) { +// is MapTiles -> data.tiles.isNotEmpty() +// else -> false +// } +// } +// .mapValues("map-tiles-packets.convert-to-map-tiles") { _, packet: FactorioEvent -> +// packet.data as MapTiles +// } + + val mapChunkUpdatesStream: KStream = + mapChunkUpdatePacketsStream + .filter { _, packet: FactorioEventUpdatePacket? -> + when (val data = packet?.update) { + is MapChunkUpdate -> !data.tileDictionary.isNullOrEmpty() + else -> false + } } - .mapValues("map-chunk-packets.convert-to-map-tiles") { _, packet: FactorioEvent -> - (packet.data as MapChunk).tiles + .mapValues("map-chunk-update-packets.convert-to-map-tiles") { _, packet: FactorioEventUpdatePacket -> + + val mapChunkUpdate: MapChunkUpdate = requireNotNull(packet.update as? MapChunkUpdate) { + "invalid UpdatePacket type $packet" + } + + val tiles: List = requireNotNull(mapChunkUpdate.tileDictionary?.toMapTileList()) { + "invalid MapChunkUpdate packet, tiles must not be null" + } + require(tiles.isNotEmpty()) { + "invalid MapChunkUpdate packet, tiles must not be empty" + } + + MapTiles(mapChunkUpdate.surfaceIndex, tiles) } - val mapTilesStream: KStream = - mapTilePacketsStream - .filter { _, packet: FactorioEvent? -> - packet != null - && packet.data is MapTiles - && !((packet.data as? MapTiles)?.tiles.isNullOrEmpty()) - } - .mapValues("map-tiles-packets.convert-to-map-tiles") { _, packet: FactorioEvent -> - packet.data as MapTiles - } // group tiles by server & surface & chunk val chunkedTilesTable: KTable> = - mapTilesStream - .merge("merge-map-chunk-tiles-with-map-tiles", mapChunkTilesStream) - .filter { _, value -> value != null && value.tiles.isNotEmpty() } + mapChunkUpdatesStream +// mapTilesStream +// .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 -> val standardChunkSize = ChunkSize.MAX @@ -102,11 +132,15 @@ 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()) + groupedAs( + "server-map-data.tiles.group", + kxsBinary.serde(), + kxsBinary.serde>() + ) ) .reduce( "server-map-data.tiles.reduce", 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 010ded3..b8fb092 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 @@ -1,6 +1,6 @@ package dev.adamko.kafkatorio.processor.topology -import dev.adamko.kafkatorio.events.schema.FactorioObjectData +import dev.adamko.kafkatorio.events.schema.FactorioEventUpdate import dev.adamko.kafkatorio.events.schema.KafkatorioPacket import dev.adamko.kafkatorio.processor.WebsocketServer import dev.adamko.kafkatorio.processor.serdes.jsonMapper @@ -15,8 +15,8 @@ fun playerUpdatesToWsServer( builder: StreamsBuilder = StreamsBuilder(), ) { builder.stream( - "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.LuaPlayer}", - consumedAs("stream-player-updates-for-ws-server", jsonMapper.serde(), jsonMapper.serde()) + "kafkatorio.${KafkatorioPacket.PacketType.UPDATE}.${FactorioEventUpdate.FactorioEventUpdateType.PLAYER}", + consumedAs("EventUpdate.Player-updates-for-ws-server", jsonMapper.serde(), jsonMapper.serde()), ) .foreach { _, value -> // println("sending ${value.packetType} packet to websocket")