delete old code, try and tidy up coroutines launching
This commit is contained in:
Adam
parent
1af5912477
commit
215ef66fbd
7 changed files with 29 additions and 623 deletions
|
|
@ -30,8 +30,6 @@ class KafkatorioTopology(
|
|||
|
||||
playerUpdatesToWsServer(websocketServer, builder)
|
||||
|
||||
// val serverMapDataTable = buildFactorioServerMap(packets, tilePrototypesTable)
|
||||
|
||||
val tileProtoColourDict = tileProtoColourDictionary(packets)
|
||||
|
||||
val tilePrototypesTable = groupTilesIntoChunksWithColours(
|
||||
|
|
@ -40,16 +38,6 @@ class KafkatorioTopology(
|
|||
)
|
||||
saveMapTiles(tilePrototypesTable)
|
||||
|
||||
|
||||
// the old way...
|
||||
// val allMapTilesTable: KTable<TileUpdateRecordKey, MapTile> = allMapTilesTable(builder)
|
||||
// val tilePrototypeColourTable: KTable<PrototypeName, Colour> = tilePrototypeColourTable(builder)
|
||||
//
|
||||
// val webMapTiles: KTable<WebMapTileChunkPosition, WebMapTileChunkPixels> =
|
||||
// aggregateWebMapTiles(allMapTilesTable, tilePrototypeColourTable)
|
||||
//
|
||||
// saveTileImages(webMapTiles)
|
||||
|
||||
val topology = builder.build()
|
||||
|
||||
val streams = KafkaStreams(topology, appProps.kafkaConfig)
|
||||
|
|
|
|||
|
|
@ -20,22 +20,6 @@ import kotlinx.serialization.Serializable
|
|||
import org.apache.kafka.streams.kstream.KStream
|
||||
import org.apache.kafka.streams.kstream.KTable
|
||||
|
||||
|
||||
//@Serializable
|
||||
//data class FactorioServerMap(
|
||||
// val serverId: FactorioServerId,
|
||||
// val tilePrototypes: Map<PrototypeName, MapTilePrototype>,
|
||||
// val surfaces: Map<SurfaceIndex, FactorioMapSurface>,
|
||||
//)
|
||||
|
||||
|
||||
//@Serializable
|
||||
//data class FactorioMapSurface(
|
||||
// val index: SurfaceIndex,
|
||||
// val tiles: Map<MapTilePosition, PrototypeName>,
|
||||
//)
|
||||
|
||||
|
||||
@Serializable
|
||||
data class ServerMapChunkId(
|
||||
val serverId: FactorioServerId,
|
||||
|
|
@ -135,62 +119,3 @@ fun groupTilesIntoChunksWithColours(
|
|||
return colourisedChunkTable
|
||||
|
||||
}
|
||||
|
||||
//@Serializable
|
||||
//private data class FactorioServerMapAggregator(
|
||||
// val tilePrototypes: SortedMap<PrototypeName, Colour> = TreeMap(),
|
||||
// val surfaces: MutableMap<SurfaceIndex, FactorioMapSurfaceAggregator> = mutableMapOf(),
|
||||
//) {
|
||||
//
|
||||
// operator fun plus(packet: KafkatorioPacket): FactorioServerMapAggregator {
|
||||
// return when (packet) {
|
||||
// is FactorioEvent -> {
|
||||
// when (val data = packet.data) {
|
||||
// is MapChunk -> plus(data)
|
||||
// is MapTiles -> plus(data)
|
||||
// else -> this
|
||||
// }
|
||||
// }
|
||||
// is FactorioPrototypes -> plus(packet)
|
||||
// is FactorioConfigurationUpdate -> this
|
||||
// }
|
||||
// }
|
||||
//
|
||||
// operator fun plus(mapTiles: MapTiles): FactorioServerMapAggregator {
|
||||
//// println("FactorioServerMapAggregator: adding ${mapTiles.tiles.size} MapTiles from surface ${mapTiles.surfaceIndex}")
|
||||
//
|
||||
// val surfaceIndex = SurfaceIndex(mapTiles.surfaceIndex)
|
||||
//
|
||||
// val surface = surfaces.getOrPut(surfaceIndex) { FactorioMapSurfaceAggregator(surfaceIndex) }
|
||||
//
|
||||
// val surfaceTiles = mapTiles.tiles.associate { it.position to PrototypeName(it.prototypeName) }
|
||||
// surface.tiles.putAll(surfaceTiles)
|
||||
//
|
||||
// return this
|
||||
// }
|
||||
//
|
||||
// operator fun plus(mapChunk: MapChunk): FactorioServerMapAggregator {
|
||||
//// println("FactorioServerMapAggregator: adding mapChunk ${mapChunk.position}")
|
||||
// return plus(mapChunk.tiles)
|
||||
// }
|
||||
//
|
||||
// operator fun plus(packet: FactorioPrototypes): FactorioServerMapAggregator {
|
||||
//// println("FactorioServerMapAggregator: adding FactorioPrototypes")
|
||||
//
|
||||
// val packetPrototypes = packet.prototypes
|
||||
// .filterIsInstance<MapTilePrototype>()
|
||||
// .associate { PrototypeName(it.name) to it.mapColour }
|
||||
//
|
||||
// tilePrototypes.putAll(packetPrototypes)
|
||||
//
|
||||
// return this
|
||||
// }
|
||||
//
|
||||
//}
|
||||
|
||||
//
|
||||
//@Serializable
|
||||
//private data class FactorioMapSurfaceAggregator(
|
||||
// val index: SurfaceIndex,
|
||||
// val tiles: MutableMap<MapTilePosition, PrototypeName> = mutableMapOf(),
|
||||
//)
|
||||
|
|
|
|||
|
|
@ -1,78 +0,0 @@
|
|||
package dev.adamko.kafkatorio.processor.topology
|
||||
|
||||
import dev.adamko.kafkatorio.events.schema.FactorioEvent
|
||||
import dev.adamko.kafkatorio.events.schema.FactorioObjectData
|
||||
import dev.adamko.kafkatorio.events.schema.KafkatorioPacket
|
||||
import dev.adamko.kafkatorio.events.schema.MapChunk
|
||||
import dev.adamko.kafkatorio.events.schema.MapTile
|
||||
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.kotka.extensions.consumedAs
|
||||
import dev.adamko.kotka.extensions.materializedWith
|
||||
import dev.adamko.kotka.extensions.streams.filter
|
||||
import dev.adamko.kotka.extensions.streams.flatMap
|
||||
import dev.adamko.kotka.extensions.streams.mapValues
|
||||
import dev.adamko.kotka.extensions.streams.toTable
|
||||
import dev.adamko.kotka.kxs.serde
|
||||
import java.time.Duration
|
||||
import kotlinx.serialization.Serializable
|
||||
import org.apache.kafka.streams.StreamsBuilder
|
||||
import org.apache.kafka.streams.kstream.KStream
|
||||
import org.apache.kafka.streams.kstream.KTable
|
||||
import org.apache.kafka.streams.kstream.Suppressed
|
||||
|
||||
|
||||
@Serializable
|
||||
data class TileUpdateRecordKey(
|
||||
val surfaceIndex: SurfaceIndex,
|
||||
val tilePosition: MapTilePosition,
|
||||
)
|
||||
|
||||
|
||||
fun allMapTilesTable(
|
||||
builder: StreamsBuilder
|
||||
): KTable<TileUpdateRecordKey, MapTile> {
|
||||
|
||||
val luaTilesUpdatesStream: KStream<FactorioServerId, MapTiles> =
|
||||
builder.stream<FactorioServerId, FactorioEvent>(
|
||||
"kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.LuaTiles}",
|
||||
consumedAs("consume-map-tiles-packets", jsonMapper.serde(), jsonMapper.serde())
|
||||
)
|
||||
.filter("events.filter.map-tiles") { _: FactorioServerId, event: FactorioEvent ->
|
||||
event.data is MapTiles
|
||||
}
|
||||
.mapValues("events.extract-map-tiles") { _, event: FactorioEvent ->
|
||||
(event.data as? MapTiles)!!
|
||||
}
|
||||
|
||||
val chunkTilesUpdateStream: KStream<FactorioServerId, MapTiles> =
|
||||
builder.stream<FactorioServerId, FactorioEvent>(
|
||||
"kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.MapChunk}",
|
||||
consumedAs("consume-map-chunk-packets", jsonMapper.serde(), jsonMapper.serde())
|
||||
)
|
||||
.filter("events.filter.map-chunks") { _: FactorioServerId, event: FactorioEvent ->
|
||||
event.data is MapChunk
|
||||
}
|
||||
.mapValues("events.extract-map-chunks") { _, event: FactorioEvent ->
|
||||
(event.data as? MapChunk)!!.tiles
|
||||
}
|
||||
|
||||
return luaTilesUpdatesStream
|
||||
.merge(chunkTilesUpdateStream)
|
||||
.flatMap("flat-map-all-tiles") { _, value ->
|
||||
value.tiles.map { tile ->
|
||||
val key = TileUpdateRecordKey(
|
||||
SurfaceIndex(value.surfaceIndex),
|
||||
tile.position
|
||||
)
|
||||
key to tile
|
||||
}
|
||||
}
|
||||
.toTable(
|
||||
"all-tiles-keyed-on-position",
|
||||
materializedWith(kxsBinary.serde(), kxsBinary.serde())
|
||||
)
|
||||
|
||||
}
|
||||
|
|
@ -59,33 +59,3 @@ fun splitFactorioServerPacketStream(
|
|||
}
|
||||
}
|
||||
}
|
||||
|
||||
//@Serializable
|
||||
//data class FactorioPacketKey(
|
||||
// val serverId: FactorioServerId,
|
||||
// /** arbitrary distinctive key */
|
||||
// val key: String,
|
||||
//)
|
||||
//
|
||||
//private fun createPacketKey(packet: KafkatorioPacket): String {
|
||||
//
|
||||
// return when (packet) {
|
||||
// is FactorioEvent -> factorioEventKey(packet)
|
||||
// is FactorioConfigurationUpdate,
|
||||
// is FactorioPrototypes -> packet.hashCode().toString()
|
||||
// }
|
||||
//}
|
||||
//
|
||||
//private fun factorioEventKey(event: FactorioEvent): String {
|
||||
// val id: String = when (val data = event.data) {
|
||||
// is ConsoleChatMessage -> data.authorPlayerIndex?.toString()
|
||||
// is EntityData -> data.type
|
||||
// is MapChunk -> data.position.toString()
|
||||
//// is MapTile -> data.position.toString()
|
||||
// is MapTiles -> data.tiles.firstOrNull()?.position?.toMapChunkPosition()?.toString()
|
||||
// is PlayerData -> (data.characterUnitNumber?.toString() ?: data.name)
|
||||
// is SurfaceData -> data.index.toString()
|
||||
// } ?: event.tick.toString()
|
||||
//
|
||||
// return id
|
||||
//}
|
||||
|
|
|
|||
|
|
@ -19,10 +19,9 @@ 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.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
|
||||
|
|
@ -39,15 +38,9 @@ private val cScope = CoroutineScope(
|
|||
Dispatchers.Default + SupervisorJob() + CoroutineName("f-server-map-updates")
|
||||
)
|
||||
|
||||
private val serverMapChunkHandler = ServerMapChunkHandler()
|
||||
|
||||
///** A bridge between Kafka and Kotlin */
|
||||
//private val chunkedTiles: MutableSharedFlow<ServerMapChunkTiles<ColourHex>> =
|
||||
// MutableSharedFlow(
|
||||
// 0,
|
||||
// 1,
|
||||
// BufferOverflow.SUSPEND
|
||||
// )
|
||||
/** A bridge between Kafka and Kotlin */
|
||||
private val serverMapChunkHandler = ServerMapChunkHandler()
|
||||
|
||||
|
||||
fun saveMapTiles(
|
||||
|
|
@ -60,90 +53,11 @@ fun saveMapTiles(
|
|||
if (value != null) {
|
||||
runBlocking(Dispatchers.Default) {
|
||||
println("emitting chunkedTiles ${value.chunkId} tiles: ${value.map.size}")
|
||||
// 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 {
|
||||
// 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<ServerMapChunkId, MutableSharedFlow<ServerMapChunkTiles<ColourHex>>>()
|
||||
// ) { accumulator, value ->
|
||||
//
|
||||
// println("folding chunkedTiles ${value.chunkId}, current flows: ${accumulator.keys}")
|
||||
//
|
||||
// val chunkFlow = accumulator.getOrElse(value.chunkId) {
|
||||
//
|
||||
// println("creating new MutableSharedFlow for ${value.chunkId}")
|
||||
//
|
||||
// val input = MutableSharedFlow<ServerMapChunkTiles<ColourHex>>(
|
||||
// replay = 0,
|
||||
// extraBufferCapacity = 1,
|
||||
// onBufferOverflow = BufferOverflow.DROP_OLDEST,
|
||||
// )
|
||||
//
|
||||
// supervisorScope {
|
||||
// println("launching MutableSharedFlow for ${value.chunkId}")
|
||||
// input
|
||||
// .conflate()
|
||||
// .distinctUntilChanged()
|
||||
// .debounce(30.seconds)
|
||||
// .mapLatest {
|
||||
// println("createMapTileImage ${it.chunkId}")
|
||||
// createMapTileImage(it)
|
||||
// }
|
||||
// .launchIn(this)
|
||||
// }
|
||||
//
|
||||
// input
|
||||
// }
|
||||
//
|
||||
// chunkFlow.emit(value)
|
||||
//
|
||||
// accumulator + (value.chunkId to chunkFlow)
|
||||
// }
|
||||
// .collect()
|
||||
//
|
||||
// println("end of cScope.launch")
|
||||
// }
|
||||
}
|
||||
|
||||
|
||||
|
|
@ -161,33 +75,6 @@ private value class TilePngFilename(
|
|||
})
|
||||
}
|
||||
|
||||
//private fun createMapTileImage(
|
||||
// chunkTiles: ServerMapChunkTiles<ColourHex>,
|
||||
//
|
||||
// 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()
|
||||
|
||||
|
|
@ -257,48 +144,36 @@ private class ServerMapChunkHandler : CoroutineScope {
|
|||
|
||||
init {
|
||||
|
||||
launch {
|
||||
supervisorScope {
|
||||
saveImagesFlow
|
||||
// .conflate()
|
||||
.onEach { (filename, img) ->
|
||||
println("saving map tile $filename")
|
||||
allChunksFlow
|
||||
.runningFold(mapOf<ServerMapChunkId, MutableSharedFlow<Cmd.ChunkSubdivide>>()) { acc, src ->
|
||||
|
||||
val chunkImageFile = File(filename.value)
|
||||
if (chunkImageFile.parentFile.mkdirs()) {
|
||||
println("created new map tile parentFile directory ${chunkImageFile.absolutePath}")
|
||||
}
|
||||
val flow = acc.getOrElse(src.chunkId) {
|
||||
println("creating new subdivide-flow for chunkId ${src.chunkId}")
|
||||
subdivisionFlow()
|
||||
}
|
||||
|
||||
val savedTile = img.output(PngWriter.NoCompression, chunkImageFile)
|
||||
println("savedTile: $savedTile")
|
||||
}
|
||||
.collect()
|
||||
flow.emit(Cmd.ChunkSubdivide(src.chunkId, src.map))
|
||||
|
||||
acc + (src.chunkId to flow)
|
||||
}
|
||||
}
|
||||
|
||||
launch {
|
||||
supervisorScope {
|
||||
allChunksFlow
|
||||
// .conflate()
|
||||
.runningFold(mapOf<ServerMapChunkId, MutableSharedFlow<Cmd.ChunkSubdivide>>()) { 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()
|
||||
.onEach {
|
||||
println("allChunksFlow has ${it.size} subdivide-flows: ${it.keys.joinToString()}")
|
||||
}
|
||||
}
|
||||
.launchIn(this)
|
||||
|
||||
saveImagesFlow
|
||||
.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")
|
||||
}
|
||||
.launchIn(this)
|
||||
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -1,173 +0,0 @@
|
|||
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.converters.toMapChunkPosition
|
||||
import dev.adamko.kafkatorio.processor.serdes.kxsBinary
|
||||
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,
|
||||
val x: Int,
|
||||
val y: Int,
|
||||
val surfaceIndex: SurfaceIndex,
|
||||
)
|
||||
|
||||
@Serializable
|
||||
private data class WebMapTilePixel(
|
||||
val mapColour: Colour,
|
||||
val tilePosition: MapTilePosition,
|
||||
)
|
||||
|
||||
@Serializable
|
||||
@JvmInline
|
||||
value class WebMapTileChunkPixels(
|
||||
val pixels: Map<MapTilePosition, Colour>,
|
||||
)
|
||||
|
||||
/** Temporary mutable accumulator */
|
||||
@Serializable
|
||||
@JvmInline
|
||||
private value class WebMapTileChunkPixelsAcc(
|
||||
val pixels: MutableMap<MapTilePosition, Colour> = 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<TileUpdateRecordKey, MapTile>,
|
||||
tilePrototypeColourTable: KTable<PrototypeName, Colour>,
|
||||
): KTable<WebMapTileChunkPosition, WebMapTileChunkPixels> {
|
||||
|
||||
|
||||
val pixels: KTable<TileUpdateRecordKey, WebMapTilePixel> =
|
||||
allMapTilesTable
|
||||
.join(
|
||||
tilePrototypeColourTable,
|
||||
"convert-tiles-to-colours",
|
||||
materializedAs(
|
||||
"store-web-map-tile-pixels",
|
||||
kxsBinary.serde(),
|
||||
kxsBinary.serde(),
|
||||
),
|
||||
{ PrototypeName(it.prototypeName) },
|
||||
) { tile: MapTile, colour: Colour ->
|
||||
WebMapTilePixel(colour, tile.position)
|
||||
}
|
||||
|
||||
|
||||
val chunkedPixelsWindowed: KTable<Windowed<WebMapTileChunkPosition>, WebMapTileChunkPixelsAcc> =
|
||||
pixels
|
||||
.toStream("stream-webmap-tile-pixels")
|
||||
.groupBy(
|
||||
groupedAs(
|
||||
"group-webmap-tile-colours-into-chunks",
|
||||
kxsBinary.serde(),
|
||||
kxsBinary.serde(),
|
||||
)
|
||||
) { pos: TileUpdateRecordKey, _: WebMapTilePixel ->
|
||||
val (chunkX, chunkY) = pos.tilePosition.toMapChunkPosition(WEB_MAP_IMAGE_TILE_SIZE)
|
||||
val chunkPos = WebMapTileChunkPosition(
|
||||
WEB_MAP_IMAGE_TILE_SIZE,
|
||||
chunkX,
|
||||
chunkY,
|
||||
pos.surfaceIndex
|
||||
)
|
||||
chunkPos
|
||||
}
|
||||
.windowedBy(
|
||||
SessionWindows.ofInactivityGapAndGrace(
|
||||
WINDOW_INACTIVITY_DURATION,
|
||||
WINDOW_INACTIVITY_DURATION,
|
||||
)
|
||||
)
|
||||
.aggregate(
|
||||
{ WebMapTileChunkPixelsAcc() },
|
||||
{ _: WebMapTileChunkPosition, newValue: WebMapTilePixel, acc: WebMapTileChunkPixelsAcc ->
|
||||
acc + newValue
|
||||
},
|
||||
{ _, aggOne: WebMapTileChunkPixelsAcc, aggTwo: WebMapTileChunkPixelsAcc ->
|
||||
aggOne + aggTwo
|
||||
},
|
||||
namedAs("web-map-tile-colours-into-$WEB_MAP_IMAGE_TILE_SIZE-chunks"),
|
||||
materializedAs(
|
||||
"web-map-tile-colour-chunks-aggregate",
|
||||
kxsBinary.serde(),
|
||||
kxsBinary.serde()
|
||||
)
|
||||
)
|
||||
.suppress(
|
||||
Suppressed.untilWindowCloses(
|
||||
Suppressed.BufferConfig.unbounded()
|
||||
).withName("chunked-pixels-debounce")
|
||||
)
|
||||
|
||||
|
||||
// merge windows
|
||||
val chunkedPixels: KTable<WebMapTileChunkPosition, WebMapTileChunkPixelsAcc> =
|
||||
chunkedPixelsWindowed
|
||||
.groupBy(
|
||||
groupedAs("merge-chunked-pixels.group-keys", kxsBinary.serde(), kxsBinary.serde())
|
||||
) { a: Windowed<WebMapTileChunkPosition>, 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())
|
||||
) { _: WebMapTileChunkPosition, v: WebMapTileChunkPixelsAcc ->
|
||||
WebMapTileChunkPixels(v.pixels)
|
||||
}
|
||||
}
|
||||
|
|
@ -1,101 +0,0 @@
|
|||
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 kotlin.math.abs
|
||||
import kotlin.math.roundToInt
|
||||
import org.apache.kafka.streams.kstream.KTable
|
||||
|
||||
//
|
||||
//val COLOUR_TRANSPARENT: RGBColor = RGBColor.fromAwt(Color(0f, 0f, 0f, 0f))
|
||||
//
|
||||
//fun saveTileImages(
|
||||
// webMapTiles: KTable<WebMapTileChunkPosition, WebMapTileChunkPixels>
|
||||
//) {
|
||||
//
|
||||
// webMapTiles
|
||||
// .toStream("get-webmap-tile-updates")
|
||||
// .foreach { position: WebMapTileChunkPosition, pixels: WebMapTileChunkPixels ->
|
||||
// runCatching {
|
||||
// saveMapTilesPng(position, pixels.pixels)
|
||||
// }.onFailure { e ->
|
||||
// println("error saving map tile png chunk: $position")
|
||||
// e.printStackTrace()
|
||||
// throw e
|
||||
// }
|
||||
// }
|
||||
//}
|
||||
//
|
||||
////val saveMapTilesContext = Dispatchers.IO.limitedParallelism(1)
|
||||
//
|
||||
//
|
||||
//private fun saveMapTilesPng(
|
||||
// chunkPosition: WebMapTileChunkPosition,
|
||||
// pixels: Map<MapTilePosition, Colour>,
|
||||
//) {
|
||||
// val chunkOriginX: Int = chunkPosition.x * chunkPosition.chunkSize // - 1
|
||||
// val chunkOriginY: Int = chunkPosition.y * chunkPosition.chunkSize // - 1
|
||||
// val surfaceIndex: Int = chunkPosition.surfaceIndex.index
|
||||
//
|
||||
// val chunkImage =
|
||||
// ImmutableImage.filled(
|
||||
// chunkPosition.chunkSize,
|
||||
// chunkPosition.chunkSize,
|
||||
// COLOUR_TRANSPARENT.awt(),
|
||||
// BufferedImage.TYPE_INT_ARGB
|
||||
// )
|
||||
//
|
||||
// pixels.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 zoom = 1u
|
||||
// val file =
|
||||
// File(tileFilename("$surfaceIndex", "$zoom", "${chunkPosition.x}", "${chunkPosition.y}"))
|
||||
//
|
||||
// if (file.parentFile.mkdirs()) {
|
||||
// println("created new map tile parentFile directory ${file.absolutePath}")
|
||||
// }
|
||||
//
|
||||
// println("saving map tile $file")
|
||||
// val savedTile = chunkImage.output(PngWriter.NoCompression, file)
|
||||
// println("savedTile: $savedTile")
|
||||
//}
|
||||
//
|
||||
//fun tileFilename(
|
||||
// surfaceIndex: String,
|
||||
// zoom: String,
|
||||
// chunkX: String,
|
||||
// chunkY: String,
|
||||
//) = "src/main/resources/kafkatorio-web-map/s${surfaceIndex}/z$zoom/x${chunkX}/y${chunkY}.png"
|
||||
//
|
||||
////fun subdivide(tile: ImmutableImage, size: Int) {
|
||||
////
|
||||
//// val halfSize = size / 2
|
||||
////
|
||||
//// val bl = tile.resizeTo(halfSize, halfSize, Position.BottomLeft)
|
||||
//// val savedTile = bl.output(PngWriter.NoCompression, file)
|
||||
////
|
||||
//// tile.resizeTo(halfSize, halfSize, Position.BottomRight)
|
||||
//// tile.resizeTo(halfSize, halfSize, Position.TopLeft)
|
||||
//// tile.resizeTo(halfSize, halfSize, Position.TopRight)
|
||||
////
|
||||
////}
|
||||
Loading…
Add table
Add a link
Reference in a new issue