packet size improvements

- change structure resource packets
- make more booleans default to false
- use encode_string - massive decrease in packet size!!
This commit is contained in:
Adam 2022-07-17 23:54:53 +02:00
parent 523f6d5914
commit 085776d42b
14 changed files with 131 additions and 134 deletions

View file

@ -167,6 +167,7 @@ internal class KafkatorioTopology(
}
/** Receive log messages from the syslog socket, and send them to a Kafka topic */
private fun syslogProducer(
id: String = "syslog-producer"
) {
@ -187,10 +188,14 @@ internal class KafkatorioTopology(
}
syslogServer.messages
.filter { it.message?.startsWith("KafkatorioPacket:") == true }
.filter {
(it.message ?: "").run {
startsWith("KafkatorioPacket:::") || startsWith("KafkatorioPacket encoded:::")
}
}
.onEach { syslogMsg ->
val kafkatorioMessage = syslogMsg.message?.substringAfter("KafkatorioPacket:")
val kafkatorioMessage = syslogMsg.message?.substringAfter(":::")
val record = ProducerRecord<String, String>(
TOPIC_SRC_SERVER_LOG,

View file

@ -14,7 +14,13 @@ import dev.adamko.kotka.extensions.producedAs
import dev.adamko.kotka.extensions.streams.map
import dev.adamko.kotka.extensions.streams.to
import dev.adamko.kotka.kxs.serde
import java.util.zip.Inflater
import kotlinx.serialization.decodeFromString
import okio.Buffer
import okio.ByteString
import okio.ByteString.Companion.decodeBase64
import okio.buffer
import okio.inflate
import org.apache.kafka.common.serialization.Serdes
import org.apache.kafka.streams.StreamsBuilder
import org.apache.kafka.streams.kstream.KStream
@ -33,7 +39,8 @@ fun factorioServerPacketStream(
)
).map("decode-packets") { serverId: String, value: String ->
val packet = runCatching {
jsonMapper.decodeFromString<KafkatorioPacket>(value)
val decoded = decodeFactorioEncodedString(value)
jsonMapper.decodeFromString<KafkatorioPacket>(decoded)
}.getOrElse { e ->
KafkatorioPacket(
modVersion = "unknown",
@ -69,3 +76,18 @@ fun splitFactorioServerPacketStream(
value.data.topicName
}
}
/**
* Factorio `game.encode_string` zlib encodes a string, then base64's it. Here we do the reverse.
*/
private fun decodeFactorioEncodedString(
source: String
): String {
val bytes: ByteString = source.decodeBase64() ?: error("failed to base64 decode $source")
return Buffer().write(bytes)
.inflate(Inflater())
.buffer()
.readString(Charsets.UTF_8)
}