From 2836c03a443a26e67747620f78ac9a983f288729 Mon Sep 17 00:00:00 2001 From: Infi Date: Fri, 21 Apr 2023 00:43:59 +0200 Subject: [PATCH] fix: switch callbacks to kotlin channels/flows = fix bugs --- .../main/java/chat/revolt/api/RevoltAPI.kt | 25 +- .../revolt/api/realtime/RealtimeSocket.kt | 95 ++--- .../chat/revolt/callbacks/ChannelCallbacks.kt | 76 ---- .../java/chat/revolt/callbacks/UiCallbacks.kt | 31 +- .../chat/sheets/MessageContextSheet.kt | 17 +- .../chat/views/channel/ChannelScreen.kt | 20 +- .../views/channel/ChannelScreenViewModel.kt | 332 ++++++------------ 7 files changed, 196 insertions(+), 400 deletions(-) delete mode 100644 app/src/main/java/chat/revolt/callbacks/ChannelCallbacks.kt diff --git a/app/src/main/java/chat/revolt/api/RevoltAPI.kt b/app/src/main/java/chat/revolt/api/RevoltAPI.kt index 291af076..436fb2d2 100644 --- a/app/src/main/java/chat/revolt/api/RevoltAPI.kt +++ b/app/src/main/java/chat/revolt/api/RevoltAPI.kt @@ -8,7 +8,6 @@ import chat.revolt.BuildConfig import chat.revolt.api.realtime.DisconnectionState import chat.revolt.api.realtime.RealtimeSocket import chat.revolt.api.routes.user.fetchSelf -import chat.revolt.api.schemas.Channel import chat.revolt.api.schemas.Emoji import chat.revolt.api.schemas.Message import chat.revolt.api.schemas.Server @@ -25,9 +24,18 @@ import io.ktor.client.plugins.logging.Logging import io.ktor.client.plugins.websocket.WebSockets import io.ktor.client.request.header import io.ktor.serialization.kotlinx.json.json +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.DelicateCoroutinesApi +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.Job +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.launch +import kotlinx.coroutines.newSingleThreadContext import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withContext import kotlinx.serialization.Serializable import kotlinx.serialization.json.Json +import chat.revolt.api.schemas.Channel as ChannelSchema const val REVOLT_BASE = "https://api.revolt.chat" const val REVOLT_SUPPORT = "https://support.revolt.chat" @@ -86,7 +94,7 @@ object RevoltAPI { // FIXME discount caching solutions! LRU would be better but this is fine for now val userCache = mutableStateMapOf() val serverCache = mutableStateMapOf() - val channelCache = mutableStateMapOf() + val channelCache = mutableStateMapOf() val emojiCache = mutableStateMapOf() val messageCache = mutableStateMapOf() @@ -97,7 +105,11 @@ object RevoltAPI { var sessionToken: String = "" private set - private var socketThread: Thread? = null + @OptIn(DelicateCoroutinesApi::class) + val realtimeContext = newSingleThreadContext("RealtimeContext") + val wsFrameChannel = Channel(Channel.UNLIMITED) + + private var socketCoroutine: Job? = null fun setSessionHeader(token: String) { sessionToken = token @@ -111,9 +123,9 @@ object RevoltAPI { } suspend fun connectWS() { - socketThread = Thread { + socketCoroutine = CoroutineScope(Dispatchers.IO).launch { try { - runBlocking { + withContext(realtimeContext) { RealtimeSocket.connect(sessionToken) } } catch (e: Exception) { @@ -125,7 +137,6 @@ object RevoltAPI { RealtimeSocket.updateDisconnectionState(DisconnectionState.Disconnected) } } - socketThread!!.start() } private suspend fun startSocketOps() { @@ -173,7 +184,7 @@ object RevoltAPI { unreads.clear() - socketThread?.interrupt() + socketCoroutine?.cancel() } /** diff --git a/app/src/main/java/chat/revolt/api/realtime/RealtimeSocket.kt b/app/src/main/java/chat/revolt/api/realtime/RealtimeSocket.kt index e9106b85..7db972e7 100644 --- a/app/src/main/java/chat/revolt/api/realtime/RealtimeSocket.kt +++ b/app/src/main/java/chat/revolt/api/realtime/RealtimeSocket.kt @@ -20,7 +20,6 @@ import chat.revolt.api.realtime.frames.receivable.ServerCreateFrame import chat.revolt.api.realtime.frames.receivable.UserUpdateFrame import chat.revolt.api.realtime.frames.sendable.AuthorizationFrame import chat.revolt.api.realtime.frames.sendable.PingFrame -import chat.revolt.callbacks.ChannelCallbacks import io.ktor.client.plugins.websocket.ws import io.ktor.websocket.CloseReason import io.ktor.websocket.Frame @@ -37,6 +36,10 @@ enum class DisconnectionState { Connected } +sealed class RealtimeSocketFrames { + data class Reconnected(val unit: Unit = Unit) : RealtimeSocketFrames() +} + object RealtimeSocket { var socket: WebSocketSession? = null @@ -61,7 +64,7 @@ object RealtimeSocket { Log.d("RealtimeSocket", "Connected to websocket.") updateDisconnectionState(DisconnectionState.Connected) - invalidateAllChannelStates() + pushReconnectEvent() // Send authorization frame val authFrame = AuthorizationFrame("Authenticate", token) @@ -99,7 +102,7 @@ object RealtimeSocket { Log.d("RealtimeSocket", "Sent ping frame with ${pingPacket.data}") } - private fun handleFrame(type: String, rawFrame: String) { + private suspend fun handleFrame(type: String, rawFrame: String) { when (type) { "Pong" -> { val pongFrame = RevoltJson.decodeFromString(PongFrame.serializer(), rawFrame) @@ -167,7 +170,7 @@ object RealtimeSocket { RevoltAPI.channelCache[it] = RevoltAPI.channelCache[it]!!.copy(lastMessageID = messageFrame.id) - ChannelCallbacks.emitMessage(it, messageFrame.id) + RevoltAPI.wsFrameChannel.send(messageFrame) } } @@ -202,51 +205,20 @@ object RealtimeSocket { Log.d( "RealtimeSocket", - "Merging message ${messageUpdateFrame.id} with partial message: $rawMessage" - ) - Log.d( - "RealtimeSocket", - "Old: $oldMessage" + "Merging message ${messageUpdateFrame.id} with updated partial." ) RevoltAPI.messageCache[messageUpdateFrame.id] = oldMessage.mergeWithPartial(rawMessage) - Log.d( - "RealtimeSocket", - "New: ${RevoltAPI.messageCache[messageUpdateFrame.id]}" - ) - messageUpdateFrame.channel.let { if (RevoltAPI.channelCache[it] == null) { Log.d("RealtimeSocket", "Channel $it not found in cache. Ignoring.") return } - - ChannelCallbacks.emitMessageUpdate(it, messageUpdateFrame.id) } - } - "ChannelStartTyping" -> { - val typingFrame = - RevoltJson.decodeFromString(ChannelStartTypingFrame.serializer(), rawFrame) - Log.d( - "RealtimeSocket", - "Received channel start typing frame for ${typingFrame.id} from ${typingFrame.user}." - ) - - ChannelCallbacks.emitStartTyping(typingFrame.id, typingFrame.user) - } - - "ChannelStopTyping" -> { - val typingFrame = - RevoltJson.decodeFromString(ChannelStopTypingFrame.serializer(), rawFrame) - Log.d( - "RealtimeSocket", - "Received channel stop typing frame for ${typingFrame.id} from ${typingFrame.user}." - ) - - ChannelCallbacks.emitStopTyping(typingFrame.id, typingFrame.user) + RevoltAPI.wsFrameChannel.send(messageUpdateFrame) } "UserUpdate" -> { @@ -298,8 +270,30 @@ object RealtimeSocket { } } + "ChannelStartTyping" -> { + val channelStartTypingFrame = + RevoltJson.decodeFromString(ChannelStartTypingFrame.serializer(), rawFrame) + Log.d( + "RealtimeSocket", + "Received channel start typing frame for ${channelStartTypingFrame.id}." + ) + + RevoltAPI.wsFrameChannel.send(channelStartTypingFrame) + } + + "ChannelStopTyping" -> { + val channelStopTypingFrame = + RevoltJson.decodeFromString(ChannelStopTypingFrame.serializer(), rawFrame) + Log.d( + "RealtimeSocket", + "Received channel stop typing frame for ${channelStopTypingFrame.id}." + ) + + RevoltAPI.wsFrameChannel.send(channelStopTypingFrame) + } + "Authenticated" -> { - // No effect + /* no-op */ } else -> { @@ -308,28 +302,7 @@ object RealtimeSocket { } } - private fun invalidateAllChannelStates() { - ChannelCallbacks.emitReconnect() + private suspend fun pushReconnectEvent() { + RevoltAPI.wsFrameChannel.send(RealtimeSocketFrames.Reconnected()) } - - /*interface ChannelCallback { - fun onStartTyping(typing: ChannelStartTypingFrame) - fun onStopTyping(typing: ChannelStopTypingFrame) - fun onMessage(message: MessageFrame) - fun onStateInvalidate() - } - - private val channelCallbacks: SnapshotStateMap = mutableStateMapOf() - - fun registerChannelCallback(channelId: String, callback: ChannelCallback) { - channelCallbacks[channelId] = callback - - Log.d("RealtimeSocket", "Registered channel callback for $channelId.") - } - - fun unregisterChannelCallback(channelId: String) { - channelCallbacks.remove(channelId) - - Log.d("RealtimeSocket", "Unregistered channel callback for $channelId") - }*/ } \ No newline at end of file diff --git a/app/src/main/java/chat/revolt/callbacks/ChannelCallbacks.kt b/app/src/main/java/chat/revolt/callbacks/ChannelCallbacks.kt deleted file mode 100644 index 2931e759..00000000 --- a/app/src/main/java/chat/revolt/callbacks/ChannelCallbacks.kt +++ /dev/null @@ -1,76 +0,0 @@ -package chat.revolt.callbacks - -object ChannelCallbacks { - interface CallbackReceiver { - fun onReconnect() - fun onStartTyping(channelId: String, userId: String) - fun onStopTyping(channelId: String, userId: String) - fun onMessage(messageId: String) - fun onMessageUpdate(messageId: String) - fun onMessageDelete(messageId: String) - fun onMessageBulkDelete(messageIds: List) - fun onMessageReactionAdd(messageId: String, emoji: String, userId: String) - fun onMessageReactionRemove(messageId: String, emoji: String, userId: String) - fun onMessageReactionRemoveAll(messageId: String) - } - - var receivers = mutableMapOf() - - fun registerReceiver(channelId: String, receiver: CallbackReceiver) { - receivers[channelId] = receiver - } - - fun unregisterReceiver(channelId: String) { - receivers.remove(channelId) - } - - fun emitReconnect() { - receivers.forEach { it.value.onReconnect() } - } - - fun emitStartTyping(channelId: String, userId: String) { - receivers[channelId]?.onStartTyping(channelId, userId) - } - - fun emitStopTyping(channelId: String, userId: String) { - receivers[channelId]?.onStopTyping(channelId, userId) - } - - fun emitMessage(channelId: String, messageId: String) { - receivers[channelId]?.onMessage(messageId) - } - - fun emitMessageUpdate(channelId: String, messageId: String) { - receivers[channelId]?.onMessageUpdate(messageId) - } - - fun emitMessageDelete(channelId: String, messageId: String) { - receivers[channelId]?.onMessageDelete(messageId) - } - - fun emitMessageBulkDelete(channelId: String, messageIds: List) { - receivers[channelId]?.onMessageBulkDelete(messageIds) - } - - fun emitMessageReactionAdd( - channelId: String, - messageId: String, - emoji: String, - userId: String - ) { - receivers[channelId]?.onMessageReactionAdd(messageId, emoji, userId) - } - - fun emitMessageReactionRemove( - channelId: String, - messageId: String, - emoji: String, - userId: String - ) { - receivers[channelId]?.onMessageReactionRemove(messageId, emoji, userId) - } - - fun emitMessageReactionRemoveAll(channelId: String, messageId: String) { - receivers[channelId]?.onMessageReactionRemoveAll(messageId) - } -} \ No newline at end of file diff --git a/app/src/main/java/chat/revolt/callbacks/UiCallbacks.kt b/app/src/main/java/chat/revolt/callbacks/UiCallbacks.kt index 2c2b7cf2..4e578ad3 100644 --- a/app/src/main/java/chat/revolt/callbacks/UiCallbacks.kt +++ b/app/src/main/java/chat/revolt/callbacks/UiCallbacks.kt @@ -1,28 +1,15 @@ package chat.revolt.callbacks -/** - * Callbacks for UI events, such as when a user selects "reply" on a message, so that the - * channel screen can add a reply to the message, for example. - * - * We do this by having a singleton object that contains all the receivers, and then - * the UI can set the callbacks to whatever it wants. - */ +import kotlinx.coroutines.flow.MutableSharedFlow + +sealed class UiCallback { + data class ReplyToMessage(val messageId: String) : UiCallback() +} + object UiCallbacks { - interface CallbackReceiver { - fun onQueueMessageForReply(messageId: String) - } + val uiCallbackFlow: MutableSharedFlow = MutableSharedFlow() - var receivers = mutableListOf() - - fun registerReceiver(receiver: CallbackReceiver) { - receivers.add(receiver) - } - - fun unregisterReceiver(receiver: CallbackReceiver) { - receivers.remove(receiver) - } - - fun emitQueueMessageForReply(messageId: String) { - receivers.forEach { it.onQueueMessageForReply(messageId) } + suspend fun replyToMessage(messageId: String) { + uiCallbackFlow.emit(UiCallback.ReplyToMessage(messageId)) } } \ No newline at end of file diff --git a/app/src/main/java/chat/revolt/screens/chat/sheets/MessageContextSheet.kt b/app/src/main/java/chat/revolt/screens/chat/sheets/MessageContextSheet.kt index 701ce144..96a37283 100644 --- a/app/src/main/java/chat/revolt/screens/chat/sheets/MessageContextSheet.kt +++ b/app/src/main/java/chat/revolt/screens/chat/sheets/MessageContextSheet.kt @@ -2,7 +2,11 @@ package chat.revolt.screens.chat.sheets import android.widget.Toast import androidx.compose.foundation.background -import androidx.compose.foundation.layout.* +import androidx.compose.foundation.layout.Box +import androidx.compose.foundation.layout.Column +import androidx.compose.foundation.layout.Spacer +import androidx.compose.foundation.layout.height +import androidx.compose.foundation.layout.padding import androidx.compose.foundation.rememberScrollState import androidx.compose.foundation.verticalScroll import androidx.compose.material.icons.Icons @@ -13,6 +17,7 @@ import androidx.compose.material3.MaterialTheme import androidx.compose.material3.Text import androidx.compose.material3.surfaceColorAtElevation import androidx.compose.runtime.Composable +import androidx.compose.runtime.rememberCoroutineScope import androidx.compose.ui.Modifier import androidx.compose.ui.draw.clip import androidx.compose.ui.platform.LocalClipboardManager @@ -28,6 +33,7 @@ import chat.revolt.api.RevoltAPI import chat.revolt.callbacks.UiCallbacks import chat.revolt.components.chat.Message import chat.revolt.components.generic.SheetClickable +import kotlinx.coroutines.launch @Composable fun MessageContextSheet( @@ -42,6 +48,7 @@ fun MessageContextSheet( val context = LocalContext.current val clipboardManager = LocalClipboardManager.current + val coroutineScope = rememberCoroutineScope() Column( modifier = Modifier @@ -80,7 +87,9 @@ fun MessageContextSheet( ) }, ) { - UiCallbacks.emitQueueMessageForReply(messageId) + coroutineScope.launch { + UiCallbacks.replyToMessage(messageId) + } navController.popBackStack() } @@ -122,7 +131,7 @@ fun MessageContextSheet( ) }, ) { - if (message.content == null || message.content.isEmpty()) { + if (message.content.isNullOrEmpty()) { Toast.makeText( context, context.getString(R.string.message_context_sheet_actions_copy_failed_empty), @@ -156,7 +165,7 @@ fun MessageContextSheet( ) }, ) { - if (message.content == null || message.content.isEmpty()) { + if (message.content.isNullOrEmpty()) { Toast.makeText( context, context.getString(R.string.message_context_sheet_actions_copy_failed_empty), diff --git a/app/src/main/java/chat/revolt/screens/chat/views/channel/ChannelScreen.kt b/app/src/main/java/chat/revolt/screens/chat/views/channel/ChannelScreen.kt index d84fbb71..0252a18a 100644 --- a/app/src/main/java/chat/revolt/screens/chat/views/channel/ChannelScreen.kt +++ b/app/src/main/java/chat/revolt/screens/chat/views/channel/ChannelScreen.kt @@ -33,8 +33,6 @@ import chat.revolt.RevoltTweenFloat import chat.revolt.RevoltTweenInt import chat.revolt.api.RevoltAPI import chat.revolt.api.routes.microservices.autumn.FileArgs -import chat.revolt.callbacks.ChannelCallbacks -import chat.revolt.callbacks.UiCallbacks import chat.revolt.components.chat.Message import chat.revolt.components.chat.MessageField import chat.revolt.components.screens.chat.AttachmentManager @@ -75,7 +73,7 @@ fun ChannelScreen( ) { uriList -> uriList.let { uris -> uris.forEach { - DocumentFile.fromSingleUri(context, it)?.let docfile@{ file -> + DocumentFile.fromSingleUri(context, it)?.let { file -> val mFile = File(context.cacheDir, file.name ?: "attachment") mFile.outputStream().use { output -> @@ -97,23 +95,19 @@ fun ChannelScreen( val scrollDownFABPadding by animateDpAsState( if (viewModel.typingUsers.isNotEmpty()) 25.dp else 0.dp, - animationSpec = RevoltTweenDp + animationSpec = RevoltTweenDp, + label = "ScrollDownFABPadding" ) LaunchedEffect(channelId) { viewModel.fetchChannel(channelId) - } - LaunchedEffect(viewModel.channel) { - if (viewModel.channel?.id != channelId) { - viewModel.fetchChannel(channelId) + coroutineScope.launch { + viewModel.listenForWsFrame() } - } - DisposableEffect(channelId) { - onDispose { - viewModel.channelCallbackReceiver?.let { ChannelCallbacks.unregisterReceiver(channelId) } - viewModel.uiCallbackReceiver?.let { UiCallbacks.unregisterReceiver(it) } + coroutineScope.launch { + viewModel.listenForUiCallbacks() } } diff --git a/app/src/main/java/chat/revolt/screens/chat/views/channel/ChannelScreenViewModel.kt b/app/src/main/java/chat/revolt/screens/chat/views/channel/ChannelScreenViewModel.kt index 8b9d4e0a..8539b8e1 100644 --- a/app/src/main/java/chat/revolt/screens/chat/views/channel/ChannelScreenViewModel.kt +++ b/app/src/main/java/chat/revolt/screens/chat/views/channel/ChannelScreenViewModel.kt @@ -8,7 +8,13 @@ import androidx.compose.runtime.setValue import androidx.lifecycle.ViewModel import androidx.lifecycle.viewModelScope import chat.revolt.api.RevoltAPI +import chat.revolt.api.RevoltJson import chat.revolt.api.internals.ULID +import chat.revolt.api.realtime.frames.receivable.ChannelStartTypingFrame +import chat.revolt.api.realtime.frames.receivable.ChannelStopTypingFrame +import chat.revolt.api.realtime.frames.receivable.MessageDeleteFrame +import chat.revolt.api.realtime.frames.receivable.MessageFrame +import chat.revolt.api.realtime.frames.receivable.MessageUpdateFrame import chat.revolt.api.routes.channel.SendMessageReply import chat.revolt.api.routes.channel.ackChannel import chat.revolt.api.routes.channel.fetchMessagesFromChannel @@ -20,13 +26,18 @@ import chat.revolt.api.routes.microservices.autumn.uploadToAutumn import chat.revolt.api.routes.user.addUserIfUnknown import chat.revolt.api.schemas.Channel import chat.revolt.api.schemas.Message -import chat.revolt.callbacks.ChannelCallbacks +import chat.revolt.callbacks.UiCallback import chat.revolt.callbacks.UiCallbacks import io.ktor.http.ContentType -import io.sentry.Sentry +import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Job import kotlinx.coroutines.delay +import kotlinx.coroutines.flow.catch +import kotlinx.coroutines.flow.flow +import kotlinx.coroutines.flow.launchIn +import kotlinx.coroutines.flow.onEach import kotlinx.coroutines.launch +import kotlinx.coroutines.withContext import kotlinx.datetime.Instant class ChannelScreenViewModel : ViewModel() { @@ -114,190 +125,8 @@ class ChannelScreenViewModel : ViewModel() { val noMoreMessages: Boolean get() = _noMoreMessages - private fun setNoMoreMessages(noMore: Boolean) { - _noMoreMessages = noMore - } - - private var _uiCallbackReceiver = mutableStateOf(null) - val uiCallbackReceiver: UiCallbacks.CallbackReceiver? - get() = _uiCallbackReceiver.value - - private var _uiCallbackRegistered by mutableStateOf(false) - - private var _channelCallbackReceiver = mutableStateOf(null) - val channelCallbackReceiver: ChannelCallbacks.CallbackReceiver? - get() = _channelCallbackReceiver.value - - private var _channelCallbackRegistered by mutableStateOf(false) - - /* - inner class ChannelScreenCallback : RealtimeSocket.ChannelCallback { - override fun onMessage(message: Message) { - viewModelScope.launch { - addUserIfUnknown(message.author!!) - } - - regroupMessages(listOf(message) + renderableMessages) - ackNewest() - } - - override fun onStartTyping(typing: ChannelStartTypingFrame) { - viewModelScope.launch { - addUserIfUnknown(typing.user) - } - - if (!_typingUsers.contains(typing.user)) { - _typingUsers.add(typing.user) - } - } - - override fun onStopTyping(typing: ChannelStopTypingFrame) { - if (_typingUsers.contains(typing.user)) { - _typingUsers.remove(typing.user) - } - } - - override fun onStateInvalidate() { - fetchMessages() - _typingUsers.clear() - } - }*/ - - inner class UiCallbackReceiver : UiCallbacks.CallbackReceiver { - override fun onQueueMessageForReply(messageId: String) { - viewModelScope.launch { - addInReplyTo(SendMessageReply(messageId, true)) - } - } - } - - inner class ChannelCallbackReceiver : ChannelCallbacks.CallbackReceiver { - override fun onReconnect() { - fetchMessages() - _typingUsers.clear() - // TODO push time rift to messages - } - - override fun onStartTyping(channelId: String, userId: String) { - viewModelScope.launch { - addUserIfUnknown(userId) - - if (!_typingUsers.contains(userId)) { - _typingUsers.add(userId) - } - } - } - - override fun onStopTyping(channelId: String, userId: String) { - if (_typingUsers.contains(userId)) { - _typingUsers.remove(userId) - } - } - - override fun onMessage(messageId: String) { - viewModelScope.launch { - val message = RevoltAPI.messageCache[messageId] ?: return@launch - - addUserIfUnknown(message.author!!) - - regroupMessages(listOf(message) + renderableMessages) - ackNewest() - } - } - - override fun onMessageUpdate(messageId: String) { - val message = RevoltAPI.messageCache[messageId] ?: return - - Log.d("ChannelScreen", "Handler Message updated: $message") - - regroupMessages(renderableMessages.map { - if (it.id == message.id) { - message - } else { - it - } - }) - } - - override fun onMessageDelete(messageId: String) { - // TODO Not implemented - Log.d("ChannelScreen", "Handler Message deleted: $messageId") - } - - override fun onMessageBulkDelete(messageIds: List) { - // TODO Not implemented - Log.d("ChannelScreen", "Handler Messages bulk deleted: $messageIds") - } - - override fun onMessageReactionAdd(messageId: String, emoji: String, userId: String) { - // TODO Not implemented - Log.d("ChannelScreen", "Handler Message reaction added: $messageId $emoji $userId") - } - - override fun onMessageReactionRemove(messageId: String, emoji: String, userId: String) { - // TODO Not implemented - Log.d("ChannelScreen", "Handler Message reaction removed: $messageId $emoji $userId") - } - - override fun onMessageReactionRemoveAll(messageId: String) { - // TODO Not implemented - Log.d("ChannelScreen", "Handler Message reactions removed: $messageId") - } - } - - private fun registerCallbacks() { - if (channel?.id == null) { - Sentry.captureException(IllegalStateException("Channel ID is null while trying to register callbacks")) - return - } - - if (!_channelCallbackRegistered) { - _channelCallbackReceiver.value = ChannelCallbackReceiver() - ChannelCallbacks.registerReceiver(channel!!.id!!, _channelCallbackReceiver.value!!) - _channelCallbackRegistered = true - } else { - Log.d( - "ChannelScreenViewModel", - "Channel Callbacks already registered but trying to register again. Ignoring but this is a bug." - ) - } - - if (!_uiCallbackRegistered) { - _uiCallbackReceiver.value = UiCallbackReceiver() - UiCallbacks.registerReceiver(_uiCallbackReceiver.value!!) - _uiCallbackRegistered = true - } else { - Log.d( - "ChannelScreenViewModel", - "UI Callbacks already registered but trying to register again. Ignoring but this is a bug." - ) - } - } - - fun fetchMessages() { - if (channel == null) { - return - } - - _renderableMessages.clear() - - viewModelScope.launch { - val messages = arrayListOf() - fetchMessagesFromChannel(channel!!.id!!, limit = 50, false).let { - if (it.messages.isNullOrEmpty() || it.messages.size < 50) { - setNoMoreMessages(true) - } - - it.messages?.forEach { message -> - addUserIfUnknown(message.author ?: return@forEach) - if (!RevoltAPI.messageCache.containsKey(message.id)) { - RevoltAPI.messageCache[message.id!!] = message - } - messages.add(message) - } - } - regroupMessages(renderableMessages + messages) - } + private fun setNoMoreMessages() { + _noMoreMessages = true } fun fetchOlderMessages() { @@ -308,38 +137,26 @@ class ChannelScreenViewModel : ViewModel() { viewModelScope.launch { val messages = arrayListOf() - if (renderableMessages.isNotEmpty()) { - fetchMessagesFromChannel( - channel!!.id!!, - limit = 50, - true, - before = renderableMessages.last().id - ).let { - if (it.messages.isNullOrEmpty() || it.messages.size < 50) { - setNoMoreMessages(true) - } - - it.messages?.forEach { message -> - addUserIfUnknown(message.author ?: return@forEach) - if (!RevoltAPI.messageCache.containsKey(message.id)) { - RevoltAPI.messageCache[message.id!!] = message - } - messages.add(message) - } + fetchMessagesFromChannel( + channel!!.id!!, + limit = 50, + true, + before = if (renderableMessages.isNotEmpty()) { + renderableMessages.first().id + } else { + null + } + ).let { + if (it.messages.isNullOrEmpty() || it.messages.size < 50) { + setNoMoreMessages() } - } else { - fetchMessagesFromChannel(channel!!.id!!, limit = 50, true).let { - if (it.messages.isNullOrEmpty() || it.messages.size < 50) { - setNoMoreMessages(true) - } - it.messages?.forEach { message -> - addUserIfUnknown(message.author ?: return@forEach) - if (!RevoltAPI.messageCache.containsKey(message.id)) { - RevoltAPI.messageCache[message.id!!] = message - } - messages.add(message) + it.messages?.forEach { message -> + addUserIfUnknown(message.author ?: return@forEach) + if (!RevoltAPI.messageCache.containsKey(message.id)) { + RevoltAPI.messageCache[message.id!!] = message } + messages.add(message) } } @@ -357,8 +174,6 @@ class ChannelScreenViewModel : ViewModel() { _channel = RevoltAPI.channelCache[id] } - registerCallbacks() - if (_channel?.lastMessageID != null) { ackNewest() } else { @@ -440,6 +255,89 @@ class ChannelScreenViewModel : ViewModel() { setRenderableMessages(groupedMessages.values.toList()) } + suspend fun listenForWsFrame() { + withContext(RevoltAPI.realtimeContext) { + flow { + while (true) { + emit(RevoltAPI.wsFrameChannel.receive()) + } + }.onEach { + when (it) { + is MessageFrame -> { + if (it.channel != channel?.id) return@onEach + + addUserIfUnknown(it.author!!) + regroupMessages(listOf(it) + renderableMessages) + ackNewest() + } + + is MessageUpdateFrame -> { + if (it.channel != channel?.id) return@onEach + + val messageFrame = + RevoltJson.decodeFromJsonElement(MessageFrame.serializer(), it.data) + + renderableMessages.find { currentMsg -> + currentMsg.id == it.id + } ?: return@onEach // Message not found, ignore. + + regroupMessages(renderableMessages.map { currentMsg -> + if (currentMsg.id == it.id) { + messageFrame + } else { + currentMsg + } + }) + } + + is MessageDeleteFrame -> { + if (it.channel != channel?.id) return@onEach + + regroupMessages(renderableMessages.filter { currentMsg -> + currentMsg.id != it.id + }) + } + + is ChannelStartTypingFrame -> { + if (it.id != channel?.id) return@onEach + if (_typingUsers.contains(it.user)) return@onEach + + addUserIfUnknown(it.user) + _typingUsers.add(it.user) + } + + is ChannelStopTypingFrame -> { + if (it.id != channel?.id) return@onEach + if (!_typingUsers.contains(it.user)) return@onEach + + _typingUsers.remove(it.user) + } + } + }.catch { + Log.e("ChannelScreen", "Failed to receive WS frame", it) + }.launchIn(this) + } + } + + suspend fun listenForUiCallbacks() { + withContext(Dispatchers.Main) { + UiCallbacks.uiCallbackFlow.onEach { + when (it) { + is UiCallback.ReplyToMessage -> { + addInReplyTo( + SendMessageReply( + id = it.messageId, + mention = false + ) + ) + } + } + }.catch { + Log.e("ChannelScreen", "Failed to receive UI callback", it) + }.launchIn(this) + } + } + private var debouncedChannelAck: Job? = null private fun ackNewest() { if (debouncedChannelAck?.isActive == true) {