lazzexus/Kafkatorio
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

[syslog-server] refactor logging and error handling (avoid runCatching) in producer

Browse source
This commit is contained in:
Adam 2022-10-18 21:25:18 +02:00
parent 8342a923a8
commit 557dc97635
1 changed files with 13 additions and 17 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -10,7 +10,6 @@ import dev.adamko.kafkatorio.schema.common.FactorioServerId
import dev.adamko.kotka.kxs.kafkaSerializer import dev.adamko.kotka.kxs.kafkaSerializer
import java.util.Properties import java.util.Properties
import kotlin.coroutines.coroutineContext import kotlin.coroutines.coroutineContext
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.flow.collect import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.filter import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.flow.onEach import kotlinx.coroutines.flow.onEach
@ -57,44 +56,41 @@ class KafkatorioPacketKafkaProducer(
suspend fun launch() { suspend fun launch() {
coroutineContext.job.invokeOnCompletion { coroutineContext.job.invokeOnCompletion {
val cause = if (it != null) "(cause: ${it::class.java.name} ${it.message})" else "" val cause = if (it != null) "(cause: ${it::class.java.name} ${it.message})" else ""
log("Closing $cause") log("Closing KafkatorioPacketKafkaProducer $cause")
this@KafkatorioPacketKafkaProducer.close() this@KafkatorioPacketKafkaProducer.close()
} }
syslogServer.messages syslogServer.messages
.filter { .filter {
(it.message ?: "").run { (it.message ?: "").run {
startsWith("KafkatorioPacket:::") || startsWith("KafkatorioPacket encoded:::") startsWith("KafkatorioPacket:::")
} }
}.onEach { syslog -> }.onEach { syslog ->
runCatching { val serverId = getFactorioServerId(syslog)
getFactorioServerId(syslog) if (serverId != null) {
}.fold( produceMessage(serverId, syslog)
onSuccess = { serverId -> produceMessage(serverId, syslog) },
onFailure = { ex ->
produceDlqMessage(syslog, "${ex::class.java.name}: ${ex.message}")
if (ex is CancellationException) throw ex
} }
)
}.collect() }.collect()
} }
private fun getFactorioServerId(syslog: SyslogMsg): FactorioServerId { private fun getFactorioServerId(syslog: SyslogMsg): FactorioServerId? {
val token = syslog.headerAppName val token = syslog.headerAppName
if (token.isNullOrBlank()) { if (token.isNullOrBlank()) {
error("no token in syslog message. msg: ${syslog.src}") produceDlqMessage(syslog, "no token in syslog message. msg: ${syslog.src}")
return null
} }
val serverToken = authenticator.verifyJwt(token) val serverToken = authenticator.verifyJwt(token)
if (serverToken.isNullOrBlank()) { if (serverToken.isNullOrBlank()) {
error("could not decode token $token. msg: ${syslog.src}") produceDlqMessage(syslog, "could not decode token $token. msg: ${syslog.src}")
return null
} }
val serverId = appProps.kafkatorioServers[serverToken] val serverId = appProps.kafkatorioServers[serverToken]
if (serverId.isNullOrBlank()) { if (serverId.isNullOrBlank()) {
error("no server ID for token $serverToken. msg: ${syslog.src}") produceDlqMessage(syslog, "no server ID for token $serverToken. msg: ${syslog.src}")
return null
} }
return serverId return serverId