diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/admin/CreateTopics.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/admin/CreateTopics.kt index 1152dcc..5f25d92 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/admin/CreateTopics.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/admin/CreateTopics.kt @@ -73,7 +73,7 @@ object CreateTopics { } private fun Admin.createTopics( - numPartitions: Int = 10, + numPartitions: Int = 3, replicationFactor: Short = 1, topicNames: () -> Collection, ): CreateTopicsResult = 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 0a68437..fdab787 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 @@ -31,6 +31,7 @@ class KafkatorioTopology( fun build() { splitFactorioServerLog(builder) + playerUpdatesToWsServer(websocketServer, builder) val allMapTilesTable: KTable = allMapTilesTable(builder) 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 634fb6b..4e9df9a 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,6 +1,7 @@ package dev.adamko.kafkatorio.processor +import dev.adamko.kafkatorio.admin.CreateTopics import dev.adamko.kafkatorio.processor.config.ApplicationProperties import dev.adamko.kafkatorio.processor.tileserver.WebMapTileServer import kotlinx.coroutines.runBlocking @@ -12,6 +13,8 @@ val appProps = ApplicationProperties() fun main() = runBlocking { + CreateTopics.create() + val wsServer = WebsocketServer() val tileServer = WebMapTileServer() diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioMapTiles.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioMapTiles.kt index 30ed815..7a48a0d 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioMapTiles.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioMapTiles.kt @@ -8,6 +8,8 @@ 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.kafkatorio.processor.serdes.serde import dev.adamko.kotka.extensions.consumedAs import dev.adamko.kotka.extensions.materializedWith import dev.adamko.kotka.extensions.streams.filter @@ -71,7 +73,7 @@ fun allMapTilesTable( } .toTable( "all-tiles-keyed-on-position", - materializedWith(jsonMapper.serde(), jsonMapper.serde()) + materializedWith(kxsBinary.serde(), kxsBinary.serde()) ) .suppress( Suppressed.untilTimeLimit( diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioServerPackets.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioServerPackets.kt index 17fbcf1..229c3f4 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioServerPackets.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioServerPackets.kt @@ -12,7 +12,6 @@ import dev.adamko.kafkatorio.events.schema.MapTiles import dev.adamko.kafkatorio.events.schema.PlayerData import dev.adamko.kafkatorio.events.schema.SurfaceData import dev.adamko.kafkatorio.events.schema.converters.toMapChunkPosition -import dev.adamko.kafkatorio.processor.serdes.KafkatorioPacketSerde import dev.adamko.kafkatorio.processor.KafkatorioTopology import dev.adamko.kafkatorio.processor.serdes.jsonMapper import dev.adamko.kotka.extensions.consumedAs @@ -56,7 +55,7 @@ fun splitFactorioServerLog( producedAs( "split-server-log", jsonMapper.serde(), - KafkatorioPacketSerde + jsonMapper.serde(), ) ) { _: FactorioPacketKey, value: KafkatorioPacket, _: RecordContext -> // println("[$key] sending event:${value.eventType} to topic:${value.data.objectName()}") diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/playerUpdates.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/playerUpdates.kt index 86f24d7..eb51ecf 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/playerUpdates.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/playerUpdates.kt @@ -2,22 +2,21 @@ package dev.adamko.kafkatorio.processor.topology import dev.adamko.kafkatorio.events.schema.FactorioObjectData import dev.adamko.kafkatorio.events.schema.KafkatorioPacket -import dev.adamko.kafkatorio.processor.serdes.KafkatorioPacketSerde import dev.adamko.kafkatorio.processor.WebsocketServer import dev.adamko.kafkatorio.processor.serdes.jsonMapper +import dev.adamko.kotka.extensions.consumedAs +import dev.adamko.kotka.kxs.serde import kotlinx.serialization.encodeToString -import org.apache.kafka.common.serialization.Serdes import org.apache.kafka.streams.StreamsBuilder -import org.apache.kafka.streams.kstream.Consumed fun playerUpdatesToWsServer( websocketServer: WebsocketServer, builder: StreamsBuilder = StreamsBuilder(), ) { - builder.stream( + builder.stream( "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.LuaPlayer}", - Consumed.with(Serdes.String(), KafkatorioPacketSerde) + consumedAs("stream-player-updates-for-ws-server", jsonMapper.serde(), jsonMapper.serde()) ) .foreach { _, value -> // println("sending ${value.packetType} packet to websocket") diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/prototypes.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/prototypes.kt index 3f44473..0327096 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/prototypes.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/prototypes.kt @@ -4,6 +4,8 @@ import dev.adamko.kafkatorio.events.schema.FactorioPrototypes import dev.adamko.kafkatorio.events.schema.KafkatorioPacket import dev.adamko.kafkatorio.events.schema.MapTilePrototype import dev.adamko.kafkatorio.processor.serdes.jsonMapper +import dev.adamko.kafkatorio.processor.serdes.kxsBinary +import dev.adamko.kafkatorio.processor.serdes.serde import dev.adamko.kotka.extensions.consumedAs import dev.adamko.kotka.extensions.materializedWith import dev.adamko.kotka.extensions.streams.flatMap @@ -22,12 +24,12 @@ value class PrototypeName(val name: String) { fun prototypesTable(builder: StreamsBuilder): KTable { - return builder.stream( + return builder.stream( "kafkatorio.${KafkatorioPacket.PacketType.PROTOTYPES}.all", consumedAs( "consume-all-prototypes-packets", - Serdes.String(), - jsonMapper.serde() + jsonMapper.serde(), + jsonMapper.serde() ) ) .flatMap("map-MapTilePrototype") { _, prototypes: FactorioPrototypes -> @@ -37,62 +39,8 @@ fun prototypesTable(builder: StreamsBuilder): KTable, -//) -// -//fun serverPrototypesTable(builder: StreamsBuilder): KTable { -// -// return builder.stream( -// "kafkatorio.${KafkatorioPacket.PacketType.PROTOTYPES}.all", -// consumedAs( -// "consume-all-prototypes-packets", -// Serdes.String(), -// jsonMapper.serde() -// ) -// ) -// .map("build-FactorioServerPrototypesRecord", ::buildPrototypesRecord) -// .peek { _, value -> -// println("tile prototypes: ${value.mapTilePrototypes.keys.joinToString()}") -// } -// .groupByKey( -// groupedAs( -// "grouping-tile-prototypes-by-surface-index", -// jsonMapper.serde(), -// jsonMapper.serde() -// ) -// ) -// .reduce( -// "group-surface-prototypes", -// materializedWith( -//// "kafkatorio.surface-prototypes", -// jsonMapper.serde(), -// jsonMapper.serde() -// ) -// ) { value1: FactorioServerPrototypesRecord, value2: FactorioServerPrototypesRecord -> -// val allTiles = value1.mapTilePrototypes + value2.mapTilePrototypes -// value1.copy(mapTilePrototypes = allTiles) -// } -//} -// -//private fun buildPrototypesRecord( -// serverId: String, -// prototypes: FactorioPrototypes -//): Pair { -// val tilePrototypes = prototypes.prototypes -// .filterIsInstance() -// .associateBy { PrototypeName(it.name) } -// -// val key = FactorioServerId(serverId) -// val data = FactorioServerPrototypesRecord(key, tilePrototypes) -// return key to data -//} diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/webMapTileAggregate.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/webMapTileAggregate.kt index c1f04fd..7583556 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/webMapTileAggregate.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/webMapTileAggregate.kt @@ -5,7 +5,8 @@ import dev.adamko.kafkatorio.events.schema.MapTile import dev.adamko.kafkatorio.events.schema.MapTilePosition import dev.adamko.kafkatorio.events.schema.MapTilePrototype import dev.adamko.kafkatorio.events.schema.converters.toMapChunkPosition -import dev.adamko.kafkatorio.processor.serdes.jsonMapper +import dev.adamko.kafkatorio.processor.serdes.kxsBinary +import dev.adamko.kafkatorio.processor.serdes.serde import dev.adamko.kotka.extensions.groupedAs import dev.adamko.kotka.extensions.materializedAs import dev.adamko.kotka.extensions.namedAs @@ -13,7 +14,6 @@ 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.toKeyValue -import dev.adamko.kotka.kxs.serde import java.time.Duration import kotlinx.serialization.Serializable import org.apache.kafka.streams.kstream.KTable @@ -60,8 +60,8 @@ fun aggregateWebMapTiles( "convert-tiles-to-colours", materializedAs( "store-web-map-tile-pixels", - jsonMapper.serde(), - jsonMapper.serde(), + kxsBinary.serde(), + kxsBinary.serde(), ), { PrototypeName(it.prototypeName) }, ) { tile: MapTile, proto: MapTilePrototype -> @@ -74,8 +74,8 @@ fun aggregateWebMapTiles( .groupBy( groupedAs( "group-webmap-tile-colours-into-chunks", - jsonMapper.serde(), - jsonMapper.serde(), + kxsBinary.serde(), + kxsBinary.serde(), ) ) { pos: TileUpdateRecordKey, px: WebMapTilePixel -> val (chunkX, chunkY) = pos.tilePosition.toMapChunkPosition(WEB_MAP_IMAGE_TILE_SIZE) @@ -100,24 +100,24 @@ fun aggregateWebMapTiles( namedAs("web-map-tile-colours-into-$WEB_MAP_IMAGE_TILE_SIZE-chunks"), materializedAs( "web-map-tile-colour-chunks-aggregate", - jsonMapper.serde(), - jsonMapper.serde() + kxsBinary.serde(), + kxsBinary.serde() ) ) .suppress( Suppressed.untilTimeLimit( Duration.ofSeconds(30), Suppressed.BufferConfig - .maxRecords(2) + .maxRecords(30) // .withMaxBytes(31457280 ) // 30MB - .emitEarlyWhenFull() +// .emitEarlyWhenFull() ).withName("web-map-tile-aggregate-debounce") ) return webMapTileChunkAggregate .mapValues( "finalise-web-map-tile-colour-chunk-aggregation", - materializedAs("web-map-tile-colour-chunks", jsonMapper.serde(), jsonMapper.serde()) + materializedAs("web-map-tile-colour-chunks", kxsBinary.serde(), kxsBinary.serde()) ) { _, v -> WebMapTileChunkPixels(v.pixels) } diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/webMapTiles.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/webMapTiles.kt index 0ffc2b0..a1011b2 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/webMapTiles.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/webMapTiles.kt @@ -20,10 +20,10 @@ fun saveTileImages( webMapTiles .suppress( - Suppressed.untilTimeLimit( + Suppressed.untilTimeLimit( Duration.ofSeconds(30), Suppressed.BufferConfig.maxRecords(30) - ) + ).withName("save-tile-images-debounce") ) .toStream("get-webmap-tile-updates") .foreach { position: WebMapTileChunkPosition, pixels: WebMapTileChunkPixels -> @@ -37,13 +37,15 @@ fun saveTileImages( } } +//val saveMapTilesContext = Dispatchers.IO.limitedParallelism(1) + private fun saveMapTilesPng( chunkPosition: WebMapTileChunkPosition, pixels: Set ) { - val chunkOriginX: Int = chunkPosition.x * chunkPosition.chunkSize - val chunkOriginY: Int = chunkPosition.y * chunkPosition.chunkSize + val chunkOriginX: Int = chunkPosition.x * chunkPosition.chunkSize // - 1 + val chunkOriginY: Int = chunkPosition.y * chunkPosition.chunkSize // - 1 val surfaceIndex: Int = chunkPosition.surfaceIndex.index val chunkImage = @@ -76,15 +78,33 @@ private fun saveMapTilesPng( val zoom = 1u val file = - File( - "src/main/resources/kafkatorio-web-map/s${surfaceIndex}/z$zoom/x${chunkPosition.x}/y${chunkPosition.y}.png" - ) + 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") - - chunkImage.output(PngWriter.MaxCompression, 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) +// +//} 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 ff73df7..14a3326 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 @@ -33,7 +33,7 @@ fun createFactorioMap() = Maps { """http://localhost:9073/tiles/s1/z{z}/x{x}/y{y}.png""" ) { attribution = "kafkatorio" -// tileSize = 256 + tileSize = 256 tms = true // minZoom = -2 // maxZoom = 1