refactor the KafkatorioPacket tag that the mod logs, and the syslog-server parses

modules/events-mod/src/generated/lua/main/emitting/PacketEmitter.lua

@@ -1,5 +1,6 @@
 local ____lualib = require("lualib_bundle")
 local __TS__Class = ____lualib.__TS__Class
+local __TS__StringTrim = ____lualib.__TS__StringTrim
 local __TS__New = ____lualib.__TS__New
 local ____exports = {}
 ____exports.KafkatorioPacketEmitter = __TS__Class()
@@ -15,9 +16,16 @@ function KafkatorioPacketEmitter.prototype.emitKeyedPacket(self, data)
 end
 function KafkatorioPacketEmitter.emitPacket(self, packet)
     local data = game.table_to_json(packet)
+    if #__TS__StringTrim(data) <= 0 then
+        print("[error] table_to_json returned empty string for packet:" .. tostring(packet))
+        return
+    end
     local encodedData = game.encode_string(data)
-    if encodedData ~= nil then
+    if encodedData == nil or #__TS__StringTrim(encodedData) <= 0 then
-        print("KafkatorioPacket encoded::: " .. encodedData)
+        print("[error] could not encode packet")
+        print("KafkatorioPacket:::" .. data)
+    else
+        print("KafkatorioPacket:::encoded:" .. encodedData)
     end
 end
 local PacketEmitter = __TS__New(____exports.KafkatorioPacketEmitter)

modules/events-mod/src/main/typescript/main/emitting/PacketEmitter.ts

@@ -27,11 +27,19 @@ export class KafkatorioPacketEmitter {
   /** Emit a serialised KafkatorioPacket */
   private static emitPacket<T extends KafkatorioPacket>(packet: T) {
     const data = game.table_to_json(packet)
-    // print(`KafkatorioPacket::: ${data}`)
+
+    if (data.trim().length <= 0) {
+      print(`[error] table_to_json returned empty string for packet:${packet}`)
+      return
+    }
+
     const encodedData = game.encode_string(data)
-    if (encodedData != null) {
+
-      print(`KafkatorioPacket encoded::: ${encodedData}`)
+    if (encodedData == null || encodedData.trim().length <= 0) {
-      // rcon.print(`KafkatorioPacket: ${data}`)
+      print(`[error] could not encode packet`)
+      print(`KafkatorioPacket:::${data}`)
+    } else {
+      print(`KafkatorioPacket:::encoded:${encodedData}`)
       // this.appendToFile(encodedData)
     }
   }

modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/factorioServerPackets.kt

@@ -1,12 +1,12 @@
 package dev.adamko.kafkatorio.server.processor.topology
 
+import dev.adamko.kafkatorio.processor.config.TOPIC_SRC_SERVER_LOG
+import dev.adamko.kafkatorio.processor.config.jsonMapper
+import dev.adamko.kafkatorio.processor.config.topicName
 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.KafkatorioPacketDataError
-import dev.adamko.kafkatorio.processor.config.TOPIC_SRC_SERVER_LOG
-import dev.adamko.kafkatorio.processor.config.jsonMapper
-import dev.adamko.kafkatorio.processor.config.topicName
 import dev.adamko.kotka.extensions.component1
 import dev.adamko.kotka.extensions.component2
 import dev.adamko.kotka.extensions.consumedAs
@@ -26,6 +26,10 @@ import org.apache.kafka.streams.StreamsBuilder
 import org.apache.kafka.streams.kstream.KStream
 
 
+private const val PACKET_TAG = "KafkatorioPacket:::"
+private const val PACKET_ENCODED_TAG = "${PACKET_TAG}encoded:"
+
+
 fun factorioServerPacketStream(
   builder: StreamsBuilder = StreamsBuilder(),
 ): KStream<FactorioServerId, KafkatorioPacket> {
@@ -39,21 +43,7 @@ fun factorioServerPacketStream(
     )
   ).mapValues("decode-packets") { _: FactorioServerId, value: String? ->
 
-    val packet = try {
+    val packet = decodeKafkatorioPacket(value)
-      val decoded = decodeFactorioEncodedString(value!!)
-      jsonMapper.decodeFromString(KafkatorioPacket.serializer(), decoded)
-    } catch (e: Exception) {
-      if (e is CancellationException) throw e
-
-      KafkatorioPacket(
-        modVersion = "unknown",
-        tick = Tick(0u),
-        data = KafkatorioPacketDataError(
-          exception = e,
-          rawValue = value,
-        )
-      )
-    }
 
     if (packet.data is KafkatorioPacketDataError) {
       println("error parsing $TOPIC_SRC_SERVER_LOG message: $packet")
@@ -81,6 +71,31 @@ fun splitFactorioServerPacketStream(
 }
 
 
+private fun decodeKafkatorioPacket(value: String?): KafkatorioPacket = try {
+  when {
+    value.isNullOrBlank()                      ->
+      createErrorPacket(message = "null/blank message", rawValue = value)
+
+    value.substringAfter(PACKET_TAG).isBlank() ->
+      createErrorPacket(message = "missing '$PACKET_TAG'", rawValue = value)
+
+    else                                       -> {
+      val packetJson = if (PACKET_ENCODED_TAG in value) {
+        val encodedValue = value.substringAfter(PACKET_ENCODED_TAG)
+        decodeFactorioEncodedString(encodedValue)
+      } else {
+        value.substringAfter(PACKET_TAG)
+      }
+
+      jsonMapper.decodeFromString(KafkatorioPacket.serializer(), packetJson)
+    }
+  }
+} catch (ex: Exception) {
+  if (ex is CancellationException) throw ex
+  errorPacket(exception = ex, rawValue = value)
+}
+
+
 /**
  * Factorio `game.encode_string` zlib encodes a string, then base64's it. Here we do the reverse.
  */
@@ -94,3 +109,31 @@ private fun decodeFactorioEncodedString(
     .buffer()
     .readString(Charsets.UTF_8)
 }
+
+
+private fun createErrorPacket(
+  message: String,
+  rawValue: String?,
+): KafkatorioPacket =
+  KafkatorioPacket(
+    modVersion = "unknown",
+    tick = Tick(0u),
+    data = KafkatorioPacketDataError(
+      message = message,
+      rawValue = rawValue,
+    )
+  )
+
+
+private fun errorPacket(
+  exception: Exception,
+  rawValue: String?,
+): KafkatorioPacket {
+  val message = """
+${exception::class.qualifiedName} ${exception.message}
+
+${exception.stackTraceToString()}
+""".trimIndent()
+
+  return createErrorPacket(message, rawValue)
+}

modules/infra-kafka-pipe/src/kafka-pipe.sh

@@ -17,7 +17,7 @@ KEY="${INSTANCE_ID:-none}"
 
 while :; do
   docker logs --tail 0 -f "$FACTORIO_SERVER_CONTAINER_NAME" |
-    sed -n -e 's/^KafkatorioPacket encoded::: //p' |
+    sed -n -e 's/^KafkatorioPacket::://p' |
     kcat -P \
       -b "$KAFKA_HOST" \
       -t kafkatorio.src.server-log \