Render buildings and resources
- create LuaJsonList to handle empty Lua JSON lists being sent as empty objects - fix exporting of resources and entities - split up map tiles into different layer directories - add 'chunk size' to MapChunkPosition - move tests to correct packages - bump tstl - decrease tstl logging - remove some defined, but unused, kafka topics - syslog, rest, websocket servers & web-map: decrease logging - webmap: add additional layers for buildings and resources
This commit is contained in:
Adam
parent
a805d2ad63
commit
b507081eac
72 changed files with 2167 additions and 1068 deletions
|
|
@ -20,12 +20,10 @@ suspend fun main(): Unit = coroutineScope {
|
|||
startWebServer(appProps, websocketServer)
|
||||
}
|
||||
|
||||
|
||||
launch {
|
||||
syslogSocketServer.start()
|
||||
}
|
||||
|
||||
|
||||
launch {
|
||||
eventsProcessor(appProps, syslogSocketServer, websocketServer)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -20,19 +20,15 @@ import org.apache.kafka.streams.state.StoreBuilder
|
|||
import org.apache.kafka.streams.state.Stores
|
||||
import org.apache.kafka.streams.state.TimestampedKeyValueStore
|
||||
import org.apache.kafka.streams.state.ValueAndTimestamp
|
||||
import java.time.Duration as DurationJvm
|
||||
import kotlin.time.Duration as DurationKt
|
||||
import kotlin.time.Duration
|
||||
|
||||
|
||||
class DebounceProcessor<K, V>(
|
||||
private val inactivityDuration: DurationKt,
|
||||
checkInterval: DurationKt,
|
||||
private val inactivityDuration: Duration,
|
||||
private val checkInterval: Duration,
|
||||
private val storeName: String,
|
||||
) : Processor<K, V, K, V> {
|
||||
|
||||
|
||||
private val checkInterval: DurationJvm = checkInterval.toJavaDuration()
|
||||
|
||||
private lateinit var state: State<K, V>
|
||||
|
||||
|
||||
|
|
@ -70,7 +66,7 @@ class DebounceProcessor<K, V>(
|
|||
) : State<K, V> {
|
||||
val store: TimestampedKeyValueStore<K, V> = context.getStateStore(storeName)
|
||||
val schedule: Cancellable = context.schedule(
|
||||
checkInterval,
|
||||
checkInterval.toJavaDuration(),
|
||||
PunctuationType.WALL_CLOCK_TIME,
|
||||
::checkRecords
|
||||
)
|
||||
|
|
@ -95,8 +91,8 @@ class DebounceProcessor<K, V>(
|
|||
sourceTopic: String,
|
||||
sinkTopic: String,
|
||||
|
||||
inactivityDuration: DurationKt,
|
||||
checkInterval: DurationKt = inactivityDuration / 10,
|
||||
inactivityDuration: Duration,
|
||||
checkInterval: Duration = inactivityDuration / 10,
|
||||
keySerde: Serde<K>,
|
||||
valueSerde: Serde<V>,
|
||||
storeName: String = "$namePrefix.debounce-store",
|
||||
|
|
@ -131,8 +127,8 @@ class DebounceProcessor<K, V>(
|
|||
}
|
||||
|
||||
class Supplier<K, V>(
|
||||
private val inactivityDuration: DurationKt,
|
||||
private val checkInterval: DurationKt,
|
||||
private val inactivityDuration: Duration,
|
||||
private val checkInterval: Duration,
|
||||
private val storeName: String,
|
||||
private val keySerde: Serde<K>,
|
||||
private val valueSerde: Serde<V>,
|
||||
|
|
|
|||
|
|
@ -1,12 +1,24 @@
|
|||
package dev.adamko.kafkatorio.server.processor
|
||||
|
||||
import com.github.palindromicity.syslog.dsl.SyslogFieldKeys
|
||||
import dev.adamko.kafkatorio.schema.common.FactorioServerId
|
||||
import dev.adamko.kafkatorio.schema.common.PrototypeHashCode
|
||||
import dev.adamko.kafkatorio.schema.common.ServerMapChunkId
|
||||
import dev.adamko.kafkatorio.schema.common.ServerMapTileLayer
|
||||
import dev.adamko.kafkatorio.schema.packets.MapChunkEntityUpdate
|
||||
import dev.adamko.kafkatorio.schema.packets.MapChunkResourceUpdate
|
||||
import dev.adamko.kafkatorio.schema.packets.PrototypesUpdate
|
||||
import dev.adamko.kafkatorio.server.config.ApplicationProperties
|
||||
import dev.adamko.kafkatorio.server.processor.topology.ServerMapChunkTiles
|
||||
import dev.adamko.kafkatorio.server.processor.topology.broadcastToWebsocket
|
||||
import dev.adamko.kafkatorio.server.processor.topology.colourEntityChunks
|
||||
import dev.adamko.kafkatorio.server.processor.topology.colourMapChunks
|
||||
import dev.adamko.kafkatorio.server.processor.topology.convertEntityUpdateToServerMapChunkTiles
|
||||
import dev.adamko.kafkatorio.server.processor.topology.convertResourceUpdateToServerMapChunkTiles
|
||||
import dev.adamko.kafkatorio.server.processor.topology.factorioServerPacketStream
|
||||
import dev.adamko.kafkatorio.server.processor.topology.saveMapTiles
|
||||
import dev.adamko.kafkatorio.server.processor.topology.splitFactorioServerPacketStream
|
||||
import dev.adamko.kafkatorio.server.processor.topology.streamPacketData
|
||||
import dev.adamko.kafkatorio.server.socket.SyslogSocketServer
|
||||
import dev.adamko.kafkatorio.server.web.websocket.WebmapWebsocketServer
|
||||
import java.util.Properties
|
||||
|
|
@ -35,6 +47,7 @@ import org.apache.kafka.streams.StreamsBuilder
|
|||
import org.apache.kafka.streams.StreamsConfig
|
||||
import org.apache.kafka.streams.Topology
|
||||
import org.apache.kafka.streams.TopologyDescription
|
||||
import org.apache.kafka.streams.kstream.KStream
|
||||
|
||||
|
||||
internal class KafkatorioTopology(
|
||||
|
|
@ -50,6 +63,19 @@ internal class KafkatorioTopology(
|
|||
splitPackets()
|
||||
|
||||
groupTilesMapChunks()
|
||||
|
||||
groupEntitiesMapChunks(
|
||||
layer = ServerMapTileLayer.BUILDING,
|
||||
colouredUpdatesStreamTopic = TOPIC_MAP_CHUNK_BUILDING_COLOURED_032_UPDATES,
|
||||
colouredChunkStateTopic = TOPIC_MAP_CHUNK_BUILDING_COLOURED_STATE,
|
||||
)
|
||||
|
||||
groupEntitiesMapChunks(
|
||||
layer = ServerMapTileLayer.RESOURCE,
|
||||
colouredUpdatesStreamTopic = TOPIC_MAP_CHUNK_RESOURCE_COLOURED_032_UPDATES,
|
||||
colouredChunkStateTopic = TOPIC_MAP_CHUNK_RESOURCE_COLOURED_STATE,
|
||||
)
|
||||
|
||||
saveTiles()
|
||||
|
||||
websocketBroadcaster()
|
||||
|
|
@ -69,10 +95,53 @@ internal class KafkatorioTopology(
|
|||
}
|
||||
|
||||
|
||||
/** Terrain tiles */
|
||||
private suspend fun groupTilesMapChunks() {
|
||||
val builder = StreamsBuilder()
|
||||
val topology = colourMapChunks(builder)
|
||||
launchTopology("groupTilesMapChunks", topology)
|
||||
|
||||
val protosStream: KStream<FactorioServerId, PrototypesUpdate> =
|
||||
builder.streamPacketData()
|
||||
|
||||
val topology = colourMapChunks(builder, protosStream)
|
||||
|
||||
launchTopology("groupTilesMapChunks.${ServerMapTileLayer.TERRAIN.dir}", topology)
|
||||
}
|
||||
|
||||
|
||||
/** Building and resource tiles */
|
||||
private suspend fun groupEntitiesMapChunks(
|
||||
layer: ServerMapTileLayer,
|
||||
colouredUpdatesStreamTopic: String,
|
||||
colouredChunkStateTopic: String,
|
||||
) {
|
||||
val builder = StreamsBuilder()
|
||||
|
||||
val entityChunksStream: KStream<ServerMapChunkId, ServerMapChunkTiles<PrototypeHashCode>> =
|
||||
when (layer) {
|
||||
ServerMapTileLayer.TERRAIN -> return
|
||||
|
||||
ServerMapTileLayer.RESOURCE ->
|
||||
builder.streamPacketData<MapChunkResourceUpdate>()
|
||||
.convertResourceUpdateToServerMapChunkTiles("groupEntitiesMapChunks.${layer.dir}")
|
||||
|
||||
ServerMapTileLayer.BUILDING ->
|
||||
builder.streamPacketData<MapChunkEntityUpdate>()
|
||||
.convertEntityUpdateToServerMapChunkTiles("groupEntitiesMapChunks.${layer.dir}")
|
||||
}
|
||||
|
||||
val protosStream: KStream<FactorioServerId, PrototypesUpdate> =
|
||||
builder.streamPacketData()
|
||||
|
||||
val topology = colourEntityChunks(
|
||||
builder,
|
||||
entityChunksStream,
|
||||
protosStream,
|
||||
layer,
|
||||
colouredUpdatesStreamTopic,
|
||||
colouredChunkStateTopic,
|
||||
)
|
||||
|
||||
launchTopology("groupTilesMapChunks.${layer.dir}", topology)
|
||||
}
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -5,7 +5,9 @@ import dev.adamko.kafkatorio.schema.packets.ConsoleChatUpdate
|
|||
import dev.adamko.kafkatorio.schema.packets.ConsoleCommandUpdate
|
||||
import dev.adamko.kafkatorio.schema.packets.EntityUpdate
|
||||
import dev.adamko.kafkatorio.schema.packets.KafkatorioPacketData
|
||||
import dev.adamko.kafkatorio.schema.packets.KafkatorioPacketDataError
|
||||
import dev.adamko.kafkatorio.schema.packets.MapChunkEntityUpdate
|
||||
import dev.adamko.kafkatorio.schema.packets.MapChunkResourceUpdate
|
||||
import dev.adamko.kafkatorio.schema.packets.MapChunkTileUpdate
|
||||
import dev.adamko.kafkatorio.schema.packets.PlayerUpdate
|
||||
import dev.adamko.kafkatorio.schema.packets.PrototypesUpdate
|
||||
|
|
@ -17,14 +19,21 @@ 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 = "$DOMAIN.state.map-chunks.grouped"
|
||||
|
||||
const val TOPIC_MAP_CHUNK_COLOURED_032_STATE = "$DOMAIN.state.map-chunk-032.colour"
|
||||
const val TOPIC_MAP_CHUNK_TERRAIN_COLOURED_032_UPDATES =
|
||||
"$DOMAIN.state-updates.map-chunk-032.terrain.colour"
|
||||
const val TOPIC_MAP_CHUNK_RESOURCE_COLOURED_032_UPDATES =
|
||||
"$DOMAIN.state-updates.map-chunk-032.resource.colour"
|
||||
const val TOPIC_MAP_CHUNK_BUILDING_COLOURED_032_UPDATES =
|
||||
"$DOMAIN.state-updates.map-chunk-032.building.colour"
|
||||
|
||||
const val TOPIC_MAP_CHUNK_COLOURED_STATE = "$DOMAIN.state.map-chunk.colour"
|
||||
const val TOPIC_MAP_CHUNK_TERRAIN_COLOURED_STATE = "$DOMAIN.state.map-chunk.terrain.colour"
|
||||
const val TOPIC_MAP_CHUNK_RESOURCE_COLOURED_STATE = "$DOMAIN.state.map-chunk.resource.colour"
|
||||
const val TOPIC_MAP_CHUNK_BUILDING_COLOURED_STATE = "$DOMAIN.state.map-chunk.building.colour"
|
||||
|
||||
const val TOPIC_SUBDIVIDED_MAP_TILES = "$DOMAIN.map-tiles.subdivided"
|
||||
const val TOPIC_SUBDIVIDED_MAP_TILES_DEBOUNCED = "$TOPIC_SUBDIVIDED_MAP_TILES.debounced"
|
||||
//const val TOPIC_SUBDIVIDED_MAP_TILES = "$DOMAIN.map-tiles.subdivided"
|
||||
//const val TOPIC_SUBDIVIDED_MAP_TILES_DEBOUNCED = "$TOPIC_SUBDIVIDED_MAP_TILES.debounced"
|
||||
|
||||
|
||||
/** Broadcast these events out, via websocket */
|
||||
|
|
@ -33,14 +42,18 @@ const val TOPIC_BROADCAST_TO_WEBSOCKET = "$DOMAIN.broadcast.websocket"
|
|||
fun allTopics(): Set<String> = buildSet {
|
||||
add(TOPIC_SRC_SERVER_LOG)
|
||||
|
||||
add(TOPIC_GROUPED_MAP_CHUNKS_STATE)
|
||||
// add(TOPIC_GROUPED_MAP_CHUNKS_STATE)
|
||||
|
||||
add(TOPIC_MAP_CHUNK_COLOURED_032_STATE)
|
||||
add(TOPIC_MAP_CHUNK_TERRAIN_COLOURED_032_UPDATES)
|
||||
add(TOPIC_MAP_CHUNK_RESOURCE_COLOURED_032_UPDATES)
|
||||
add(TOPIC_MAP_CHUNK_BUILDING_COLOURED_032_UPDATES)
|
||||
|
||||
add(TOPIC_MAP_CHUNK_COLOURED_STATE)
|
||||
add(TOPIC_MAP_CHUNK_TERRAIN_COLOURED_STATE)
|
||||
add(TOPIC_MAP_CHUNK_RESOURCE_COLOURED_STATE)
|
||||
add(TOPIC_MAP_CHUNK_BUILDING_COLOURED_STATE)
|
||||
|
||||
add(TOPIC_SUBDIVIDED_MAP_TILES)
|
||||
add(TOPIC_SUBDIVIDED_MAP_TILES_DEBOUNCED)
|
||||
// add(TOPIC_SUBDIVIDED_MAP_TILES)
|
||||
// add(TOPIC_SUBDIVIDED_MAP_TILES_DEBOUNCED)
|
||||
|
||||
add(TOPIC_BROADCAST_TO_WEBSOCKET)
|
||||
|
||||
|
|
@ -49,16 +62,17 @@ fun allTopics(): Set<String> = buildSet {
|
|||
|
||||
val KafkatorioPacketData.topicName: String
|
||||
get() = when (this) {
|
||||
is ConfigurationUpdate -> packetTopicNames.getValue(ConfigurationUpdate::class)
|
||||
is ConsoleChatUpdate -> packetTopicNames.getValue(ConsoleChatUpdate::class)
|
||||
is ConsoleCommandUpdate -> packetTopicNames.getValue(ConsoleCommandUpdate::class)
|
||||
is PrototypesUpdate -> packetTopicNames.getValue(PrototypesUpdate::class)
|
||||
is SurfaceUpdate -> packetTopicNames.getValue(SurfaceUpdate::class)
|
||||
is EntityUpdate -> packetTopicNames.getValue(EntityUpdate::class)
|
||||
is MapChunkTileUpdate -> packetTopicNames.getValue(MapChunkTileUpdate::class)
|
||||
is PlayerUpdate -> packetTopicNames.getValue(PlayerUpdate::class)
|
||||
is KafkatorioPacketData.Error -> packetTopicNames.getValue(KafkatorioPacketData.Error::class)
|
||||
is MapChunkEntityUpdate -> packetTopicNames.getValue(MapChunkEntityUpdate::class)
|
||||
is ConfigurationUpdate -> packetTopicNames.getValue(ConfigurationUpdate::class)
|
||||
is ConsoleChatUpdate -> packetTopicNames.getValue(ConsoleChatUpdate::class)
|
||||
is ConsoleCommandUpdate -> packetTopicNames.getValue(ConsoleCommandUpdate::class)
|
||||
is PrototypesUpdate -> packetTopicNames.getValue(PrototypesUpdate::class)
|
||||
is SurfaceUpdate -> packetTopicNames.getValue(SurfaceUpdate::class)
|
||||
is EntityUpdate -> packetTopicNames.getValue(EntityUpdate::class)
|
||||
is MapChunkTileUpdate -> packetTopicNames.getValue(MapChunkTileUpdate::class)
|
||||
is PlayerUpdate -> packetTopicNames.getValue(PlayerUpdate::class)
|
||||
is KafkatorioPacketDataError -> packetTopicNames.getValue(KafkatorioPacketDataError::class)
|
||||
is MapChunkEntityUpdate -> packetTopicNames.getValue(MapChunkEntityUpdate::class)
|
||||
is MapChunkResourceUpdate -> packetTopicNames.getValue(MapChunkResourceUpdate::class)
|
||||
}
|
||||
|
||||
|
||||
|
|
@ -72,7 +86,8 @@ private val packetTopicNames: Map<KClass<out KafkatorioPacketData>, String> = ma
|
|||
MapChunkTileUpdate::class to "map-chunk-tile",
|
||||
PlayerUpdate::class to "player",
|
||||
MapChunkEntityUpdate::class to "map-chunk-entity",
|
||||
KafkatorioPacketData.Error::class to "error",
|
||||
MapChunkResourceUpdate::class to "map-chunk-resource",
|
||||
KafkatorioPacketDataError::class to "error",
|
||||
).mapValues { (_, v) ->
|
||||
"$DOMAIN.packet.$v"
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,17 +1,20 @@
|
|||
package dev.adamko.kafkatorio.server.processor.topology
|
||||
|
||||
import dev.adamko.kafkatorio.library.kxsBinary
|
||||
import dev.adamko.kafkatorio.library.resize
|
||||
import dev.adamko.kafkatorio.library.toMapChunkPosition
|
||||
import dev.adamko.kafkatorio.schema.common.ChunkSize
|
||||
import dev.adamko.kafkatorio.schema.common.ColourHex
|
||||
import dev.adamko.kafkatorio.schema.common.FactorioServerId
|
||||
import dev.adamko.kafkatorio.schema.common.MapTilePosition
|
||||
import dev.adamko.kafkatorio.schema.common.PrototypeHashCode
|
||||
import dev.adamko.kafkatorio.schema.common.ServerMapChunkId
|
||||
import dev.adamko.kafkatorio.schema.common.leftTopTile
|
||||
import dev.adamko.kafkatorio.schema.common.toMapChunkPosition
|
||||
import dev.adamko.kafkatorio.schema.common.ServerMapTileLayer
|
||||
import dev.adamko.kafkatorio.schema.packets.MapChunkTileUpdate
|
||||
import dev.adamko.kafkatorio.schema.packets.PrototypesUpdate
|
||||
import dev.adamko.kafkatorio.server.processor.DebounceProcessor.Companion.addDebounceProcessor
|
||||
import dev.adamko.kafkatorio.server.processor.TOPIC_MAP_CHUNK_COLOURED_032_STATE
|
||||
import dev.adamko.kafkatorio.server.processor.TOPIC_MAP_CHUNK_COLOURED_STATE
|
||||
import dev.adamko.kafkatorio.server.processor.TOPIC_MAP_CHUNK_TERRAIN_COLOURED_032_UPDATES
|
||||
import dev.adamko.kafkatorio.server.processor.TOPIC_MAP_CHUNK_TERRAIN_COLOURED_STATE
|
||||
import dev.adamko.kotka.extensions.consumedAs
|
||||
import dev.adamko.kotka.extensions.groupedAs
|
||||
import dev.adamko.kotka.extensions.materializedAs
|
||||
|
|
@ -19,8 +22,8 @@ import dev.adamko.kotka.extensions.producedAs
|
|||
import dev.adamko.kotka.extensions.repartitionedAs
|
||||
import dev.adamko.kotka.extensions.stream
|
||||
import dev.adamko.kotka.extensions.streams.filter
|
||||
import dev.adamko.kotka.extensions.streams.flatMap
|
||||
import dev.adamko.kotka.extensions.streams.map
|
||||
import dev.adamko.kotka.extensions.streams.mapValues
|
||||
import dev.adamko.kotka.extensions.streams.peek
|
||||
import dev.adamko.kotka.extensions.streams.reduce
|
||||
import dev.adamko.kotka.extensions.tableJoined
|
||||
|
|
@ -36,27 +39,27 @@ import org.apache.kafka.streams.kstream.KStream
|
|||
import org.apache.kafka.streams.kstream.KTable
|
||||
|
||||
|
||||
private const val pid: String = "colourMapChunks"
|
||||
|
||||
|
||||
/** Add colour to each [MapChunkTileUpdate] */
|
||||
fun colourMapChunks(builder: StreamsBuilder): Topology {
|
||||
fun colourMapChunks(
|
||||
builder: StreamsBuilder,
|
||||
protosStream: KStream<FactorioServerId, PrototypesUpdate>
|
||||
): Topology {
|
||||
val pid = "colourMapChunks.terrain"
|
||||
|
||||
val mapChunksStream: KStream<FactorioServerId, MapChunkTileUpdate> = builder.streamPacketData()
|
||||
val protosStream: KStream<FactorioServerId, PrototypesUpdate> = builder.streamPacketData()
|
||||
|
||||
val protosTable: KTable<FactorioServerId, TileColourDict> =
|
||||
protosStream.createTilePrototypeTable()
|
||||
|
||||
val chunkTilesColouredTable032: KTable<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> =
|
||||
mapChunksStream
|
||||
.convertToServerMapChunkTiles()
|
||||
.reduceServerMapChunkTilesToTable()
|
||||
.enrichWithColourData(protosTable)
|
||||
.convertToServerMapChunkTiles(pid)
|
||||
.reduceServerMapChunkTilesToTable(pid)
|
||||
.enrichWithColourData(pid, protosTable)
|
||||
|
||||
chunkTilesColouredTable032.streamMapChunkColouredTo(
|
||||
ChunkSize.CHUNK_032,
|
||||
TOPIC_MAP_CHUNK_COLOURED_032_STATE,
|
||||
"$pid.${ChunkSize.CHUNK_032}",
|
||||
TOPIC_MAP_CHUNK_TERRAIN_COLOURED_032_UPDATES,
|
||||
)
|
||||
|
||||
val chunkTilesColoured: KStream<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> =
|
||||
|
|
@ -66,7 +69,7 @@ fun colourMapChunks(builder: StreamsBuilder): Topology {
|
|||
kxsBinary.serde(ServerMapChunkId.serializer()),
|
||||
kxsBinary.serde(ServerMapChunkTiles.serializer(ColourHex.serializer())),
|
||||
),
|
||||
TOPIC_MAP_CHUNK_COLOURED_STATE,
|
||||
TOPIC_MAP_CHUNK_TERRAIN_COLOURED_STATE,
|
||||
)
|
||||
|
||||
mapOf(
|
||||
|
|
@ -77,71 +80,78 @@ fun colourMapChunks(builder: StreamsBuilder): Topology {
|
|||
|
||||
ChunkSize.CHUNK_032 to ChunkSize.CHUNK_256,
|
||||
ChunkSize.CHUNK_256 to ChunkSize.CHUNK_512,
|
||||
|
||||
).forEach { (from, to) ->
|
||||
).forEach { (from, to) ->
|
||||
chunkTilesColoured
|
||||
.filter("$pid.combine-chunks.filter-$from-to-$to") { key, _ ->
|
||||
.filter("$pid.combine-chunks.filter-$from-to-$to") { key: ServerMapChunkId, _: ServerMapChunkTiles<ColourHex> ->
|
||||
key.chunkSize == from
|
||||
}.groupByChunkPosition(to)
|
||||
.streamMapChunkColouredTo(to, TOPIC_MAP_CHUNK_COLOURED_STATE)
|
||||
}.groupByChunkPosition("$pid.from-$from-to-$to", to)
|
||||
.streamMapChunkColouredTo(
|
||||
"$pid.from-$from-to-$to",
|
||||
TOPIC_MAP_CHUNK_TERRAIN_COLOURED_STATE
|
||||
)
|
||||
}
|
||||
|
||||
return builder.build()
|
||||
.addDebounceProcessor(
|
||||
namePrefix = "groupTilesMapChunks",
|
||||
sourceTopic = TOPIC_MAP_CHUNK_COLOURED_032_STATE,
|
||||
sinkTopic = TOPIC_MAP_CHUNK_COLOURED_STATE,
|
||||
inactivityDuration = 5.seconds,
|
||||
namePrefix = "groupMapChunks.terrain",
|
||||
sourceTopic = TOPIC_MAP_CHUNK_TERRAIN_COLOURED_032_UPDATES,
|
||||
sinkTopic = TOPIC_MAP_CHUNK_TERRAIN_COLOURED_STATE,
|
||||
inactivityDuration = 15.seconds,
|
||||
keySerde = kxsBinary.serde(ServerMapChunkId.serializer()),
|
||||
valueSerde = kxsBinary.serde(ServerMapChunkTiles.serializer(ColourHex.serializer())),
|
||||
)
|
||||
}
|
||||
|
||||
|
||||
private fun KStream<FactorioServerId, MapChunkTileUpdate>.convertToServerMapChunkTiles()
|
||||
: KStream<ServerMapChunkId, ServerMapChunkTiles<TileProtoHashCode>> {
|
||||
private fun KStream<FactorioServerId, MapChunkTileUpdate>.convertToServerMapChunkTiles(
|
||||
callerPid: String
|
||||
): KStream<ServerMapChunkId, ServerMapChunkTiles<PrototypeHashCode>> {
|
||||
|
||||
val pid = "$pid.convertToMapTiles"
|
||||
val pid = "$callerPid.convertToMapTiles"
|
||||
|
||||
return map("$pid.map") { serverId: FactorioServerId, update: MapChunkTileUpdate ->
|
||||
|
||||
val mapTileList = update.tileDictionary?.toMapTileList() ?: emptyList()
|
||||
|
||||
val (validMapTilesList, invalidMapTilesList) = mapTileList.partition {
|
||||
val (validMapTiles, invalidMapTiles) = mapTileList.partition {
|
||||
it.position in update.key.chunkPosition
|
||||
}
|
||||
|
||||
if (invalidMapTilesList.isNotEmpty()) {
|
||||
println("WARNING [reduceMapTilesToTable] MapChunkTileUpdate contained out-of-bounds tiles $invalidMapTilesList")
|
||||
if (invalidMapTiles.isNotEmpty()) {
|
||||
println(
|
||||
"WARNING [$pid] MapChunkTileUpdate contained ${invalidMapTiles.size} out-of-bounds tiles, " +
|
||||
"key: ${update.key},\n" +
|
||||
"invalid tiles:" + invalidMapTiles.joinToString(limit = 10)
|
||||
)
|
||||
}
|
||||
|
||||
val validMapTilesToProtoHashCode =
|
||||
validMapTilesList.associate { it.position to TileProtoHashCode(it) }
|
||||
val validMapTilesToProtoHashCode = validMapTiles.associate {
|
||||
it.position to PrototypeHashCode(it.protoId)
|
||||
}
|
||||
|
||||
val chunkId = ServerMapChunkId(
|
||||
serverId = serverId,
|
||||
layer = ServerMapTileLayer.TERRAIN,
|
||||
chunkPosition = update.key.chunkPosition,
|
||||
surfaceIndex = update.key.surfaceIndex,
|
||||
chunkSize = ChunkSize.CHUNK_032,
|
||||
)
|
||||
|
||||
val chunkTiles = ServerMapChunkTiles(chunkId, validMapTilesToProtoHashCode)
|
||||
|
||||
chunkId to chunkTiles
|
||||
chunkId to ServerMapChunkTiles(chunkId, validMapTilesToProtoHashCode)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
private fun KStream<ServerMapChunkId, ServerMapChunkTiles<TileProtoHashCode>>.reduceServerMapChunkTilesToTable()
|
||||
: KTable<ServerMapChunkId, ServerMapChunkTiles<TileProtoHashCode>> {
|
||||
fun KStream<ServerMapChunkId, ServerMapChunkTiles<PrototypeHashCode>>.reduceServerMapChunkTilesToTable(
|
||||
callerPid: String,
|
||||
): KTable<ServerMapChunkId, ServerMapChunkTiles<PrototypeHashCode>> {
|
||||
|
||||
val pid = "$pid.reduceServerMapChunkTilesToTable"
|
||||
val pid = "$callerPid.reduceServerMapChunkTilesToTable"
|
||||
|
||||
return repartition(
|
||||
repartitionedAs(
|
||||
"$pid.pre-table-repartition",
|
||||
kxsBinary.serde<ServerMapChunkId>(),
|
||||
kxsBinary.serde<ServerMapChunkTiles<TileProtoHashCode>>(),
|
||||
kxsBinary.serde<ServerMapChunkTiles<PrototypeHashCode>>(),
|
||||
// force, otherwise KTable-KTable FK join doesn't work
|
||||
numberOfPartitions = 1,
|
||||
)
|
||||
|
|
@ -149,14 +159,14 @@ private fun KStream<ServerMapChunkId, ServerMapChunkTiles<TileProtoHashCode>>.re
|
|||
groupedAs(
|
||||
"$pid.group-by-key",
|
||||
kxsBinary.serde<ServerMapChunkId>(),
|
||||
kxsBinary.serde<ServerMapChunkTiles<TileProtoHashCode>>(),
|
||||
kxsBinary.serde<ServerMapChunkTiles<PrototypeHashCode>>(),
|
||||
)
|
||||
).reduce(
|
||||
"$pid.reduce",
|
||||
materializedAs(
|
||||
"$pid.reduce.store",
|
||||
kxsBinary.serde<ServerMapChunkId>(),
|
||||
kxsBinary.serde<ServerMapChunkTiles<TileProtoHashCode>>(),
|
||||
kxsBinary.serde<ServerMapChunkTiles<PrototypeHashCode>>(),
|
||||
)
|
||||
) { chunkTiles, otherChunkTiles ->
|
||||
chunkTiles + otherChunkTiles
|
||||
|
|
@ -164,13 +174,14 @@ private fun KStream<ServerMapChunkId, ServerMapChunkTiles<TileProtoHashCode>>.re
|
|||
}
|
||||
|
||||
|
||||
private fun KTable<ServerMapChunkId, ServerMapChunkTiles<TileProtoHashCode>>.enrichWithColourData(
|
||||
private fun KTable<ServerMapChunkId, ServerMapChunkTiles<PrototypeHashCode>>.enrichWithColourData(
|
||||
callerPid: String,
|
||||
protosTable: KTable<FactorioServerId, TileColourDict>,
|
||||
): KTable<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> {
|
||||
|
||||
val pid = "$pid.enrichWithColourData"
|
||||
val pid = "$callerPid.enrichWithColourData"
|
||||
|
||||
return this.join(
|
||||
return join(
|
||||
other = protosTable,
|
||||
tableJoined = tableJoined("$pid.join-tiles-with-prototypes"),
|
||||
materialized = materializedAs(
|
||||
|
|
@ -178,26 +189,30 @@ private fun KTable<ServerMapChunkId, ServerMapChunkTiles<TileProtoHashCode>>.enr
|
|||
kxsBinary.serde(),
|
||||
kxsBinary.serde(),
|
||||
),
|
||||
foreignKeyExtractor = { chunkTiles: ServerMapChunkTiles<TileProtoHashCode> ->
|
||||
foreignKeyExtractor = { chunkTiles: ServerMapChunkTiles<PrototypeHashCode> ->
|
||||
chunkTiles.chunkId.serverId
|
||||
}
|
||||
) { tiles: ServerMapChunkTiles<TileProtoHashCode>, colourDict: TileColourDict ->
|
||||
// println("joining tiles:${tiles.map.size} with colourDict:${colourDict.map.size}")
|
||||
) { tiles: ServerMapChunkTiles<PrototypeHashCode>, colourDict: TileColourDict ->
|
||||
println("joining tiles:${tiles.map.size} with colourDict:${colourDict.size}")
|
||||
|
||||
val missingProtos = mutableSetOf<TileProtoHashCode>()
|
||||
val missingProtos = mutableSetOf<PrototypeHashCode>()
|
||||
|
||||
val tileColours = tiles.map.mapValues { (_, code) ->
|
||||
colourDict.getOrElse(code) {
|
||||
missingProtos.add(code)
|
||||
ColourHex.TRANSPARENT
|
||||
}
|
||||
}
|
||||
val tileColours: Map<MapTilePosition, ColourHex> =
|
||||
tiles.map.mapNotNull { (position, code) ->
|
||||
when (val colour = colourDict[code]) {
|
||||
null -> {
|
||||
missingProtos.add(code)
|
||||
null
|
||||
}
|
||||
else -> position to colour
|
||||
}
|
||||
}.toMap()
|
||||
|
||||
if (missingProtos.isNotEmpty()) {
|
||||
println("[$pid] missing ${missingProtos.size} tile prototypes: ${missingProtos.joinToString { "${it.code}" }}")
|
||||
}
|
||||
|
||||
println("[$pid] Set tile colours for chunk ${tiles.chunkId}")
|
||||
println("[$pid] Set ${tileColours.size} tile colours for chunk ${tiles.chunkId}")
|
||||
|
||||
ServerMapChunkTiles(tiles.chunkId, tileColours)
|
||||
}
|
||||
|
|
@ -205,20 +220,16 @@ private fun KTable<ServerMapChunkId, ServerMapChunkTiles<TileProtoHashCode>>.enr
|
|||
|
||||
|
||||
fun KTable<ServerMapChunkId, ServerMapChunkTiles<ColourHex>>.streamMapChunkColouredTo(
|
||||
chunkSize: ChunkSize,
|
||||
callerPid: String,
|
||||
outputTopic: String,
|
||||
) {
|
||||
val pid = "$pid.output-chunk.${chunkSize.name}"
|
||||
val pid = "$callerPid.output-chunk"
|
||||
|
||||
toStream("$pid.stream")
|
||||
.filter("$pid.filter-tiles-not-empty") { _, chunkTiles ->
|
||||
!chunkTiles?.map.isNullOrEmpty()
|
||||
}
|
||||
.mapValues("$pid.map-not-null") { _, chunkTiles ->
|
||||
requireNotNull(chunkTiles)
|
||||
}
|
||||
.peek("$pid.print-group-result") { _, chunkTiles ->
|
||||
println("Grouping map tiles result: ${chunkTiles.chunkId} / size:${chunkTiles.map.size}")
|
||||
.flatMap("$pid.group-by-chunks") { _: ServerMapChunkId, tiles: ServerMapChunkTiles<ColourHex>? ->
|
||||
reChunkTiles(pid, tiles)
|
||||
}.peek("$pid.print-group-result") { _, chunkTiles ->
|
||||
println("[$pid] Grouping map tiles result: ${chunkTiles.chunkId} / size:${chunkTiles.map.size}")
|
||||
}.to(
|
||||
outputTopic,
|
||||
producedAs(
|
||||
|
|
@ -235,26 +246,22 @@ fun KTable<ServerMapChunkId, ServerMapChunkTiles<ColourHex>>.streamMapChunkColou
|
|||
*
|
||||
* They should already be grouped, but do it again to make sure, and to filter out empty updates.
|
||||
*/
|
||||
private fun KStream<ServerMapChunkId, ServerMapChunkTiles<ColourHex>>.groupByChunkPosition(
|
||||
chunkSize: ChunkSize = ChunkSize.CHUNK_032
|
||||
fun KStream<ServerMapChunkId, ServerMapChunkTiles<ColourHex>>.groupByChunkPosition(
|
||||
callerPid: String,
|
||||
targetChunkSize: ChunkSize = ChunkSize.CHUNK_032
|
||||
): KTable<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> {
|
||||
|
||||
val pid = "$pid.groupByChunkPosition.${chunkSize.name}"
|
||||
val pid = "$callerPid.groupByChunkPosition.${targetChunkSize.name}"
|
||||
|
||||
return map("$pid.change-chunk-size") { chunkId: ServerMapChunkId, chunkTiles: ServerMapChunkTiles<ColourHex>? ->
|
||||
return flatMap("$pid.change-chunk-size") { _: ServerMapChunkId, chunkTiles: ServerMapChunkTiles<ColourHex>? ->
|
||||
// println("[$pid] changing chunk size for $chunkId")
|
||||
chunkTiles ?: return@flatMap emptyList()
|
||||
|
||||
val newChunkPosition = chunkId.chunkPosition
|
||||
.leftTopTile(chunkId.chunkSize)
|
||||
.toMapChunkPosition(chunkSize)
|
||||
val newChunkPosition = chunkTiles.chunkId.chunkPosition.resize(targetChunkSize)
|
||||
val newId = chunkTiles.chunkId.copy(chunkPosition = newChunkPosition)
|
||||
val newChunkTiles = chunkTiles.copy(chunkId = newId)
|
||||
|
||||
val newId = chunkId.copy(
|
||||
chunkSize = chunkSize,
|
||||
chunkPosition = newChunkPosition,
|
||||
)
|
||||
|
||||
val newChunkTiles = chunkTiles?.map ?: emptyMap()
|
||||
|
||||
newId to ServerMapChunkTiles(chunkId = newId, map = newChunkTiles)
|
||||
reChunkTiles(pid, newChunkTiles)
|
||||
}.groupByKey(
|
||||
groupedAs(
|
||||
"$pid.group-by-key",
|
||||
|
|
@ -272,3 +279,51 @@ private fun KStream<ServerMapChunkId, ServerMapChunkTiles<ColourHex>>.groupByChu
|
|||
chunkTiles + otherChunkTiles
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
fun <T : Any> reChunkTiles(
|
||||
pid: String,
|
||||
tiles: ServerMapChunkTiles<T>?,
|
||||
): List<Pair<ServerMapChunkId, ServerMapChunkTiles<T>>> {
|
||||
|
||||
return when {
|
||||
tiles == null || tiles.map.isEmpty() -> emptyList()
|
||||
|
||||
tiles.map.keys.all { it in tiles.chunkId.chunkPosition } -> listOf(tiles.chunkId to tiles)
|
||||
|
||||
else -> {
|
||||
val chunkId = tiles.chunkId
|
||||
|
||||
val (boundedTiles, unboundedTiles) = tiles.map.entries.partition { (position, _) ->
|
||||
position in chunkId.chunkPosition
|
||||
}
|
||||
|
||||
if (unboundedTiles.isNotEmpty()) {
|
||||
val unboundedEntities = unboundedTiles.map { (_, entity) -> entity }
|
||||
println(
|
||||
"WARNING [$pid] MapChunkEntityUpdate $chunkId "
|
||||
+ "contained ${unboundedEntities.size} out-of-bounds entities "
|
||||
+ unboundedEntities.joinToString(limit = 10)
|
||||
)
|
||||
}
|
||||
|
||||
// the existing chunkId and its tiles
|
||||
val boundedTilesChunk =
|
||||
chunkId to ServerMapChunkTiles(chunkId, boundedTiles.associate { it.key to it.value })
|
||||
|
||||
// re-calculate a chunk for the out-of-bounds tiles
|
||||
val chunkedUnboundedTiles = unboundedTiles
|
||||
.map { it.key to it.value }
|
||||
.groupBy { (position, _) ->
|
||||
position.toMapChunkPosition(chunkId.chunkSize)
|
||||
}.map { (newChunkPosition, tiles) ->
|
||||
val newChunkId = chunkId.copy(chunkPosition = newChunkPosition)
|
||||
newChunkId to ServerMapChunkTiles(newChunkId, tiles.toMap())
|
||||
}
|
||||
|
||||
println("[$pid] re-bounded ${unboundedTiles.size} tiles into ${chunkedUnboundedTiles.size} chunks in $chunkId")
|
||||
|
||||
listOf(boundedTilesChunk) + chunkedUnboundedTiles
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,442 @@
|
|||
package dev.adamko.kafkatorio.server.processor.topology
|
||||
|
||||
import dev.adamko.kafkatorio.library.kxsBinary
|
||||
import dev.adamko.kafkatorio.library.toMapTilePosition
|
||||
import dev.adamko.kafkatorio.schema.common.ChunkSize
|
||||
import dev.adamko.kafkatorio.schema.common.ColourHex
|
||||
import dev.adamko.kafkatorio.schema.common.FactorioPrototype
|
||||
import dev.adamko.kafkatorio.schema.common.FactorioServerId
|
||||
import dev.adamko.kafkatorio.schema.common.MapBoundingBox
|
||||
import dev.adamko.kafkatorio.schema.common.MapTilePosition
|
||||
import dev.adamko.kafkatorio.schema.common.PrototypeHashCode
|
||||
import dev.adamko.kafkatorio.schema.common.ServerMapChunkId
|
||||
import dev.adamko.kafkatorio.schema.common.ServerMapTileLayer
|
||||
import dev.adamko.kafkatorio.schema.common.toHex
|
||||
import dev.adamko.kafkatorio.schema.packets.MapChunkEntityUpdate
|
||||
import dev.adamko.kafkatorio.schema.packets.MapChunkResourceUpdate
|
||||
import dev.adamko.kafkatorio.schema.packets.MapChunkTileUpdate
|
||||
import dev.adamko.kafkatorio.schema.packets.PrototypesUpdate
|
||||
import dev.adamko.kafkatorio.server.processor.DebounceProcessor.Companion.addDebounceProcessor
|
||||
import dev.adamko.kotka.extensions.consumedAs
|
||||
import dev.adamko.kotka.extensions.materializedAs
|
||||
import dev.adamko.kotka.extensions.stream
|
||||
import dev.adamko.kotka.extensions.streams.filter
|
||||
import dev.adamko.kotka.extensions.streams.flatMap
|
||||
import dev.adamko.kotka.extensions.tableJoined
|
||||
import dev.adamko.kotka.extensions.tables.join
|
||||
import dev.adamko.kotka.kxs.serde
|
||||
import kotlin.collections.component1
|
||||
import kotlin.collections.component2
|
||||
import kotlin.math.ceil
|
||||
import kotlin.math.floor
|
||||
import kotlin.time.Duration.Companion.seconds
|
||||
import org.apache.kafka.streams.StreamsBuilder
|
||||
import org.apache.kafka.streams.Topology
|
||||
import org.apache.kafka.streams.kstream.KStream
|
||||
import org.apache.kafka.streams.kstream.KTable
|
||||
|
||||
|
||||
/**
|
||||
* Add colour to each [MapChunkTileUpdate]
|
||||
*
|
||||
* @param[colouredUpdatesStreamTopic] Output topic for the result of converting input updates into 32 sized chunks
|
||||
* @param[colouredChunkStateTopic] Post-debounce state topic, used to recursively combine chunks
|
||||
*/
|
||||
fun colourEntityChunks(
|
||||
builder: StreamsBuilder,
|
||||
entityChunksStream: KStream<ServerMapChunkId, ServerMapChunkTiles<PrototypeHashCode>>,
|
||||
protosStream: KStream<FactorioServerId, PrototypesUpdate>,
|
||||
layer: ServerMapTileLayer,
|
||||
colouredUpdatesStreamTopic: String,
|
||||
colouredChunkStateTopic: String,
|
||||
): Topology {
|
||||
val pid = "colourEntityChunks.${layer.dir}"
|
||||
|
||||
val protosTable: KTable<FactorioServerId, EntityPrototypeMap> =
|
||||
protosStream.createEntityPrototypeMap(pid)
|
||||
|
||||
val chunkEntitiesColouredTable032: KTable<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> =
|
||||
entityChunksStream
|
||||
.reduceServerMapChunkTilesToTable(pid)
|
||||
.enrichWithColourData(pid, protosTable)
|
||||
|
||||
chunkEntitiesColouredTable032.streamMapChunkColouredTo(
|
||||
"$pid.${ChunkSize.CHUNK_032}",
|
||||
colouredUpdatesStreamTopic,
|
||||
)
|
||||
|
||||
val chunkTilesColoured: KStream<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> =
|
||||
builder.stream(
|
||||
consumedAs(
|
||||
"$pid.consume.map-chunks-debounced",
|
||||
kxsBinary.serde(ServerMapChunkId.serializer()),
|
||||
kxsBinary.serde(ServerMapChunkTiles.serializer(ColourHex.serializer())),
|
||||
),
|
||||
colouredChunkStateTopic,
|
||||
)
|
||||
|
||||
mapOf(
|
||||
// ChunkSize.CHUNK_032 to ChunkSize.CHUNK_064,
|
||||
// ChunkSize.CHUNK_064 to ChunkSize.CHUNK_128,
|
||||
// ChunkSize.CHUNK_128 to ChunkSize.CHUNK_256,
|
||||
// ChunkSize.CHUNK_256 to ChunkSize.CHUNK_512,
|
||||
|
||||
ChunkSize.CHUNK_032 to ChunkSize.CHUNK_256,
|
||||
ChunkSize.CHUNK_256 to ChunkSize.CHUNK_512,
|
||||
).forEach { (from, to) ->
|
||||
chunkTilesColoured
|
||||
.filter("$pid.combine-chunks.filter-$from-to-$to") { key, _ ->
|
||||
key.chunkSize == from && key.layer == layer
|
||||
}.groupByChunkPosition(pid, to)
|
||||
.streamMapChunkColouredTo(
|
||||
"$pid.from-$from-to-$to",
|
||||
colouredChunkStateTopic,
|
||||
)
|
||||
}
|
||||
|
||||
return builder.build().addDebounceProcessor(
|
||||
namePrefix = "groupMapChunks.${layer.dir}",
|
||||
sourceTopic = colouredUpdatesStreamTopic,
|
||||
sinkTopic = colouredChunkStateTopic,
|
||||
inactivityDuration = 15.seconds,
|
||||
keySerde = kxsBinary.serde(ServerMapChunkId.serializer()),
|
||||
valueSerde = kxsBinary.serde(ServerMapChunkTiles.serializer(ColourHex.serializer())),
|
||||
)
|
||||
}
|
||||
|
||||
|
||||
fun KStream<FactorioServerId, MapChunkEntityUpdate>.convertEntityUpdateToServerMapChunkTiles(
|
||||
callerPid: String,
|
||||
): KStream<ServerMapChunkId, ServerMapChunkTiles<PrototypeHashCode>> {
|
||||
|
||||
val pid = "$callerPid.convertToServerMapChunkTiles"
|
||||
|
||||
return flatMap("$pid.map") { serverId: FactorioServerId, update: MapChunkEntityUpdate ->
|
||||
|
||||
// val entities = update.entitiesXY.convert { entity, position ->
|
||||
// FactorioEntityData.Standard(
|
||||
// protoId = update.key.protoId,
|
||||
// position = position,
|
||||
// element = entity,
|
||||
// )
|
||||
// }
|
||||
|
||||
// val (validEntityPositions, unboundedEntityPositions) = entities.map { entity ->
|
||||
// entity.position.toMapTilePosition() to entity
|
||||
// }.partition { (position, _) ->
|
||||
// position in update.key.chunkPosition
|
||||
// }
|
||||
//
|
||||
// if (unboundedEntityPositions.isNotEmpty()) {
|
||||
// val unboundedEntities = unboundedEntityPositions.map { (_, entity) -> entity }
|
||||
// println(
|
||||
// "WARNING [convertToServerMapChunkTiles] MapChunkEntityUpdate ${update.key} " + "contained ${unboundedEntities.size} out-of-bounds entities " + unboundedEntities.joinToString(
|
||||
// limit = 10
|
||||
// )
|
||||
// )
|
||||
// }
|
||||
//
|
||||
// // TODO There might be multiple resources per tile position, therefore this will arbitrarily
|
||||
// // exclude resources. Maybe change it from a Map to a List? Or have multiple resources per
|
||||
// // tile position?
|
||||
// val mapValidEntitiesToPrototypeHash = validEntityPositions.associate { (pos, resource) ->
|
||||
// pos to PrototypeHashCode(resource.protoId)
|
||||
// }
|
||||
|
||||
val entitiesToPrototypeHash = update.entities().associate { entity ->
|
||||
entity.position.toMapTilePosition() to PrototypeHashCode(entity.protoId)
|
||||
}
|
||||
|
||||
val chunkId = ServerMapChunkId(
|
||||
serverId = serverId,
|
||||
layer = ServerMapTileLayer.BUILDING,
|
||||
chunkPosition = update.key.chunkPosition,
|
||||
surfaceIndex = update.key.surfaceIndex,
|
||||
)
|
||||
|
||||
val tiles = ServerMapChunkTiles(chunkId, entitiesToPrototypeHash)
|
||||
|
||||
reChunkTiles(pid, tiles)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
fun KStream<FactorioServerId, MapChunkResourceUpdate>.convertResourceUpdateToServerMapChunkTiles(
|
||||
callerPid: String,
|
||||
): KStream<ServerMapChunkId, ServerMapChunkTiles<PrototypeHashCode>> {
|
||||
|
||||
val pid = "$callerPid.convertResourceUpdateToChunkTiles"
|
||||
|
||||
return flatMap("$pid.map") { serverId: FactorioServerId, update: MapChunkResourceUpdate ->
|
||||
|
||||
// // TODO There might be multiple resources per tile position, therefore this will arbitrarily
|
||||
// // exclude resources. Maybe change it from a Map to a List? Or have multiple resources per
|
||||
// // tile position?
|
||||
// val mapValidEntitiesToPrototypeHash = validEntityPositions.associate { (pos, resource) ->
|
||||
// pos to PrototypeHashCode(resource.protoId)
|
||||
// }
|
||||
|
||||
val entitiesToPrototypeHash = update.resources().associate { entity ->
|
||||
entity.position.toMapTilePosition() to PrototypeHashCode(entity.protoId)
|
||||
}
|
||||
|
||||
val chunkId = ServerMapChunkId(
|
||||
serverId = serverId,
|
||||
layer = ServerMapTileLayer.RESOURCE,
|
||||
chunkPosition = update.key.chunkPosition,
|
||||
surfaceIndex = update.key.surfaceIndex,
|
||||
)
|
||||
|
||||
val tiles = ServerMapChunkTiles(chunkId, entitiesToPrototypeHash)
|
||||
|
||||
reChunkTiles(pid, tiles)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
//private fun KStream<ServerMapChunkId, ServerMapChunkTiles<PrototypeHashCode>>.reduceServerMapResourceTilesToTable()
|
||||
// : KTable<ServerMapChunkId, ServerMapChunkTiles<PrototypeHashCode>> {
|
||||
//
|
||||
// val pid = "$pid.reduceServerMapResourceTilesToTable"
|
||||
//
|
||||
// return repartition(
|
||||
// repartitionedAs(
|
||||
// "$pid.pre-table-repartition",
|
||||
// kxsBinary.serde<ServerMapChunkId>(),
|
||||
// kxsBinary.serde<ServerMapChunkTiles<PrototypeHashCode>>(),
|
||||
// // force, otherwise KTable-KTable FK join doesn't work
|
||||
// numberOfPartitions = 1,
|
||||
// )
|
||||
// ).groupByKey(
|
||||
// groupedAs(
|
||||
// "$pid.group-by-key",
|
||||
// kxsBinary.serde<ServerMapChunkId>(),
|
||||
// kxsBinary.serde<ServerMapChunkTiles<PrototypeHashCode>>(),
|
||||
// )
|
||||
// ).reduce(
|
||||
// "$pid.reduce",
|
||||
// materializedAs(
|
||||
// "$pid.reduce.store",
|
||||
// kxsBinary.serde<ServerMapChunkId>(),
|
||||
// kxsBinary.serde<ServerMapChunkTiles<PrototypeHashCode>>(),
|
||||
// )
|
||||
// ) { chunkTiles, otherChunkTiles ->
|
||||
// chunkTiles + otherChunkTiles
|
||||
// }
|
||||
//}
|
||||
|
||||
|
||||
private fun KTable<ServerMapChunkId, ServerMapChunkTiles<PrototypeHashCode>>.enrichWithColourData(
|
||||
callerPid: String,
|
||||
protosTable: KTable<FactorioServerId, EntityPrototypeMap>,
|
||||
): KTable<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> {
|
||||
|
||||
val pid = "$callerPid.enrichWithColourData"
|
||||
|
||||
return join(
|
||||
other = protosTable,
|
||||
tableJoined = tableJoined("$pid.join-tiles-with-prototypes"),
|
||||
materialized = materializedAs(
|
||||
"$pid.join-tiles-with-prototypes.store",
|
||||
kxsBinary.serde(),
|
||||
kxsBinary.serde(),
|
||||
),
|
||||
foreignKeyExtractor = { chunkTiles: ServerMapChunkTiles<PrototypeHashCode> ->
|
||||
chunkTiles.chunkId.serverId
|
||||
}) { tiles: ServerMapChunkTiles<PrototypeHashCode>, prototypes: EntityPrototypeMap ->
|
||||
println("joining tiles:${tiles.map.size} with prototypes:${prototypes.size}")
|
||||
|
||||
val missingProtos = mutableSetOf<PrototypeHashCode>()
|
||||
|
||||
val tileColours: Map<MapTilePosition, ColourHex> = buildList {
|
||||
tiles.map.forEach { (position, code) ->
|
||||
when (val proto = prototypes[code]) {
|
||||
|
||||
null -> missingProtos += code
|
||||
|
||||
else -> when (val box = proto.collisionBox) {
|
||||
|
||||
null -> add(position to proto.mapColour())
|
||||
|
||||
else -> box.iterator(position).forEach { entitySubTile ->
|
||||
add(entitySubTile to proto.mapColour())
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}.fold(mapOf()) { acc, (pos, colour) ->
|
||||
// merge overlapping colours
|
||||
val newColour = when (val existing = acc[pos]) {
|
||||
null -> colour
|
||||
else -> (existing / 2u) + (colour / 2u)
|
||||
}
|
||||
acc + (pos to newColour)
|
||||
}
|
||||
|
||||
if (missingProtos.isNotEmpty()) {
|
||||
println("[$pid] missing ${missingProtos.size} tile prototypes: ${missingProtos.joinToString { "${it.code}" }}")
|
||||
}
|
||||
|
||||
println("[$pid] Set ${tileColours.size} tile colours for chunk ${tiles.chunkId}")
|
||||
|
||||
ServerMapChunkTiles(tiles.chunkId, tileColours)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
private fun MapBoundingBox.iterator(origin: MapTilePosition): Iterator<MapTilePosition> {
|
||||
|
||||
val xOffset: Int = floor(tileWidth.toDouble() / 2).toInt()
|
||||
val startX = origin.x - xOffset
|
||||
val endX = origin.x + xOffset
|
||||
|
||||
val yOffset: Int = floor(tileHeight.toDouble() / 2).toInt()
|
||||
val startY = origin.y - yOffset
|
||||
val endY = origin.y + yOffset
|
||||
|
||||
return iterator {
|
||||
for (x in (startX..endX)) {
|
||||
for (y in (startY..endY)) { // screen coordinates = +y points to the floor, -y to the ceiling
|
||||
yield(MapTilePosition(x, y))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Rounds a number 'up', away from zero.
|
||||
* (Towards positive infinity if positive, and towards negative infinity if negative.)
|
||||
*/
|
||||
private fun Double.roundToInfinity(): Int = when {
|
||||
this > 0 -> ceil(this).toInt()
|
||||
this < 0 -> floor(this).toInt()
|
||||
else -> 0
|
||||
}
|
||||
// ceil(abs(this)).toInt() * this.sign.toInt()
|
||||
|
||||
|
||||
private fun FactorioPrototype.Entity.mapColour(): ColourHex =
|
||||
(colour ?: mapColour ?: mapColourFriend)?.toHex() ?: ColourHex.WHITE
|
||||
|
||||
|
||||
private operator fun ColourHex.div(divisor: UInt): ColourHex = copy(
|
||||
red = (red / divisor).toUByte(),
|
||||
green = (green / divisor).toUByte(),
|
||||
blue = (blue / divisor).toUByte(),
|
||||
alpha = (alpha / divisor).toUByte(),
|
||||
)
|
||||
|
||||
|
||||
private operator fun ColourHex.plus(other: ColourHex): ColourHex = copy(
|
||||
red = (red + other.red).toUByte(),
|
||||
green = (green + other.green).toUByte(),
|
||||
blue = (blue + other.blue).toUByte(),
|
||||
alpha = (alpha + other.alpha).toUByte(),
|
||||
)
|
||||
|
||||
|
||||
//// https://stackoverflow.com/a/9355778/4161471
|
||||
//private operator fun ColourHex.plus(other: ColourHex): ColourHex {
|
||||
//
|
||||
// @Suppress("LocalVariableName")
|
||||
// val `255` = UByte.MAX_VALUE
|
||||
//
|
||||
// val compositeAlpha: UByte = (
|
||||
// `255` - ((`255` - other.alpha) * (`255` - this.alpha)) / `255`
|
||||
// ).toUByte()
|
||||
//
|
||||
// if (compositeAlpha == UByte.MIN_VALUE) {
|
||||
// return ColourHex.TRANSPARENT
|
||||
// } else {
|
||||
//
|
||||
// fun compositeColour(
|
||||
// colour: (ColourHex) -> UByte,
|
||||
// ): UByte {
|
||||
// val colour1 = colour(this)
|
||||
// val colour2 = colour(other)
|
||||
// return ((
|
||||
// (
|
||||
// ( colour2 * other.alpha) + (colour1 * this.alpha * ( other.alpha))
|
||||
// ) / compositeAlpha
|
||||
// ) / `255`
|
||||
// ).toUByte()
|
||||
// }
|
||||
//
|
||||
// return ColourHex(
|
||||
// compositeColour(ColourHex::red),
|
||||
// compositeColour(ColourHex::green),
|
||||
// compositeColour(ColourHex::blue),
|
||||
// compositeAlpha,
|
||||
// )
|
||||
// }
|
||||
//}
|
||||
|
||||
//private fun KTable<ServerMapChunkId, ServerMapChunkTiles<ColourHex>>.streamMapChunkColouredTo(
|
||||
// chunkSize: ChunkSize,
|
||||
// outputTopic: String,
|
||||
//) {
|
||||
// val pid = "$pid.output-chunk.${chunkSize.name}"
|
||||
//
|
||||
// toStream("$pid.stream")
|
||||
// .filter("$pid.filter-tiles-not-empty") { _, chunkTiles ->
|
||||
// !chunkTiles?.map.isNullOrEmpty()
|
||||
// }
|
||||
// .mapValues("$pid.map-not-null") { _, chunkTiles ->
|
||||
// requireNotNull(chunkTiles)
|
||||
// }
|
||||
// .peek("$pid.print-group-result") { _, chunkTiles ->
|
||||
// println("Grouping map tiles result: ${chunkTiles.chunkId} / size:${chunkTiles.map.size}")
|
||||
// }.to(
|
||||
// outputTopic,
|
||||
// producedAs(
|
||||
// "$pid.grouped-map-chunks",
|
||||
// kxsBinary.serde<ServerMapChunkId>(),
|
||||
// kxsBinary.serde<ServerMapChunkTiles<ColourHex>>(),
|
||||
// )
|
||||
// )
|
||||
//}
|
||||
|
||||
|
||||
///**
|
||||
// * Group [MapChunkTileUpdate]s by the Chunk position.
|
||||
// *
|
||||
// * They should already be grouped, but do it again to make sure, and to filter out empty updates.
|
||||
// */
|
||||
//private fun KStream<ServerMapChunkId, ServerMapChunkTiles<ColourHex>>.groupByChunkPosition(
|
||||
// chunkSize: ChunkSize = ChunkSize.CHUNK_032
|
||||
//): KTable<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> {
|
||||
//
|
||||
// val pid = "$pid.groupByChunkPosition.${chunkSize.name}"
|
||||
//
|
||||
// return map("$pid.change-chunk-size") { chunkId: ServerMapChunkId, chunkTiles: ServerMapChunkTiles<ColourHex>? ->
|
||||
//
|
||||
// val newChunkPosition = chunkId.chunkPosition
|
||||
// .leftTopTile(chunkId.chunkSize)
|
||||
// .toMapChunkPosition(chunkSize)
|
||||
//
|
||||
// val newId = chunkId.copy(
|
||||
// chunkSize = chunkSize,
|
||||
// chunkPosition = newChunkPosition,
|
||||
// )
|
||||
//
|
||||
// val newChunkTiles = chunkTiles?.map ?: emptyMap()
|
||||
//
|
||||
// newId to ServerMapChunkTiles(chunkId = newId, map = newChunkTiles)
|
||||
// }.groupByKey(
|
||||
// groupedAs(
|
||||
// "$pid.group-by-key",
|
||||
// kxsBinary.serde<ServerMapChunkId>(),
|
||||
// kxsBinary.serde<ServerMapChunkTiles<ColourHex>>(),
|
||||
// )
|
||||
// ).reduce(
|
||||
// "$pid.reduce",
|
||||
// materializedAs(
|
||||
// "$pid.reduce.store",
|
||||
// kxsBinary.serde<ServerMapChunkId>(),
|
||||
// kxsBinary.serde<ServerMapChunkTiles<ColourHex>>(),
|
||||
// )
|
||||
// ) { chunkTiles, otherChunkTiles ->
|
||||
// chunkTiles + otherChunkTiles
|
||||
// }
|
||||
//}
|
||||
|
|
@ -3,7 +3,7 @@ package dev.adamko.kafkatorio.server.processor.topology
|
|||
import dev.adamko.kafkatorio.schema.common.FactorioServerId
|
||||
import dev.adamko.kafkatorio.schema.common.Tick
|
||||
import dev.adamko.kafkatorio.schema.packets.KafkatorioPacket
|
||||
import dev.adamko.kafkatorio.schema.packets.KafkatorioPacketData
|
||||
import dev.adamko.kafkatorio.schema.packets.KafkatorioPacketDataError
|
||||
import dev.adamko.kafkatorio.server.config.jsonMapper
|
||||
import dev.adamko.kafkatorio.server.processor.TOPIC_SRC_SERVER_LOG
|
||||
import dev.adamko.kafkatorio.server.processor.topicName
|
||||
|
|
@ -38,20 +38,18 @@ fun factorioServerPacketStream(
|
|||
KafkatorioPacket(
|
||||
modVersion = "unknown",
|
||||
tick = Tick(0u),
|
||||
data = KafkatorioPacketData.Error(
|
||||
data = KafkatorioPacketDataError(
|
||||
message = e.message,
|
||||
rawValue = value,
|
||||
)
|
||||
)
|
||||
}
|
||||
|
||||
if (packet.data is KafkatorioPacketData.Error) {
|
||||
if (packet.data is KafkatorioPacketDataError) {
|
||||
println("error parsing $TOPIC_SRC_SERVER_LOG message: ${packet.data}")
|
||||
}
|
||||
|
||||
FactorioServerId(serverId) to packet
|
||||
}.filter { _, value ->
|
||||
value.data !is KafkatorioPacketData.Error
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -10,7 +10,7 @@ import kotlinx.serialization.Serializable
|
|||
|
||||
|
||||
@Serializable
|
||||
data class ServerMapChunkTiles<Data>(
|
||||
data class ServerMapChunkTiles<Data : Any>(
|
||||
val chunkId: ServerMapChunkId,
|
||||
val map: Map<MapTilePosition, Data>,
|
||||
) {
|
||||
|
|
@ -19,7 +19,7 @@ data class ServerMapChunkTiles<Data>(
|
|||
|
||||
|
||||
val MapTile.position
|
||||
get() = MapTilePosition(x, y)
|
||||
get() = MapTilePosition(x = x, y = y)
|
||||
|
||||
|
||||
val TRANSPARENT_AWT: Color = ColourHex.TRANSPARENT.toRgbColor().awt()
|
||||
|
|
|
|||
|
|
@ -2,36 +2,34 @@ package dev.adamko.kafkatorio.server.processor.topology
|
|||
|
||||
import dev.adamko.kafkatorio.library.kxsBinary
|
||||
import dev.adamko.kafkatorio.schema.common.ColourHex
|
||||
import dev.adamko.kafkatorio.schema.common.FactorioEntityData
|
||||
import dev.adamko.kafkatorio.schema.common.FactorioPrototype
|
||||
import dev.adamko.kafkatorio.schema.common.FactorioServerId
|
||||
import dev.adamko.kafkatorio.schema.common.MapTile
|
||||
import dev.adamko.kafkatorio.schema.common.PrototypeHashCode
|
||||
import dev.adamko.kafkatorio.schema.common.PrototypeId
|
||||
import dev.adamko.kafkatorio.schema.common.toHex
|
||||
import dev.adamko.kafkatorio.schema.packets.PrototypesUpdate
|
||||
import dev.adamko.kotka.extensions.groupedAs
|
||||
import dev.adamko.kotka.extensions.materializedAs
|
||||
import dev.adamko.kotka.extensions.repartitionedAs
|
||||
import dev.adamko.kotka.extensions.streams.filter
|
||||
import dev.adamko.kotka.extensions.materializedWith
|
||||
import dev.adamko.kotka.extensions.*
|
||||
import dev.adamko.kotka.extensions.streams.*
|
||||
import dev.adamko.kotka.extensions.streams.mapValues
|
||||
import dev.adamko.kotka.extensions.streams.toTable
|
||||
import dev.adamko.kotka.extensions.tables.*
|
||||
import dev.adamko.kotka.kxs.serde
|
||||
import kotlinx.serialization.Serializable
|
||||
import org.apache.kafka.streams.kstream.KStream
|
||||
import org.apache.kafka.streams.kstream.KTable
|
||||
|
||||
|
||||
/** Use the hashcode of Prototypes names as keys - more efficient storage */
|
||||
@JvmInline
|
||||
@Serializable
|
||||
value class TileProtoHashCode private constructor(val code: Int) {
|
||||
constructor(tile: MapTile) : this(tile.protoId.hashCode())
|
||||
constructor(tile: FactorioPrototype.MapTile) : this(tile.protoId.hashCode())
|
||||
}
|
||||
|
||||
|
||||
@JvmInline
|
||||
@Serializable
|
||||
value class TileColourDict(
|
||||
private val map: Map<TileProtoHashCode, ColourHex>
|
||||
) : Map<TileProtoHashCode, ColourHex> by map
|
||||
private val map: Map<PrototypeHashCode, ColourHex>
|
||||
) : Map<PrototypeHashCode, ColourHex> by map {
|
||||
operator fun plus(other: TileColourDict): TileColourDict = TileColourDict(this.map + other.map)
|
||||
}
|
||||
|
||||
|
||||
/** Get the latest map-tile colours per server. */
|
||||
|
|
@ -46,7 +44,7 @@ fun KStream<FactorioServerId, PrototypesUpdate>.createTilePrototypeTable()
|
|||
val map = protoPacket
|
||||
.prototypes
|
||||
.filterIsInstance<FactorioPrototype.MapTile>()
|
||||
.associate { TileProtoHashCode(it) to it.mapColour.toHex() }
|
||||
.associate { PrototypeHashCode(it) to it.mapColour.toHex() }
|
||||
|
||||
TileColourDict(map)
|
||||
}.filter("$pid.filterMapTileProtos") { _, dict: TileColourDict ->
|
||||
|
|
@ -70,3 +68,71 @@ fun KStream<FactorioServerId, PrototypesUpdate>.createTilePrototypeTable()
|
|||
)
|
||||
)
|
||||
}
|
||||
|
||||
|
||||
@JvmInline
|
||||
@Serializable
|
||||
value class EntityPrototypeMap(
|
||||
private val map: Map<PrototypeHashCode, FactorioPrototype.Entity>
|
||||
) : Map<PrototypeHashCode, FactorioPrototype.Entity> by map {
|
||||
operator fun plus(other: EntityPrototypeMap): EntityPrototypeMap =
|
||||
EntityPrototypeMap(this.map + other.map)
|
||||
|
||||
operator fun get(protoId: PrototypeId): FactorioPrototype.Entity? =
|
||||
map[PrototypeHashCode(protoId)]
|
||||
|
||||
operator fun get(entity: FactorioEntityData): FactorioPrototype.Entity? =
|
||||
map[PrototypeHashCode(entity.protoId)]
|
||||
}
|
||||
|
||||
|
||||
/** Get the latest entity colours per server. */
|
||||
fun KStream<FactorioServerId, PrototypesUpdate>.createEntityPrototypeMap(
|
||||
callerPid: String
|
||||
): KTable<FactorioServerId, EntityPrototypeMap> {
|
||||
|
||||
val pid = "$callerPid.createEntityPrototypeMap"
|
||||
|
||||
return mapValues("$pid.colour-dict.build") { _, protoPacket: PrototypesUpdate ->
|
||||
val map = protoPacket
|
||||
.prototypes
|
||||
.filterIsInstance<FactorioPrototype.Entity>()
|
||||
.associateBy { entityProto ->
|
||||
PrototypeHashCode(entityProto)
|
||||
}
|
||||
|
||||
EntityPrototypeMap(map)
|
||||
}.filter("$pid.filterEntityProtos") { _, dict: EntityPrototypeMap ->
|
||||
dict.isNotEmpty()
|
||||
}.groupByKey(groupedAs("$pid.group-by-key"))
|
||||
.reduce(
|
||||
"$pid.merge",
|
||||
materializedWith(
|
||||
kxsBinary.serde(),
|
||||
kxsBinary.serde(),
|
||||
)
|
||||
) { dict1: EntityPrototypeMap, dict2: EntityPrototypeMap ->
|
||||
dict1 + dict2
|
||||
}.toStream("$pid.")
|
||||
.filter("$pid.not-null.filter") { _, v -> v != null }
|
||||
.mapValues("$pid.not-null.require") { _, value ->
|
||||
requireNotNull(value)
|
||||
}.peek { serverId, dict: EntityPrototypeMap ->
|
||||
println("$pid: server $serverId has EntityPrototypeMap (size:${dict.size}): ${dict.entries.joinToString()}")
|
||||
}.repartition(
|
||||
repartitionedAs(
|
||||
"$pid.pre-table-repartition",
|
||||
kxsBinary.serde(),
|
||||
kxsBinary.serde(),
|
||||
// force, otherwise KTable-KTable FK join doesn't work
|
||||
numberOfPartitions = 1,
|
||||
)
|
||||
).toTable(
|
||||
"$pid.create-table",
|
||||
materializedAs(
|
||||
"$pid.output-store",
|
||||
kxsBinary.serde(),
|
||||
kxsBinary.serde(),
|
||||
)
|
||||
)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -8,11 +8,13 @@ import dev.adamko.kafkatorio.schema.common.ChunkSize
|
|||
import dev.adamko.kafkatorio.schema.common.ColourHex
|
||||
import dev.adamko.kafkatorio.schema.common.FactorioServerId
|
||||
import dev.adamko.kafkatorio.schema.common.ServerMapChunkId
|
||||
import dev.adamko.kafkatorio.schema.common.TilePngFilename
|
||||
import dev.adamko.kafkatorio.schema.common.ServerMapTilePngFilename
|
||||
import dev.adamko.kafkatorio.schema.packets.EventServerPacket
|
||||
import dev.adamko.kafkatorio.server.config.jsonMapper
|
||||
import dev.adamko.kafkatorio.server.processor.TOPIC_BROADCAST_TO_WEBSOCKET
|
||||
import dev.adamko.kafkatorio.server.processor.TOPIC_MAP_CHUNK_COLOURED_STATE
|
||||
import dev.adamko.kafkatorio.server.processor.TOPIC_MAP_CHUNK_BUILDING_COLOURED_STATE
|
||||
import dev.adamko.kafkatorio.server.processor.TOPIC_MAP_CHUNK_RESOURCE_COLOURED_STATE
|
||||
import dev.adamko.kafkatorio.server.processor.TOPIC_MAP_CHUNK_TERRAIN_COLOURED_STATE
|
||||
import dev.adamko.kotka.extensions.consumedAs
|
||||
import dev.adamko.kotka.extensions.producedAs
|
||||
import dev.adamko.kotka.extensions.stream
|
||||
|
|
@ -51,12 +53,14 @@ fun saveMapTiles(
|
|||
kxsBinary.serde(ServerMapChunkId.serializer()),
|
||||
kxsBinary.serde(ServerMapChunkTiles.serializer(ColourHex.serializer())),
|
||||
),
|
||||
TOPIC_MAP_CHUNK_COLOURED_STATE,
|
||||
TOPIC_MAP_CHUNK_TERRAIN_COLOURED_STATE,
|
||||
TOPIC_MAP_CHUNK_RESOURCE_COLOURED_STATE,
|
||||
TOPIC_MAP_CHUNK_BUILDING_COLOURED_STATE,
|
||||
).filter("$pid.consume.grouped-map-chunks.filter-size") { _, y ->
|
||||
y.chunkId.chunkSize.lengthInTiles >= ChunkSize.CHUNK_256.lengthInTiles
|
||||
}
|
||||
|
||||
val savedChunkFilenames: KStream<ServerMapChunkId, TilePngFilename> =
|
||||
val savedChunkFilenames: KStream<ServerMapChunkId, ServerMapTilePngFilename> =
|
||||
subdividedMapChunkTilesDebounced.saveChunkAsImage(serverDataDir)
|
||||
|
||||
savedChunkFilenames.broadcastSavedTileUpdates()
|
||||
|
|
@ -67,7 +71,7 @@ fun saveMapTiles(
|
|||
|
||||
private fun KStream<ServerMapChunkId, ServerMapChunkTiles<ColourHex>>.saveChunkAsImage(
|
||||
tileDirectory: Path,
|
||||
): KStream<ServerMapChunkId, TilePngFilename> = mapValues(
|
||||
): KStream<ServerMapChunkId, ServerMapTilePngFilename> = mapValues(
|
||||
"$pid.save-tiles"
|
||||
) { chunkId: ServerMapChunkId, tileColours: ServerMapChunkTiles<ColourHex> ->
|
||||
|
||||
|
|
@ -81,19 +85,36 @@ private fun KStream<ServerMapChunkId, ServerMapChunkTiles<ColourHex>>.saveChunkA
|
|||
val chunkOriginX: Int = chunkId.chunkPosition.x * chunkId.chunkSize.lengthInTiles
|
||||
val chunkOriginY: Int = chunkId.chunkPosition.y * chunkId.chunkSize.lengthInTiles
|
||||
|
||||
val validPixelRangeX = 0 until chunkImage.width
|
||||
val validPixelRangeY = 0 until chunkImage.height
|
||||
|
||||
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)
|
||||
if (!(pixelX in validPixelRangeX && pixelY in validPixelRangeY)) {
|
||||
// TODO figure out why buildings on chunk boundaries aren't re-chunked, and so cause this error
|
||||
println(
|
||||
"ERROR - pixel XY:[$pixelX,$pixelY] was out of bounds (x:$validPixelRangeX, y:$validPixelRangeY). $tilePosition, $colour, chunkOriginXY[$chunkOriginX,$chunkOriginY], $chunkId"
|
||||
)
|
||||
} else {
|
||||
try {
|
||||
chunkImage.setColor(pixelX, pixelY, rgbColour)
|
||||
} catch (e: Exception) {
|
||||
println(
|
||||
"ERROR $e - pixel XY:[$pixelX,$pixelY] (x:$validPixelRangeX, y:$validPixelRangeY). $tilePosition, $colour, chunkOriginXY[$chunkOriginX,$chunkOriginY], $chunkId"
|
||||
)
|
||||
throw e
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
val scaledImage =
|
||||
chunkImage.scaleTo(WEB_MAP_TILE_SIZE_PX, WEB_MAP_TILE_SIZE_PX, ScaleMethod.FastScale)
|
||||
|
||||
val filename = TilePngFilename(chunkId)
|
||||
val filename = ServerMapTilePngFilename(chunkId)
|
||||
saveTile(filename, tileDirectory, scaledImage)
|
||||
filename
|
||||
}
|
||||
|
|
@ -101,7 +122,7 @@ private fun KStream<ServerMapChunkId, ServerMapChunkTiles<ColourHex>>.saveChunkA
|
|||
|
||||
@Synchronized
|
||||
private fun saveTile(
|
||||
filename: TilePngFilename,
|
||||
filename: ServerMapTilePngFilename,
|
||||
serverDataDir: Path,
|
||||
img: ImmutableImage,
|
||||
) {
|
||||
|
|
@ -124,8 +145,8 @@ private fun saveTile(
|
|||
}
|
||||
|
||||
|
||||
private fun KStream<ServerMapChunkId, TilePngFilename>.broadcastSavedTileUpdates() {
|
||||
map<ServerMapChunkId, TilePngFilename, FactorioServerId, EventServerPacket>(
|
||||
private fun KStream<ServerMapChunkId, ServerMapTilePngFilename>.broadcastSavedTileUpdates() {
|
||||
map<ServerMapChunkId, ServerMapTilePngFilename, FactorioServerId, EventServerPacket>(
|
||||
"$pid.broadcast.filename-update.convert",
|
||||
) { id, filename ->
|
||||
id.serverId to EventServerPacket.ChunkTileSaved(id, filename)
|
||||
|
|
|
|||
|
|
@ -69,7 +69,7 @@ class SyslogSocketServer(
|
|||
log("received 'null' from ${socket.description()} - closing connection")
|
||||
socket.close()
|
||||
}
|
||||
else -> log("received '$line' from ${socket.description()}")
|
||||
// else -> log("received '$line' from ${socket.description()}")
|
||||
}
|
||||
line
|
||||
},
|
||||
|
|
|
|||
|
|
@ -2,8 +2,9 @@ package dev.adamko.kafkatorio.server.web.rest
|
|||
|
||||
import dev.adamko.kafkatorio.schema.common.FactorioServerId
|
||||
import dev.adamko.kafkatorio.schema.common.PlayerIndex
|
||||
import dev.adamko.kafkatorio.schema.common.ServerMapTileLayer
|
||||
import dev.adamko.kafkatorio.schema.common.ServerMapTilePngFilename
|
||||
import dev.adamko.kafkatorio.schema.common.SurfaceIndex
|
||||
import dev.adamko.kafkatorio.schema.common.TilePngFilename
|
||||
import dev.adamko.kafkatorio.server.config.ApplicationProperties
|
||||
import io.ktor.http.HttpStatusCode
|
||||
import io.ktor.http.content.EntityTagVersion
|
||||
|
|
@ -41,8 +42,9 @@ private suspend fun PipelineContext<Unit, ApplicationCall>.handleTileRequest(
|
|||
appProps: ApplicationProperties,
|
||||
) {
|
||||
|
||||
val tileFilename = TilePngFilename(
|
||||
val tileFilename = ServerMapTilePngFilename(
|
||||
tile.serverId,
|
||||
tile.layer,
|
||||
tile.surfaceIndex,
|
||||
tile.zoomLevel,
|
||||
tile.chunkX,
|
||||
|
|
@ -60,6 +62,19 @@ private suspend fun PipelineContext<Unit, ApplicationCall>.handleTileRequest(
|
|||
}
|
||||
} else {
|
||||
call.respond(HttpStatusCode.NotFound)
|
||||
// // It's not a client error to request this tile, it just doesn't exist (yet). So return 304?
|
||||
// call.response.cacheControl(CacheControl.NoStore(null))
|
||||
//
|
||||
// HeadersBuilder(size = 2).apply {
|
||||
// EntityTagVersion(System.currentTimeMillis().hashCode().toString())
|
||||
// .appendHeadersTo(this)
|
||||
// LastModifiedVersion(System.currentTimeMillis())
|
||||
// .appendHeadersTo(this)
|
||||
// }.build().flattenForEach { name, value ->
|
||||
// call.response.header(name, value)
|
||||
// }
|
||||
//
|
||||
// call.respond(HttpStatusCode.NotModified)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -82,9 +97,11 @@ data class FactorioServer(
|
|||
) {
|
||||
|
||||
@Serializable
|
||||
@Resource("tiles/s{surfaceIndex}/z{zoomLevel}/x{chunkX}/y{chunkY}.png")
|
||||
@Resource("layers/{layer}/s{surfaceIndex}/z{zoomLevel}/x{chunkX}/y{chunkY}.png")
|
||||
data class Tile(
|
||||
val mapData: MapData,
|
||||
|
||||
val layer: ServerMapTileLayer,
|
||||
val surfaceIndex: SurfaceIndex,
|
||||
val zoomLevel: Int,
|
||||
val chunkX: Int,
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@ package dev.adamko.kafkatorio.server.web.rest
|
|||
import dev.adamko.kafkatorio.server.config.ApplicationProperties
|
||||
import io.ktor.http.ContentType
|
||||
import io.ktor.http.HttpStatusCode
|
||||
import io.ktor.http.isSuccess
|
||||
import io.ktor.serialization.kotlinx.json.json
|
||||
import io.ktor.server.application.Application
|
||||
import io.ktor.server.application.install
|
||||
|
|
@ -24,7 +25,12 @@ import io.ktor.server.routing.routing
|
|||
fun Application.webmapTileServer(appProps: ApplicationProperties) {
|
||||
|
||||
install(AutoHeadResponse)
|
||||
install(CallLogging)
|
||||
install(CallLogging) {
|
||||
filter { call ->
|
||||
// only log failures
|
||||
!(call.response.status()?.isSuccess() ?: false)
|
||||
}
|
||||
}
|
||||
install(Compression) {
|
||||
default()
|
||||
minimumSize(1024)
|
||||
|
|
|
|||
|
|
@ -97,7 +97,9 @@ class WebmapWebsocketServer {
|
|||
override suspend fun send(frame: Frame): Unit = runCatching {
|
||||
super.send(frame)
|
||||
}.fold(
|
||||
onSuccess = { log("sent message to client $name: $frame") },
|
||||
onSuccess = {
|
||||
// log("sent message to client $name: $frame")
|
||||
},
|
||||
onFailure = { e -> log("failed to send message to client $name: $e") }
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -5,7 +5,7 @@ server-data-dir: ${SERVER_DATA_DIR:-./server-data}
|
|||
|
||||
|
||||
socket-server-host: ${SOCKET_SERVER_HOST:-127.0.0.1}
|
||||
socket-server-port: ${SOCKET_SERVER_PORT:-12092}
|
||||
#socket-server-port: ${SOCKET_SERVER_PORT:-12092}
|
||||
|
||||
|
||||
kafka-admin-config:
|
||||
|
|
|
|||
|
|
@ -7,6 +7,7 @@ import dev.adamko.kafkatorio.schema.common.ColourHex
|
|||
import dev.adamko.kafkatorio.schema.common.FactorioServerId
|
||||
import dev.adamko.kafkatorio.schema.common.MapChunkPosition
|
||||
import dev.adamko.kafkatorio.schema.common.MapTilePosition
|
||||
import dev.adamko.kafkatorio.schema.common.ServerMapTileLayer
|
||||
import dev.adamko.kafkatorio.schema.common.SurfaceIndex
|
||||
import dev.adamko.kafkatorio.server.processor.topology.ServerMapChunkTiles
|
||||
import java.util.UUID
|
||||
|
|
@ -22,9 +23,9 @@ class Blah {
|
|||
val tiles = ServerMapChunkTiles<ColourHex>(
|
||||
chunkId = ServerMapChunkId(
|
||||
serverId = FactorioServerId(UUID.randomUUID().toString()),
|
||||
chunkPosition = MapChunkPosition(Int.MAX_VALUE, Int.MAX_VALUE),
|
||||
layer = ServerMapTileLayer.BUILDING,
|
||||
chunkPosition = MapChunkPosition(Int.MAX_VALUE, Int.MAX_VALUE, ChunkSize.MAX),
|
||||
surfaceIndex = SurfaceIndex(UInt.MAX_VALUE),
|
||||
chunkSize = ChunkSize.MAX,
|
||||
),
|
||||
buildMap {
|
||||
repeat(512) { x ->
|
||||
|
|
|
|||
|
|
@ -1,15 +1,16 @@
|
|||
package dev.adamko.kafkatorio.server.processor.topology
|
||||
|
||||
import dev.adamko.kafkatorio.library.iterateTiles
|
||||
import dev.adamko.kafkatorio.library.kxsBinary
|
||||
import dev.adamko.kafkatorio.library.toMapChunkPosition
|
||||
import dev.adamko.kafkatorio.schema.common.ChunkSize
|
||||
import dev.adamko.kafkatorio.schema.common.ColourHex
|
||||
import dev.adamko.kafkatorio.schema.common.FactorioServerId
|
||||
import dev.adamko.kafkatorio.schema.common.MapChunkPosition
|
||||
import dev.adamko.kafkatorio.schema.common.MapTilePosition
|
||||
import dev.adamko.kafkatorio.schema.common.ServerMapChunkId
|
||||
import dev.adamko.kafkatorio.schema.common.ServerMapTileLayer
|
||||
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
|
||||
|
|
@ -55,11 +56,11 @@ class SaveMapChunksTest : FunSpec({
|
|||
scenario += 1.minutes
|
||||
delay(5.seconds)
|
||||
scenario += 1.minutes
|
||||
// delay(5.seconds)
|
||||
// delay(5.seconds)
|
||||
scenario += 1.minutes
|
||||
// delay(5.seconds)
|
||||
// delay(5.seconds)
|
||||
scenario += 1.minutes
|
||||
// delay(5.seconds)
|
||||
// delay(5.seconds)
|
||||
|
||||
println("file:///${scenario.outputDir.absolutePathString()}")
|
||||
}
|
||||
|
|
@ -71,7 +72,7 @@ class SaveMapChunksTest : FunSpec({
|
|||
class Scenario(
|
||||
streamsBuilder: StreamsBuilder = StreamsBuilder(),
|
||||
|
||||
groupedMapChunksInputTopicName: String = "kafkatorio.state.map-chunk.colour",
|
||||
groupedMapChunksInputTopicName: String = "kafkatorio.state.map-chunk.terrain.colour",
|
||||
|
||||
val serverId: FactorioServerId = FactorioServerId("test-server-id"),
|
||||
) : AutoCloseable {
|
||||
|
|
@ -104,8 +105,9 @@ class SaveMapChunksTest : FunSpec({
|
|||
private fun mapChunkPositionArb(
|
||||
xArb: Arb<Int> = Arb.int(-10..10),
|
||||
yArb: Arb<Int> = Arb.int(-10..10),
|
||||
chunkSize: ChunkSize = ChunkSize.MAX,
|
||||
): Arb<MapChunkPosition> = arbitrary {
|
||||
MapChunkPosition(xArb.bind(), yArb.bind())
|
||||
MapChunkPosition(xArb.bind(), yArb.bind(), chunkSize)
|
||||
}
|
||||
|
||||
private fun colourHexArb(
|
||||
|
|
@ -124,15 +126,15 @@ class SaveMapChunksTest : FunSpec({
|
|||
|
||||
private fun serverMapChunkIdArb(
|
||||
serverId: FactorioServerId,
|
||||
chunkPosition: Arb<MapChunkPosition> = mapChunkPositionArb(),
|
||||
surfaceIndex: SurfaceIndex = SurfaceIndex(1u),
|
||||
chunkSize: ChunkSize = ChunkSize.MAX,
|
||||
chunkPosition: Arb<MapChunkPosition> = mapChunkPositionArb(chunkSize = chunkSize),
|
||||
surfaceIndex: SurfaceIndex = SurfaceIndex(1u),
|
||||
): Arb<ServerMapChunkId> = arbitrary {
|
||||
ServerMapChunkId(
|
||||
serverId = serverId,
|
||||
layer = ServerMapTileLayer.RESOURCE,
|
||||
chunkPosition = chunkPosition.bind(),
|
||||
surfaceIndex = surfaceIndex,
|
||||
chunkSize = chunkSize,
|
||||
)
|
||||
}
|
||||
|
||||
|
|
@ -145,7 +147,7 @@ class SaveMapChunksTest : FunSpec({
|
|||
println("chunk size: ${chunkId.chunkSize} ${chunkId.chunkSize.lengthInTiles}")
|
||||
|
||||
val map = chunkId.chunkPosition
|
||||
.iterateTiles(chunkId.chunkSize)
|
||||
.iterateTiles()
|
||||
.asSequence()
|
||||
.map { tilePos ->
|
||||
val colour =
|
||||
|
|
@ -154,7 +156,7 @@ class SaveMapChunksTest : FunSpec({
|
|||
}
|
||||
tilePos to colour
|
||||
}
|
||||
// .take(2500)
|
||||
// .take(100)
|
||||
.toMap()
|
||||
|
||||
println(map.size)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue