From ee3c7956cd5fd9285b41468da2d8fbb0e80f779f Mon Sep 17 00:00:00 2001 From: Adam <897017+aSemy@users.noreply.github.com> Date: Sun, 1 May 2022 15:50:44 +0200 Subject: [PATCH] events-server: fix tile grouping, chunking, saving - debounce chunk and tile updates - make the tile output-dir configurable - basic test for manual checking --- .../kafkatorio/schema/common/positions.kt | 20 +- .../processor/KafkatorioTopology.kt | 2 +- .../processor/admin/kafkatorioTopics.kt | 7 + .../processor/topology/flowMapTiles.kt | 14 +- .../processor/topology/flowMapTiles2.kt | 167 ++++++++++----- .../processor/topology/groupMapChunks.kt | 32 ++- .../kafkatorio/processor/topology/model.kt | 13 +- .../processor/topology/SaveMapChunksTest.kt | 195 ++++++++++++++++++ 8 files changed, 370 insertions(+), 80 deletions(-) create mode 100644 modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/topology/SaveMapChunksTest.kt diff --git a/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/schema/common/positions.kt b/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/schema/common/positions.kt index f6a4463..16a0739 100644 --- a/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/schema/common/positions.kt +++ b/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/schema/common/positions.kt @@ -134,9 +134,27 @@ fun MapChunkPosition.leftTopTile(chunkSize: Int = MAP_CHUNK_SIZE): MapTilePositi y * chunkSize, ) + +fun MapChunkPosition.rightBottomTile(chunkSize: Int = MAP_CHUNK_SIZE): MapTilePosition = + leftTopTile(chunkSize) + (chunkSize - 1) + + +fun MapChunkPosition.iterateTiles(chunkSize: Int = MAP_CHUNK_SIZE): Iterator { + val leftTop = leftTopTile(chunkSize) + val rightBottom = rightBottomTile(chunkSize) + return iterator { + (leftTop.x..rightBottom.x).forEach { x -> + (leftTop.y..rightBottom.y).forEach { y -> + yield(MapTilePosition(x, y)) + } + } + } +} + + //val MapChunkPosition.rightBottomTile: MapTilePosition // get() = leftTopTile + (MAP_CHUNK_SIZE - 1) -// + // //val MapChunkPosition.boundingBox: MapBoundingBox // get() = MapBoundingBox(leftTopTile, rightBottomTile) 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 d69b557..dbc940e 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 @@ -61,7 +61,7 @@ internal class KafkatorioTopology( private fun saveTiles() { val builder = StreamsBuilder() // val topology = saveMapTiles(builder) - val topology = saveMapTiles2(builder) + val topology = saveMapTiles2(builder, appProps.webmapTileDir) launchTopology("saveTiles", topology) } diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/admin/kafkatorioTopics.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/admin/kafkatorioTopics.kt index 18e483a..d4c808f 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/admin/kafkatorioTopics.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/admin/kafkatorioTopics.kt @@ -17,11 +17,18 @@ private const val DOMAIN = "kafkatorio" const val TOPIC_SRC_SERVER_LOG = "$DOMAIN.src.server-log" const val TOPIC_GROUPED_MAP_CHUNKS_STATE = "$DOMAIN.state.map-chunks.grouped" +const val TOPIC_GROUPED_MAP_CHUNKS_STATE_DEBOUNCED = "$TOPIC_GROUPED_MAP_CHUNKS_STATE.debounced" + +const val TOPIC_SUBDIVIDED_MAP_TILES = "$DOMAIN.map-tiles.subdivided" +const val TOPIC_SUBDIVIDED_MAP_TILES_DEBOUNCED = "$TOPIC_SUBDIVIDED_MAP_TILES.debounced" fun allTopics(): Set = buildSet { add(TOPIC_SRC_SERVER_LOG) add(TOPIC_GROUPED_MAP_CHUNKS_STATE) + add(TOPIC_GROUPED_MAP_CHUNKS_STATE_DEBOUNCED) + add(TOPIC_SUBDIVIDED_MAP_TILES) + add(TOPIC_SUBDIVIDED_MAP_TILES_DEBOUNCED) addAll(packetTopicNames.values) } 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 cd4b72e..e771b0b 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 @@ -4,6 +4,7 @@ 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 @@ -82,8 +83,7 @@ value class TilePngFilename( val value: String, ) { constructor(id: ServerMapChunkId) : this(buildString { - append("src/main/resources/kafkatorio-web-map") - append("/s${id.surfaceIndex}") + append("s${id.surfaceIndex}") append("/z${id.chunkSize.zoomLevel}") append("/x${id.chunkPosition.x}") append("/y${id.chunkPosition.y}") @@ -198,7 +198,7 @@ private class ServerMapChunkHandler : CoroutineScope { .tiles .entries .groupBy( - { (tile, _) -> tile.toMapChunkPosition(chunkSize.tilesPerChunk) } + { (tile, _) -> tile.toMapChunkPosition(chunkSize.lengthInTiles) } ) { (tilePosition, colour) -> tilePosition to colour } @@ -245,14 +245,14 @@ private class ServerMapChunkHandler : CoroutineScope { ): ImmutableImage { val chunkImage = ImmutableImage.filled( - chunkSize.tilesPerChunk, - chunkSize.tilesPerChunk, + chunkSize.lengthInTiles, + chunkSize.lengthInTiles, TRANSPARENT_AWT, BufferedImage.TYPE_INT_ARGB ) - val chunkOriginX: Int = chunkPosition.x * chunkSize.tilesPerChunk - val chunkOriginY: Int = chunkPosition.y * chunkSize.tilesPerChunk + val chunkOriginX: Int = chunkPosition.x * chunkSize.lengthInTiles + val chunkOriginY: Int = chunkPosition.y * chunkSize.lengthInTiles chunkColours.forEach { (tilePosition, colour) -> 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/flowMapTiles2.kt index 5cdd67e..e1aa3f0 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/flowMapTiles2.kt @@ -3,18 +3,25 @@ package dev.adamko.kafkatorio.processor.topology import com.sksamuel.scrimage.ImmutableImage import com.sksamuel.scrimage.ScaleMethod import com.sksamuel.scrimage.nio.PngWriter -import dev.adamko.kafkatorio.processor.admin.TOPIC_GROUPED_MAP_CHUNKS_STATE +import dev.adamko.kafkatorio.processor.admin.TOPIC_GROUPED_MAP_CHUNKS_STATE_DEBOUNCED +import dev.adamko.kafkatorio.processor.admin.TOPIC_SUBDIVIDED_MAP_TILES +import dev.adamko.kafkatorio.processor.admin.TOPIC_SUBDIVIDED_MAP_TILES_DEBOUNCED +import dev.adamko.kafkatorio.processor.misc.DebounceProcessor.Companion.addDebounceProcessor import dev.adamko.kafkatorio.processor.serdes.kxsBinary import dev.adamko.kafkatorio.schema.common.ColourHex +import dev.adamko.kafkatorio.schema.common.MapTilePosition import dev.adamko.kafkatorio.schema.common.toMapChunkPosition import dev.adamko.kotka.extensions.consumedAs +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.image.BufferedImage -import java.io.File +import java.nio.file.Path import kotlin.io.path.fileSize import kotlin.math.abs +import kotlin.time.Duration.Companion.seconds +import kotlin.time.measureTimedValue import org.apache.kafka.streams.StreamsBuilder import org.apache.kafka.streams.Topology import org.apache.kafka.streams.kstream.KStream @@ -22,78 +29,130 @@ import org.apache.kafka.streams.kstream.KStream private const val WEB_MAP_TILE_SIZE_PX = 256 - +/** + * @param[tileDirectory] [dev.adamko.kafkatorio.processor.config.ApplicationProperties.webmapTileDir] + */ fun saveMapTiles2( builder: StreamsBuilder, + tileDirectory: Path, ): Topology { val pid = "saveMapTiles" val groupedMapChunkTiles: KStream> = builder.stream( - TOPIC_GROUPED_MAP_CHUNKS_STATE, - consumedAs("$pid.consume.grouped-map-chunks", kxsBinary.serde(), kxsBinary.serde()) + TOPIC_GROUPED_MAP_CHUNKS_STATE_DEBOUNCED, + consumedAs("$pid.consume.grouped-map-chunks", kxsBinary.serde(), kxsBinary.serde()), ) groupedMapChunkTiles.flatMap("$pid.subdivide-chunks") { initialKey, initialChunk -> - ChunkSize.entries.flatMap { chunkSize -> - initialChunk.map.entries - .groupBy( - { (tile, _) -> tile.toMapChunkPosition(chunkSize.tilesPerChunk) } - ) { (tilePosition, colour) -> - tilePosition to colour - } - .map { (chunkPos, tiles) -> - val key = initialKey.copy( - chunkSize = chunkSize, - chunkPosition = chunkPos - ) - val value = ServerMapChunkTiles(key, tiles.toMap()) + 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) + } + } - key to value - } } - }.forEach("$pid.save-tiles") { key, value -> - val filename = TilePngFilename(key) - val chunkColours = value.map + println("subdivided ${initialKey.chunkPosition} to ${chunks.size} chunks in $time") - val chunkImage = ImmutableImage.filled( - key.chunkSize.tilesPerChunk, - key.chunkSize.tilesPerChunk, - TRANSPARENT_AWT, - BufferedImage.TYPE_INT_ARGB + chunks + }.to( + TOPIC_SUBDIVIDED_MAP_TILES, + producedAs( + "$pid.produce.grouped-map-chunks", + kxsBinary.serde(), + kxsBinary.serde(), + ) + ) + + val subdividedMapChunkTilesDebounced: KStream> = + builder.stream( + TOPIC_SUBDIVIDED_MAP_TILES_DEBOUNCED, + consumedAs( + "$pid.consume.debounced-subdivided-chunks", + kxsBinary.serde(), + kxsBinary.serde(), + ), ) - val chunkOriginX: Int = key.chunkPosition.x * key.chunkSize.tilesPerChunk - val chunkOriginY: Int = key.chunkPosition.y * key.chunkSize.tilesPerChunk + subdividedMapChunkTilesDebounced + .forEach("$pid.save-tiles") { chunkId: ServerMapChunkId, tileColours: ServerMapChunkTiles -> - chunkColours.forEach { (tilePosition, colour) -> - val rgbColour = colour.toRgbColor() + val chunkImage = ImmutableImage.filled( + chunkId.chunkSize.lengthInTiles, + chunkId.chunkSize.lengthInTiles, + TRANSPARENT_AWT, + BufferedImage.TYPE_INT_ARGB + ) - val pixelX = abs(abs(tilePosition.x) - abs(chunkOriginX)) - val pixelY = abs(abs(tilePosition.y) - abs(chunkOriginY)) + val chunkOriginX: Int = chunkId.chunkPosition.x * chunkId.chunkSize.lengthInTiles + val chunkOriginY: Int = chunkId.chunkPosition.y * chunkId.chunkSize.lengthInTiles - chunkImage.setColor(pixelX, pixelY, rgbColour) + 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) } - val img = chunkImage.scaleTo(WEB_MAP_TILE_SIZE_PX, WEB_MAP_TILE_SIZE_PX, ScaleMethod.FastScale) - - 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") - } - } - return builder.build() + .addDebounceProcessor>( + namePrefix = pid, + sourceTopic = TOPIC_SUBDIVIDED_MAP_TILES, + sinkTopic = TOPIC_SUBDIVIDED_MAP_TILES_DEBOUNCED, + inactivityDuration = 15.seconds, + keySerde = kxsBinary.serde(), + valueSerde = kxsBinary.serde(), + ) +} + +@Synchronized +fun saveTile( + filename: TilePngFilename, + tileDirectory: Path, + img: ImmutableImage, +) { + val chunkImageFile = tileDirectory.resolve(filename.value).toFile() + 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() + + val sizeDescription = if (sizeBefore != sizeAfter && sizeBefore != null) { + "$sizeBefore/$sizeAfter" + } else { + "$sizeAfter" + } + println("savedTile $sizeDescription: $savedTile") } 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 aae4413..319c59f 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,6 +1,8 @@ package dev.adamko.kafkatorio.processor.topology import dev.adamko.kafkatorio.processor.admin.TOPIC_GROUPED_MAP_CHUNKS_STATE +import dev.adamko.kafkatorio.processor.admin.TOPIC_GROUPED_MAP_CHUNKS_STATE_DEBOUNCED +import dev.adamko.kafkatorio.processor.misc.DebounceProcessor.Companion.addDebounceProcessor import dev.adamko.kafkatorio.processor.serdes.kxsBinary import dev.adamko.kafkatorio.schema.common.ColourHex import dev.adamko.kafkatorio.schema.common.MapChunkPosition @@ -13,7 +15,7 @@ import dev.adamko.kafkatorio.schema.common.toMapChunkPosition import dev.adamko.kafkatorio.schema.packets.MapChunkUpdate import dev.adamko.kafkatorio.schema.packets.PrototypesUpdate import dev.adamko.kotka.extensions.groupedAs -import dev.adamko.kotka.extensions.materializedWith +import dev.adamko.kotka.extensions.materializedAs import dev.adamko.kotka.extensions.producedAs import dev.adamko.kotka.extensions.repartitionedAs import dev.adamko.kotka.extensions.streams.filter @@ -27,6 +29,7 @@ import dev.adamko.kotka.extensions.tables.toStream import dev.adamko.kotka.kxs.serde import kotlin.collections.component1 import kotlin.collections.component2 +import kotlin.time.Duration.Companion.seconds import kotlinx.serialization.Serializable import org.apache.kafka.streams.StreamsBuilder import org.apache.kafka.streams.Topology @@ -82,12 +85,20 @@ fun groupMapChunks(builder: StreamsBuilder): Topology { TOPIC_GROUPED_MAP_CHUNKS_STATE, producedAs( "$pid.grouped-map-chunks", - kxsBinary.serde(), - kxsBinary.serde>() + kxsBinary.serde(), + kxsBinary.serde(), ) ) return builder.build() + .addDebounceProcessor>( + namePrefix = pid, + sourceTopic = TOPIC_GROUPED_MAP_CHUNKS_STATE, + sinkTopic = TOPIC_GROUPED_MAP_CHUNKS_STATE_DEBOUNCED, + inactivityDuration = 15.seconds, + keySerde = kxsBinary.serde(), + valueSerde = kxsBinary.serde(), + ) } @@ -122,7 +133,7 @@ private fun groupTilesIntoChunksWithColours( mapTiles .tiles .groupBy { tile -> - tile.position.toMapChunkPosition(standardChunkSize.tilesPerChunk) + tile.position.toMapChunkPosition(standardChunkSize.lengthInTiles) } .map { (chunkPos, tiles) -> @@ -165,9 +176,8 @@ private fun groupTilesIntoChunksWithColours( ) .reduce( "server-map-data.tiles.reduce", - materializedWith( -// materializedAs( -// "server-map-data.tiles.reduce.store", + materializedAs( + "server-map-data.tiles.reduce.store", kxsBinary.serde(), kxsBinary.serde>(), ) @@ -185,10 +195,10 @@ private fun groupTilesIntoChunksWithColours( .join( other = tileProtoColourDict, tableJoined = tableJoined("server-map-data.join-tiles-with-prototypes"), - materialized = materializedWith( -// "server-map-data.join-tiles-with-prototypes.store", - kxsBinary.serde(), - kxsBinary.serde>(), + materialized = materializedAs( + "server-map-data.join-tiles-with-prototypes.store", + kxsBinary.serde(), + kxsBinary.serde(), ), foreignKeyExtractor = { chunkTiles: ServerMapChunkTiles -> chunkTiles.chunkId.serverId 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 376e63d..1060f56 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 @@ -23,14 +23,15 @@ enum class ChunkSize( CHUNK_032(3), ; - val tilesPerChunk: Int = 2f.pow(8 - zoomLevel).roundToInt() + /** 32, 64, 128, 256, or 512 */ + val lengthInTiles: Int = 2f.pow(8 - zoomLevel).roundToInt() init { - require(tilesPerChunk > 0) { "tilesPerChunk $tilesPerChunk must be positive" } + require(lengthInTiles > 0) { "tilesPerChunk $lengthInTiles must be positive" } // 1000 & 0111 = 0000 => pow^2 // 1001 & 1000 = 1000 => !pow^2 - require(tilesPerChunk and (tilesPerChunk - 1) == 0) { - "tilesPerChunk $tilesPerChunk must be a power-of-two number" + require(lengthInTiles and (lengthInTiles - 1) == 0) { + "tilesPerChunk $lengthInTiles must be a power-of-two number" } } @@ -38,8 +39,8 @@ enum class ChunkSize( // cache values for better performance. KT-48872 val entries: Set = values().toSet() - val MAX: ChunkSize = entries.maxByOrNull { it.tilesPerChunk }!! - val MIN: ChunkSize = entries.minByOrNull { it.tilesPerChunk }!! + val MAX: ChunkSize = entries.maxByOrNull { it.lengthInTiles }!! + val MIN: ChunkSize = entries.minByOrNull { it.lengthInTiles }!! val STANDARD: ChunkSize = entries.first { it.zoomLevel == 0 } } 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 new file mode 100644 index 0000000..3f4eaf2 --- /dev/null +++ b/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/topology/SaveMapChunksTest.kt @@ -0,0 +1,195 @@ +package dev.adamko.kafkatorio.processor.topology + +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.iterateTiles +import dev.adamko.kafkatorio.schema.common.toMapChunkPosition +import dev.adamko.kotka.kxs.serde +import io.kotest.core.spec.style.FunSpec +import io.kotest.property.Arb +import io.kotest.property.PropTestConfig +import io.kotest.property.RandomSource +import io.kotest.property.arbitrary.arbitrary +import io.kotest.property.arbitrary.int +import io.kotest.property.arbitrary.uByte +import io.kotest.property.checkAll +import java.nio.file.Path +import java.time.LocalTime +import java.time.format.DateTimeFormatter +import kotlin.io.path.Path +import kotlin.io.path.absolutePathString +import kotlin.time.Duration +import kotlin.time.Duration.Companion.minutes +import kotlin.time.Duration.Companion.seconds +import kotlin.time.toJavaDuration +import kotlinx.coroutines.delay +import org.apache.kafka.streams.StreamsBuilder +import org.apache.kafka.streams.TestInputTopic +import org.apache.kafka.streams.TopologyTestDriver + + +class SaveMapChunksTest : FunSpec({ + + test("save tile") { + + + Scenario().use { scenario -> + + val chunkTilesArb = serverMapChunkTilesArb(scenario.serverId) + + checkAll( + config = PropTestConfig(iterations = 1), + chunkTilesArb, + ) { chunkTiles -> + + scenario.groupedMapChunksInputTopic.pipeInput( + chunkTiles.chunkId, + chunkTiles, + ) + + scenario += 1.minutes + delay(5.seconds) + scenario += 1.minutes +// delay(5.seconds) + scenario += 1.minutes +// delay(5.seconds) + scenario += 1.minutes +// delay(5.seconds) + + println("file:///${scenario.outputDir.absolutePathString()}") + } + } + } + +}) { + + class Scenario( + streamsBuilder: StreamsBuilder = StreamsBuilder(), + + groupedMapChunksInputTopicName: String = "kafkatorio.state.map-chunks.grouped.debounced", + + val serverId: FactorioServerId = FactorioServerId("test-server-id"), + ) : AutoCloseable { + + val outputDir: Path = run { + val now = LocalTime.now().format(DateTimeFormatter.ISO_TIME).filter { it.isLetterOrDigit() } + Path("build/test/save-map-chunks-${now}") + } + private val topology = saveMapTiles2(streamsBuilder, outputDir) + val testDriver: TopologyTestDriver = TopologyTestDriver(topology) + + val groupedMapChunksInputTopic: TestInputTopic> = + testDriver.createInputTopic( + groupedMapChunksInputTopicName, + kxsBinary.serde().serializer(), + kxsBinary.serde>().serializer(), + ) + + operator fun plusAssign(wallClockTime: Duration) { + testDriver.advanceWallClockTime(wallClockTime.toJavaDuration()) + } + + override fun close() { + testDriver.close() + } + } + + companion object { + + fun mapTilePositionArb( + xArb: Arb = Arb.int(), + yArb: Arb = Arb.int(), + ): Arb = arbitrary { + MapTilePosition(xArb.bind(), yArb.bind()) + } + + fun mapChunkPositionArb( + xArb: Arb = Arb.int(-10..10), + yArb: Arb = Arb.int(-10..10), + ): Arb = arbitrary { + MapChunkPosition(xArb.bind(), yArb.bind()) + } + + fun colourHexArb( + redArb: Arb = Arb.uByte(), + greenArb: Arb = Arb.uByte(), + blueArb: Arb = Arb.uByte(), + alphaArb: Arb = Arb.uByte(), + ): Arb = arbitrary { + ColourHex( + redArb.bind(), + greenArb.bind(), + blueArb.bind(), +// alphaArb.bind(), + 255u, + ) +// ColourHex( +// 255u, +// 0u, +// 0u, +// 255u, +// ) + } + + fun serverMapChunkIdArb( + serverId: FactorioServerId, + chunkPosition: Arb = mapChunkPositionArb(), + surfaceIndex: SurfaceIndex = SurfaceIndex(1u), + chunkSize: ChunkSize = ChunkSize.MAX, + ): Arb = arbitrary { + ServerMapChunkId( + serverId = serverId, + chunkPosition = chunkPosition.bind(), + surfaceIndex = surfaceIndex, + chunkSize = chunkSize, + ) + } + + fun tilesInChunkArb( + chunkId: ServerMapChunkId + ): Arb> = arbitrary { + + val tileColours = mutableMapOf() + + println("chunk size: ${chunkId.chunkSize} ${chunkId.chunkSize.lengthInTiles}") + + val map = chunkId.chunkPosition + .iterateTiles(chunkId.chunkSize.lengthInTiles) + .asSequence() + .map { tilePos -> + val colour = + tileColours.getOrPut(tilePos.toMapChunkPosition(ChunkSize.MIN.lengthInTiles)) { + colourHexArb().sample(RandomSource.default()).value + } + tilePos to colour + } +// .take(2500) + .toMap() + + println(map.size) + + map + } + + fun serverMapChunkTilesArb( + serverId: FactorioServerId, + ): Arb> { + + val serverMapChunkId = serverMapChunkIdArb(serverId) + + return arbitrary { + val chunkId = serverMapChunkId.bind() + + val tileColoursArb = tilesInChunkArb(serverMapChunkId.bind()) + + ServerMapChunkTiles( + chunkId = chunkId, + map = tileColoursArb.bind() + ) + } + } + } +}