From afe6a16287bc442ccca070535ff157df1fc103f2 Mon Sep 17 00:00:00 2001 From: Adam <897017+aSemy@users.noreply.github.com> Date: Mon, 2 May 2022 23:35:19 +0200 Subject: [PATCH] tidy up topology --- .../processor/KafkatorioTopology.kt | 7 +- .../processor/topology/flowMapTiles.kt | 309 ------------------ .../processor/topology/groupMapChunks.kt | 5 - .../kafkatorio/processor/topology/model.kt | 3 + .../{flowMapTiles2.kt => saveMapTiles.kt} | 175 ++++++---- .../processor/topology/SaveMapChunksTest.kt | 2 +- 6 files changed, 114 insertions(+), 387 deletions(-) delete mode 100644 modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/flowMapTiles.kt rename modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/{flowMapTiles2.kt => saveMapTiles.kt} (65%) 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 dbc940e..5bea0c6 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 @@ -4,7 +4,7 @@ import dev.adamko.kafkatorio.processor.config.ApplicationProperties import dev.adamko.kafkatorio.processor.topology.factorioServerPacketStream import dev.adamko.kafkatorio.processor.topology.groupMapChunks import dev.adamko.kafkatorio.processor.topology.playerUpdatesToWsServer -import dev.adamko.kafkatorio.processor.topology.saveMapTiles2 +import dev.adamko.kafkatorio.processor.topology.saveMapTiles import dev.adamko.kafkatorio.processor.topology.splitFactorioServerPacketStream import java.time.Duration import java.util.Properties @@ -60,8 +60,7 @@ internal class KafkatorioTopology( private fun saveTiles() { val builder = StreamsBuilder() -// val topology = saveMapTiles(builder) - val topology = saveMapTiles2(builder, appProps.webmapTileDir) + val topology = saveMapTiles(builder, appProps.webmapTileDir) launchTopology("saveTiles", topology) } @@ -87,8 +86,6 @@ internal class KafkatorioTopology( val appId = props.compute(StreamsConfig.APPLICATION_ID_CONFIG) { _, v -> "$v.$id" } as? String props.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, appId) -// props.setProperty(StreamsConfig.MAX_TASK_IDLE_MS_CONFIG, "${30.minutes.inWholeMilliseconds}") - val streams = KafkaStreams(topology, props) streams.setUncaughtExceptionHandler(StreamsExceptionHandler { coroutineContext.job.cancel(CancellationException(cause = it)) 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 deleted file mode 100644 index e771b0b..0000000 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/flowMapTiles.kt +++ /dev/null @@ -1,309 +0,0 @@ -package dev.adamko.kafkatorio.processor.topology - -import com.sksamuel.scrimage.ImmutableImage -import com.sksamuel.scrimage.ScaleMethod -import com.sksamuel.scrimage.color.RGBColor -import com.sksamuel.scrimage.nio.PngWriter -import dev.adamko.kafkatorio.processor.KafkatorioTopology.Companion.rootJob -import dev.adamko.kafkatorio.processor.admin.TOPIC_GROUPED_MAP_CHUNKS_STATE -import dev.adamko.kafkatorio.processor.serdes.kxsBinary -import dev.adamko.kafkatorio.schema.common.ColourHex -import dev.adamko.kafkatorio.schema.common.MapChunkPosition -import dev.adamko.kafkatorio.schema.common.MapTilePosition -import dev.adamko.kafkatorio.schema.common.SurfaceIndex -import dev.adamko.kafkatorio.schema.common.toMapChunkPosition -import dev.adamko.kotka.extensions.consumedAs -import dev.adamko.kotka.extensions.streams.forEach -import dev.adamko.kotka.kxs.serde -import java.awt.Color -import java.awt.image.BufferedImage -import java.io.File -import kotlin.coroutines.CoroutineContext -import kotlin.io.path.fileSize -import kotlin.math.abs -import kotlin.time.Duration.Companion.seconds -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.Flow -import kotlinx.coroutines.flow.MutableSharedFlow -import kotlinx.coroutines.flow.asFlow -import kotlinx.coroutines.flow.debounce -import kotlinx.coroutines.flow.distinctUntilChangedBy -import kotlinx.coroutines.flow.emitAll -import kotlinx.coroutines.flow.flatMapConcat -import kotlinx.coroutines.flow.launchIn -import kotlinx.coroutines.flow.map -import kotlinx.coroutines.flow.onEach -import kotlinx.coroutines.flow.runningFold -import kotlinx.coroutines.launch -import kotlinx.coroutines.supervisorScope -import kotlinx.serialization.Serializable -import org.apache.kafka.streams.StreamsBuilder -import org.apache.kafka.streams.Topology -import org.apache.kafka.streams.kstream.KStream - - -private const val WEB_MAP_TILE_SIZE_PX = 256 - - -/** A bridge between Kafka and Kotlin */ -private val serverMapChunkHandler = ServerMapChunkHandler() - - -fun saveMapTiles( - builder: StreamsBuilder, -): Topology { - val pid = "saveMapTiles" - - val groupedMapChunkTiles: KStream> = - builder.stream( - TOPIC_GROUPED_MAP_CHUNKS_STATE, - consumedAs("$pid.consume.grouped-map-chunks", kxsBinary.serde(), kxsBinary.serde()) - ) - - groupedMapChunkTiles - .forEach("save-chunked-tiles") { _, value -> - if (value != null) { - println("emitting chunkedTiles ${value.chunkId} tiles: ${value.map.size}") - serverMapChunkHandler.emit(value) - } - } - - return builder.build() -} - - -@JvmInline -value class TilePngFilename( - val value: String, -) { - constructor(id: ServerMapChunkId) : this(buildString { - append("s${id.surfaceIndex}") - append("/z${id.chunkSize.zoomLevel}") - append("/x${id.chunkPosition.x}") - append("/y${id.chunkPosition.y}") - append(".png") - }) -} - - -val TRANSPARENT_AWT: Color = ColourHex.TRANSPARENT.toRgbColor().awt() - - -fun ColourHex.toRgbColor(): RGBColor { - return RGBColor( - red.toInt(), - green.toInt(), - blue.toInt(), - alpha.toInt(), - ) -} - - -private class ServerMapChunkHandler : CoroutineScope { - - override val coroutineContext: CoroutineContext = - Dispatchers.Default + SupervisorJob(rootJob) + 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 - .runningFold(mutableMapOf>()) { acc, src -> - - val flow = acc.getOrPut(src.chunkId) { - println("creating new subdivide-flow for chunkId ${src.chunkId}") - createSubdivisionFlow() - } - - flow.emit(Cmd.ChunkSubdivide(src.chunkId, src.map)) - - acc - } - .distinctUntilChangedBy { it.size } - .onEach { - println("allChunksFlow has ${it.size} subdivide-flows: ${it.keys.joinToString()}") - } - .launchIn(this) - - - saveImagesFlow - .onEach { (filename, img) -> - - val chunkImageFile = File(filename.value) - if (chunkImageFile.parentFile.mkdirs()) { - println("created new map tile parentFile directory ${chunkImageFile.absolutePath}") - } - - val sizeBefore = chunkImageFile.takeIf { it.exists() }?.toPath()?.fileSize() - - val savedTile = img.output(PngWriter.NoCompression, chunkImageFile) - - val sizeAfter = savedTile.takeIf { it.exists() }?.toPath()?.fileSize() - if (sizeBefore != sizeAfter) { -// println("savedTile $sizeBefore/$sizeAfter: $savedTile") - } else { - println("savedTile NO-SIZE-CHANGE $sizeBefore/$sizeAfter: $savedTile") - } - } - .launchIn(this) - - } - - - /** - * 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 subdivisionFlow = MutableSharedFlow( - replay = 0, - extraBufferCapacity = 10, - onBufferOverflow = BufferOverflow.SUSPEND, - ) - - val saveImgCommands: Flow = - subdivisionFlow - .debounce(30.seconds) - .flatMapConcat { subdivideCmd -> - ChunkSize - .entries - .flatMap { chunkSize -> - subdivideCmd - .tiles - .entries - .groupBy( - { (tile, _) -> tile.toMapChunkPosition(chunkSize.lengthInTiles) } - ) { (tilePosition, colour) -> - tilePosition to colour - } - .map { (chunkPos, tiles) -> - async { - Cmd.CreateImage( - surfaceIndex = subdivideCmd.chunkId.surfaceIndex, - chunkSize = chunkSize, - chunkPosition = chunkPos, - tiles = tiles.toMap() - ) - } - } - } - .awaitAll() - .asFlow() - } - .map { createCmd -> - val img = createMapTileImage( - createCmd.chunkPosition, - createCmd.tiles, - createCmd.chunkSize, - ) - - Cmd.SaveImage(createCmd.tilePngFilename, img) - } - - launch { - supervisorScope { - println("emitting all saveImgCommands to saveImagesFlow") - saveImagesFlow.emitAll(saveImgCommands) - } - } - - 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.lengthInTiles, - chunkSize.lengthInTiles, - TRANSPARENT_AWT, - BufferedImage.TYPE_INT_ARGB - ) - - val chunkOriginX: Int = chunkPosition.x * chunkSize.lengthInTiles - val chunkOriginY: Int = chunkPosition.y * chunkSize.lengthInTiles - - 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 { - - @Serializable - data class ChunkSubdivide( - val chunkId: ServerMapChunkId, - val tiles: Map, - ) : Cmd - - @Serializable - data class CreateImage( - val surfaceIndex: SurfaceIndex, - val chunkSize: ChunkSize, - val chunkPosition: MapChunkPosition, - val tiles: Map, - ) : Cmd { - val tilePngFilename: TilePngFilename by lazy { - TilePngFilename(buildString { - append("src/main/resources/kafkatorio-web-map") - append("/s${surfaceIndex}") - append("/z${chunkSize.zoomLevel}") - append("/x${chunkPosition.x}") - append("/y${chunkPosition.y}") - append(".png") - }) - } - } - - - data class SaveImage( - val filename: TilePngFilename, - val image: ImmutableImage, - ) : 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 319c59f..eb87d07 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 @@ -107,7 +107,6 @@ private fun groupTilesIntoChunksWithColours( tileProtoColourDict: KTable, ): KTable> { - val mapTilesStream: KStream = mapChunksStream.filter { _, value -> !value.tileDictionary.isNullOrEmpty() @@ -117,10 +116,6 @@ private fun groupTilesIntoChunksWithColours( }.filter("map-chunk-update-packets.filter-out.no-tiles") { _, value -> value.tiles.isNotEmpty() } -// .peek { _, value -> -// println("[${System.currentTimeMillis()}] mapTilesStream: surface:${value.surfaceIndex}, tile count: ${value.tiles.size}") -// } - val chunkedTilesTable: KTable> = mapTilesStream 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 1060f56..44fb966 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.schema.common.ColourHex +import java.awt.Color import kotlin.math.pow import kotlin.math.roundToInt import kotlinx.serialization.Serializable diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/flowMapTiles2.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/saveMapTiles.kt similarity index 65% rename from modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/flowMapTiles2.kt rename to modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/saveMapTiles.kt index e1aa3f0..71f062b 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/flowMapTiles2.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/saveMapTiles.kt @@ -2,6 +2,7 @@ package dev.adamko.kafkatorio.processor.topology import com.sksamuel.scrimage.ImmutableImage import com.sksamuel.scrimage.ScaleMethod +import com.sksamuel.scrimage.color.RGBColor import com.sksamuel.scrimage.nio.PngWriter import dev.adamko.kafkatorio.processor.admin.TOPIC_GROUPED_MAP_CHUNKS_STATE_DEBOUNCED import dev.adamko.kafkatorio.processor.admin.TOPIC_SUBDIVIDED_MAP_TILES @@ -16,6 +17,7 @@ import dev.adamko.kotka.extensions.producedAs import dev.adamko.kotka.extensions.streams.flatMap import dev.adamko.kotka.extensions.streams.forEach import dev.adamko.kotka.kxs.serde +import java.awt.Color import java.awt.image.BufferedImage import java.nio.file.Path import kotlin.io.path.fileSize @@ -27,16 +29,18 @@ import org.apache.kafka.streams.Topology import org.apache.kafka.streams.kstream.KStream +private const val pid = "saveMapTiles" + private const val WEB_MAP_TILE_SIZE_PX = 256 + /** * @param[tileDirectory] [dev.adamko.kafkatorio.processor.config.ApplicationProperties.webmapTileDir] */ -fun saveMapTiles2( +fun saveMapTiles( builder: StreamsBuilder, tileDirectory: Path, ): Topology { - val pid = "saveMapTiles" val groupedMapChunkTiles: KStream> = builder.stream( @@ -44,44 +48,16 @@ fun saveMapTiles2( consumedAs("$pid.consume.grouped-map-chunks", kxsBinary.serde(), kxsBinary.serde()), ) - groupedMapChunkTiles.flatMap("$pid.subdivide-chunks") { initialKey, initialChunk -> - - val (chunks, time) = measureTimedValue { - ChunkSize.entries.flatMap { chunkSize -> - initialChunk.map - .entries - .groupingBy { (tilePosition, _) -> - val chunkPosition = tilePosition.toMapChunkPosition(chunkSize.lengthInTiles) -// println("$tilePosition groupingBy $chunkPosition") - initialKey.copy( - chunkSize = chunkSize, - chunkPosition = chunkPosition, - ) - }.fold( - initialValueSelector = { _, _ -> mutableMapOf() } - ) { _, accumulator, (tilePosition, colour) -> -// println("fold $tilePosition $colour") - accumulator[tilePosition] = colour - accumulator - } - .map { (chunkId, tiles) -> -// println("map $chunkId ${tiles.size}") - chunkId to ServerMapChunkTiles(chunkId, tiles) - } - } - - } - println("subdivided ${initialKey.chunkPosition} to ${chunks.size} chunks in $time") - - chunks - }.to( - TOPIC_SUBDIVIDED_MAP_TILES, - producedAs( - "$pid.produce.grouped-map-chunks", - kxsBinary.serde(), - kxsBinary.serde(), + groupedMapChunkTiles + .flatMapSubdividedChunk() + .to( + TOPIC_SUBDIVIDED_MAP_TILES, + producedAs( + "$pid.produce.grouped-map-chunks", + kxsBinary.serde(), + kxsBinary.serde(), + ) ) - ) val subdividedMapChunkTilesDebounced: KStream> = builder.stream( @@ -93,34 +69,7 @@ fun saveMapTiles2( ), ) - subdividedMapChunkTilesDebounced - .forEach("$pid.save-tiles") { chunkId: ServerMapChunkId, tileColours: ServerMapChunkTiles -> - - val chunkImage = ImmutableImage.filled( - chunkId.chunkSize.lengthInTiles, - chunkId.chunkSize.lengthInTiles, - TRANSPARENT_AWT, - BufferedImage.TYPE_INT_ARGB - ) - - val chunkOriginX: Int = chunkId.chunkPosition.x * chunkId.chunkSize.lengthInTiles - val chunkOriginY: Int = chunkId.chunkPosition.y * chunkId.chunkSize.lengthInTiles - - tileColours.map.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) - } - - val scaledImage = - chunkImage.scaleTo(WEB_MAP_TILE_SIZE_PX, WEB_MAP_TILE_SIZE_PX, ScaleMethod.FastScale) - - val filename = TilePngFilename(chunkId) - saveTile(filename, tileDirectory, scaledImage) - } + subdividedMapChunkTilesDebounced.forEachChunkSaveImage(tileDirectory) return builder.build() .addDebounceProcessor>( @@ -133,6 +82,71 @@ fun saveMapTiles2( ) } + +private fun KStream>.flatMapSubdividedChunk() = + flatMap("$pid.subdivide-chunks") { initialKey, initialChunk -> + + val (chunks, time) = measureTimedValue { + ChunkSize.entries.flatMap { chunkSize -> + initialChunk.map + .entries + .groupingBy { (tilePosition, _) -> + val chunkPosition = tilePosition.toMapChunkPosition(chunkSize.lengthInTiles) + initialKey.copy( + chunkSize = chunkSize, + chunkPosition = chunkPosition, + ) + }.fold( + initialValueSelector = { _, _ -> mutableMapOf() } + ) { _, accumulator, (tilePosition, colour) -> + accumulator[tilePosition] = colour + accumulator + } + .map { (chunkId, tiles) -> + chunkId to ServerMapChunkTiles(chunkId, tiles) + } + } + } + + println("subdivided ${initialKey.chunkPosition} to ${chunks.size} chunks in $time") + + chunks + } + + +private fun KStream>.forEachChunkSaveImage( + tileDirectory: Path, +) = forEach( + "$pid.save-tiles" +) { chunkId: ServerMapChunkId, tileColours: ServerMapChunkTiles -> + + val chunkImage = ImmutableImage.filled( + chunkId.chunkSize.lengthInTiles, + chunkId.chunkSize.lengthInTiles, + TRANSPARENT_AWT, + BufferedImage.TYPE_INT_ARGB + ) + + val chunkOriginX: Int = chunkId.chunkPosition.x * chunkId.chunkSize.lengthInTiles + val chunkOriginY: Int = chunkId.chunkPosition.y * chunkId.chunkSize.lengthInTiles + + tileColours.map.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) + } + + val scaledImage = + chunkImage.scaleTo(WEB_MAP_TILE_SIZE_PX, WEB_MAP_TILE_SIZE_PX, ScaleMethod.FastScale) + + val filename = TilePngFilename(chunkId) + saveTile(filename, tileDirectory, scaledImage) +} + + @Synchronized fun saveTile( filename: TilePngFilename, @@ -156,3 +170,30 @@ fun saveTile( } println("savedTile $sizeDescription: $savedTile") } + + +@JvmInline +value class TilePngFilename( + val value: String, +) { + constructor(id: ServerMapChunkId) : this(buildString { + append("s${id.surfaceIndex}") + append("/z${id.chunkSize.zoomLevel}") + append("/x${id.chunkPosition.x}") + append("/y${id.chunkPosition.y}") + append(".png") + }) +} + + +val TRANSPARENT_AWT: Color = ColourHex.TRANSPARENT.toRgbColor().awt() + + +fun ColourHex.toRgbColor(): RGBColor { + return RGBColor( + red.toInt(), + green.toInt(), + blue.toInt(), + alpha.toInt(), + ) +} diff --git a/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/topology/SaveMapChunksTest.kt b/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/topology/SaveMapChunksTest.kt index 3f4eaf2..9ca2067 100644 --- a/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/topology/SaveMapChunksTest.kt +++ b/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/topology/SaveMapChunksTest.kt @@ -78,7 +78,7 @@ class SaveMapChunksTest : FunSpec({ val now = LocalTime.now().format(DateTimeFormatter.ISO_TIME).filter { it.isLetterOrDigit() } Path("build/test/save-map-chunks-${now}") } - private val topology = saveMapTiles2(streamsBuilder, outputDir) + private val topology = saveMapTiles(streamsBuilder, outputDir) val testDriver: TopologyTestDriver = TopologyTestDriver(topology) val groupedMapChunksInputTopic: TestInputTopic> =