From c53d1d7c8783040386a3b8fecab94691427d45f7 Mon Sep 17 00:00:00 2001 From: Adam <897017+aSemy@users.noreply.github.com> Date: Sun, 20 Feb 2022 12:36:53 +0100 Subject: [PATCH] refactoring (I can't remember what... it's untested for sure), and using the latest things from event-schema --- .../kafkatorio/processor/eventProcessor.kt | 3 +- .../processor/topology/flowMapTiles.kt | 137 ++++++++++-------- .../processor/topology/groupMapChunks.kt | 11 +- .../kafkatorio/processor/topology/model.kt | 43 +++--- .../kafkatorio/processor/misc/serdeSize.kt | 9 +- 5 files changed, 111 insertions(+), 92 deletions(-) diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/eventProcessor.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/eventProcessor.kt index e53bf6d..921e401 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/eventProcessor.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/eventProcessor.kt @@ -1,8 +1,7 @@ package dev.adamko.kafkatorio.processor -import dev.adamko.kafkatorio.admin.KafkatorioKafkaAdmin -import dev.adamko.kafkatorio.processor.config.ApplicationProperties +import dev.adamko.kafkatorio.processor.admin.KafkatorioKafkaAdmin import dev.adamko.kafkatorio.processor.tileserver.WebMapTileServer import kotlinx.coroutines.runBlocking import org.http4k.server.PolyHandler 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 25b6f4d..e604844 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 @@ -7,6 +7,7 @@ import com.sksamuel.scrimage.nio.PngWriter import dev.adamko.kafkatorio.events.schema.ColourHex import dev.adamko.kafkatorio.events.schema.MapChunkPosition import dev.adamko.kafkatorio.events.schema.MapTilePosition +import dev.adamko.kafkatorio.events.schema.SurfaceIndex import dev.adamko.kafkatorio.events.schema.converters.toMapChunkPosition import java.awt.image.BufferedImage import java.io.File @@ -16,6 +17,8 @@ import kotlinx.coroutines.CoroutineName import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.async +import kotlinx.coroutines.awaitAll import kotlinx.coroutines.channels.BufferOverflow import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.asFlow @@ -67,8 +70,8 @@ private value class TilePngFilename( ) { constructor(cmd: Cmd.CreateImage) : this(buildString { append("src/main/resources/kafkatorio-web-map") - append("/s${cmd.surfaceIndex.index}") - append("/z${cmd.zoomLevel.level}") + append("/s${cmd.surfaceIndex.value}") + append("/z${cmd.chunkSize.zoomLevel}") append("/x${cmd.chunkPosition.x}") append("/y${cmd.chunkPosition.y}") append(".png") @@ -79,40 +82,6 @@ private value class TilePngFilename( private val TRANSPARENT_AWT = ColourHex.TRANSPARENT.toRgbColor().awt() -private fun createMapTileImage( - chunkPosition: MapChunkPosition, - chunkColours: Map, - z: ZoomLevel -): ImmutableImage { - - val chunkImage = ImmutableImage.filled( - z.chunkSize, - z.chunkSize, - TRANSPARENT_AWT, - BufferedImage.TYPE_INT_ARGB - ) - - val chunkOriginX: Int = chunkPosition.x * z.chunkSize - val chunkOriginY: Int = chunkPosition.y * z.chunkSize - - chunkColours.forEach { (tilePosition, colour) -> - - val rgbColour = colour.toRgbColor() - - val pixelX = abs(abs(tilePosition.x) - abs(chunkOriginX)) - val pixelY = abs(abs(tilePosition.y) - abs(chunkOriginY)) - - chunkImage.setColor( - pixelX, - pixelY, - rgbColour - ) - } - - return chunkImage.scaleTo(WEB_MAP_TILE_SIZE_PX, WEB_MAP_TILE_SIZE_PX, ScaleMethod.FastScale) -} - - private fun ColourHex.toRgbColor(): RGBColor { return RGBColor( red.toInt(), @@ -122,6 +91,7 @@ private fun ColourHex.toRgbColor(): RGBColor { ) } + private class ServerMapChunkHandler : CoroutineScope { override val coroutineContext: CoroutineContext = Dispatchers.Default + @@ -129,19 +99,26 @@ private class ServerMapChunkHandler : CoroutineScope { CoroutineName("ServerMapChunkHandler") + /** The inbox for this handler. */ private val allChunksFlow = MutableSharedFlow>( replay = 0, extraBufferCapacity = 10, onBufferOverflow = BufferOverflow.SUSPEND, ) + fun emit(mapChunkTiles: ServerMapChunkTiles) { + launch { + allChunksFlow.emit(mapChunkTiles) + } + } + + /** This flow saves images one-by-one */ private val saveImagesFlow = MutableSharedFlow( replay = 0, extraBufferCapacity = 10, onBufferOverflow = BufferOverflow.SUSPEND, ) - init { allChunksFlow @@ -149,7 +126,7 @@ private class ServerMapChunkHandler : CoroutineScope { val flow = acc.getOrElse(src.chunkId) { println("creating new subdivide-flow for chunkId ${src.chunkId}") - subdivisionFlow() + createSubdivisionFlow() } flow.emit(Cmd.ChunkSubdivide(src.chunkId, src.map)) @@ -161,6 +138,7 @@ private class ServerMapChunkHandler : CoroutineScope { } .launchIn(this) + saveImagesFlow .onEach { (filename, img) -> @@ -176,42 +154,45 @@ private class ServerMapChunkHandler : CoroutineScope { } - fun emit(mapChunkTiles: ServerMapChunkTiles) { - launch { - allChunksFlow.emit(mapChunkTiles) - } - } - private fun subdivisionFlow(): MutableSharedFlow { + /** + * Given a chunk, split it into multiple, smaller sub-tiles. + * + * Send each sub-tile to [saveImagesFlow] to be saved as a PNG + */ + private fun createSubdivisionFlow(): MutableSharedFlow { - val flow = MutableSharedFlow( + val subdivisionFlow = MutableSharedFlow( replay = 0, extraBufferCapacity = 10, onBufferOverflow = BufferOverflow.SUSPEND, ) - val saveImgCommands = flow -// .conflate() + val saveImgCommands = subdivisionFlow .flatMapConcat { subdivideCmd -> - ZoomLevel.values - .asFlow() - .flatMapConcat { zoom -> + ChunkSize + .values + .flatMap { chunkSize -> subdivideCmd.tiles .entries - .groupBy({ (tile, _) -> tile.toMapChunkPosition(zoom.chunkSize) } + .groupBy( + { (tile, _) -> tile.toMapChunkPosition(chunkSize.tilesPerChunk) } ) { it.key to it.value } .map { (chunkPos, tiles) -> - Cmd.CreateImage( - surfaceIndex = subdivideCmd.chunkId.surfaceIndex, - zoomLevel = zoom, - chunkPosition = chunkPos, - tiles = tiles.toMap() - ) + async { + Cmd.CreateImage( + surfaceIndex = subdivideCmd.chunkId.surfaceIndex, + chunkSize = chunkSize, + chunkPosition = chunkPos, + tiles = tiles.toMap() + ) + } } - .asFlow() } + .awaitAll() + .asFlow() } .map { createCmd -> val filename = TilePngFilename(createCmd) @@ -219,7 +200,7 @@ private class ServerMapChunkHandler : CoroutineScope { val img = createMapTileImage( createCmd.chunkPosition, createCmd.tiles, - createCmd.zoomLevel, + createCmd.chunkSize, ) Cmd.SaveImage(filename, img) @@ -232,12 +213,46 @@ private class ServerMapChunkHandler : CoroutineScope { } } - return flow + return subdivisionFlow } + /** Create (but don't save) a PNG for the given chunk. */ + private fun createMapTileImage( + chunkPosition: MapChunkPosition, + chunkColours: Map, + chunkSize: ChunkSize + ): ImmutableImage { + + val chunkImage = ImmutableImage.filled( + chunkSize.tilesPerChunk, + chunkSize.tilesPerChunk, + TRANSPARENT_AWT, + BufferedImage.TYPE_INT_ARGB + ) + + val chunkOriginX: Int = chunkPosition.x * chunkSize.tilesPerChunk + val chunkOriginY: Int = chunkPosition.y * chunkSize.tilesPerChunk + + chunkColours.forEach { (tilePosition, colour) -> + + val rgbColour = colour.toRgbColor() + + val pixelX = abs(abs(tilePosition.x) - abs(chunkOriginX)) + val pixelY = abs(abs(tilePosition.y) - abs(chunkOriginY)) + + chunkImage.setColor( + pixelX, + pixelY, + rgbColour + ) + } + + return chunkImage.scaleTo(WEB_MAP_TILE_SIZE_PX, WEB_MAP_TILE_SIZE_PX, ScaleMethod.FastScale) + } } + private sealed interface Cmd { data class ChunkSubdivide( @@ -247,7 +262,7 @@ private sealed interface Cmd { data class CreateImage( val surfaceIndex: SurfaceIndex, - val zoomLevel: ZoomLevel, + val chunkSize: ChunkSize, val chunkPosition: MapChunkPosition, val tiles: Map, ) : Cmd 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 2130f77..17a1e82 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 @@ -6,6 +6,7 @@ import dev.adamko.kafkatorio.events.schema.MapChunk import dev.adamko.kafkatorio.events.schema.MapChunkPosition import dev.adamko.kafkatorio.events.schema.MapTilePosition import dev.adamko.kafkatorio.events.schema.MapTiles +import dev.adamko.kafkatorio.events.schema.SurfaceIndex import dev.adamko.kafkatorio.events.schema.converters.toMapChunkPosition import dev.adamko.kafkatorio.processor.serdes.kxsBinary import dev.adamko.kotka.extensions.groupedAs @@ -28,7 +29,7 @@ data class ServerMapChunkId( val serverId: FactorioServerId, val chunkPosition: MapChunkPosition, val surfaceIndex: SurfaceIndex, - val zoomLevel: ZoomLevel, + val chunkSize: ChunkSize, ) @@ -77,20 +78,20 @@ fun groupTilesIntoChunksWithColours( .filter { _, value -> value != null && value.tiles.isNotEmpty() } .flatMap("server-map-data.tiles.flatMapByChunk") { key: FactorioServerId, mapTiles: MapTiles -> - val zoomLevel = ZoomLevel.ZOOM_0 + val standardChunkSize = ChunkSize.MAX mapTiles .tiles .groupBy { tile -> - tile.position.toMapChunkPosition(zoomLevel.chunkSize) + tile.position.toMapChunkPosition(standardChunkSize.tilesPerChunk) } .map { (chunkPos, tiles) -> val chunkId = ServerMapChunkId( key, chunkPos, - SurfaceIndex(mapTiles.surfaceIndex), - zoomLevel, + mapTiles.surfaceIndex, + standardChunkSize, ) val tilesMap = tiles.associate { it.position to TileProtoHashCode(it) } 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 8bd855d..93375ee 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 @@ -5,15 +5,6 @@ import kotlin.math.roundToInt import kotlinx.serialization.Serializable -@Serializable -@JvmInline -value class SurfaceIndex( - val index: Int -) { - override fun toString() = "$index" -} - - @JvmInline @Serializable value class FactorioServerId(val id: String) { @@ -21,23 +12,35 @@ value class FactorioServerId(val id: String) { } -enum class ZoomLevel( - val level: Int -) { - ZOOM_0(0), // 512 - ZOOM_1(1), // 256 - ZOOM_2(2), // 128 - ZOOM_3(3), // 64 - ZOOM_4(4), // 32 +@Serializable +enum class ChunkSize( + val zoomLevel: Int, +) : Comparable { + CHUNK_512(-1), // 512 + CHUNK_256(0), // 256 + CHUNK_128(1), // 128 + CHUNK_064(2), // 64 + CHUNK_032(3), // 32 ; - val chunkSize: Int = 2f.pow(9 - level).roundToInt() + val tilesPerChunk: Int = 2f.pow(8 - zoomLevel).roundToInt() init { - require(chunkSize > 0) { "chunkSize must be positive" } + require(tilesPerChunk > 0) { "chunkSize must be positive" } + // 1000 & 0111 = 0000 => pow^2 + // 1001 & 1000 = 1000 => !pow^2 + require(tilesPerChunk and (tilesPerChunk - 1) == 0) { + "tilesPerChunk must be a power-of-two number" + } } companion object { - val values: List = values().asList() // better performance, KT-48872 + // cache values for better performance. KT-48872 + val values: List = values().asList() + + val MAX: ChunkSize = values.maxByOrNull { it.tilesPerChunk }!! + val MIN: ChunkSize = values.minByOrNull { it.tilesPerChunk }!! + val STANDARD: ChunkSize = values.first { it.zoomLevel == 0 } + } } diff --git a/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/misc/serdeSize.kt b/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/misc/serdeSize.kt index 941cc09..747ed65 100644 --- a/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/misc/serdeSize.kt +++ b/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/misc/serdeSize.kt @@ -3,16 +3,17 @@ package dev.adamko.kafkatorio.processor.misc import dev.adamko.kafkatorio.events.schema.ColourHex import dev.adamko.kafkatorio.events.schema.MapChunkPosition import dev.adamko.kafkatorio.events.schema.MapTilePosition +import dev.adamko.kafkatorio.events.schema.SurfaceIndex import dev.adamko.kafkatorio.processor.serdes.kxsBinary +import dev.adamko.kafkatorio.processor.topology.ChunkSize import dev.adamko.kafkatorio.processor.topology.FactorioServerId import dev.adamko.kafkatorio.processor.topology.ServerMapChunkId import dev.adamko.kafkatorio.processor.topology.ServerMapChunkTiles -import dev.adamko.kafkatorio.processor.topology.SurfaceIndex -import dev.adamko.kafkatorio.processor.topology.ZoomLevel import java.util.UUID import kotlinx.serialization.encodeToByteArray import org.junit.jupiter.api.Test + class Blah { @Test @@ -22,8 +23,8 @@ class Blah { chunkId = ServerMapChunkId( serverId = FactorioServerId(UUID.randomUUID().toString()), chunkPosition = MapChunkPosition(Int.MAX_VALUE, Int.MAX_VALUE), - surfaceIndex = SurfaceIndex(Int.MAX_VALUE), - zoomLevel = ZoomLevel.ZOOM_0 + surfaceIndex = SurfaceIndex(UInt.MAX_VALUE), + chunkSize = ChunkSize.MAX ), buildMap { repeat(512) { x ->