biggest refactor ever...
- use kxs-ts-gen for TypeScript schemas
- update schema files, so they're tidier (no need for typescript-generator workarounds)
- version bumps (kotlin, gradle, tstl, factorio, kotest)
- improve kxs binary format (use okio)
- move schema files package ('events' wasn't needed)
- tidy up topic creation, naming. base it off classes, make util functions
This commit is contained in:
Adam
parent
db14865ca9
commit
3fff57d46d
82 changed files with 2545 additions and 1914 deletions
|
|
@ -41,6 +41,9 @@ dependencies {
|
|||
|
||||
implementation(libs.scrimage.core)
|
||||
|
||||
implementation(platform(libs.okio.bom))
|
||||
implementation(libs.okio.core)
|
||||
|
||||
implementation(libs.kafka.kotkaStreams) {
|
||||
isChanging = true
|
||||
}
|
||||
|
|
|
|||
|
|
@ -7,9 +7,9 @@ import dev.adamko.kafkatorio.processor.topology.playerUpdatesToWsServer
|
|||
import dev.adamko.kafkatorio.processor.topology.saveMapTiles
|
||||
import dev.adamko.kafkatorio.processor.topology.splitFactorioServerPacketStream
|
||||
import java.time.Duration
|
||||
import java.util.Properties
|
||||
import kotlin.coroutines.CoroutineContext
|
||||
import kotlin.coroutines.cancellation.CancellationException
|
||||
import kotlin.time.Duration.Companion.minutes
|
||||
import kotlinx.coroutines.CoroutineName
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
|
|
@ -31,15 +31,12 @@ internal class KafkatorioTopology(
|
|||
override val coroutineContext: CoroutineContext =
|
||||
Dispatchers.Default + SupervisorJob() + CoroutineName("KafkatorioTopology")
|
||||
|
||||
companion object {
|
||||
const val TOPIC_SRC_SERVER_LOG = "factorio-server-log"
|
||||
const val TOPIC_GROUPED_MAP_CHUNKS = "kafkatorio.MAP.grouped-map-chunks"
|
||||
}
|
||||
|
||||
fun start() {
|
||||
splitPackets()
|
||||
|
||||
groupTilesMapChunks()
|
||||
saveTiles()
|
||||
splitPackets()
|
||||
|
||||
playerUpdates()
|
||||
}
|
||||
|
||||
|
|
@ -56,7 +53,6 @@ internal class KafkatorioTopology(
|
|||
}
|
||||
|
||||
private fun splitPackets() {
|
||||
|
||||
val builder = StreamsBuilder()
|
||||
val packets = factorioServerPacketStream(builder)
|
||||
splitFactorioServerPacketStream(packets)
|
||||
|
|
@ -75,12 +71,19 @@ internal class KafkatorioTopology(
|
|||
launchTopology("playerUpdates", topology)
|
||||
}
|
||||
|
||||
private fun launchTopology(id: String, topology: Topology) {
|
||||
private fun launchTopology(
|
||||
id: String,
|
||||
topology: Topology,
|
||||
config: Map<String, String> = mapOf()
|
||||
) {
|
||||
|
||||
val props = appProps.kafkaConfig
|
||||
val props: Properties = appProps.kafkaConfig
|
||||
|
||||
config.forEach { (k, v) -> props.setProperty(k, v) }
|
||||
|
||||
val appId = props.compute(StreamsConfig.APPLICATION_ID_CONFIG) { _, v -> "$v.$id" } as? String
|
||||
props.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, appId)
|
||||
|
||||
// props.setProperty(StreamsConfig.MAX_TASK_IDLE_MS_CONFIG, "${30.minutes.inWholeMilliseconds}")
|
||||
|
||||
val streams = KafkaStreams(topology, props)
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@ package dev.adamko.kafkatorio.processor
|
|||
import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler
|
||||
import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse
|
||||
|
||||
|
||||
class StreamsExceptionHandler(
|
||||
val shutdown: (e: Throwable) -> Unit,
|
||||
) : StreamsUncaughtExceptionHandler {
|
||||
|
|
|
|||
|
|
@ -1,9 +1,5 @@
|
|||
package dev.adamko.kafkatorio.processor.admin
|
||||
|
||||
import dev.adamko.kafkatorio.events.schema.FactorioEventUpdate
|
||||
import dev.adamko.kafkatorio.events.schema.FactorioObjectData
|
||||
import dev.adamko.kafkatorio.events.schema.KafkatorioPacket
|
||||
import dev.adamko.kafkatorio.processor.KafkatorioTopology
|
||||
import java.util.concurrent.TimeUnit
|
||||
import mu.KotlinLogging
|
||||
import org.apache.kafka.clients.admin.Admin
|
||||
|
|
@ -28,34 +24,9 @@ object KafkatorioKafkaAdmin {
|
|||
|
||||
val currentTopics = currentTopics().map { it.name() }
|
||||
|
||||
val kafkatorioTopics = buildSet {
|
||||
KafkatorioPacket.PacketType.values.forEach { packetType ->
|
||||
|
||||
add(KafkatorioTopology.TOPIC_GROUPED_MAP_CHUNKS)
|
||||
add(KafkatorioTopology.TOPIC_SRC_SERVER_LOG)
|
||||
|
||||
val base = topicNameBase(packetType)
|
||||
|
||||
when (packetType) {
|
||||
KafkatorioPacket.PacketType.EVENT ->
|
||||
FactorioObjectData.ObjectName.values.forEach { objectName ->
|
||||
add("$base.${objectName.name}")
|
||||
}
|
||||
KafkatorioPacket.PacketType.CONFIG -> add("$base.FactorioConfigurationUpdate")
|
||||
KafkatorioPacket.PacketType.PROTOTYPES -> add("$base.all")
|
||||
KafkatorioPacket.PacketType.UPDATE -> {
|
||||
add("$base.all")
|
||||
FactorioEventUpdate.FactorioEventUpdateType.values.forEach { objectName ->
|
||||
add("$base.${objectName.name}")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
}.minus(currentTopics.toSet())
|
||||
val kafkatorioTopics = allTopics() - currentTopics.toSet()
|
||||
|
||||
if (kafkatorioTopics.isNotEmpty()) {
|
||||
|
||||
logger.info("Creating ${kafkatorioTopics.size} topics")
|
||||
val result = kafkaAdmin.createTopics { kafkatorioTopics }
|
||||
|
||||
|
|
@ -83,11 +54,14 @@ object KafkatorioKafkaAdmin {
|
|||
replicationFactor: Short = 1,
|
||||
topicNames: () -> Collection<String>,
|
||||
): CreateTopicsResult =
|
||||
createTopics(topicNames()
|
||||
.distinct()
|
||||
.map {
|
||||
NewTopic(it, numPartitions, replicationFactor)
|
||||
})
|
||||
createTopics(
|
||||
topicNames()
|
||||
.distinct()
|
||||
.map {
|
||||
NewTopic(it, numPartitions, replicationFactor)
|
||||
// .configs(mapOf("cleanup.policy" to "compact"))
|
||||
}
|
||||
)
|
||||
|
||||
private fun currentTopics(): MutableCollection<TopicListing> {
|
||||
return kafkaAdmin.listTopics()
|
||||
|
|
@ -95,6 +69,6 @@ object KafkatorioKafkaAdmin {
|
|||
.get()
|
||||
}
|
||||
|
||||
fun topicNameBase(packetType: KafkatorioPacket.PacketType): String =
|
||||
"kafkatorio.${packetType.name}"
|
||||
// fun topicNameBase(packetType: KafkatorioPacketDataType): String =
|
||||
// "kafkatorio.${packetType.name}"
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,104 @@
|
|||
package dev.adamko.kafkatorio.processor.admin
|
||||
|
||||
import dev.adamko.kafkatorio.schema2.ConfigurationUpdate
|
||||
import dev.adamko.kafkatorio.schema2.ConsoleChatUpdate
|
||||
import dev.adamko.kafkatorio.schema2.ConsoleCommandUpdate
|
||||
import dev.adamko.kafkatorio.schema2.EntityUpdate
|
||||
import dev.adamko.kafkatorio.schema2.KafkatorioPacketData2
|
||||
import dev.adamko.kafkatorio.schema2.MapChunkUpdate
|
||||
import dev.adamko.kafkatorio.schema2.PlayerUpdate
|
||||
import dev.adamko.kafkatorio.schema2.PrototypesUpdate
|
||||
import dev.adamko.kafkatorio.schema2.SurfaceUpdate
|
||||
import kotlin.reflect.KClass
|
||||
|
||||
|
||||
private const val DOMAIN = "kafkatorio"
|
||||
|
||||
|
||||
const val TOPIC_SRC_SERVER_LOG = "$DOMAIN.src.server-log"
|
||||
// "factorio-server-log"
|
||||
const val TOPIC_GROUPED_MAP_CHUNKS_STATE = "$DOMAIN.state.map-chunks.grouped"
|
||||
|
||||
|
||||
fun allTopics(): Set<String> = buildSet {
|
||||
add(TOPIC_SRC_SERVER_LOG)
|
||||
add(TOPIC_GROUPED_MAP_CHUNKS_STATE)
|
||||
|
||||
addAll(packetTopicNames.values)
|
||||
}
|
||||
|
||||
//fun KafkatorioPacket2.topicName(): String = data.topicName()
|
||||
|
||||
//fun KafkatorioPacketData2.topicName(): String = "$DOMAIN.packet.${topicName}"
|
||||
|
||||
|
||||
val KafkatorioPacketData2.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 MapChunkUpdate -> packetTopicNames.getValue(MapChunkUpdate::class)
|
||||
is PlayerUpdate -> packetTopicNames.getValue(PlayerUpdate::class)
|
||||
}
|
||||
|
||||
|
||||
private val packetTopicNames: Map<KClass<out KafkatorioPacketData2>, String> = mapOf(
|
||||
ConfigurationUpdate::class to "configuration",
|
||||
ConsoleChatUpdate::class to "console-chat",
|
||||
ConsoleCommandUpdate::class to "console-command",
|
||||
PrototypesUpdate::class to "prototypes",
|
||||
SurfaceUpdate::class to "surface",
|
||||
EntityUpdate::class to "entity",
|
||||
MapChunkUpdate::class to "map-chunk",
|
||||
PlayerUpdate::class to "player",
|
||||
).mapValues { (_, v) ->
|
||||
"$DOMAIN.packet.$v"
|
||||
}
|
||||
|
||||
val KClass<out KafkatorioPacketData2>.topicName: String
|
||||
get() = packetTopicNames.getValue(this)
|
||||
|
||||
//fun packetTopic(
|
||||
// dataType: KafkatorioKeyedPacketData.KafkatorioKeyedPacketDataType,
|
||||
//): String = packetTopic(KafkatorioPacketType.KEYED, dataType.name.lowercase())
|
||||
//
|
||||
//private fun packetTopic(
|
||||
// packetType: KafkatorioPacketType,
|
||||
// dataType: String,
|
||||
//): String {
|
||||
// return "$DOMAIN.packet.$packetType.$dataType"
|
||||
//}
|
||||
|
||||
//sealed interface KafkatorioTopic {
|
||||
//
|
||||
// val type: String
|
||||
// val entity: String
|
||||
// val description: String
|
||||
//
|
||||
// companion object {
|
||||
// private const val DOMAIN = "kafkatorio"
|
||||
//
|
||||
// val KafkatorioTopic.topic
|
||||
// get() = listOf(DOMAIN, type, entity, description).joinToString(".")
|
||||
//
|
||||
// }
|
||||
//
|
||||
// data class Event(
|
||||
// override val entity: String,
|
||||
// override val description: String,
|
||||
// ) : KafkatorioTopic {
|
||||
// override val type: String = "event"
|
||||
// }
|
||||
//
|
||||
//
|
||||
// data class State(
|
||||
// override val entity: String,
|
||||
// override val description: String,
|
||||
// ) : KafkatorioTopic {
|
||||
// override val type: String = "state"
|
||||
// }
|
||||
//
|
||||
//}
|
||||
|
|
@ -1,11 +1,5 @@
|
|||
package dev.adamko.kafkatorio.processor.serdes
|
||||
|
||||
import java.io.ByteArrayInputStream
|
||||
import java.io.ByteArrayOutputStream
|
||||
import java.io.DataInput
|
||||
import java.io.DataInputStream
|
||||
import java.io.DataOutput
|
||||
import java.io.DataOutputStream
|
||||
import kotlinx.serialization.BinaryFormat
|
||||
import kotlinx.serialization.DeserializationStrategy
|
||||
import kotlinx.serialization.SerializationStrategy
|
||||
|
|
@ -17,6 +11,9 @@ import kotlinx.serialization.encoding.CompositeDecoder
|
|||
import kotlinx.serialization.encoding.CompositeEncoder
|
||||
import kotlinx.serialization.modules.EmptySerializersModule
|
||||
import kotlinx.serialization.modules.SerializersModule
|
||||
import okio.Buffer
|
||||
import okio.BufferedSink
|
||||
import okio.BufferedSource
|
||||
|
||||
|
||||
val kxsBinary: KxsBinary = KxsBinary()
|
||||
|
|
@ -27,18 +24,18 @@ class KxsBinary(
|
|||
) : BinaryFormat {
|
||||
|
||||
override fun <T> encodeToByteArray(serializer: SerializationStrategy<T>, value: T): ByteArray {
|
||||
val output = ByteArrayOutputStream()
|
||||
val encoder = KxsDataOutputEncoder(DataOutputStream(output), serializersModule)
|
||||
val output = Buffer()
|
||||
val encoder = KxsDataOutputEncoder(output, serializersModule)
|
||||
encoder.encodeSerializableValue(serializer, value)
|
||||
return output.toByteArray()
|
||||
return output.readByteArray()
|
||||
}
|
||||
|
||||
override fun <T> decodeFromByteArray(
|
||||
deserializer: DeserializationStrategy<T>,
|
||||
bytes: ByteArray
|
||||
): T {
|
||||
val input = ByteArrayInputStream(bytes)
|
||||
val decoder = KxsDataInputDecoder(DataInputStream(input), serializersModule)
|
||||
val input = Buffer().write(bytes)
|
||||
val decoder = KxsDataInputDecoder(input, serializersModule)
|
||||
return decoder.decodeSerializableValue(deserializer)
|
||||
}
|
||||
}
|
||||
|
|
@ -46,28 +43,31 @@ class KxsBinary(
|
|||
private val byteArraySerializer = ByteArraySerializer()
|
||||
|
||||
class KxsDataOutputEncoder(
|
||||
private val output: DataOutput,
|
||||
private val output: BufferedSink,
|
||||
override val serializersModule: SerializersModule = EmptySerializersModule,
|
||||
) : AbstractEncoder() {
|
||||
|
||||
override fun encodeBoolean(value: Boolean) = output.writeByte(if (value) 1 else 0)
|
||||
override fun encodeByte(value: Byte) = output.writeByte(value.toInt())
|
||||
override fun encodeChar(value: Char) = output.writeChar(value.code)
|
||||
override fun encodeDouble(value: Double) = output.writeDouble(value)
|
||||
override fun encodeEnum(enumDescriptor: SerialDescriptor, index: Int) = output.writeInt(index)
|
||||
override fun encodeFloat(value: Float) = output.writeFloat(value)
|
||||
override fun encodeInt(value: Int) = output.writeInt(value)
|
||||
override fun encodeLong(value: Long) = output.writeLong(value)
|
||||
override fun encodeShort(value: Short) = output.writeShort(value.toInt())
|
||||
override fun encodeString(value: String) = output.writeUTF(value)
|
||||
// override fun encodeString(value: String) = encodeByteArray(value.toByteArray(Charsets.UTF_8))
|
||||
private operator fun BufferedSink.invoke(action: BufferedSink.() -> Unit): Unit = action()
|
||||
|
||||
override fun encodeBoolean(value: Boolean) = output { writeByte(if (value) 1 else 0) }
|
||||
override fun encodeByte(value: Byte) = output { writeByte(value.toInt()) }
|
||||
override fun encodeChar(value: Char) = output { writeUtf8CodePoint(value.code) }
|
||||
override fun encodeDouble(value: Double) = output { writeLong(value.toRawBits()) }
|
||||
override fun encodeEnum(enumDescriptor: SerialDescriptor, index: Int) = output { writeInt(index) }
|
||||
override fun encodeFloat(value: Float) = output { writeInt(value.toRawBits()) }
|
||||
override fun encodeInt(value: Int) = output { writeInt(value) }
|
||||
override fun encodeLong(value: Long) = output { writeLong(value) }
|
||||
override fun encodeShort(value: Short) = output { writeShort(value.toInt()) }
|
||||
override fun encodeString(value: String) = output {
|
||||
encodeCompactSize(value.length)
|
||||
writeUtf8(value)
|
||||
}
|
||||
|
||||
override fun beginCollection(
|
||||
descriptor: SerialDescriptor,
|
||||
collectionSize: Int
|
||||
): CompositeEncoder {
|
||||
encodeInt(collectionSize)
|
||||
// encodeCompactSize(collectionSize)
|
||||
encodeCompactSize(collectionSize)
|
||||
return this
|
||||
}
|
||||
|
||||
|
|
@ -98,7 +98,7 @@ class KxsDataOutputEncoder(
|
|||
|
||||
|
||||
class KxsDataInputDecoder(
|
||||
private val input: DataInput,
|
||||
private val input: BufferedSource,
|
||||
override val serializersModule: SerializersModule = EmptySerializersModule,
|
||||
private var elementsCount: Int = 0,
|
||||
) : AbstractDecoder() {
|
||||
|
|
@ -107,15 +107,17 @@ class KxsDataInputDecoder(
|
|||
|
||||
override fun decodeBoolean(): Boolean = input.readByte().toInt() != 0
|
||||
override fun decodeByte(): Byte = input.readByte()
|
||||
override fun decodeChar(): Char = input.readChar()
|
||||
override fun decodeDouble(): Double = input.readDouble()
|
||||
override fun decodeChar(): Char = input.readUtf8CodePoint().toChar()
|
||||
override fun decodeDouble(): Double = Double.fromBits(input.readLong())
|
||||
override fun decodeEnum(enumDescriptor: SerialDescriptor): Int = input.readInt()
|
||||
override fun decodeFloat(): Float = input.readFloat()
|
||||
override fun decodeFloat(): Float = Float.fromBits(input.readInt())
|
||||
override fun decodeInt(): Int = input.readInt()
|
||||
override fun decodeLong(): Long = input.readLong()
|
||||
override fun decodeShort(): Short = input.readShort()
|
||||
override fun decodeString(): String = input.readUTF()
|
||||
// override fun decodeString(): String = decodeByteArray().toString(Charsets.UTF_8)
|
||||
override fun decodeString(): String {
|
||||
val size = decodeCompactSize()
|
||||
return input.readUtf8(size.toLong())
|
||||
}
|
||||
|
||||
override fun decodeElementIndex(descriptor: SerialDescriptor): Int {
|
||||
if (elementIndex == elementsCount) return CompositeDecoder.DECODE_DONE
|
||||
|
|
@ -128,8 +130,7 @@ class KxsDataInputDecoder(
|
|||
override fun decodeSequentially(): Boolean = true
|
||||
|
||||
override fun decodeCollectionSize(descriptor: SerialDescriptor): Int =
|
||||
decodeInt().also { elementsCount = it }
|
||||
// decodeCompactSize().also { elementsCount = it }
|
||||
decodeCompactSize().also { elementsCount = it }
|
||||
|
||||
override fun decodeNotNullMark(): Boolean = decodeBoolean()
|
||||
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
package dev.adamko.kafkatorio.processor.serdes
|
||||
|
||||
import dev.adamko.kafkatorio.events.schema.KafkatorioPacket
|
||||
import dev.adamko.kafkatorio.events.schema.jsonMapperKafkatorio
|
||||
import dev.adamko.kafkatorio.schema.jsonMapperKafkatorio
|
||||
import dev.adamko.kafkatorio.schema2.KafkatorioPacket2
|
||||
import kotlinx.serialization.decodeFromString
|
||||
import kotlinx.serialization.encodeToString
|
||||
import kotlinx.serialization.json.Json
|
||||
|
|
@ -13,15 +13,15 @@ import org.http4k.format.asConfigurable
|
|||
import org.http4k.format.withStandardMappings
|
||||
|
||||
|
||||
object KafkatorioPacketSerde : Serde<KafkatorioPacket> {
|
||||
override fun serializer() = Serializer<KafkatorioPacket> { _, message ->
|
||||
jsonMapper.encodeToString<KafkatorioPacket>(message).encodeToByteArray()
|
||||
}
|
||||
|
||||
override fun deserializer() = Deserializer<KafkatorioPacket> { _, bytes ->
|
||||
jsonMapper.decodeFromString(bytes.decodeToString())
|
||||
}
|
||||
}
|
||||
//object KafkatorioPacketSerde : Serde<KafkatorioPacket2> {
|
||||
// override fun serializer() = Serializer<KafkatorioPacket> { _, message ->
|
||||
// jsonMapper.encodeToString<KafkatorioPacket>(message).encodeToByteArray()
|
||||
// }
|
||||
//
|
||||
// override fun deserializer() = Deserializer<KafkatorioPacket> { _, bytes ->
|
||||
// jsonMapper.decodeFromString(bytes.decodeToString())
|
||||
// }
|
||||
//}
|
||||
|
||||
val jsonMapper: Json = Json {
|
||||
prettyPrint = true
|
||||
|
|
|
|||
|
|
@ -1,13 +1,10 @@
|
|||
package dev.adamko.kafkatorio.processor.topology
|
||||
|
||||
import dev.adamko.kafkatorio.events.schema.FactorioConfigurationUpdate
|
||||
import dev.adamko.kafkatorio.events.schema.FactorioEvent
|
||||
import dev.adamko.kafkatorio.events.schema.FactorioEventUpdatePacket
|
||||
import dev.adamko.kafkatorio.events.schema.FactorioPrototypes
|
||||
import dev.adamko.kafkatorio.events.schema.KafkatorioPacket
|
||||
import dev.adamko.kafkatorio.processor.KafkatorioTopology
|
||||
import dev.adamko.kafkatorio.processor.admin.KafkatorioKafkaAdmin
|
||||
|
||||
import dev.adamko.kafkatorio.processor.admin.TOPIC_SRC_SERVER_LOG
|
||||
import dev.adamko.kafkatorio.processor.admin.topicName
|
||||
import dev.adamko.kafkatorio.processor.serdes.jsonMapper
|
||||
import dev.adamko.kafkatorio.schema2.KafkatorioPacket2
|
||||
import dev.adamko.kotka.extensions.consumedAs
|
||||
import dev.adamko.kotka.extensions.producedAs
|
||||
import dev.adamko.kotka.extensions.streams.map
|
||||
|
|
@ -22,25 +19,24 @@ import org.apache.kafka.streams.processor.RecordContext
|
|||
|
||||
fun factorioServerPacketStream(
|
||||
builder: StreamsBuilder = StreamsBuilder(),
|
||||
): KStream<FactorioServerId, KafkatorioPacket> {
|
||||
): KStream<FactorioServerId, KafkatorioPacket2> {
|
||||
|
||||
return builder.stream(
|
||||
KafkatorioTopology.TOPIC_SRC_SERVER_LOG,
|
||||
TOPIC_SRC_SERVER_LOG,
|
||||
consumedAs(
|
||||
"read-raw-packets-from-server",
|
||||
Serdes.String(),
|
||||
Serdes.String(),
|
||||
)
|
||||
)
|
||||
.map("decode-packets") { serverId: String, value: String ->
|
||||
val packet = jsonMapper.decodeFromString<KafkatorioPacket>(value)
|
||||
FactorioServerId(serverId) to packet
|
||||
}
|
||||
).map("decode-packets") { serverId: String, value: String ->
|
||||
val packet = jsonMapper.decodeFromString<KafkatorioPacket2>(value)
|
||||
FactorioServerId(serverId) to packet
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
fun splitFactorioServerPacketStream(
|
||||
factorioServerPacketStream: KStream<FactorioServerId, KafkatorioPacket>
|
||||
factorioServerPacketStream: KStream<FactorioServerId, KafkatorioPacket2>
|
||||
) {
|
||||
factorioServerPacketStream
|
||||
.to(
|
||||
|
|
@ -49,18 +45,8 @@ fun splitFactorioServerPacketStream(
|
|||
jsonMapper.serde(),
|
||||
jsonMapper.serde(),
|
||||
)
|
||||
) { _: FactorioServerId, value: KafkatorioPacket, _: RecordContext ->
|
||||
) { _: FactorioServerId, value: KafkatorioPacket2, _: RecordContext ->
|
||||
// println("[$key] sending event:${value.eventType} to topic:${value.data.objectName()}")
|
||||
|
||||
val base = KafkatorioKafkaAdmin.topicNameBase(value.packetType)
|
||||
|
||||
val type: String = when (value) {
|
||||
is FactorioEvent -> value.data.objectName.name
|
||||
is FactorioConfigurationUpdate -> "FactorioConfigurationUpdate"
|
||||
is FactorioPrototypes -> "all"
|
||||
is FactorioEventUpdatePacket -> value.update.updateType.name
|
||||
}
|
||||
|
||||
"$base.$type"
|
||||
value.data.topicName
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -4,12 +4,8 @@ import com.sksamuel.scrimage.ImmutableImage
|
|||
import com.sksamuel.scrimage.ScaleMethod
|
||||
import com.sksamuel.scrimage.color.RGBColor
|
||||
import com.sksamuel.scrimage.nio.PngWriter
|
||||
import dev.adamko.kafkatorio.events.schema.ColourHex
|
||||
import dev.adamko.kafkatorio.events.schema.MapChunkPosition
|
||||
import dev.adamko.kafkatorio.events.schema.MapTilePosition
|
||||
import dev.adamko.kafkatorio.events.schema.SurfaceIndex
|
||||
import dev.adamko.kafkatorio.events.schema.converters.toMapChunkPosition
|
||||
import dev.adamko.kafkatorio.processor.KafkatorioTopology
|
||||
import dev.adamko.kafkatorio.schema.common.*
|
||||
import dev.adamko.kafkatorio.processor.admin.TOPIC_GROUPED_MAP_CHUNKS_STATE
|
||||
import dev.adamko.kafkatorio.processor.serdes.kxsBinary
|
||||
import dev.adamko.kotka.extensions.consumedAs
|
||||
import dev.adamko.kotka.extensions.streams.forEach
|
||||
|
|
@ -17,7 +13,9 @@ import dev.adamko.kotka.kxs.serde
|
|||
import java.awt.image.BufferedImage
|
||||
import java.io.File
|
||||
import kotlin.coroutines.CoroutineContext
|
||||
import kotlin.io.path.fileSize
|
||||
import kotlin.math.abs
|
||||
import kotlin.time.Duration.Companion.seconds
|
||||
import kotlinx.coroutines.CoroutineName
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
|
|
@ -25,17 +23,18 @@ import kotlinx.coroutines.SupervisorJob
|
|||
import kotlinx.coroutines.async
|
||||
import kotlinx.coroutines.awaitAll
|
||||
import kotlinx.coroutines.channels.BufferOverflow
|
||||
import kotlinx.coroutines.flow.Flow
|
||||
import kotlinx.coroutines.flow.MutableSharedFlow
|
||||
import kotlinx.coroutines.flow.asFlow
|
||||
import kotlinx.coroutines.flow.debounce
|
||||
import kotlinx.coroutines.flow.distinctUntilChangedBy
|
||||
import kotlinx.coroutines.flow.emitAll
|
||||
import kotlinx.coroutines.flow.flatMapConcat
|
||||
import kotlinx.coroutines.flow.launchIn
|
||||
import kotlinx.coroutines.flow.map
|
||||
import kotlinx.coroutines.flow.onEach
|
||||
import kotlinx.coroutines.flow.runningFold
|
||||
import kotlinx.coroutines.job
|
||||
import kotlinx.coroutines.launch
|
||||
import kotlinx.coroutines.runBlocking
|
||||
import kotlinx.coroutines.supervisorScope
|
||||
import org.apache.kafka.streams.StreamsBuilder
|
||||
import org.apache.kafka.streams.Topology
|
||||
|
|
@ -44,10 +43,6 @@ import org.apache.kafka.streams.kstream.KStream
|
|||
|
||||
private const val WEB_MAP_TILE_SIZE_PX = 256
|
||||
|
||||
private val cScope = CoroutineScope(
|
||||
Dispatchers.Default + SupervisorJob() + CoroutineName("f-server-map-updates")
|
||||
)
|
||||
|
||||
|
||||
/** A bridge between Kafka and Kotlin */
|
||||
private val serverMapChunkHandler = ServerMapChunkHandler()
|
||||
|
|
@ -59,17 +54,15 @@ fun saveMapTiles(
|
|||
|
||||
val groupedMapChunkTiles: KStream<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> =
|
||||
builder.stream(
|
||||
KafkatorioTopology.TOPIC_GROUPED_MAP_CHUNKS,
|
||||
TOPIC_GROUPED_MAP_CHUNKS_STATE,
|
||||
consumedAs("consume.grouped-map-chunks", kxsBinary.serde(), kxsBinary.serde())
|
||||
)
|
||||
|
||||
groupedMapChunkTiles
|
||||
.forEach("save-chunked-tiles") { _, value ->
|
||||
if (value != null) {
|
||||
runBlocking(Dispatchers.Default) {
|
||||
println("emitting chunkedTiles ${value.chunkId} tiles: ${value.map.size}")
|
||||
serverMapChunkHandler.emit(value)
|
||||
}
|
||||
println("emitting chunkedTiles ${value.chunkId} tiles: ${value.map.size}")
|
||||
serverMapChunkHandler.emit(value)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -107,9 +100,8 @@ private fun ColourHex.toRgbColor(): RGBColor {
|
|||
|
||||
private class ServerMapChunkHandler : CoroutineScope {
|
||||
|
||||
override val coroutineContext: CoroutineContext = Dispatchers.Default +
|
||||
SupervisorJob(cScope.coroutineContext.job) +
|
||||
CoroutineName("ServerMapChunkHandler")
|
||||
override val coroutineContext: CoroutineContext =
|
||||
Dispatchers.Default + SupervisorJob() + CoroutineName("ServerMapChunkHandler")
|
||||
|
||||
|
||||
/** The inbox for this handler. */
|
||||
|
|
@ -119,10 +111,8 @@ private class ServerMapChunkHandler : CoroutineScope {
|
|||
onBufferOverflow = BufferOverflow.SUSPEND,
|
||||
)
|
||||
|
||||
fun emit(mapChunkTiles: ServerMapChunkTiles<ColourHex>) {
|
||||
launch {
|
||||
allChunksFlow.emit(mapChunkTiles)
|
||||
}
|
||||
fun emit(mapChunkTiles: ServerMapChunkTiles<ColourHex>) = launch {
|
||||
allChunksFlow.emit(mapChunkTiles)
|
||||
}
|
||||
|
||||
/** This flow saves images one-by-one */
|
||||
|
|
@ -135,17 +125,18 @@ private class ServerMapChunkHandler : CoroutineScope {
|
|||
init {
|
||||
|
||||
allChunksFlow
|
||||
.runningFold(mapOf<ServerMapChunkId, MutableSharedFlow<Cmd.ChunkSubdivide>>()) { acc, src ->
|
||||
.runningFold(mutableMapOf<ServerMapChunkId, MutableSharedFlow<Cmd.ChunkSubdivide>>()) { acc, src ->
|
||||
|
||||
val flow = acc.getOrElse(src.chunkId) {
|
||||
val flow = acc.getOrPut(src.chunkId) {
|
||||
println("creating new subdivide-flow for chunkId ${src.chunkId}")
|
||||
createSubdivisionFlow()
|
||||
}
|
||||
|
||||
flow.emit(Cmd.ChunkSubdivide(src.chunkId, src.map))
|
||||
|
||||
acc + (src.chunkId to flow)
|
||||
acc
|
||||
}
|
||||
.distinctUntilChangedBy { it.size }
|
||||
.onEach {
|
||||
println("allChunksFlow has ${it.size} subdivide-flows: ${it.keys.joinToString()}")
|
||||
}
|
||||
|
|
@ -160,9 +151,15 @@ private class ServerMapChunkHandler : CoroutineScope {
|
|||
println("created new map tile parentFile directory ${chunkImageFile.absolutePath}")
|
||||
}
|
||||
|
||||
val sizeBefore = chunkImageFile.takeIf { it.exists() }?.toPath()?.fileSize()
|
||||
|
||||
val savedTile = img.output(PngWriter.NoCompression, chunkImageFile)
|
||||
if (filename.value.contains("z-1")) {
|
||||
println("savedTile: $savedTile")
|
||||
|
||||
val sizeAfter = savedTile.takeIf { it.exists() }?.toPath()?.fileSize()
|
||||
if (sizeBefore != sizeAfter) {
|
||||
// println("savedTile $sizeBefore/$sizeAfter: $savedTile")
|
||||
} else {
|
||||
println("savedTile NO-SIZE-CHANGE $sizeBefore/$sizeAfter: $savedTile")
|
||||
}
|
||||
}
|
||||
.launchIn(this)
|
||||
|
|
@ -183,43 +180,46 @@ private class ServerMapChunkHandler : CoroutineScope {
|
|||
onBufferOverflow = BufferOverflow.SUSPEND,
|
||||
)
|
||||
|
||||
val saveImgCommands = subdivisionFlow
|
||||
.flatMapConcat { subdivideCmd ->
|
||||
ChunkSize
|
||||
.values
|
||||
.flatMap { chunkSize ->
|
||||
subdivideCmd.tiles
|
||||
.entries
|
||||
.groupBy(
|
||||
{ (tile, _) -> tile.toMapChunkPosition(chunkSize.tilesPerChunk) }
|
||||
) {
|
||||
it.key to it.value
|
||||
}
|
||||
.map { (chunkPos, tiles) ->
|
||||
async {
|
||||
Cmd.CreateImage(
|
||||
surfaceIndex = subdivideCmd.chunkId.surfaceIndex,
|
||||
chunkSize = chunkSize,
|
||||
chunkPosition = chunkPos,
|
||||
tiles = tiles.toMap()
|
||||
)
|
||||
val saveImgCommands: Flow<Cmd.SaveImage> =
|
||||
subdivisionFlow
|
||||
.debounce(30.seconds)
|
||||
.flatMapConcat { subdivideCmd ->
|
||||
ChunkSize
|
||||
.entries
|
||||
.flatMap { chunkSize ->
|
||||
subdivideCmd
|
||||
.tiles
|
||||
.entries
|
||||
.groupBy(
|
||||
{ (tile, _) -> tile.toMapChunkPosition(chunkSize.tilesPerChunk) }
|
||||
) { (tilePosition, colour) ->
|
||||
tilePosition to colour
|
||||
}
|
||||
}
|
||||
}
|
||||
.awaitAll()
|
||||
.asFlow()
|
||||
}
|
||||
.map { createCmd ->
|
||||
val filename = TilePngFilename(createCmd)
|
||||
.map { (chunkPos, tiles) ->
|
||||
async {
|
||||
Cmd.CreateImage(
|
||||
surfaceIndex = subdivideCmd.chunkId.surfaceIndex,
|
||||
chunkSize = chunkSize,
|
||||
chunkPosition = chunkPos,
|
||||
tiles = tiles.toMap()
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
.awaitAll()
|
||||
.asFlow()
|
||||
}
|
||||
.map { createCmd ->
|
||||
val filename = TilePngFilename(createCmd)
|
||||
|
||||
val img = createMapTileImage(
|
||||
createCmd.chunkPosition,
|
||||
createCmd.tiles,
|
||||
createCmd.chunkSize,
|
||||
)
|
||||
val img = createMapTileImage(
|
||||
createCmd.chunkPosition,
|
||||
createCmd.tiles,
|
||||
createCmd.chunkSize,
|
||||
)
|
||||
|
||||
Cmd.SaveImage(filename, img)
|
||||
}
|
||||
Cmd.SaveImage(filename, img)
|
||||
}
|
||||
|
||||
launch {
|
||||
supervisorScope {
|
||||
|
|
|
|||
|
|
@ -1,52 +1,31 @@
|
|||
package dev.adamko.kafkatorio.processor.topology
|
||||
|
||||
import dev.adamko.kafkatorio.events.schema.ColourHex
|
||||
import dev.adamko.kafkatorio.events.schema.FactorioEventUpdate
|
||||
import dev.adamko.kafkatorio.events.schema.FactorioEventUpdatePacket
|
||||
import dev.adamko.kafkatorio.events.schema.FactorioPrototypes
|
||||
import dev.adamko.kafkatorio.events.schema.KafkatorioPacket
|
||||
import dev.adamko.kafkatorio.events.schema.MapChunkPosition
|
||||
import dev.adamko.kafkatorio.events.schema.MapChunkUpdate
|
||||
import dev.adamko.kafkatorio.events.schema.MapTile
|
||||
import dev.adamko.kafkatorio.events.schema.MapTileDictionary.Companion.isNullOrEmpty
|
||||
import dev.adamko.kafkatorio.events.schema.MapTilePosition
|
||||
import dev.adamko.kafkatorio.events.schema.MapTiles
|
||||
import dev.adamko.kafkatorio.events.schema.SurfaceIndex
|
||||
import dev.adamko.kafkatorio.events.schema.converters.toMapChunkPosition
|
||||
import dev.adamko.kafkatorio.processor.KafkatorioTopology
|
||||
import dev.adamko.kafkatorio.processor.serdes.jsonMapper
|
||||
import dev.adamko.kafkatorio.processor.admin.TOPIC_GROUPED_MAP_CHUNKS_STATE
|
||||
import dev.adamko.kafkatorio.processor.serdes.kxsBinary
|
||||
import dev.adamko.kotka.extensions.consumedAs
|
||||
import dev.adamko.kafkatorio.schema.common.ColourHex
|
||||
import dev.adamko.kafkatorio.schema.common.MapChunkPosition
|
||||
import dev.adamko.kafkatorio.schema.common.MapTile
|
||||
import dev.adamko.kafkatorio.schema.common.MapTileDictionary.Companion.isNullOrEmpty
|
||||
import dev.adamko.kafkatorio.schema.common.MapTilePosition
|
||||
import dev.adamko.kafkatorio.schema.common.MapTiles
|
||||
import dev.adamko.kafkatorio.schema.common.SurfaceIndex
|
||||
import dev.adamko.kafkatorio.schema.common.toMapChunkPosition
|
||||
import dev.adamko.kafkatorio.schema2.MapChunkUpdate
|
||||
import dev.adamko.kafkatorio.schema2.PrototypesUpdate
|
||||
import dev.adamko.kotka.extensions.groupedAs
|
||||
import dev.adamko.kotka.extensions.materializedWith
|
||||
import dev.adamko.kotka.extensions.producedAs
|
||||
import dev.adamko.kotka.extensions.repartitionedAs
|
||||
import dev.adamko.kotka.extensions.streams.filter
|
||||
import dev.adamko.kotka.extensions.streams.flatMap
|
||||
import dev.adamko.kotka.extensions.streams.mapValues
|
||||
import dev.adamko.kotka.extensions.streams.peek
|
||||
import dev.adamko.kotka.extensions.streams.reduce
|
||||
import dev.adamko.kotka.extensions.streams.toTable
|
||||
import dev.adamko.kotka.extensions.tableJoined
|
||||
import dev.adamko.kotka.extensions.tables.filter
|
||||
import dev.adamko.kotka.extensions.tables.join
|
||||
import dev.adamko.kotka.extensions.tables.mapValues
|
||||
import dev.adamko.kotka.extensions.tables.toStream
|
||||
import dev.adamko.kotka.kxs.serde
|
||||
import kotlin.collections.List
|
||||
import kotlin.collections.Map
|
||||
import kotlin.collections.associate
|
||||
import kotlin.collections.component1
|
||||
import kotlin.collections.component2
|
||||
import kotlin.collections.getOrElse
|
||||
import kotlin.collections.groupBy
|
||||
import kotlin.collections.isNotEmpty
|
||||
import kotlin.collections.isNullOrEmpty
|
||||
import kotlin.collections.joinToString
|
||||
import kotlin.collections.map
|
||||
import kotlin.collections.mapValues
|
||||
import kotlin.collections.mutableSetOf
|
||||
import kotlin.collections.plus
|
||||
import kotlinx.serialization.Serializable
|
||||
import org.apache.kafka.streams.StreamsBuilder
|
||||
import org.apache.kafka.streams.Topology
|
||||
|
|
@ -72,33 +51,39 @@ data class ServerMapChunkTiles<Data>(
|
|||
}
|
||||
|
||||
|
||||
fun groupMapChunks(
|
||||
builder: StreamsBuilder,
|
||||
prototypesTopicName: String = "kafkatorio.${KafkatorioPacket.PacketType.PROTOTYPES}.all",
|
||||
mapChunkUpdatesTopicName: String = "kafkatorio.${KafkatorioPacket.PacketType.UPDATE}.${FactorioEventUpdate.FactorioEventUpdateType.MAP_CHUNK}",
|
||||
): Topology {
|
||||
fun groupMapChunks(builder: StreamsBuilder): Topology {
|
||||
|
||||
val mapChunksStream: KStream<FactorioServerId, MapChunkUpdate> = builder.streamPacketData()
|
||||
val protosStream: KStream<FactorioServerId, PrototypesUpdate> = builder.streamPacketData()
|
||||
|
||||
// val protosStream: KStream<FactorioServerId, KafkatorioPacket2> =
|
||||
// builder.stream<FactorioServerId?, KafkatorioPacket2?>(
|
||||
// prototypesTopicName,
|
||||
// consumedAs("consume.factorio-protos.all", jsonMapper.serde(), jsonMapper.serde())
|
||||
// ).filter { key: FactorioServerId?, value: KafkatorioPacket2? ->
|
||||
// key != null
|
||||
// && value != null
|
||||
// && value.data ==
|
||||
// }
|
||||
// .map
|
||||
|
||||
|
||||
val protosStream: KStream<FactorioServerId, FactorioPrototypes> =
|
||||
builder.stream(
|
||||
prototypesTopicName,
|
||||
consumedAs("consume.factorio-protos.all", jsonMapper.serde(), jsonMapper.serde())
|
||||
)
|
||||
val tileProtoColourDict: KTable<FactorioServerId, TileColourDict> =
|
||||
tileProtoColourDictionary(protosStream)
|
||||
|
||||
val mapChunksUpdatesStream: KStream<FactorioServerId, FactorioEventUpdatePacket> =
|
||||
builder.stream(
|
||||
mapChunkUpdatesTopicName,
|
||||
consumedAs(
|
||||
"consume.map-chunk-updates",
|
||||
jsonMapper.serde<FactorioServerId>(),
|
||||
jsonMapper.serde<FactorioEventUpdatePacket>(),
|
||||
)
|
||||
)
|
||||
// val mapChunksUpdatesStream: KStream<FactorioServerId, FactorioEventUpdatePacket> =
|
||||
// builder.stream(
|
||||
// mapChunkUpdatesTopicName,
|
||||
// consumedAs(
|
||||
// "consume.map-chunk-updates",
|
||||
// jsonMapper.serde<FactorioServerId>(),
|
||||
// jsonMapper.serde<FactorioEventUpdatePacket>(),
|
||||
// )
|
||||
// )
|
||||
|
||||
val groupedMapChunkTiles: KTable<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> =
|
||||
groupTilesIntoChunksWithColours(
|
||||
mapChunksUpdatesStream,
|
||||
mapChunksStream,
|
||||
tileProtoColourDict,
|
||||
)
|
||||
|
||||
|
|
@ -108,13 +93,13 @@ fun groupMapChunks(
|
|||
!v?.map.isNullOrEmpty()
|
||||
}
|
||||
.mapValues("stream-grouped-map-tiles.map-not-null") { _, v ->
|
||||
v as ServerMapChunkTiles<ColourHex>
|
||||
requireNotNull(v)
|
||||
}
|
||||
.peek("stream-grouped-map-tiles.peek") { _, v ->
|
||||
println("Grouping map tiles result: ${v.chunkId} / ${v.map.size}")
|
||||
}
|
||||
.to(
|
||||
KafkatorioTopology.TOPIC_GROUPED_MAP_CHUNKS,
|
||||
TOPIC_GROUPED_MAP_CHUNKS_STATE,
|
||||
producedAs(
|
||||
"produce.grouped-map-chunks",
|
||||
kxsBinary.serde<ServerMapChunkId>(),
|
||||
|
|
@ -127,37 +112,24 @@ fun groupMapChunks(
|
|||
|
||||
|
||||
private fun groupTilesIntoChunksWithColours(
|
||||
mapChunkUpdatePacketsStream: KStream<FactorioServerId, FactorioEventUpdatePacket>,
|
||||
mapChunksStream: KStream<FactorioServerId, MapChunkUpdate>,
|
||||
tileProtoColourDict: KTable<FactorioServerId, TileColourDict>,
|
||||
): KTable<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> {
|
||||
|
||||
val mapChunkUpdatesStream: KStream<FactorioServerId, MapTiles> =
|
||||
mapChunkUpdatePacketsStream
|
||||
.filter { _, packet: FactorioEventUpdatePacket? ->
|
||||
when (val data = packet?.update) {
|
||||
is MapChunkUpdate -> !data.tileDictionary.isNullOrEmpty()
|
||||
else -> false
|
||||
}
|
||||
}
|
||||
.mapValues("map-chunk-update-packets.convert-to-map-tiles") { _, packet: FactorioEventUpdatePacket ->
|
||||
|
||||
val mapChunkUpdate: MapChunkUpdate = requireNotNull(packet.update as? MapChunkUpdate) {
|
||||
"invalid UpdatePacket type $packet"
|
||||
}
|
||||
|
||||
val tiles: List<MapTile> = requireNotNull(mapChunkUpdate.tileDictionary?.toMapTileList()) {
|
||||
"invalid MapChunkUpdate packet, tiles must not be null"
|
||||
}
|
||||
require(tiles.isNotEmpty()) {
|
||||
"invalid MapChunkUpdate packet, tiles must not be empty"
|
||||
}
|
||||
|
||||
MapTiles(mapChunkUpdate.surfaceIndex, tiles)
|
||||
}
|
||||
val mapTilesStream: KStream<FactorioServerId, MapTiles> =
|
||||
mapChunksStream.filter { _, value ->
|
||||
!value.tileDictionary.isNullOrEmpty()
|
||||
}.mapValues("map-chunk-update-packets.convert-to-map-tiles") { _, packet: MapChunkUpdate ->
|
||||
val mapTiles = packet.tileDictionary?.toMapTileList() ?: emptyList()
|
||||
MapTiles(packet.key.surfaceIndex, mapTiles)
|
||||
}.filter { _, value ->
|
||||
value.tiles.isNotEmpty()
|
||||
}
|
||||
|
||||
|
||||
val chunkedTilesTable: KTable<ServerMapChunkId, ServerMapChunkTiles<TileProtoHashCode>> =
|
||||
mapChunkUpdatesStream
|
||||
mapTilesStream
|
||||
// group tiles by server & surface & chunk
|
||||
.flatMap("server-map-data.tiles.flatMapByChunk") { serverId: FactorioServerId, mapTiles: MapTiles ->
|
||||
|
||||
|
|
@ -208,65 +180,65 @@ private fun groupTilesIntoChunksWithColours(
|
|||
kxsBinary.serde<ServerMapChunkTiles<TileProtoHashCode>>(),
|
||||
)
|
||||
) { chunkTiles, other -> chunkTiles + other }
|
||||
.toStream("server-map-data.tiles.repartition.to-stream")
|
||||
.repartition(
|
||||
repartitionedAs(
|
||||
"server-map-data.tiles.repartition.config",
|
||||
kxsBinary.serde<ServerMapChunkId>(),
|
||||
kxsBinary.serde<ServerMapChunkTiles<TileProtoHashCode>?>(),
|
||||
3,
|
||||
)
|
||||
).toTable(
|
||||
"server-map-data.tiles.repartition.to-table",
|
||||
materializedWith(
|
||||
kxsBinary.serde<ServerMapChunkId>(),
|
||||
kxsBinary.serde<ServerMapChunkTiles<TileProtoHashCode>?>(),
|
||||
)
|
||||
)
|
||||
.filter("server-map-data.tiles.repartition.filter-not-null") { _: ServerMapChunkId, v: ServerMapChunkTiles<TileProtoHashCode>? ->
|
||||
v != null
|
||||
}
|
||||
.mapValues(
|
||||
"server-map-data.tiles.repartition.map-not-null",
|
||||
materializedWith(
|
||||
kxsBinary.serde<ServerMapChunkId>(),
|
||||
kxsBinary.serde<ServerMapChunkTiles<TileProtoHashCode>>(),
|
||||
)
|
||||
) { _: ServerMapChunkId, v: ServerMapChunkTiles<TileProtoHashCode>? ->
|
||||
v as ServerMapChunkTiles<TileProtoHashCode>
|
||||
}
|
||||
// .toStream("server-map-data.tiles.repartition.to-stream")
|
||||
// .repartition(
|
||||
// repartitionedAs(
|
||||
// "server-map-data.tiles.repartition.config",
|
||||
// kxsBinary.serde<ServerMapChunkId>(),
|
||||
// kxsBinary.serde<ServerMapChunkTiles<TileProtoHashCode>?>(),
|
||||
// 3,
|
||||
// )
|
||||
// ).toTable(
|
||||
// "server-map-data.tiles.repartition.to-table",
|
||||
// materializedWith(
|
||||
// kxsBinary.serde<ServerMapChunkId>(),
|
||||
// kxsBinary.serde<ServerMapChunkTiles<TileProtoHashCode>?>(),
|
||||
// )
|
||||
// )
|
||||
// .filter("server-map-data.tiles.repartition.filter-not-null") { _: ServerMapChunkId, v: ServerMapChunkTiles<TileProtoHashCode>? ->
|
||||
// v != null
|
||||
// }
|
||||
// .mapValues(
|
||||
// "server-map-data.tiles.repartition.map-not-null",
|
||||
// materializedWith(
|
||||
// kxsBinary.serde<ServerMapChunkId>(),
|
||||
// kxsBinary.serde<ServerMapChunkTiles<TileProtoHashCode>>(),
|
||||
// )
|
||||
// ) { _: ServerMapChunkId, v: ServerMapChunkTiles<TileProtoHashCode>? ->
|
||||
// v as ServerMapChunkTiles<TileProtoHashCode>
|
||||
// }
|
||||
|
||||
|
||||
// repartition to try and fix KTable-KTable joining skipping records
|
||||
val tileProtoColourDictRepartitioned =
|
||||
tileProtoColourDict
|
||||
.toStream("tileProtoColourDict.repartition.to-stream")
|
||||
.repartition(
|
||||
repartitionedAs(
|
||||
"tileProtoColourDict.repartition.config",
|
||||
kxsBinary.serde<FactorioServerId>(),
|
||||
kxsBinary.serde<TileColourDict?>(),
|
||||
3,
|
||||
)
|
||||
).toTable(
|
||||
"tileProtoColourDict.repartition.to-table",
|
||||
materializedWith(
|
||||
kxsBinary.serde<FactorioServerId>(),
|
||||
kxsBinary.serde<TileColourDict?>(),
|
||||
)
|
||||
)
|
||||
.filter("tileProtoColourDict.repartition.filter-not-null") { _: FactorioServerId, v: TileColourDict? ->
|
||||
v != null
|
||||
}
|
||||
.mapValues(
|
||||
"tileProtoColourDict.repartition.map-not-null",
|
||||
materializedWith(
|
||||
kxsBinary.serde<FactorioServerId>(),
|
||||
kxsBinary.serde<TileColourDict>(),
|
||||
)
|
||||
) { _: FactorioServerId, v: TileColourDict? ->
|
||||
v as TileColourDict
|
||||
}
|
||||
// .toStream("tileProtoColourDict.repartition.to-stream")
|
||||
// .repartition(
|
||||
// repartitionedAs(
|
||||
// "tileProtoColourDict.repartition.config",
|
||||
// kxsBinary.serde<FactorioServerId>(),
|
||||
// kxsBinary.serde<TileColourDict?>(),
|
||||
// 3,
|
||||
// )
|
||||
// ).toTable(
|
||||
// "tileProtoColourDict.repartition.to-table",
|
||||
// materializedWith(
|
||||
// kxsBinary.serde<FactorioServerId>(),
|
||||
// kxsBinary.serde<TileColourDict?>(),
|
||||
// )
|
||||
// )
|
||||
// .filter("tileProtoColourDict.repartition.filter-not-null") { _: FactorioServerId, v: TileColourDict? ->
|
||||
// v != null
|
||||
// }
|
||||
// .mapValues(
|
||||
// "tileProtoColourDict.repartition.map-not-null",
|
||||
// materializedWith(
|
||||
// kxsBinary.serde<FactorioServerId>(),
|
||||
// kxsBinary.serde<TileColourDict>(),
|
||||
// )
|
||||
// ) { _: FactorioServerId, v: TileColourDict? ->
|
||||
// v as TileColourDict
|
||||
// }
|
||||
|
||||
|
||||
val colourisedChunkTable: KTable<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> =
|
||||
|
|
|
|||
|
|
@ -36,11 +36,11 @@ enum class ChunkSize(
|
|||
|
||||
companion object {
|
||||
// cache values for better performance. KT-48872
|
||||
val values: List<ChunkSize> = values().asList()
|
||||
val entries: Set<ChunkSize> = values().toSet()
|
||||
|
||||
val MAX: ChunkSize = values.maxByOrNull { it.tilesPerChunk }!!
|
||||
val MIN: ChunkSize = values.minByOrNull { it.tilesPerChunk }!!
|
||||
val STANDARD: ChunkSize = values.first { it.zoomLevel == 0 }
|
||||
val MAX: ChunkSize = entries.maxByOrNull { it.tilesPerChunk }!!
|
||||
val MIN: ChunkSize = entries.minByOrNull { it.tilesPerChunk }!!
|
||||
val STANDARD: ChunkSize = entries.first { it.zoomLevel == 0 }
|
||||
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,27 @@
|
|||
package dev.adamko.kafkatorio.processor.topology
|
||||
|
||||
import dev.adamko.kafkatorio.processor.admin.topicName
|
||||
import dev.adamko.kafkatorio.processor.serdes.jsonMapper
|
||||
import dev.adamko.kafkatorio.schema2.KafkatorioPacket2
|
||||
import dev.adamko.kafkatorio.schema2.KafkatorioPacketData2
|
||||
import dev.adamko.kotka.extensions.consumedAs
|
||||
import dev.adamko.kotka.extensions.streams.mapValues
|
||||
import dev.adamko.kotka.kxs.serde
|
||||
import org.apache.kafka.streams.StreamsBuilder
|
||||
import org.apache.kafka.streams.kstream.KStream
|
||||
|
||||
|
||||
inline fun <reified T : KafkatorioPacketData2> StreamsBuilder.streamPacketData(): KStream<FactorioServerId, T> {
|
||||
|
||||
val simpleName = T::class.simpleName!!
|
||||
val topicName = T::class.topicName
|
||||
|
||||
return stream<FactorioServerId, KafkatorioPacket2>(
|
||||
topicName,
|
||||
consumedAs("consume.${simpleName}", jsonMapper.serde(), jsonMapper.serde())
|
||||
).filter { key: FactorioServerId?, value: KafkatorioPacket2? ->
|
||||
key != null && value?.data is T
|
||||
}.mapValues("consume.map-values.${simpleName}") { _: FactorioServerId, packet: KafkatorioPacket2 ->
|
||||
packet.data as T
|
||||
}
|
||||
}
|
||||
|
|
@ -1,9 +1,10 @@
|
|||
package dev.adamko.kafkatorio.processor.topology
|
||||
|
||||
import dev.adamko.kafkatorio.events.schema.FactorioEventUpdate
|
||||
import dev.adamko.kafkatorio.events.schema.KafkatorioPacket
|
||||
import dev.adamko.kafkatorio.processor.WebsocketServer
|
||||
import dev.adamko.kafkatorio.processor.admin.topicName
|
||||
import dev.adamko.kafkatorio.processor.serdes.jsonMapper
|
||||
import dev.adamko.kafkatorio.schema2.KafkatorioPacket2
|
||||
import dev.adamko.kafkatorio.schema2.PlayerUpdate
|
||||
import dev.adamko.kotka.extensions.consumedAs
|
||||
import dev.adamko.kotka.kxs.serde
|
||||
import kotlinx.serialization.encodeToString
|
||||
|
|
@ -14,12 +15,11 @@ fun playerUpdatesToWsServer(
|
|||
websocketServer: WebsocketServer,
|
||||
builder: StreamsBuilder = StreamsBuilder(),
|
||||
) {
|
||||
builder.stream<FactorioServerId, KafkatorioPacket>(
|
||||
"kafkatorio.${KafkatorioPacket.PacketType.UPDATE}.${FactorioEventUpdate.FactorioEventUpdateType.PLAYER}",
|
||||
builder.stream<FactorioServerId, KafkatorioPacket2>(
|
||||
PlayerUpdate::class.topicName,
|
||||
consumedAs("EventUpdate.Player-updates-for-ws-server", jsonMapper.serde(), jsonMapper.serde()),
|
||||
)
|
||||
.foreach { _, value ->
|
||||
).foreach { _, value ->
|
||||
// println("sending ${value.packetType} packet to websocket")
|
||||
websocketServer.sendMessage(jsonMapper.encodeToString(value))
|
||||
}
|
||||
websocketServer.sendMessage(jsonMapper.encodeToString(value))
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,15 +1,13 @@
|
|||
package dev.adamko.kafkatorio.processor.topology
|
||||
|
||||
import dev.adamko.kafkatorio.events.schema.ColourHex
|
||||
import dev.adamko.kafkatorio.events.schema.FactorioPrototypes
|
||||
import dev.adamko.kafkatorio.events.schema.KafkatorioPacket
|
||||
import dev.adamko.kafkatorio.events.schema.MapTile
|
||||
import dev.adamko.kafkatorio.events.schema.MapTilePrototype
|
||||
import dev.adamko.kafkatorio.events.schema.converters.toHex
|
||||
import dev.adamko.kafkatorio.processor.serdes.jsonMapper
|
||||
import dev.adamko.kafkatorio.processor.serdes.kxsBinary
|
||||
import dev.adamko.kafkatorio.schema.common.ColourHex
|
||||
import dev.adamko.kafkatorio.schema.common.MapTile
|
||||
import dev.adamko.kafkatorio.schema.common.toHex
|
||||
import dev.adamko.kafkatorio.schema.prototypes.FactorioPrototype2
|
||||
import dev.adamko.kafkatorio.schema.prototypes.MapTilePrototype
|
||||
import dev.adamko.kafkatorio.schema2.PrototypesUpdate
|
||||
import dev.adamko.kotka.extensions.materializedAs
|
||||
import dev.adamko.kotka.extensions.materializedWith
|
||||
import dev.adamko.kotka.extensions.streams.filter
|
||||
import dev.adamko.kotka.extensions.streams.mapValues
|
||||
import dev.adamko.kotka.extensions.streams.toTable
|
||||
|
|
@ -25,6 +23,7 @@ import org.apache.kafka.streams.kstream.KTable
|
|||
value class TileProtoHashCode private constructor(val code: Int) {
|
||||
constructor(prototype: MapTilePrototype) : this(prototype.name.hashCode())
|
||||
constructor(tile: MapTile) : this(tile.proto.hashCode())
|
||||
constructor(tile: FactorioPrototype2.MapTile) : this(tile.name.hashCode())
|
||||
}
|
||||
|
||||
|
||||
|
|
@ -35,32 +34,53 @@ value class TileColourDict(val map: Map<TileProtoHashCode, ColourHex>)
|
|||
|
||||
/** Get the latest map-tile colours per server. */
|
||||
fun tileProtoColourDictionary(
|
||||
factorioServerPacketStream: KStream<FactorioServerId, FactorioPrototypes>
|
||||
factorioServerPacketStream: KStream<FactorioServerId, PrototypesUpdate>
|
||||
): KTable<FactorioServerId, TileColourDict> {
|
||||
|
||||
// get all the prototypes
|
||||
return factorioServerPacketStream
|
||||
.mapValues("server-map-data.tile-prototypes.mapValues") { _, packet: KafkatorioPacket ->
|
||||
val map = when (packet) {
|
||||
is FactorioPrototypes -> packet
|
||||
.prototypes
|
||||
.filterIsInstance<MapTilePrototype>()
|
||||
.associate { TileProtoHashCode(it) to it.mapColour.toHex() }
|
||||
else -> mapOf()
|
||||
}
|
||||
TileColourDict(map)
|
||||
}.filter("server-map-data.tile-prototypes.filterMapTileProtos") { _, dict: TileColourDict ->
|
||||
dict.map.isNotEmpty()
|
||||
}.peek { serverId, dict: TileColourDict ->
|
||||
println("server $serverId has TileColourDict[${dict.map.size}]: ${dict.map.entries.joinToString()}")
|
||||
}
|
||||
.toTable(
|
||||
"server-map-data.tile-prototypes",
|
||||
materializedAs(
|
||||
"server-map-data.tile-prototypes.store",
|
||||
kxsBinary.serde(),
|
||||
kxsBinary.serde(),
|
||||
)
|
||||
)
|
||||
return factorioServerPacketStream.mapValues("server-map-data.tile-prototypes.map-values") { _, protoPacket: PrototypesUpdate ->
|
||||
val map = protoPacket
|
||||
.prototypes
|
||||
.filterIsInstance<FactorioPrototype2.MapTile>()
|
||||
.associate { TileProtoHashCode(it) to it.mapColour.toHex() }
|
||||
|
||||
TileColourDict(map)
|
||||
}.filter("server-map-data.tile-prototypes.filterMapTileProtos") { _, dict: TileColourDict ->
|
||||
dict.map.isNotEmpty()
|
||||
}.peek { serverId, dict: TileColourDict ->
|
||||
println("server $serverId has TileColourDict[${dict.map.size}]: ${dict.map.entries.joinToString()}")
|
||||
}.toTable(
|
||||
"server-map-data.tile-prototypes",
|
||||
materializedAs(
|
||||
"server-map-data.tile-prototypes.store",
|
||||
kxsBinary.serde(),
|
||||
kxsBinary.serde(),
|
||||
)
|
||||
)
|
||||
}
|
||||
|
||||
// get all the prototypes
|
||||
// return factorioServerPacketStream
|
||||
// .mapValues("server-map-data.tile-prototypes.mapValues") { _, packet: KafkatorioPacket ->
|
||||
// val map = when (packet) {
|
||||
// is FactorioPrototype -> packet
|
||||
// .prototypes
|
||||
// .filterIsInstance<MapTilePrototype>()
|
||||
// .associate { TileProtoHashCode(it) to it.mapColour.toHex() }
|
||||
// else -> mapOf()
|
||||
// }
|
||||
// TileColourDict(map)
|
||||
// }.filter("server-map-data.tile-prototypes.filterMapTileProtos") { _, dict: TileColourDict ->
|
||||
// dict.map.isNotEmpty()
|
||||
// }.peek { serverId, dict: TileColourDict ->
|
||||
// println("server $serverId has TileColourDict[${dict.map.size}]: ${dict.map.entries.joinToString()}")
|
||||
// }
|
||||
// .toTable(
|
||||
// "server-map-data.tile-prototypes",
|
||||
// materializedAs(
|
||||
// "server-map-data.tile-prototypes.store",
|
||||
// kxsBinary.serde(),
|
||||
// kxsBinary.serde(),
|
||||
// )
|
||||
// )
|
||||
//
|
||||
//}
|
||||
|
|
|
|||
|
|
@ -1,24 +1,59 @@
|
|||
package dev.adamko.kafkatorio.processor.misc
|
||||
|
||||
import dev.adamko.kafkatorio.events.schema.MapChunkPosition
|
||||
import dev.adamko.kafkatorio.events.schema.SurfaceIndex
|
||||
import dev.adamko.kafkatorio.processor.serdes.kxsBinary
|
||||
import dev.adamko.kafkatorio.processor.topology.ChunkSize
|
||||
import dev.adamko.kafkatorio.processor.topology.FactorioServerId
|
||||
import dev.adamko.kafkatorio.processor.topology.ServerMapChunkId
|
||||
import dev.adamko.kafkatorio.processor.topology.ServerMapChunkTiles
|
||||
import dev.adamko.kafkatorio.schema.common.ColourHex
|
||||
import dev.adamko.kafkatorio.schema.common.MapChunkPosition
|
||||
import dev.adamko.kafkatorio.schema.common.MapTilePosition
|
||||
import dev.adamko.kafkatorio.schema.common.SurfaceIndex
|
||||
import dev.adamko.kafkatorio.schema.jsonMapperKafkatorio
|
||||
import io.kotest.assertions.withClue
|
||||
import io.kotest.core.spec.style.FunSpec
|
||||
import io.kotest.matchers.collections.shouldContainExactly
|
||||
import io.kotest.matchers.comparables.shouldBeEqualComparingTo
|
||||
import io.kotest.matchers.equality.shouldBeEqualToComparingFields
|
||||
import io.kotest.matchers.ints.shouldBeExactly
|
||||
import io.kotest.matchers.longs.shouldBeExactly
|
||||
import io.kotest.matchers.shouldBe
|
||||
import io.kotest.property.Arb
|
||||
import io.kotest.property.arbitrary.boolean
|
||||
import io.kotest.property.arbitrary.booleanArray
|
||||
import io.kotest.property.arbitrary.byte
|
||||
import io.kotest.property.arbitrary.byteArray
|
||||
import io.kotest.property.arbitrary.char
|
||||
import io.kotest.property.arbitrary.charArray
|
||||
import io.kotest.property.arbitrary.double
|
||||
import io.kotest.property.arbitrary.doubleArray
|
||||
import io.kotest.property.arbitrary.float
|
||||
import io.kotest.property.arbitrary.floatArray
|
||||
import io.kotest.property.arbitrary.int
|
||||
import io.kotest.property.arbitrary.intArray
|
||||
import io.kotest.property.arbitrary.long
|
||||
import io.kotest.property.arbitrary.longArray
|
||||
import io.kotest.property.arbitrary.positiveInt
|
||||
import io.kotest.property.arbitrary.short
|
||||
import io.kotest.property.arbitrary.shortArray
|
||||
import io.kotest.property.arbitrary.string
|
||||
import io.kotest.property.arbitrary.uByte
|
||||
import io.kotest.property.arbitrary.uInt
|
||||
import io.kotest.property.arbitrary.uLong
|
||||
import io.kotest.property.arbitrary.uShort
|
||||
import io.kotest.property.checkAll
|
||||
import kotlinx.serialization.builtins.serializer
|
||||
import kotlinx.serialization.decodeFromByteArray
|
||||
import kotlinx.serialization.decodeFromString
|
||||
import kotlinx.serialization.encodeToByteArray
|
||||
import kotlinx.serialization.encodeToString
|
||||
|
||||
|
||||
class KxsBinaryTests : FunSpec({
|
||||
|
||||
test("byte array") {
|
||||
val ba = byteArrayOf(
|
||||
0, 4, 116, 101, 115, 116, 0, 0, 0, 1, 0, 0, 0, 2, 0, 0, 0, 1, 0, 0, 0, 3
|
||||
4, 116, 101, 115, 116, 2, 0, 0, 0, 1, 0, 0, 0, 2, 0, 0, 0, 1, 0, 0, 0, 3
|
||||
)
|
||||
|
||||
val result = kxsBinary.decodeFromByteArray(ServerMapChunkId.serializer(), ba)
|
||||
|
|
@ -47,7 +82,7 @@ class KxsBinaryTests : FunSpec({
|
|||
|
||||
withClue(ba.toString(Charsets.UTF_8)) {
|
||||
ba shouldBe byteArrayOf(
|
||||
0, 4, 116, 101, 115, 116, 0, 0, 0, 1, 0, 0, 0, 2, 0, 0, 0, 1, 0, 0, 0, 3
|
||||
4, 116, 101, 115, 116, 2, 0, 0, 0, 1, 0, 0, 0, 2, 0, 0, 0, 1, 0, 0, 0, 3
|
||||
)
|
||||
}
|
||||
}
|
||||
|
|
@ -55,7 +90,10 @@ class KxsBinaryTests : FunSpec({
|
|||
context("inline value class") {
|
||||
val uuid = "fa858d73-6205-44cf-aeeb-6bbf3148c4a7"
|
||||
//@formatter:off
|
||||
val expectedBytes = byteArrayOf(0, 36, 102, 97, 56, 53, 56, 100, 55, 51, 45, 54, 50, 48, 53, 45, 52, 52, 99, 102, 45, 97, 101, 101, 98, 45, 54, 98, 98, 102, 51, 49, 52, 56, 99, 52, 97, 55)
|
||||
val expectedBytes = byteArrayOf(
|
||||
36, 102, 97, 56, 53, 56, 100, 55, 51, 45, 54, 50, 48, 53, 45, 52, 52, 99,
|
||||
102, 45, 97, 101, 101, 98, 45, 54, 98, 98, 102, 51, 49, 52, 56, 99, 52, 97, 55,
|
||||
)
|
||||
//@formatter:on
|
||||
|
||||
test("encode") {
|
||||
|
|
@ -73,4 +111,208 @@ class KxsBinaryTests : FunSpec({
|
|||
}
|
||||
}
|
||||
|
||||
context("ServerMapChunkTiles") {
|
||||
val initial = ServerMapChunkTiles(
|
||||
ServerMapChunkId(
|
||||
FactorioServerId("server-id"),
|
||||
MapChunkPosition(33, 44),
|
||||
SurfaceIndex(4u),
|
||||
ChunkSize.CHUNK_032,
|
||||
),
|
||||
mapOf(
|
||||
MapTilePosition(1, 2) to ColourHex(1u, 2u, 3u, 4u),
|
||||
)
|
||||
)
|
||||
|
||||
test("!json round trip") {
|
||||
val jsonEncoded = jsonMapperKafkatorio.encodeToString(initial)
|
||||
val jsonDecoded =
|
||||
jsonMapperKafkatorio.decodeFromString<ServerMapChunkTiles<ColourHex>>(jsonEncoded)
|
||||
|
||||
jsonDecoded shouldBeEqualToComparingFields initial
|
||||
}
|
||||
|
||||
test("binary round trip") {
|
||||
val binaryEncoded = kxsBinary.encodeToByteArray(initial)
|
||||
val binaryDecoded = kxsBinary.decodeFromByteArray<ServerMapChunkTiles<ColourHex>>(binaryEncoded)
|
||||
|
||||
binaryDecoded shouldBeEqualToComparingFields initial
|
||||
}
|
||||
}
|
||||
|
||||
context("primitives round trip") {
|
||||
test("double") {
|
||||
checkAll(Arb.double()) { input ->
|
||||
val encoded: ByteArray = kxsBinary.encodeToByteArray(input)
|
||||
val decoded: Double = kxsBinary.decodeFromByteArray(encoded)
|
||||
decoded shouldBeEqualComparingTo input
|
||||
}
|
||||
}
|
||||
|
||||
test("float") {
|
||||
checkAll(Arb.float()) { input ->
|
||||
val encoded: ByteArray = kxsBinary.encodeToByteArray(input)
|
||||
val decoded: Float = kxsBinary.decodeFromByteArray(encoded)
|
||||
decoded shouldBeEqualComparingTo input
|
||||
}
|
||||
}
|
||||
|
||||
test("int") {
|
||||
checkAll(Arb.int()) { input ->
|
||||
val encoded: ByteArray = kxsBinary.encodeToByteArray(input)
|
||||
val decoded: Int = kxsBinary.decodeFromByteArray(encoded)
|
||||
decoded shouldBeExactly input
|
||||
}
|
||||
}
|
||||
|
||||
test("long") {
|
||||
checkAll(Arb.long()) { input ->
|
||||
val encoded: ByteArray = kxsBinary.encodeToByteArray(input)
|
||||
val decoded: Long = kxsBinary.decodeFromByteArray(encoded)
|
||||
decoded shouldBeExactly input
|
||||
}
|
||||
}
|
||||
|
||||
test("short") {
|
||||
checkAll(Arb.short()) { input ->
|
||||
val encoded: ByteArray = kxsBinary.encodeToByteArray(input)
|
||||
val decoded: Short = kxsBinary.decodeFromByteArray(encoded)
|
||||
decoded shouldBeEqualComparingTo input
|
||||
}
|
||||
}
|
||||
|
||||
test("byte") {
|
||||
checkAll(Arb.byte()) { input ->
|
||||
val encoded: ByteArray = kxsBinary.encodeToByteArray(input)
|
||||
val decoded: Byte = kxsBinary.decodeFromByteArray(encoded)
|
||||
decoded shouldBeEqualComparingTo input
|
||||
}
|
||||
}
|
||||
|
||||
test("boolean") {
|
||||
checkAll(Arb.boolean()) { input ->
|
||||
val encoded: ByteArray = kxsBinary.encodeToByteArray(input)
|
||||
val decoded: Boolean = kxsBinary.decodeFromByteArray(encoded)
|
||||
decoded shouldBeEqualComparingTo input
|
||||
}
|
||||
}
|
||||
|
||||
test("char") {
|
||||
checkAll(Arb.char()) { input ->
|
||||
val encoded: ByteArray = kxsBinary.encodeToByteArray(input)
|
||||
val decoded: Char = kxsBinary.decodeFromByteArray(encoded)
|
||||
decoded shouldBeEqualComparingTo input
|
||||
}
|
||||
}
|
||||
|
||||
test("string") {
|
||||
checkAll(Arb.string()) { input ->
|
||||
val encoded: ByteArray = kxsBinary.encodeToByteArray(input)
|
||||
val decoded: String = kxsBinary.decodeFromByteArray(encoded)
|
||||
decoded shouldBeEqualComparingTo input
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@OptIn(ExperimentalUnsignedTypes::class)
|
||||
context("unsigned numbers round trip") {
|
||||
test("uint") {
|
||||
checkAll(Arb.uInt()) { input ->
|
||||
val encoded: ByteArray = kxsBinary.encodeToByteArray(UInt.serializer(), input)
|
||||
val decoded: UInt = kxsBinary.decodeFromByteArray(UInt.serializer(), encoded)
|
||||
decoded shouldBeEqualComparingTo input
|
||||
}
|
||||
}
|
||||
|
||||
test("ulong") {
|
||||
checkAll(Arb.uLong()) { input ->
|
||||
val encoded: ByteArray = kxsBinary.encodeToByteArray(ULong.serializer(), input)
|
||||
val decoded: ULong = kxsBinary.decodeFromByteArray(ULong.serializer(), encoded)
|
||||
decoded shouldBeEqualComparingTo input
|
||||
}
|
||||
}
|
||||
|
||||
test("ushort") {
|
||||
checkAll(Arb.uShort()) { input ->
|
||||
val encoded: ByteArray = kxsBinary.encodeToByteArray(UShort.serializer(), input)
|
||||
val decoded: UShort = kxsBinary.decodeFromByteArray(UShort.serializer(), encoded)
|
||||
decoded shouldBeEqualComparingTo input
|
||||
}
|
||||
}
|
||||
|
||||
test("ubyte") {
|
||||
checkAll(Arb.uByte()) { input ->
|
||||
val encoded: ByteArray = kxsBinary.encodeToByteArray(UByte.serializer(), input)
|
||||
val decoded: UByte = kxsBinary.decodeFromByteArray(UByte.serializer(), encoded)
|
||||
decoded shouldBeEqualComparingTo input
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
context("primitive arrays round trip") {
|
||||
test("double array") {
|
||||
checkAll(Arb.doubleArray(Arb.positiveInt(50), Arb.double())) { input ->
|
||||
val encoded: ByteArray = kxsBinary.encodeToByteArray(input)
|
||||
val decoded: DoubleArray = kxsBinary.decodeFromByteArray(encoded)
|
||||
decoded.asList() shouldContainExactly input.asList()
|
||||
}
|
||||
}
|
||||
|
||||
test("float array") {
|
||||
checkAll(Arb.floatArray(Arb.positiveInt(50), Arb.float())) { input ->
|
||||
val encoded: ByteArray = kxsBinary.encodeToByteArray(input)
|
||||
val decoded: FloatArray = kxsBinary.decodeFromByteArray(encoded)
|
||||
decoded.asList() shouldContainExactly input.asList()
|
||||
}
|
||||
}
|
||||
|
||||
test("int array") {
|
||||
checkAll(Arb.intArray(Arb.positiveInt(50), Arb.int())) { input ->
|
||||
val encoded: ByteArray = kxsBinary.encodeToByteArray(input)
|
||||
val decoded: IntArray = kxsBinary.decodeFromByteArray(encoded)
|
||||
decoded.asList() shouldContainExactly input.asList()
|
||||
}
|
||||
}
|
||||
|
||||
test("long array") {
|
||||
checkAll(Arb.longArray(Arb.positiveInt(50), Arb.long())) { input ->
|
||||
val encoded: ByteArray = kxsBinary.encodeToByteArray(input)
|
||||
val decoded: LongArray = kxsBinary.decodeFromByteArray(encoded)
|
||||
decoded.asList() shouldContainExactly input.asList()
|
||||
}
|
||||
}
|
||||
|
||||
test("short array") {
|
||||
checkAll(Arb.shortArray(Arb.positiveInt(50), Arb.short())) { input ->
|
||||
val encoded: ByteArray = kxsBinary.encodeToByteArray(input)
|
||||
val decoded: ShortArray = kxsBinary.decodeFromByteArray(encoded)
|
||||
decoded.asList() shouldContainExactly input.asList()
|
||||
}
|
||||
}
|
||||
|
||||
test("byte array") {
|
||||
checkAll(Arb.byteArray(Arb.positiveInt(50), Arb.byte())) { input ->
|
||||
val encoded: ByteArray = kxsBinary.encodeToByteArray(input)
|
||||
val decoded: ByteArray = kxsBinary.decodeFromByteArray(encoded)
|
||||
decoded.asList() shouldContainExactly input.asList()
|
||||
}
|
||||
}
|
||||
|
||||
test("boolean array") {
|
||||
checkAll(Arb.booleanArray(Arb.positiveInt(50), Arb.boolean())) { input ->
|
||||
val encoded: ByteArray = kxsBinary.encodeToByteArray(input)
|
||||
val decoded: BooleanArray = kxsBinary.decodeFromByteArray(encoded)
|
||||
decoded.asList() shouldContainExactly input.asList()
|
||||
}
|
||||
}
|
||||
|
||||
test("char array") {
|
||||
checkAll(Arb.charArray(Arb.positiveInt(50), Arb.char())) { input ->
|
||||
val encoded: ByteArray = kxsBinary.encodeToByteArray(input)
|
||||
val decoded: CharArray = kxsBinary.decodeFromByteArray(encoded)
|
||||
decoded.asList() shouldContainExactly input.asList()
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
|
|
|
|||
|
|
@ -1,14 +1,11 @@
|
|||
package dev.adamko.kafkatorio.processor.misc
|
||||
|
||||
import dev.adamko.kafkatorio.events.schema.ColourHex
|
||||
import dev.adamko.kafkatorio.events.schema.MapChunkPosition
|
||||
import dev.adamko.kafkatorio.events.schema.MapTilePosition
|
||||
import dev.adamko.kafkatorio.events.schema.SurfaceIndex
|
||||
import dev.adamko.kafkatorio.processor.serdes.kxsBinary
|
||||
import dev.adamko.kafkatorio.processor.topology.ChunkSize
|
||||
import dev.adamko.kafkatorio.processor.topology.FactorioServerId
|
||||
import dev.adamko.kafkatorio.processor.topology.ServerMapChunkId
|
||||
import dev.adamko.kafkatorio.processor.topology.ServerMapChunkTiles
|
||||
import dev.adamko.kafkatorio.schema.common.*
|
||||
import java.util.UUID
|
||||
import kotlinx.serialization.encodeToByteArray
|
||||
import org.junit.jupiter.api.Test
|
||||
|
|
|
|||
|
|
@ -1,21 +1,15 @@
|
|||
package dev.adamko.kafkatorio.processor.topology
|
||||
|
||||
import dev.adamko.kafkatorio.events.schema.Colour
|
||||
import dev.adamko.kafkatorio.events.schema.ColourHex
|
||||
import dev.adamko.kafkatorio.events.schema.FactorioEventUpdatePacket
|
||||
import dev.adamko.kafkatorio.events.schema.FactorioPrototypes
|
||||
import dev.adamko.kafkatorio.events.schema.MapChunkPosition
|
||||
import dev.adamko.kafkatorio.events.schema.MapChunkUpdate
|
||||
import dev.adamko.kafkatorio.events.schema.MapTileDictionary
|
||||
import dev.adamko.kafkatorio.events.schema.MapTilePosition
|
||||
import dev.adamko.kafkatorio.events.schema.MapTilePrototype
|
||||
import dev.adamko.kafkatorio.events.schema.PrototypeKey
|
||||
import dev.adamko.kafkatorio.events.schema.PrototypeName
|
||||
import dev.adamko.kafkatorio.events.schema.SurfaceIndex
|
||||
import dev.adamko.kafkatorio.events.schema.Tick
|
||||
import dev.adamko.kafkatorio.processor.KafkatorioTopology
|
||||
import dev.adamko.kafkatorio.processor.admin.TOPIC_GROUPED_MAP_CHUNKS_STATE
|
||||
import dev.adamko.kafkatorio.processor.serdes.jsonMapper
|
||||
import dev.adamko.kafkatorio.processor.serdes.kxsBinary
|
||||
import dev.adamko.kafkatorio.schema.common.*
|
||||
import dev.adamko.kafkatorio.schema.common.MapTileDictionary.PrototypeKey
|
||||
import dev.adamko.kafkatorio.schema.prototypes.FactorioPrototype2
|
||||
import dev.adamko.kafkatorio.schema2.MapChunkUpdate
|
||||
import dev.adamko.kafkatorio.schema2.MapChunkUpdateKey
|
||||
import dev.adamko.kafkatorio.schema2.PrototypesUpdate
|
||||
import dev.adamko.kotka.kxs.serde
|
||||
import io.kotest.assertions.withClue
|
||||
import io.kotest.core.spec.style.FunSpec
|
||||
|
|
@ -32,34 +26,29 @@ import org.apache.kafka.streams.TopologyTestDriver
|
|||
class GroupMapChunksTest : FunSpec({
|
||||
|
||||
test("basic test") {
|
||||
val streamsBuilder = StreamsBuilder()
|
||||
|
||||
val prototypesTopicName = "prototypes-topic"
|
||||
val mapChunkUpdatesTopicName = "map-chunk-updates-topic"
|
||||
|
||||
val topology = groupMapChunks(
|
||||
StreamsBuilder(),
|
||||
prototypesTopicName,
|
||||
mapChunkUpdatesTopicName,
|
||||
)
|
||||
val topology = groupMapChunks(streamsBuilder)
|
||||
|
||||
val testDriver = TopologyTestDriver(topology)
|
||||
|
||||
val prototypesTestInputTopic: TestInputTopic<FactorioServerId, FactorioPrototypes> =
|
||||
val prototypesTestInputTopic: TestInputTopic<FactorioServerId, PrototypesUpdate> =
|
||||
testDriver.createInputTopic(
|
||||
prototypesTopicName,
|
||||
jsonMapper.serde<FactorioServerId>().serializer(),
|
||||
jsonMapper.serde<FactorioPrototypes>().serializer(),
|
||||
jsonMapper.serde<PrototypesUpdate>().serializer(),
|
||||
)
|
||||
|
||||
val serverId = FactorioServerId("test-server-id")
|
||||
|
||||
prototypesTestInputTopic.pipeInput(
|
||||
serverId,
|
||||
FactorioPrototypes(
|
||||
modVersion = "1.2.3",
|
||||
tick = Tick(456u),
|
||||
PrototypesUpdate(
|
||||
listOf(
|
||||
MapTilePrototype(
|
||||
FactorioPrototype2.MapTile(
|
||||
PrototypeName("fake-proto"),
|
||||
layer = 1u,
|
||||
mapColour = Colour(0.3f, 0.2f, 0.1f, 1.0f),
|
||||
|
|
@ -71,26 +60,25 @@ class GroupMapChunksTest : FunSpec({
|
|||
)
|
||||
)
|
||||
|
||||
val mapChunkUpdatesInputTopic: TestInputTopic<FactorioServerId, FactorioEventUpdatePacket> =
|
||||
val mapChunkUpdatesInputTopic: TestInputTopic<FactorioServerId, MapChunkUpdate> =
|
||||
testDriver.createInputTopic(
|
||||
mapChunkUpdatesTopicName,
|
||||
jsonMapper.serde<FactorioServerId>().serializer(),
|
||||
jsonMapper.serde<FactorioEventUpdatePacket>().serializer(),
|
||||
jsonMapper.serde<MapChunkUpdate>().serializer(),
|
||||
)
|
||||
|
||||
mapChunkUpdatesInputTopic.pipeInput(
|
||||
serverId,
|
||||
FactorioEventUpdatePacket(
|
||||
modVersion = "1.2.3",
|
||||
tick = Tick(854u),
|
||||
update = MapChunkUpdate(
|
||||
MapChunkUpdate(
|
||||
key = MapChunkUpdateKey(
|
||||
MapChunkPosition(11, 22),
|
||||
SurfaceIndex(1u),
|
||||
tileDictionary = MapTileDictionary(
|
||||
tilesXY = mapOf(
|
||||
"1" to mapOf("2" to PrototypeKey("33"))
|
||||
),
|
||||
protos = mapOf(PrototypeKey("33") to PrototypeName("fake-proto"))
|
||||
)
|
||||
),
|
||||
tileDictionary = MapTileDictionary(
|
||||
tilesXY = mapOf(
|
||||
"1" to mapOf("2" to PrototypeKey(33))
|
||||
),
|
||||
protos = mapOf(PrototypeName("fake-proto") to PrototypeKey(33))
|
||||
)
|
||||
)
|
||||
)
|
||||
|
|
@ -98,7 +86,7 @@ class GroupMapChunksTest : FunSpec({
|
|||
val outputTopic: TestOutputTopic<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> =
|
||||
testDriver
|
||||
.createOutputTopic(
|
||||
KafkatorioTopology.TOPIC_GROUPED_MAP_CHUNKS,
|
||||
TOPIC_GROUPED_MAP_CHUNKS_STATE,
|
||||
kxsBinary.serde<ServerMapChunkId>().deserializer(),
|
||||
kxsBinary.serde<ServerMapChunkTiles<ColourHex>>().deserializer(),
|
||||
)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue