Implements notifications

backend/src/main/java/net/buzzert/kordophone/backend/server/ChatRepository.kt

@@ -10,6 +10,7 @@ import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.SharedFlow
+import kotlinx.coroutines.flow.asSharedFlow
 import kotlinx.coroutines.flow.collect
 import kotlinx.coroutines.flow.collectLatest
 import kotlinx.coroutines.flow.onEach
@@ -53,16 +54,20 @@ class ChatRepository(
 
     // Channel that's signaled when an outgoing message is delivered.
     val messageDeliveredChannel: SharedFlow<MessageDeliveredEvent>
-        get() = _messageDeliveredChannel
+        get() = _messageDeliveredChannel.asSharedFlow()
 
     // Changes Flow
     val conversationChanges: Flow<List<Conversation>>
         get() = database.conversationChanges
             .onEach { Log.d(REPO_LOG, "Got database conversations changed") }
 
+    // New Messages
+    val newMessages: SharedFlow<Message>
+        get() = _newMessageChannel.asSharedFlow()
+
     // Errors channel
     val errorEncounteredChannel: SharedFlow<Error>
-        get() = _errorEncounteredChannel
+        get() = _errorEncounteredChannel.asSharedFlow()
 
     fun messagesChanged(conversation: Conversation): Flow<List<Message>> =
         database.messagesChanged(conversation)
@@ -91,6 +96,7 @@ class ChatRepository(
     private var outgoingMessageThread: Thread? = null
     private val _messageDeliveredChannel = MutableSharedFlow<MessageDeliveredEvent>()
     private val _errorEncounteredChannel = MutableSharedFlow<Error>()
+    private val _newMessageChannel = MutableSharedFlow<Message>()
 
     private var updateMonitor = UpdateMonitor(apiClient)
     private var updateWatchJob: Job? = null
@@ -218,17 +224,18 @@ class ChatRepository(
             .onEach { it.conversation = conversation }
     }
 
-    private fun handleConversationChangedUpdate(conversation: Conversation) {
+    private suspend fun handleConversationChangedUpdate(conversation: Conversation) {
         Log.d(REPO_LOG, "Handling conversation changed update")
         database.writeConversations(listOf(conversation))
     }
 
-    private fun handleMessageAddedUpdate(message: Message) {
+    private suspend fun handleMessageAddedUpdate(message: Message) {
         Log.d(REPO_LOG, "Handling messages added update")
         database.writeMessages(listOf(message), message.conversation)
+        _newMessageChannel.emit(message)
     }
 
-    private fun handleMessageDelivered(event: MessageDeliveredEvent) {
+    private suspend fun handleMessageDelivered(event: MessageDeliveredEvent) {
         Log.d(REPO_LOG, "Handling successful delivery event")
 
         // Unfortunate protocol reality: the server doesn't tell us about new messages that are from us,