move waitForNext to Server.kt

This commit is contained in:
2026-09-15 03:06:55 +01:00
parent 1599a20583
commit 95b27aa0af
2 changed files with 18 additions and 18 deletions
@@ -6,10 +6,15 @@ import io.github.oshai.kotlinlogging.KotlinLogging
import io.github.oshai.kotlinlogging.slf4j.logger import io.github.oshai.kotlinlogging.slf4j.logger
import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.flow.SharedFlow import kotlinx.coroutines.flow.SharedFlow
import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.flow.filterIsInstance import kotlinx.coroutines.flow.filterIsInstance
import kotlinx.coroutines.flow.first
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.onSubscription import kotlinx.coroutines.flow.onSubscription
import kotlinx.coroutines.flow.takeWhile import kotlinx.coroutines.flow.takeWhile
import org.slf4j.helpers.NOPLogger import org.slf4j.helpers.NOPLogger
import kotlin.reflect.KClass
import kotlin.reflect.cast
class Server( class Server(
private val msgBuffer: SharedFlow<ServerMessage>, private val msgBuffer: SharedFlow<ServerMessage>,
@@ -24,10 +29,11 @@ class Server(
var channels: Set<IRCChannel> = setOf() var channels: Set<IRCChannel> = setOf()
private set private set
val users = mutableListOf<User>()
suspend fun fetchChannels(): Set<IRCChannel> { suspend fun fetchChannels(): Set<IRCChannel> {
logger.debug { "Fetching IRC channels" } logger.debug { "Fetching IRC channels" }
val buffer = mutableSetOf<IRCChannel>() val buffer = mutableSetOf<IRCChannel>()
msgBuffer msgBuffer
.onSubscription { outgoingMessages.send(ClientMessage.List()) } .onSubscription { outgoingMessages.send(ClientMessage.List()) }
.filterIsInstance<ServerMessage.Numeric>() .filterIsInstance<ServerMessage.Numeric>()
@@ -40,4 +46,15 @@ class Server(
fun fetchFeatures() { fun fetchFeatures() {
} }
suspend fun <T : ServerMessage> waitForNext(
kClass: KClass<T>,
predicate: (T) -> Boolean = { true },
onSuscribedLambda: suspend () -> Unit = {},
): T =
msgBuffer
.onSubscription { onSuscribedLambda() }
.filter { kClass.isInstance(it) }
.map { kClass.cast(it) }
.first(predicate)
} }
@@ -11,13 +11,8 @@ 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.asSharedFlow
import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.flow.first
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.launch import kotlinx.coroutines.launch
import org.slf4j.helpers.NOPLogger import org.slf4j.helpers.NOPLogger
import kotlin.reflect.KClass
import kotlin.reflect.cast
class MessageReader(socket: Socket, scope: CoroutineScope, enableLogging: Boolean = false) : MessageIO(socket, scope) { class MessageReader(socket: Socket, scope: CoroutineScope, enableLogging: Boolean = false) : MessageIO(socket, scope) {
private lateinit var receive: ByteReadChannel private lateinit var receive: ByteReadChannel
@@ -45,16 +40,4 @@ class MessageReader(socket: Socket, scope: CoroutineScope, enableLogging: Boolea
} }
} }
} }
suspend fun <T : ServerMessage> waitForNext(
kClass: KClass<T>,
predicate: (T) -> Boolean = { true },
): T {
return messageBuffer
.filter {
kClass.isInstance(it)
}.map {
kClass.cast(it)
}.first(predicate)
}
} }