more fixing of Kafkatorio packets...

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

@@ -26,8 +26,7 @@ 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:"
+private const val ENCODED_TAG = "encoded:"
 
 
 fun factorioServerPacketStream(
@@ -43,7 +42,7 @@ fun factorioServerPacketStream(
     )
   ).mapValues("decode-packets") { _: FactorioServerId, value: String? ->
 
-    val packet = decodeKafkatorioPacket(value)
+    val packet = decodeKafkatorioPacket(value?.trim())
 
     if (packet.data is KafkatorioPacketDataError) {
       println("error parsing $TOPIC_SRC_SERVER_LOG message: $packet")
@@ -73,20 +72,16 @@ fun splitFactorioServerPacketStream(
 
 private fun decodeKafkatorioPacket(value: String?): KafkatorioPacket = try {
   when {
-    value.isNullOrBlank()                      ->
+    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)
+    else                  -> {
+      val packetJson = if (value.startsWith(ENCODED_TAG)) {
+        val encodedValue = value.substringAfter(ENCODED_TAG)
         decodeFactorioEncodedString(encodedValue)
       } else {
-        value.substringAfter(PACKET_TAG)
+        value
       }
-
       jsonMapper.decodeFromString(KafkatorioPacket.serializer(), packetJson)
     }
   }

modules/events-server-syslog/src/main/kotlin/dev/adamko/kafkatorio/processor/syslog/KafkatorioPacketKafkaProducer.kt

@@ -101,9 +101,26 @@ class KafkatorioPacketKafkaProducer(
     serverId: FactorioServerId,
     syslog: SyslogMsg,
   ) {
-    val kafkatorioMessage = syslog.message?.substringAfter(":::")
+    val syslogMessage = syslog.message?.trim()
 
-    val record = ProducerRecord<FactorioServerId, String>(
+    if (syslogMessage.isNullOrBlank()) {
+      // ignore blank messages
+      return
+    }
+
+    if (!syslogMessage.startsWith(KAFKATORIO_MESSAGE_TAG)) {
+      // ignore non-Kafkatorio message
+      return
+    }
+
+    val kafkatorioMessage = syslogMessage.substringAfter(KAFKATORIO_MESSAGE_TAG)
+
+    if (kafkatorioMessage.isBlank()) {
+      produceDlqMessage(syslog, "error - Kafkatorio message was blank $syslogMessage")
+      return
+    }
+
+    val record = ProducerRecord(
       TOPIC_SRC_SERVER_LOG,
       serverId,
       kafkatorioMessage,
@@ -136,6 +153,9 @@ class KafkatorioPacketKafkaProducer(
 
 
   companion object {
+    private const val KAFKATORIO_MESSAGE_TAG = "KafkatorioPacket:::"
+
+
     private fun log(msg: String) = println("[KafkatorioPacketKafkaProducer] $msg")