diff --git a/modules/events-server/build.gradle.kts b/modules/events-server/build.gradle.kts index 0567e7f..23e9470 100644 --- a/modules/events-server/build.gradle.kts +++ b/modules/events-server/build.gradle.kts @@ -78,6 +78,6 @@ val kafkatorioEventsServerKafkaForceReset by tasks.registering(Exec::class) { commandLine = parseSpaceSeparatedArgs(""" docker exec -d kafka $cmd """) } -tasks.run.configure { - dependsOn(kafkatorioEventsServerKafkaForceReset) -} +//tasks.run.configure { +// dependsOn(kafkatorioEventsServerKafkaForceReset) +//} 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 307c915..5431617 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,21 +7,26 @@ 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.converters.toMapChunkPosition import java.awt.image.BufferedImage import java.io.File +import kotlin.coroutines.CoroutineContext 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.channels.BufferOverflow import kotlinx.coroutines.flow.MutableSharedFlow +import kotlinx.coroutines.flow.asFlow import kotlinx.coroutines.flow.collect import kotlinx.coroutines.flow.conflate -import kotlinx.coroutines.flow.debounce -import kotlinx.coroutines.flow.distinctUntilChanged +import kotlinx.coroutines.flow.emitAll +import kotlinx.coroutines.flow.flatMapConcat import kotlinx.coroutines.flow.map +import kotlinx.coroutines.flow.onEach +import kotlinx.coroutines.flow.runningFold +import kotlinx.coroutines.job import kotlinx.coroutines.launch import kotlinx.coroutines.runBlocking import kotlinx.coroutines.supervisorScope @@ -34,13 +39,15 @@ private val cScope = CoroutineScope( Dispatchers.Default + SupervisorJob() + CoroutineName("f-server-map-updates") ) -/** A bridge between Kafka and Kotlin */ -private val chunkedTiles: MutableSharedFlow> = - MutableSharedFlow( - 0, - 1, - BufferOverflow.SUSPEND - ) +private val serverMapChunkHandler = ServerMapChunkHandler() + +///** A bridge between Kafka and Kotlin */ +//private val chunkedTiles: MutableSharedFlow> = +// MutableSharedFlow( +// 0, +// 1, +// BufferOverflow.SUSPEND +// ) fun saveMapTiles( @@ -53,29 +60,50 @@ fun saveMapTiles( if (value != null) { runBlocking(Dispatchers.Default) { println("emitting chunkedTiles ${value.chunkId} tiles: ${value.map.size}") - chunkedTiles.emit(value) +// chunkedTiles.emit(value) + serverMapChunkHandler.emit(value) } } } - cScope.launch { - supervisorScope { - chunkedTiles - .conflate() - .distinctUntilChanged() - .debounce(1.seconds) - .map { - println("createMapTileImage ${it.chunkId}") - createMapTileImage(it) - } - .collect() - } - } +// cScope.launch { +// supervisorScope { +// chunkedTiles +// .conflate() +// .distinctUntilChanged() +// .debounce(1.seconds) +// .map { +// println("createMapTileImage ${it.chunkId}") +// createMapTileImage(it) +// } +// .collect() +// } +// } // cScope.launch { // println("launching serverMapDataFlow merging") // // chunkedTiles +// .conflate() +// .flatMapConcat { chunk -> +// ZoomLevel.values +// .asFlow() +// .flatMapConcat { zoom -> +// chunk.map.entries +// .groupBy({ (tile, _) -> tile.toMapChunkPosition(zoom.chunkSize) } +// ) { +// it.key to it.value +// } +// .map { (chunkPos, tiles) -> +// val chunkId = chunk.chunkId.copy( +// zoomLevel = zoom, +// chunkPosition = chunkPos +// ) +// ServerMapChunkTiles(chunkId, tiles.toMap()) +// } +// .asFlow() +// } +// } // .runningFold( // mapOf>>() // ) { accumulator, value -> @@ -95,11 +123,14 @@ fun saveMapTiles( // supervisorScope { // println("launching MutableSharedFlow for ${value.chunkId}") // input +// .conflate() +// .distinctUntilChanged() // .debounce(30.seconds) // .mapLatest { // println("createMapTileImage ${it.chunkId}") // createMapTileImage(it) -// }.launchIn(this) +// } +// .launchIn(this) // } // // input @@ -120,49 +151,43 @@ fun saveMapTiles( private value class TilePngFilename( val value: String, ) { - - constructor( - chunkId: ServerMapChunkId - ) : this(buildString { - append("src/main/resources/kafkatorio-web-map/s") - append(chunkId.surfaceIndex.index) - append("/z") - append(chunkId.zoomLevel.level) - append("/x") - append(chunkId.chunkPosition.x) - append("/y") - append(chunkId.chunkPosition.y) + constructor(cmd: Cmd.CreateImage) : this(buildString { + append("src/main/resources/kafkatorio-web-map") + append("/s${cmd.surfaceIndex.index}") + append("/z${cmd.zoomLevel.level}") + append("/x${cmd.chunkPosition.x}") + append("/y${cmd.chunkPosition.y}") append(".png") }) } -private fun createMapTileImage( - chunkTiles: ServerMapChunkTiles -) { - - println("emitting ChunkData ${chunkTiles.chunkId.chunkPosition}") - - println("starting to save chunkData ${chunkTiles.chunkId.chunkPosition}...") - - val filename = TilePngFilename(chunkTiles.chunkId) - - val chunkImageFile = File(filename.value) - if (chunkImageFile.parentFile.mkdirs()) { - println("created new map tile parentFile directory ${chunkImageFile.absolutePath}") - } - - val image = - createMapTileImage( - chunkTiles.chunkId.chunkPosition, - chunkTiles.map, - chunkTiles.chunkId.zoomLevel - ) - - println("saving map tile $chunkImageFile") - val savedTile = image.output(PngWriter.NoCompression, chunkImageFile) - println("savedTile: $savedTile") -} - +//private fun createMapTileImage( +// chunkTiles: ServerMapChunkTiles, +// +// zoomLevel: ZoomLevel +//) { +// +// println("creating chunkData ${chunkTiles.chunkId.chunkPosition}...") +// +// val filename = TilePngFilename(chunkTiles.chunkId) +// +// val chunkImageFile = File(filename.value) +// if (chunkImageFile.parentFile.mkdirs()) { +// println("created new map tile parentFile directory ${chunkImageFile.absolutePath}") +// } +// +// val image = +// createMapTileImage( +// chunkTiles.chunkId.chunkPosition, +// chunkTiles.map, +// chunkTiles.chunkId.zoomLevel +// ) +// +// println("saving map tile $chunkImageFile") +// val savedTile = image.output(PngWriter.NoCompression, chunkImageFile) +// println("savedTile: $savedTile") +//} +// private val TRANSPARENT_AWT = ColourHex.TRANSPARENT.toRgbColor().awt() @@ -197,7 +222,7 @@ private fun createMapTileImage( ) } - return chunkImage.scaleTo(WEB_MAP_TILE_SIZE_PX, WEB_MAP_TILE_SIZE_PX, ScaleMethod.Progressive) + return chunkImage.scaleTo(WEB_MAP_TILE_SIZE_PX, WEB_MAP_TILE_SIZE_PX, ScaleMethod.FastScale) } @@ -209,3 +234,153 @@ private fun ColourHex.toRgbColor(): RGBColor { alpha.toInt(), ) } + +private class ServerMapChunkHandler : CoroutineScope { + + override val coroutineContext: CoroutineContext = Dispatchers.Default + + SupervisorJob(cScope.coroutineContext.job) + + CoroutineName("ServerMapChunkHandler") + + + private val allChunksFlow = MutableSharedFlow>( + replay = 0, + extraBufferCapacity = 10, + onBufferOverflow = BufferOverflow.SUSPEND, + ) + + private val saveImagesFlow = MutableSharedFlow( + replay = 0, + extraBufferCapacity = 10, + onBufferOverflow = BufferOverflow.SUSPEND, + ) + + + init { + + launch { + supervisorScope { + saveImagesFlow +// .conflate() + .onEach { (filename, img) -> + println("saving map tile $filename") + + val chunkImageFile = File(filename.value) + if (chunkImageFile.parentFile.mkdirs()) { + println("created new map tile parentFile directory ${chunkImageFile.absolutePath}") + } + + val savedTile = img.output(PngWriter.NoCompression, chunkImageFile) + println("savedTile: $savedTile") + } + .collect() + } + } + + launch { + supervisorScope { + allChunksFlow +// .conflate() + .runningFold(mapOf>()) { acc, src -> + + val flow = acc.getOrElse(src.chunkId) { + println("creating new subdivide-flow for chunkId ${src.chunkId}") + subdivisionFlow() + } + + launch { + flow.emit(Cmd.ChunkSubdivide(src.chunkId, src.map)) + } + + acc + (src.chunkId to flow) + } + .onEach { + println("allChunksFlow has ${it.size} subdivide-flows: ${it.keys.joinToString()}") + } + .collect() + } + } + + } + + fun emit(mapChunkTiles: ServerMapChunkTiles) { + launch { + allChunksFlow.emit(mapChunkTiles) + } + } + + private fun subdivisionFlow(): MutableSharedFlow { + + val flow = MutableSharedFlow( + replay = 0, + extraBufferCapacity = 10, + onBufferOverflow = BufferOverflow.SUSPEND, + ) + + val saveImgCommands = flow +// .conflate() + .flatMapConcat { subdivideCmd -> + ZoomLevel.values + .asFlow() + .flatMapConcat { zoom -> + subdivideCmd.tiles + .entries + .groupBy({ (tile, _) -> tile.toMapChunkPosition(zoom.chunkSize) } + ) { + it.key to it.value + } + .map { (chunkPos, tiles) -> + Cmd.CreateImage( + surfaceIndex = subdivideCmd.chunkId.surfaceIndex, + zoomLevel = zoom, + chunkPosition = chunkPos, + tiles = tiles.toMap() + ) + } + .asFlow() + } + } + .map { createCmd -> + val filename = TilePngFilename(createCmd) + + val img = createMapTileImage( + createCmd.chunkPosition, + createCmd.tiles, + createCmd.zoomLevel, + ) + + Cmd.SaveImage(filename, img) + } + + launch { + supervisorScope { + println("emitting all saveImgCommands to saveImagesFlow") + saveImagesFlow.emitAll(saveImgCommands) + } + } + + return flow + } + + +} + +private sealed interface Cmd { + + data class ChunkSubdivide( + val chunkId: ServerMapChunkId, + val tiles: Map, + ) : Cmd + + data class CreateImage( + val surfaceIndex: SurfaceIndex, + val zoomLevel: ZoomLevel, + val chunkPosition: MapChunkPosition, + val tiles: Map, + ) : Cmd + + data class SaveImage( + val filename: TilePngFilename, + val image: ImmutableImage, + ) : Cmd + +} diff --git a/modules/web-map/src/main/kotlin/dev/adamko/kafkatorio/webmap/map.kt b/modules/web-map/src/main/kotlin/dev/adamko/kafkatorio/webmap/map.kt index 934ebb6..a0ef35b 100644 --- a/modules/web-map/src/main/kotlin/dev/adamko/kafkatorio/webmap/map.kt +++ b/modules/web-map/src/main/kotlin/dev/adamko/kafkatorio/webmap/map.kt @@ -36,8 +36,8 @@ fun createFactorioMap() = Maps { tms = true // minZoom = -2 // maxZoom = 1 - maxNativeZoom = 5 - minNativeZoom = 1 + maxNativeZoom = 4 + minNativeZoom = 0 noWrap = true updateWhenIdle = true updateWhenZooming = true