diff --git a/modules/events-server-syslog/src/main/kotlin/dev/adamko/kafkatorio/processor/syslog/KafkatorioPacketKafkaProducer.kt b/modules/events-server-syslog/src/main/kotlin/dev/adamko/kafkatorio/processor/syslog/KafkatorioPacketKafkaProducer.kt index d60a436..f69eb9d 100644 --- a/modules/events-server-syslog/src/main/kotlin/dev/adamko/kafkatorio/processor/syslog/KafkatorioPacketKafkaProducer.kt +++ b/modules/events-server-syslog/src/main/kotlin/dev/adamko/kafkatorio/processor/syslog/KafkatorioPacketKafkaProducer.kt @@ -2,11 +2,11 @@ package dev.adamko.kafkatorio.processor.syslog import com.github.palindromicity.syslog.dsl.SyslogFieldKeys import dev.adamko.kafkatorio.processor.config.ApplicationProperties +import dev.adamko.kafkatorio.processor.config.TOPIC_SRC_SERVER_LOG +import dev.adamko.kafkatorio.processor.config.TOPIC_SRC_SERVER_LOG_DLQ import dev.adamko.kafkatorio.processor.core.Authenticator import dev.adamko.kafkatorio.processor.syslog.config.jsonMapper import dev.adamko.kafkatorio.schema.common.FactorioServerId -import dev.adamko.kafkatorio.processor.config.TOPIC_SRC_SERVER_LOG -import dev.adamko.kafkatorio.processor.config.TOPIC_SRC_SERVER_LOG_DLQ import dev.adamko.kotka.kxs.kafkaSerializer import java.util.Properties import kotlin.coroutines.coroutineContext @@ -56,6 +56,8 @@ class KafkatorioPacketKafkaProducer( suspend fun launch() { coroutineContext.job.invokeOnCompletion { + val cause = if (it != null) "(cause: ${it::class.java.name} ${it.message})" else "" + log("Closing $cause") this@KafkatorioPacketKafkaProducer.close() } diff --git a/modules/events-server-syslog/src/main/kotlin/dev/adamko/kafkatorio/processor/syslog/SyslogSocketServer.kt b/modules/events-server-syslog/src/main/kotlin/dev/adamko/kafkatorio/processor/syslog/SyslogSocketServer.kt index cd9780f..49a9fdc 100644 --- a/modules/events-server-syslog/src/main/kotlin/dev/adamko/kafkatorio/processor/syslog/SyslogSocketServer.kt +++ b/modules/events-server-syslog/src/main/kotlin/dev/adamko/kafkatorio/processor/syslog/SyslogSocketServer.kt @@ -7,6 +7,7 @@ import com.github.palindromicity.syslog.NilPolicy import com.github.palindromicity.syslog.SyslogParser import com.github.palindromicity.syslog.SyslogParserBuilder import com.github.palindromicity.syslog.SyslogSpecification +import com.github.palindromicity.syslog.dsl.ParseException import dev.adamko.kafkatorio.processor.config.ApplicationProperties import io.ktor.network.selector.SelectorManager import io.ktor.network.sockets.Socket @@ -83,16 +84,23 @@ class SyslogSocketServer( "received 'null' line from ${socket.description()}" } - val syslog = SyslogMsg(syslogParser.parseLine(line)) - _messages.emit(syslog) + try { + val syslog = SyslogMsg(syslogParser.parseLine(line)) + _messages.emit(syslog) + } catch (ex: ParseException) { + log("ignoring non-syslog message: $line") + } yield() } } catch (e: Throwable) { - if (e is CancellationException) throw e - log("closing socket ${socket.description()}, cause: ${e::class.java.name} ${e.message}") - e.printStackTrace() withContext(Dispatchers.IO) { socket.close() } + if (e is CancellationException) { + throw e + } else { + log("closing socket ${socket.description()}, cause: ${e::class.java.name} ${e.message}") + e.printStackTrace() + } } }