massive refactor to export entities (buildings and resources)
- minor version bumps - new Factorio version - use both prototype name and type to identify a prototype - kxs-ts-gen: use BrandTyping, but exclude some identifiers - kxs-ts-gen: manually create zip dist, so it can run after the project compiles - mod: use typed-factorio strict types - mod: refactor event handlers to be a little more consistent and object-oriented
This commit is contained in:
Adam
parent
88982d818a
commit
90dc520d22
62 changed files with 2629 additions and 713 deletions
|
|
@ -2,11 +2,13 @@ package dev.adamko.kafkatorio.server.processor
|
|||
|
||||
import com.github.palindromicity.syslog.dsl.SyslogFieldKeys
|
||||
import dev.adamko.kafkatorio.server.config.ApplicationProperties
|
||||
import dev.adamko.kafkatorio.server.processor.topology.broadcastToWebsocket
|
||||
import dev.adamko.kafkatorio.server.processor.topology.colourMapChunks
|
||||
import dev.adamko.kafkatorio.server.processor.topology.*
|
||||
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.socket.SyslogSocketServer
|
||||
import dev.adamko.kafkatorio.server.web.websocket.WebmapWebsocketServer
|
||||
import dev.adamko.kafkatorio.server.web.websocket.broadcastToWebsocket
|
||||
import java.util.Properties
|
||||
import kotlin.coroutines.resume
|
||||
import kotlin.coroutines.resumeWithException
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
package dev.adamko.kafkatorio.server.processor.admin
|
||||
|
||||
import dev.adamko.kafkatorio.server.processor.allTopics
|
||||
import dev.adamko.kafkatorio.server.config.ApplicationProperties
|
||||
import dev.adamko.kafkatorio.server.processor.allTopics
|
||||
import java.util.concurrent.TimeUnit
|
||||
import mu.KotlinLogging
|
||||
import org.apache.kafka.clients.admin.Admin
|
||||
|
|
|
|||
|
|
@ -1,9 +1,9 @@
|
|||
package dev.adamko.kafkatorio.server.processor
|
||||
|
||||
|
||||
import dev.adamko.kafkatorio.server.socket.SyslogSocketServer
|
||||
import dev.adamko.kafkatorio.server.processor.admin.KafkatorioKafkaAdmin
|
||||
import dev.adamko.kafkatorio.server.config.ApplicationProperties
|
||||
import dev.adamko.kafkatorio.server.processor.admin.KafkatorioKafkaAdmin
|
||||
import dev.adamko.kafkatorio.server.socket.SyslogSocketServer
|
||||
import dev.adamko.kafkatorio.server.web.websocket.WebmapWebsocketServer
|
||||
import kotlinx.coroutines.awaitCancellation
|
||||
import kotlinx.coroutines.coroutineScope
|
||||
|
|
@ -27,11 +27,11 @@ suspend fun eventsProcessor(
|
|||
val topology = KafkatorioTopology(appProps, wsServer, syslogServer)
|
||||
|
||||
|
||||
launch {
|
||||
topology.start()
|
||||
launch {
|
||||
topology.start()
|
||||
|
||||
println("launched KafkatorioTopology")
|
||||
println("launched KafkatorioTopology")
|
||||
|
||||
awaitCancellation()
|
||||
}
|
||||
awaitCancellation()
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -5,7 +5,8 @@ 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.MapChunkUpdate
|
||||
import dev.adamko.kafkatorio.schema.packets.MapChunkEntityUpdate
|
||||
import dev.adamko.kafkatorio.schema.packets.MapChunkTileUpdate
|
||||
import dev.adamko.kafkatorio.schema.packets.PlayerUpdate
|
||||
import dev.adamko.kafkatorio.schema.packets.PrototypesUpdate
|
||||
import dev.adamko.kafkatorio.schema.packets.SurfaceUpdate
|
||||
|
|
@ -54,9 +55,10 @@ val KafkatorioPacketData.topicName: String
|
|||
is PrototypesUpdate -> packetTopicNames.getValue(PrototypesUpdate::class)
|
||||
is SurfaceUpdate -> packetTopicNames.getValue(SurfaceUpdate::class)
|
||||
is EntityUpdate -> packetTopicNames.getValue(EntityUpdate::class)
|
||||
is MapChunkUpdate -> packetTopicNames.getValue(MapChunkUpdate::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)
|
||||
}
|
||||
|
||||
|
||||
|
|
@ -67,12 +69,14 @@ private val packetTopicNames: Map<KClass<out KafkatorioPacketData>, String> = ma
|
|||
PrototypesUpdate::class to "prototypes",
|
||||
SurfaceUpdate::class to "surface",
|
||||
EntityUpdate::class to "entity",
|
||||
MapChunkUpdate::class to "map-chunk",
|
||||
MapChunkTileUpdate::class to "map-chunk-tile",
|
||||
PlayerUpdate::class to "player",
|
||||
MapChunkEntityUpdate::class to "map-chunk-entity",
|
||||
KafkatorioPacketData.Error::class to "error",
|
||||
).mapValues { (_, v) ->
|
||||
"$DOMAIN.packet.$v"
|
||||
}
|
||||
|
||||
|
||||
val KClass<out KafkatorioPacketData>.topicName: String
|
||||
get() = packetTopicNames.getValue(this)
|
||||
|
|
|
|||
|
|
@ -1,8 +1,5 @@
|
|||
package dev.adamko.kafkatorio.server.processor.topology
|
||||
|
||||
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.DebounceProcessor.Companion.addDebounceProcessor
|
||||
import dev.adamko.kafkatorio.library.kxsBinary
|
||||
import dev.adamko.kafkatorio.schema.common.ChunkSize
|
||||
import dev.adamko.kafkatorio.schema.common.ColourHex
|
||||
|
|
@ -10,8 +7,11 @@ import dev.adamko.kafkatorio.schema.common.FactorioServerId
|
|||
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.packets.MapChunkUpdate
|
||||
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.kotka.extensions.consumedAs
|
||||
import dev.adamko.kotka.extensions.groupedAs
|
||||
import dev.adamko.kotka.extensions.materializedAs
|
||||
|
|
@ -39,10 +39,10 @@ import org.apache.kafka.streams.kstream.KTable
|
|||
private const val pid: String = "colourMapChunks"
|
||||
|
||||
|
||||
/** Add colour to each [MapChunkUpdate] */
|
||||
/** Add colour to each [MapChunkTileUpdate] */
|
||||
fun colourMapChunks(builder: StreamsBuilder): Topology {
|
||||
|
||||
val mapChunksStream: KStream<FactorioServerId, MapChunkUpdate> = builder.streamPacketData()
|
||||
val mapChunksStream: KStream<FactorioServerId, MapChunkTileUpdate> = builder.streamPacketData()
|
||||
val protosStream: KStream<FactorioServerId, PrototypesUpdate> = builder.streamPacketData()
|
||||
|
||||
val protosTable: KTable<FactorioServerId, TileColourDict> =
|
||||
|
|
@ -78,7 +78,7 @@ 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, _ ->
|
||||
key.chunkSize == from
|
||||
|
|
@ -98,12 +98,12 @@ fun colourMapChunks(builder: StreamsBuilder): Topology {
|
|||
}
|
||||
|
||||
|
||||
private fun KStream<FactorioServerId, MapChunkUpdate>.convertToServerMapChunkTiles()
|
||||
private fun KStream<FactorioServerId, MapChunkTileUpdate>.convertToServerMapChunkTiles()
|
||||
: KStream<ServerMapChunkId, ServerMapChunkTiles<TileProtoHashCode>> {
|
||||
|
||||
val pid = "$pid.convertToMapTiles"
|
||||
|
||||
return map("$pid.map") { serverId: FactorioServerId, update: MapChunkUpdate ->
|
||||
return map("$pid.map") { serverId: FactorioServerId, update: MapChunkTileUpdate ->
|
||||
|
||||
val mapTileList = update.tileDictionary?.toMapTileList() ?: emptyList()
|
||||
|
||||
|
|
@ -112,7 +112,7 @@ private fun KStream<FactorioServerId, MapChunkUpdate>.convertToServerMapChunkTil
|
|||
}
|
||||
|
||||
if (invalidMapTilesList.isNotEmpty()) {
|
||||
println("WARNING [reduceMapTilesToTable] MapChunkUpdate contained out-of-bounds tiles $invalidMapTilesList")
|
||||
println("WARNING [reduceMapTilesToTable] MapChunkTileUpdate contained out-of-bounds tiles $invalidMapTilesList")
|
||||
}
|
||||
|
||||
val validMapTilesToProtoHashCode =
|
||||
|
|
@ -232,7 +232,7 @@ fun KTable<ServerMapChunkId, ServerMapChunkTiles<ColourHex>>.streamMapChunkColou
|
|||
|
||||
|
||||
/**
|
||||
* Group [MapChunkUpdate]s by the Chunk position.
|
||||
* 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.
|
||||
*/
|
||||
|
|
|
|||
|
|
@ -1,12 +1,12 @@
|
|||
package dev.adamko.kafkatorio.server.processor.topology
|
||||
|
||||
import dev.adamko.kafkatorio.server.processor.TOPIC_SRC_SERVER_LOG
|
||||
import dev.adamko.kafkatorio.server.processor.topicName
|
||||
import dev.adamko.kafkatorio.server.config.jsonMapper
|
||||
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.server.config.jsonMapper
|
||||
import dev.adamko.kafkatorio.server.processor.TOPIC_SRC_SERVER_LOG
|
||||
import dev.adamko.kafkatorio.server.processor.topicName
|
||||
import dev.adamko.kotka.extensions.component1
|
||||
import dev.adamko.kotka.extensions.component2
|
||||
import dev.adamko.kotka.extensions.consumedAs
|
||||
|
|
@ -35,7 +35,7 @@ fun factorioServerPacketStream(
|
|||
val packet = runCatching {
|
||||
jsonMapper.decodeFromString<KafkatorioPacket>(value)
|
||||
}.getOrElse { e ->
|
||||
KafkatorioPacket(
|
||||
KafkatorioPacket(
|
||||
modVersion = "unknown",
|
||||
tick = Tick(0u),
|
||||
data = KafkatorioPacketData.Error(
|
||||
|
|
|
|||
|
|
@ -1,10 +1,10 @@
|
|||
package dev.adamko.kafkatorio.server.processor.topology
|
||||
|
||||
import dev.adamko.kafkatorio.server.processor.topicName
|
||||
import dev.adamko.kafkatorio.server.config.jsonMapper
|
||||
import dev.adamko.kafkatorio.schema.common.FactorioServerId
|
||||
import dev.adamko.kafkatorio.schema.packets.KafkatorioPacket
|
||||
import dev.adamko.kafkatorio.schema.packets.KafkatorioPacketData
|
||||
import dev.adamko.kafkatorio.server.config.jsonMapper
|
||||
import dev.adamko.kafkatorio.server.processor.topicName
|
||||
import dev.adamko.kotka.extensions.consumedAs
|
||||
import dev.adamko.kotka.extensions.streams.filter
|
||||
import dev.adamko.kotka.extensions.streams.mapValues
|
||||
|
|
|
|||
|
|
@ -22,8 +22,8 @@ import org.apache.kafka.streams.kstream.KTable
|
|||
@JvmInline
|
||||
@Serializable
|
||||
value class TileProtoHashCode private constructor(val code: Int) {
|
||||
constructor(tile: MapTile) : this(tile.proto.hashCode())
|
||||
constructor(tile: FactorioPrototype.MapTile) : this(tile.name.hashCode())
|
||||
constructor(tile: MapTile) : this(tile.protoId.hashCode())
|
||||
constructor(tile: FactorioPrototype.MapTile) : this(tile.protoId.hashCode())
|
||||
}
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -3,15 +3,15 @@ package dev.adamko.kafkatorio.server.processor.topology
|
|||
import com.sksamuel.scrimage.ImmutableImage
|
||||
import com.sksamuel.scrimage.ScaleMethod
|
||||
import com.sksamuel.scrimage.nio.PngWriter
|
||||
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.config.jsonMapper
|
||||
import dev.adamko.kafkatorio.library.kxsBinary
|
||||
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.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.kotka.extensions.consumedAs
|
||||
import dev.adamko.kotka.extensions.producedAs
|
||||
import dev.adamko.kotka.extensions.streams.map
|
||||
|
|
@ -32,7 +32,7 @@ private const val WEB_MAP_TILE_SIZE_PX = 256
|
|||
|
||||
|
||||
/**
|
||||
* @param[serverDataDir] [dev.adamko.kafkatorio.processor.config.ApplicationProperties.serverDataDir]
|
||||
* @param[serverDataDir] [dev.adamko.kafkatorio.server.config.ApplicationProperties.serverDataDir]
|
||||
*/
|
||||
fun saveMapTiles(
|
||||
builder: StreamsBuilder,
|
||||
|
|
|
|||
|
|
@ -1,13 +1,14 @@
|
|||
package dev.adamko.kafkatorio.server.web.websocket
|
||||
package dev.adamko.kafkatorio.server.processor.topology
|
||||
|
||||
import dev.adamko.kafkatorio.library.jsonMapperKafkatorio
|
||||
import dev.adamko.kafkatorio.server.processor.TOPIC_BROADCAST_TO_WEBSOCKET
|
||||
import dev.adamko.kafkatorio.server.processor.topicName
|
||||
import dev.adamko.kafkatorio.schema.common.FactorioServerId
|
||||
import dev.adamko.kafkatorio.schema.packets.EventServerPacket
|
||||
import dev.adamko.kafkatorio.schema.packets.KafkatorioPacket
|
||||
import dev.adamko.kafkatorio.schema.packets.PlayerUpdate
|
||||
import dev.adamko.kafkatorio.schema.packets.SurfaceUpdate
|
||||
import dev.adamko.kafkatorio.server.processor.TOPIC_BROADCAST_TO_WEBSOCKET
|
||||
import dev.adamko.kafkatorio.server.processor.topicName
|
||||
import dev.adamko.kafkatorio.server.web.websocket.WebmapWebsocketServer
|
||||
import dev.adamko.kotka.extensions.consumedAs
|
||||
import dev.adamko.kotka.extensions.producedAs
|
||||
import dev.adamko.kotka.extensions.stream
|
||||
|
|
|
|||
|
|
@ -59,12 +59,14 @@ class SyslogSocketServer(
|
|||
launch {
|
||||
val channel = socket.openReadChannel()
|
||||
try {
|
||||
while (socket.socketContext.isActive) {
|
||||
while (socket.socketContext.isActive && !channel.isClosedForRead) {
|
||||
val line = runCatching {
|
||||
channel.readUTF8Line()
|
||||
}.fold(
|
||||
onSuccess = { line ->
|
||||
log("received '$line' from $socket")
|
||||
if (line != null) {
|
||||
log("received '$line' from ${socket.description()}")
|
||||
}
|
||||
line
|
||||
},
|
||||
onFailure = { e ->
|
||||
|
|
|
|||
|
|
@ -10,9 +10,7 @@ import io.ktor.http.content.EntityTagVersion
|
|||
import io.ktor.http.content.versions
|
||||
import io.ktor.resources.Resource
|
||||
import io.ktor.server.application.ApplicationCall
|
||||
import io.ktor.server.application.application
|
||||
import io.ktor.server.application.call
|
||||
import io.ktor.server.application.log
|
||||
import io.ktor.server.http.content.LastModifiedVersion
|
||||
import io.ktor.server.resources.get
|
||||
import io.ktor.server.response.respond
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue