From f1084310335d90ade39eed0590e5c830adddb805 Mon Sep 17 00:00:00 2001 From: Adam <897017+aSemy@users.noreply.github.com> Date: Sat, 5 Feb 2022 12:20:25 +0100 Subject: [PATCH] - bump kafka sdk version - fix kafka admin so it doesn't try making topics always - move some stuff to Kotka Streams - trying to optimise and debounce with windows... but it didn't work --- gradle/libs.versions.toml | 2 +- modules/events-server/build.gradle.kts | 8 +- .../adamko/kafkatorio/admin/CreateTopics.kt | 53 ++++----- .../processor/KafkatorioTopology.kt | 8 +- .../processor/serdes/binaryFormat.kt | 82 ++++++------- .../processor/topology/factorioMapTiles.kt | 10 +- .../kafkatorio/processor/topology/model.kt | 12 ++ .../processor/topology/playerUpdates.kt | 2 +- .../processor/topology/prototypes.kt | 33 +++++- .../processor/topology/webMapTileAggregate.kt | 111 +++++++++++++----- .../processor/topology/webMapTiles.kt | 30 ++--- 11 files changed, 216 insertions(+), 135 deletions(-) diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index 7dd2fe9..625e7dd 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -11,7 +11,7 @@ kotlin = "1.6.10" ## libs ## camel-kafkaConnectors = "0.11.0" http4k = "4.17.6.0" -kafka = "3.0.0" +kafka = "3.1.0" kotlinx-serialization = "1.3.2" kvision = "5.7.1" log4j2 = "2.14.1" diff --git a/modules/events-server/build.gradle.kts b/modules/events-server/build.gradle.kts index 491230d..64c0fcf 100644 --- a/modules/events-server/build.gradle.kts +++ b/modules/events-server/build.gradle.kts @@ -38,7 +38,8 @@ dependencies { implementation("com.sksamuel.scrimage:scrimage-core:4.0.24") - val kotkaVersion = "0.0.8" +// val kotkaVersion = "0.0.11" + val kotkaVersion = "store-iterators-SNAPSHOT" implementation("com.github.adamko-dev.kotka-streams:kotka-streams:$kotkaVersion") { isChanging = true } @@ -49,6 +50,10 @@ dependencies { } +//configurations.all { +// resolutionStrategy.cacheDynamicVersionsFor(1, TimeUnit.MINUTES) +//} + application { mainClass.set("dev.adamko.kafkatorio.processor.EventProcessorKt") } @@ -56,5 +61,6 @@ application { tasks.withType { kotlinOptions.freeCompilerArgs += listOf( "-opt-in=kotlinx.coroutines.ExperimentalCoroutinesApi", + "-opt-in=kotlinx.coroutines.FlowPreview", ) } 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 0857e83..02d4520 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 @@ -28,11 +28,10 @@ object CreateTopics { fun create() { - KafkatorioPacket.PacketType.values().forEach { packetType -> + val currentTopics = currentTopics().map { it.name() } - val currentTopics = currentTopics().map { it.name() } - - val kafkatorioTopics = buildSet { + val kafkatorioTopics = buildSet { + KafkatorioPacket.PacketType.values().forEach { packetType -> add(KafkatorioTopology.sourceTopic) when (packetType) { KafkatorioPacket.PacketType.EVENT -> @@ -44,30 +43,30 @@ object CreateTopics { KafkatorioPacket.PacketType.PROTOTYPES -> add("kafkatorio.${packetType.name}.all") } - }.minus(currentTopics.toSet()) - - if (currentTopics.isNotEmpty()) { - - logger.info("Creating ${kafkatorioTopics.size} topics") - val result = kafkaAdmin.createTopics { kafkatorioTopics } - - // wait for all brokers to become aware of the created topics - Thread.sleep(TimeUnit.SECONDS.toMillis(5)) - - result.values().forEach { - logger.info("Created topic: ${it.key} ${it.value.get()}") - } - - Thread.sleep(TimeUnit.SECONDS.toMillis(5)) - - logger.info("Listing topics") - kafkaAdmin.listTopics() - .listings() - .get() - .forEach { - logger.info("Topic: $it") - } } + }.minus(currentTopics.toSet()) + + if (kafkatorioTopics.isNotEmpty()) { + + logger.info("Creating ${kafkatorioTopics.size} topics") + val result = kafkaAdmin.createTopics { kafkatorioTopics } + + // wait for all brokers to become aware of the created topics + Thread.sleep(TimeUnit.SECONDS.toMillis(5)) + + result.values().forEach { + logger.info("Created topic: ${it.key} ${it.value.get()}") + } + + Thread.sleep(TimeUnit.SECONDS.toMillis(5)) + + logger.info("Listing topics") + kafkaAdmin.listTopics() + .listings() + .get() + .forEach { + logger.info("Topic: $it") + } } } 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 fdab787..89e9438 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,7 +1,7 @@ package dev.adamko.kafkatorio.processor +import dev.adamko.kafkatorio.events.schema.Colour import dev.adamko.kafkatorio.events.schema.MapTile -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 @@ -9,9 +9,9 @@ import dev.adamko.kafkatorio.processor.topology.WebMapTileChunkPosition import dev.adamko.kafkatorio.processor.topology.aggregateWebMapTiles import dev.adamko.kafkatorio.processor.topology.allMapTilesTable import dev.adamko.kafkatorio.processor.topology.playerUpdatesToWsServer -import dev.adamko.kafkatorio.processor.topology.prototypesTable import dev.adamko.kafkatorio.processor.topology.saveTileImages import dev.adamko.kafkatorio.processor.topology.splitFactorioServerLog +import dev.adamko.kafkatorio.processor.topology.tilePrototypeColourTable import java.time.Duration import org.apache.kafka.streams.KafkaStreams import org.apache.kafka.streams.StreamsBuilder @@ -35,10 +35,10 @@ class KafkatorioTopology( playerUpdatesToWsServer(websocketServer, builder) val allMapTilesTable: KTable = allMapTilesTable(builder) - val tilePrototypesTable: KTable = prototypesTable(builder) + val tilePrototypeColourTable: KTable = tilePrototypeColourTable(builder) val webMapTiles: KTable = - aggregateWebMapTiles(allMapTilesTable, tilePrototypesTable) + aggregateWebMapTiles(allMapTilesTable, tilePrototypeColourTable) saveTileImages(webMapTiles) diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes/binaryFormat.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes/binaryFormat.kt index a1587f6..180db50 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes/binaryFormat.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes/binaryFormat.kt @@ -7,44 +7,44 @@ import org.apache.kafka.common.serialization.Deserializer import org.apache.kafka.common.serialization.Serde import org.apache.kafka.common.serialization.Serializer - -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 BinaryFormat.serde() = object : Serde { - override fun serializer(): Serializer = kafkaSerializer() - override fun deserializer(): Deserializer = kafkaDeserializer() -} +// +//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 BinaryFormat.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/factorioMapTiles.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioMapTiles.kt index 7a48a0d..6a6ee07 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 @@ -9,7 +9,6 @@ import dev.adamko.kafkatorio.events.schema.MapTilePosition import dev.adamko.kafkatorio.events.schema.MapTiles import dev.adamko.kafkatorio.processor.serdes.jsonMapper import dev.adamko.kafkatorio.processor.serdes.kxsBinary -import dev.adamko.kafkatorio.processor.serdes.serde import dev.adamko.kotka.extensions.consumedAs import dev.adamko.kotka.extensions.materializedWith import dev.adamko.kotka.extensions.streams.filter @@ -75,12 +74,5 @@ fun allMapTilesTable( "all-tiles-keyed-on-position", materializedWith(kxsBinary.serde(), kxsBinary.serde()) ) - .suppress( - Suppressed.untilTimeLimit( - Duration.ofSeconds(30), - Suppressed.BufferConfig - .maxRecords(30) - .emitEarlyWhenFull() - ) - ) + } 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 0aa1fe7..fb785c6 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 @@ -1,5 +1,8 @@ package dev.adamko.kafkatorio.processor.topology +import com.sksamuel.scrimage.color.RGBColor +import dev.adamko.kafkatorio.events.schema.Colour +import kotlin.math.roundToInt import kotlinx.serialization.Serializable @Serializable @@ -15,3 +18,12 @@ value class SurfaceIndex( value class FactorioServerId(val id: String) { override fun toString() = id } + +fun Colour.toRgbColor(): RGBColor { + return RGBColor( + red.roundToInt(), + green.roundToInt(), + blue.roundToInt(), + alpha.roundToInt(), + ) +} 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 2a51006..0843190 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 @@ -19,7 +19,7 @@ fun playerUpdatesToWsServer( consumedAs("stream-player-updates-for-ws-server", jsonMapper.serde(), jsonMapper.serde()) ) .foreach { _, value -> - println("sending ${value.packetType} packet to websocket") +// println("sending ${value.packetType} packet to websocket") websocketServer.sendMessage(jsonMapper.encodeToString(value)) } } 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 0327096..13e9ae1 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/prototypes.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/prototypes.kt @@ -1,11 +1,12 @@ package dev.adamko.kafkatorio.processor.topology +import com.sksamuel.scrimage.color.RGBColor +import dev.adamko.kafkatorio.events.schema.Colour import dev.adamko.kafkatorio.events.schema.FactorioPrototypes import dev.adamko.kafkatorio.events.schema.KafkatorioPacket import dev.adamko.kafkatorio.events.schema.MapTilePrototype import dev.adamko.kafkatorio.processor.serdes.jsonMapper import dev.adamko.kafkatorio.processor.serdes.kxsBinary -import dev.adamko.kafkatorio.processor.serdes.serde import dev.adamko.kotka.extensions.consumedAs import dev.adamko.kotka.extensions.materializedWith import dev.adamko.kotka.extensions.streams.flatMap @@ -23,7 +24,31 @@ value class PrototypeName(val name: String) { } -fun prototypesTable(builder: StreamsBuilder): KTable { +//fun prototypesTable(builder: StreamsBuilder): KTable { +// return builder.stream( +// "kafkatorio.${KafkatorioPacket.PacketType.PROTOTYPES}.all", +// consumedAs( +// "consume-all-prototypes-packets", +// jsonMapper.serde(), +// jsonMapper.serde() +// ) +// ) +// .flatMap("map-MapTilePrototype") { _, prototypes: FactorioPrototypes -> +// prototypes.prototypes +// .filterIsInstance() +// .map { PrototypeName(it.name) to it } +// } +// .toTable( +// materializedWith( +// kxsBinary.serde(), +// kxsBinary.serde() +// ) +// ) +//} + + +/** Get the [Colour] of each [MapTilePrototype] */ +fun tilePrototypeColourTable(builder: StreamsBuilder): KTable { return builder.stream( "kafkatorio.${KafkatorioPacket.PacketType.PROTOTYPES}.all", consumedAs( @@ -35,7 +60,9 @@ fun prototypesTable(builder: StreamsBuilder): KTable prototypes.prototypes .filterIsInstance() - .map { PrototypeName(it.name) to it } + .map { + PrototypeName(it.name) to it.mapColour + } } .toTable( materializedWith( 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 f266978..9f6115e 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 @@ -1,28 +1,37 @@ package dev.adamko.kafkatorio.processor.topology +import com.sksamuel.scrimage.color.RGBColor import dev.adamko.kafkatorio.events.schema.Colour import dev.adamko.kafkatorio.events.schema.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.serdes.kxsBinary -import dev.adamko.kafkatorio.processor.serdes.serde import dev.adamko.kotka.extensions.groupedAs import dev.adamko.kotka.extensions.materializedAs +import dev.adamko.kotka.extensions.materializedWith import dev.adamko.kotka.extensions.namedAs +import dev.adamko.kotka.extensions.streams.groupBy import dev.adamko.kotka.extensions.tables.groupBy import dev.adamko.kotka.extensions.tables.join import dev.adamko.kotka.extensions.tables.mapValues +import dev.adamko.kotka.extensions.tables.toStream import dev.adamko.kotka.extensions.toKeyValue +import dev.adamko.kotka.kxs.serde +import java.awt.Color import java.time.Duration import kotlinx.serialization.Serializable import org.apache.kafka.streams.kstream.KTable +import org.apache.kafka.streams.kstream.SessionWindows import org.apache.kafka.streams.kstream.Suppressed +import org.apache.kafka.streams.kstream.Windowed /** The tiles of the webmap will be this many pixels high/wide */ const val WEB_MAP_IMAGE_TILE_SIZE = 256 +/** Debounce duration */ +private val WINDOW_INACTIVITY_DURATION = Duration.ofSeconds(60) + @Serializable data class WebMapTileChunkPosition( val chunkSize: Int, @@ -32,31 +41,52 @@ data class WebMapTileChunkPosition( ) @Serializable -data class WebMapTilePixel( +private data class WebMapTilePixel( val mapColour: Colour, val tilePosition: MapTilePosition, ) @Serializable -data class WebMapTileChunkPixels( - val pixels: Set = mutableSetOf(), +@JvmInline +value class WebMapTileChunkPixels( + val pixels: Map, ) -/** Temporary mutable aggregator */ +/** Temporary mutable accumulator */ @Serializable -private data class WebMapTileChunkPixelsAcc( - val pixels: MutableSet = mutableSetOf(), -) +@JvmInline +private value class WebMapTileChunkPixelsAcc( + val pixels: MutableMap = mutableMapOf(), +) { + + operator fun plus(pixel: WebMapTilePixel): WebMapTileChunkPixelsAcc { + pixels[pixel.tilePosition] = pixel.mapColour + return this + } + + operator fun plus(other: WebMapTileChunkPixelsAcc): WebMapTileChunkPixelsAcc { + this.pixels += other.pixels + return this + } + + operator fun minus(other: WebMapTileChunkPixelsAcc): WebMapTileChunkPixelsAcc { + this.pixels -= other.pixels.keys + return this + } + +} + fun aggregateWebMapTiles( allMapTilesTable: KTable, - tilePrototypesTable: KTable, + tilePrototypeColourTable: KTable, ): KTable { - val webmapTileColours: KTable = + + val pixels: KTable = allMapTilesTable .join( - tilePrototypesTable, + tilePrototypeColourTable, "convert-tiles-to-colours", materializedAs( "store-web-map-tile-pixels", @@ -64,13 +94,14 @@ fun aggregateWebMapTiles( kxsBinary.serde(), ), { PrototypeName(it.prototypeName) }, - ) { tile: MapTile, proto: MapTilePrototype -> - WebMapTilePixel(proto.mapColour, tile.position) + ) { tile: MapTile, colour: Colour -> + WebMapTilePixel(colour, tile.position) } - val webMapTileChunkAggregate: KTable = - webmapTileColours + val chunkedPixelsWindowed: KTable, WebMapTileChunkPixelsAcc> = + pixels + .toStream("stream-webmap-tile-pixels") .groupBy( groupedAs( "group-webmap-tile-colours-into-chunks", @@ -85,17 +116,21 @@ fun aggregateWebMapTiles( chunkY, pos.surfaceIndex ) - (chunkPos to px).toKeyValue() + chunkPos } + .windowedBy( + SessionWindows.ofInactivityGapAndGrace( + WINDOW_INACTIVITY_DURATION, + WINDOW_INACTIVITY_DURATION, + ) + ) .aggregate( { WebMapTileChunkPixelsAcc() }, - /* adder */ { _: WebMapTileChunkPosition, newValue: WebMapTilePixel, acc: WebMapTileChunkPixelsAcc -> - acc.apply { pixels.add(newValue) } + acc + newValue }, - /* subtractor */ - { _: WebMapTileChunkPosition, oldValue: WebMapTilePixel, acc: WebMapTileChunkPixelsAcc -> - acc.apply { pixels.minus(oldValue) } + { _, aggOne: WebMapTileChunkPixelsAcc, aggTwo: WebMapTileChunkPixelsAcc -> + aggOne + aggTwo }, namedAs("web-map-tile-colours-into-$WEB_MAP_IMAGE_TILE_SIZE-chunks"), materializedAs( @@ -105,20 +140,34 @@ fun aggregateWebMapTiles( ) ) .suppress( - Suppressed.untilTimeLimit( - Duration.ofSeconds(30), - Suppressed.BufferConfig - .maxRecords(30) -// .withMaxBytes(31457280 ) // 30MB - .emitEarlyWhenFull() - ).withName("web-map-tile-aggregate-debounce") + Suppressed.untilWindowCloses( + Suppressed.BufferConfig.unbounded() + ).withName("chunked-pixels-debounce") ) - return webMapTileChunkAggregate + + // merge windows + val chunkedPixels: KTable = + chunkedPixelsWindowed + .groupBy( + groupedAs("merge-chunked-pixels.group-keys", kxsBinary.serde(), kxsBinary.serde()) + ) { a: Windowed, b: WebMapTileChunkPixelsAcc -> + (a.key() to b).toKeyValue() + } + .reduce( + /* adder */ + { pixels1, pixels2 -> pixels1 + pixels2 }, + /* subtractor */ + { pixels1, pixels2 -> pixels1 - pixels2 }, + namedAs("merge-chunked-pixels.reduce-values"), + materializedAs("merge-chunked-pixels.reduce-values.store", kxsBinary.serde(), kxsBinary.serde()) + ) + + return chunkedPixels .mapValues( "finalise-web-map-tile-colour-chunk-aggregation", materializedAs("web-map-tile-colour-chunks", kxsBinary.serde(), kxsBinary.serde()) - ) { _, v -> + ) { _: WebMapTileChunkPosition, v: WebMapTileChunkPixelsAcc -> WebMapTileChunkPixels(v.pixels) } } 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 a1011b2..942c517 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,28 +3,24 @@ 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.Colour +import dev.adamko.kafkatorio.events.schema.MapTilePosition import dev.adamko.kotka.extensions.tables.toStream import java.awt.Color import java.awt.image.BufferedImage import java.io.File -import java.time.Duration import kotlin.math.abs import kotlin.math.roundToInt import org.apache.kafka.streams.kstream.KTable -import org.apache.kafka.streams.kstream.Suppressed +private val COLOUR_TRANSPARENT = RGBColor.fromAwt(Color(0f, 0f, 0f, 0f)) + fun saveTileImages( webMapTiles: KTable ) { webMapTiles - .suppress( - Suppressed.untilTimeLimit( - Duration.ofSeconds(30), - Suppressed.BufferConfig.maxRecords(30) - ).withName("save-tile-images-debounce") - ) .toStream("get-webmap-tile-updates") .foreach { position: WebMapTileChunkPosition, pixels: WebMapTileChunkPixels -> runCatching { @@ -42,7 +38,7 @@ fun saveTileImages( private fun saveMapTilesPng( chunkPosition: WebMapTileChunkPosition, - pixels: Set + pixels: Map, ) { val chunkOriginX: Int = chunkPosition.x * chunkPosition.chunkSize // - 1 val chunkOriginY: Int = chunkPosition.y * chunkPosition.chunkSize // - 1 @@ -52,21 +48,21 @@ private fun saveMapTilesPng( ImmutableImage.filled( chunkPosition.chunkSize, chunkPosition.chunkSize, - Color.BLACK, + COLOUR_TRANSPARENT.awt(), BufferedImage.TYPE_INT_ARGB ) - pixels.forEach { pixel -> + pixels.forEach { (tilePosition, colour) -> val rgbColour = RGBColor( - pixel.mapColour.red.roundToInt(), - pixel.mapColour.green.roundToInt(), - pixel.mapColour.blue.roundToInt(), - pixel.mapColour.alpha.roundToInt(), + colour.red.roundToInt(), + colour.green.roundToInt(), + colour.blue.roundToInt(), + colour.alpha.roundToInt(), ) - val pixelX = abs(abs(pixel.tilePosition.x) - abs(chunkOriginX)) - val pixelY = abs(abs(pixel.tilePosition.y) - abs(chunkOriginY)) + val pixelX = abs(abs(tilePosition.x) - abs(chunkOriginX)) + val pixelY = abs(abs(tilePosition.y) - abs(chunkOriginY)) chunkImage.setColor( pixelX,