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

[syslog-server] minor styling

Browse source
This commit is contained in:
Adam 2022-10-18 21:26:30 +02:00
parent 557dc97635
commit ecf680d9ce
1 changed files with 9 additions and 10 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -73,19 +73,18 @@ class SyslogSocketServer(
private suspend fun handleConnection( private suspend fun handleConnection(
socket: Socket socket: Socket
) = supervisorScope { ): Unit = supervisorScope {
try { try {
val channel = socket.openReadChannel() val channel = socket.openReadChannel()
while (socket.socketContext.isActive && !channel.isClosedForRead) { while (socket.socketContext.isActive && !channel.isClosedForRead) {
val line = channel.readUTF8Line() val line = channel.readUTF8Line()
requireNotNull(line) { requireNotNull(line) { "received 'null' line from ${socket.description()}" }
"received 'null' line from ${socket.description()}"
}
try { try {
val syslog = SyslogMsg(syslogParser.parseLine(line)) val messageContents = syslogParser.parseLine(line)
val syslog = SyslogMsg(messageContents)
_messages.emit(syslog) _messages.emit(syslog)
} catch (ex: ParseException) { } catch (ex: ParseException) {
log("ignoring non-syslog message: $line") log("ignoring non-syslog message: $line")
@ -93,13 +92,13 @@ class SyslogSocketServer(
yield() yield()
} }
} catch (e: Throwable) { } catch (ex: Throwable) {
withContext(Dispatchers.IO) { socket.close() } withContext(Dispatchers.IO) { socket.close() }
if (e is CancellationException) { if (ex is CancellationException) {
throw e throw ex
} else { } else {
log("closing socket ${socket.description()}, cause: ${e::class.java.name} ${e.message}") log("closing socket ${socket.description()}, cause: ${ex::class.qualifiedName} ${ex.message}")
e.printStackTrace() ex.printStackTrace()
} }
} }
} }