Update Server.kt with new helpers for waiting on a specific message, and new SMs

src/main/kotlin/com/shr4pnel/ferretirc/irc/Server.kt

@@ -2,46 +2,131 @@ package com.shr4pnel.ferretirc.irc
 
 import com.shr4pnel.ferretirc.net.messages.ClientMessage
 import com.shr4pnel.ferretirc.net.messages.ServerMessage
+import com.shr4pnel.ferretirc.net.messages.ServerMessage.Numeric
 import io.github.oshai.kotlinlogging.KotlinLogging
 import io.github.oshai.kotlinlogging.slf4j.logger
+import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.SharedFlow
+import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.flow.asStateFlow
 import kotlinx.coroutines.flow.filter
 import kotlinx.coroutines.flow.filterIsInstance
 import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.onSubscription
 import kotlinx.coroutines.flow.takeWhile
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.withTimeout
+import kotlinx.coroutines.withTimeoutOrNull
 import org.slf4j.helpers.NOPLogger
 import kotlin.reflect.KClass
 import kotlin.reflect.cast
 
 class Server(
-    private val msgBuffer: SharedFlow<ServerMessage>,
+    @PublishedApi internal val msgBuffer: SharedFlow<ServerMessage>,
-    private val outgoingMessages: Channel<ClientMessage>,
+    @PublishedApi internal val outgoingMessages: Channel<ClientMessage>,
+    private val scope: CoroutineScope,
     enableLogging: Boolean = false,
 ) {
-    private val logger = if (enableLogging) KotlinLogging.logger {} else KotlinLogging.logger(NOPLogger.NOP_LOGGER)
+    @PublishedApi
+    internal val logger = if (enableLogging) KotlinLogging.logger {} else KotlinLogging.logger(NOPLogger.NOP_LOGGER)
 
-    lateinit var features: ISupportFeatures
+    private val _features = MutableStateFlow<List<ISupportFeature>>(emptyList())
-        private set
+    val features: StateFlow<List<ISupportFeature>> = _features.asStateFlow()
 
-    var channels: Set<IRCChannel> = setOf()
+    /** Backing field for channels storing the current state of visible channels */
-        private set
+    private val _channels = MutableStateFlow<List<IRCChannel>>(emptyList())
+    val channels: StateFlow<List<IRCChannel>> = _channels.asStateFlow()
 
-    val users = mutableListOf<User>()
+    private val _users = MutableStateFlow<List<User>>(emptyList())
+    val users: StateFlow<List<User>> = _users.asStateFlow()
 
-    suspend fun fetchChannels(): Set<IRCChannel> {
+    init {
+        listen()
+    }
+
+    private companion object {
+        fun parseISupportFeature(msg: Numeric.RPL_ISUPPORT): List<ISupportFeature> {
+            val mutableFeatures = mutableListOf<ISupportFeature>()
+            msg.keypairs.forEach {
+                if (it.contains("=")) {
+                    val split = it.split("=")
+                    val key = split.first().uppercase()
+                    val value = split.last()
+
+                    val feature = when (key) {
+                        "ACCEPT" -> ISupportFeature.ACCEPT(value.toInt())
+                        "AWAYLEN" -> ISupportFeature.AWAYLEN(value.toInt())
+                        "BOT" -> ISupportFeature.BOT(value.single())
+                        "CALLERID" -> ISupportFeature.CALLERID(value.single())
+                        "CASEMAPPING" -> ISupportFeature.CASEMAPPING(value)
+                        "MAXNICKLEN" -> ISupportFeature.MAXNICKLEN(value.toInt())
+                        "CHARSET" -> ISupportFeature.CHARSET(value)
+                        else -> null
+                    }
+
+                    if (feature != null)
+                        mutableFeatures.add(feature)
+                }
+            }
+            return mutableFeatures.toList()
+        }
+    }
+
+    suspend fun fetchChannels(timeoutMillis: Long = 5000): List<IRCChannel> {
         logger.debug { "Fetching IRC channels" }
-        val buffer = mutableSetOf<IRCChannel>()
+        val buffer = mutableListOf<IRCChannel>()
         msgBuffer
             .onSubscription { outgoingMessages.send(ClientMessage.List()) }
-            .filterIsInstance<ServerMessage.Numeric>()
+            .filterIsInstance<Numeric>()
-            .takeWhile { it !is ServerMessage.Numeric.RPL_LISTEND }
+            .takeWhile { it !is Numeric.RPL_LISTEND }
-            .filterIsInstance<ServerMessage.Numeric.RPL_LIST>()
+            .filterIsInstance<Numeric.RPL_LIST>()
             .collect { buffer.add(IRCChannel(it.chanName, it.clientCount, it.topic)) }
-        channels = buffer.toSortedSet(compareBy { it.chanName })
+        val sortedChannels = buffer.sortedBy { it.chanName }
-        return channels
+        _channels.value = sortedChannels
+        return sortedChannels
+    }
+
+    private fun listen() = scope.launch {
+        msgBuffer.collect { msg ->
+            when (msg) {
+                is Numeric.RPL_ISUPPORT -> {
+                    val feats = parseISupportFeature(msg)
+                    _features.value += feats
+                    feats.forEach { logger.debug { "Parsed ISUPPORT: $it" } }
+                }
+                is ServerMessage.Ping -> {
+                    val pong = ClientMessage.Pong(msg.token)
+                    logger.debug { "[C->S]: Responding to server $msg with $pong" }
+                    scope.launch { outgoingMessages.send(pong) }
+                }
+                else -> {}
+            }
+        }
+    }
+
+    /**
+     * Fetch messages of type T until untilMessage appears in buffer
+     * @param subscriptionMessage A lambda to execute before waiting for U
+     * @param U The server message to wait for before returning the Set of messages
+     * @param T The type of ServerMessage to fetch
+     * @return A set of messages of type T in the buffer that arrive before untilMessage appears
+     */
+    suspend inline fun <reified T : ServerMessage, reified U : ServerMessage> fetchMessagesOfType(
+        timeoutMillis: Long = 5000,
+        noinline subscriptionMessage: suspend () -> Unit = {},
+    ): List<T> {
+        val buffer = mutableListOf<T>()
+        withTimeoutOrNull(timeoutMillis) {
+            msgBuffer
+                .onSubscription { subscriptionMessage() }
+                .takeWhile { it !is U }
+                .filterIsInstance<T>()
+                .collect { buffer.add(it) }
+        } ?: logger.error { "Timed out while waiting to receive ${U::class}" }
+        return buffer.toList()
     }
 
     fun fetchFeatures() {
@@ -50,11 +135,14 @@ class Server(
     suspend fun <T : ServerMessage> waitForNext(
         kClass: KClass<T>,
         predicate: (T) -> Boolean = { true },
-        onSuscribedLambda: suspend () -> Unit = {},
+        timeoutMillis: Long = 5000,
+        onSubscribedLambda: suspend () -> Unit = {},
     ): T =
-        msgBuffer
+        withTimeout(timeoutMillis) {
-            .onSubscription { onSuscribedLambda() }
+            msgBuffer
-            .filter { kClass.isInstance(it) }
+                .onSubscription { onSubscribedLambda() }
-            .map { kClass.cast(it) }
+                .filter { kClass.isInstance(it) }
-            .first(predicate)
+                .map { kClass.cast(it) }
+                .first(predicate)
+        }
 }

src/main/kotlin/com/shr4pnel/ferretirc/net/messages/ServerMessage.kt

@@ -4,25 +4,28 @@ package com.shr4pnel.ferretirc.net.messages
  * Representation of responses from the IRC server
  */
 sealed class ServerMessage {
-    data class Cap(val placeholder: String? = null) : ServerMessage()
+    data class Cap(val capabilities: String) : ServerMessage()
-
     data class Pong(val token: String? = null) : ServerMessage()
-
     data class Mode(val operatorName: String, val pMask: String) : ServerMessage()
-
     data class PrivMsg(val targets: List<String>, val message: String) : ServerMessage()
-
     data class Notice(val targets: List<String>, val message: String) : ServerMessage()
-
     data class UNIMPLEMENTED(val msg: String) : ServerMessage()
+    data class Join(val target: String) : ServerMessage()
+    data class Ping(val token: String) : ServerMessage()
+    data class Error(val message: String) : ServerMessage()
 
     sealed class Numeric(val number: Short) : ServerMessage() {
-        data class RPL_ISUPPORT(val keypairs: String) : Numeric(5)
+        data class RPL_WELCOME(val message: String) : Numeric(1)
-
+        data class RPL_YOURHOST(val message: String) : Numeric(2)
-        class RPL_LISTSTART : Numeric(321)
+        data class RPL_CREATED(val message: String) : Numeric(3)
-
+        data class RPL_MYINFO(val message: String) : Numeric(4)
+        data class RPL_ISUPPORT(val keypairs: List<String>) : Numeric(5)
+        data class RPL_LUSER(val message: String) : Numeric(251)
+        data object RPL_LISTSTART : Numeric(321)
         data class RPL_LIST(val chanName: String, val clientCount: Int, val topic: String) : Numeric(322)
-
+        data object RPL_LISTEND : Numeric(323)
-        class RPL_LISTEND : Numeric(323)
+        data class RPL_TOPIC(val topic: String) : Numeric(332)
+        data class RPL_NAMREPLY(val message: String) : Numeric(353)
+        data object RPL_ENDOFNAMES : Numeric(366)
     }
 }