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

refactor syslog server to use Dispatchers.IO and SupervisorScope (try to fix connection refused error)

Browse source
This commit is contained in:
Adam 2022-10-16 12:34:07 +02:00
parent 71386ded3b
commit 0f9e51df31
1 changed files with 32 additions and 38 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -16,12 +16,12 @@ import io.ktor.utils.io.readUTF8Line
import java.util.EnumSet
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.SharedFlow
import kotlinx.coroutines.flow.asSharedFlow
import kotlinx.coroutines.isActive
import kotlinx.coroutines.launch
import kotlinx.coroutines.supervisorScope
import kotlinx.coroutines.withContext
import kotlinx.coroutines.yield
@ -56,53 +56,47 @@ class SyslogSocketServer(
get() = _messages.asSharedFlow()
suspend fun start(): Unit = coroutineScope {
suspend fun start(): Unit = withContext(Dispatchers.IO) {
log("started listening: ${serverSocketTcp.localAddress}")
while (isActive) {
val socket: Socket = serverSocketTcp.accept()
log("new socket ${socket.description()}")
launch {
val channel = socket.openReadChannel()
try {
while (socket.socketContext.isActive && !channel.isClosedForRead) {
val line = runCatching {
channel.readUTF8Line()
}.fold(
onSuccess = { line ->
if (line == null) {
log("received 'null' from ${socket.description()} - closing connection")
socket.close()
}
// log("received '$line' from ${socket.description()}")
line
},
onFailure = { ex ->
if (ex is CancellationException) throw ex
log("reading line from socket failed ${socket.description()}, cause: $ex")
null
},
)
if (line != null) {
val syslog = SyslogMsg(syslogParser.parseLine(line))
_messages.emit(syslog)
}
yield()
}
} catch (e: Exception) {
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() }
}
log("new socket ${socket.description()}")
handleConnection(socket)
}
yield()
}
}
private suspend fun handleConnection(
socket: Socket
) = supervisorScope {
try {
val channel = socket.openReadChannel()
while (socket.socketContext.isActive && !channel.isClosedForRead) {
val line = channel.readUTF8Line()
requireNotNull(line) {
"received 'null' line from ${socket.description()}"
}
val syslog = SyslogMsg(syslogParser.parseLine(line))
_messages.emit(syslog)
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() }
}
}
companion object {
private fun log(msg: String) = println("[SyslogSocketServer] $msg")