Use shared flow (buffer) in reader
This commit is contained in:
@@ -11,6 +11,7 @@ import io.ktor.utils.io.ByteReadChannel
|
|||||||
import io.ktor.utils.io.readLineStrict
|
import io.ktor.utils.io.readLineStrict
|
||||||
import kotlinx.coroutines.CoroutineScope
|
import kotlinx.coroutines.CoroutineScope
|
||||||
import kotlinx.coroutines.flow.MutableSharedFlow
|
import kotlinx.coroutines.flow.MutableSharedFlow
|
||||||
|
import kotlinx.coroutines.flow.asSharedFlow
|
||||||
import kotlinx.coroutines.flow.filter
|
import kotlinx.coroutines.flow.filter
|
||||||
import kotlinx.coroutines.flow.first
|
import kotlinx.coroutines.flow.first
|
||||||
import kotlinx.coroutines.flow.map
|
import kotlinx.coroutines.flow.map
|
||||||
@@ -21,8 +22,8 @@ import kotlin.reflect.cast
|
|||||||
|
|
||||||
class MessageReader(socket: Socket, scope: CoroutineScope) : MessageIO(socket, scope) {
|
class MessageReader(socket: Socket, scope: CoroutineScope) : MessageIO(socket, scope) {
|
||||||
private lateinit var receive: ByteReadChannel
|
private lateinit var receive: ByteReadChannel
|
||||||
private val listenedMessages = mutableListOf<ServerMessage>()
|
private val messageBuffer = MutableSharedFlow<ServerMessage>(16, 64)
|
||||||
val messageBuffer = MutableSharedFlow<ServerMessage>(16, 64)
|
val sharedMessageBuffer = messageBuffer.asSharedFlow()
|
||||||
val parser = MessageParser(incomingMessages)
|
val parser = MessageParser(incomingMessages)
|
||||||
private val logger = if (Helpers.enableLogging) KotlinLogging.logger {} else KotlinLogging.logger(NOPLogger.NOP_LOGGER)
|
private val logger = if (Helpers.enableLogging) KotlinLogging.logger {} else KotlinLogging.logger(NOPLogger.NOP_LOGGER)
|
||||||
|
|
||||||
@@ -32,16 +33,19 @@ class MessageReader(socket: Socket, scope: CoroutineScope) : MessageIO(socket, s
|
|||||||
parser.start()
|
parser.start()
|
||||||
}
|
}
|
||||||
scope.launch {
|
scope.launch {
|
||||||
for (msg in parser.incomingParsedMessages) messageBuffer.emit(msg)
|
for (msg in parser.incomingParsedMessages)
|
||||||
|
messageBuffer.emit(msg)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
override fun start() = scope.launch {
|
override fun start() {
|
||||||
receive = socket.openReadChannel()
|
scope.launch {
|
||||||
while (true) {
|
receive = socket.openReadChannel()
|
||||||
val line = receive.readLineStrict() ?: break // >:( no LineEnding option for just CRLF? charlatans...
|
while (true) {
|
||||||
incomingMessages.send(line)
|
val line = receive.readLineStrict() ?: break // >:( no LineEnding option for just CRLF? charlatans...
|
||||||
logger.debug { "Receive: $line" }
|
logger.debug { "Receive: $line" }
|
||||||
|
incomingMessages.send(line)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user