Group Chat Streaming (#166)

* first pass at all the pieces needed for threading

* a few more places

* make signing key extend inboxOwner

* get it decoding messages

* dump the latest v3 code

* write a test for creating a v3 client

* use created At

* write test for creating libxmtp client and confirm it works

* move these change to a different branch

* dont pass a conversation

* fix linter

* point to local not dev

* feature flag the client creating of libxmtp while in alpha

* change to local

* fix up the test helper

* feat: fix up the example app

* fix up the 22 compat issue

* Revert "move these change to a different branch"

This reverts commit 8998d13.

* try and get some tests running

* setup local database

* have it create correctly

* write tests for functionality

* test sending

* send encoded content

* add updates to the v3 bindings

* add updates to the v3 bindings

* store in a keystore

* move to preferences

* fix lint

* Fix build issues

* new libxmtp updates

* dump the latest schema

* update to the latest client creation flow

* get the create working again

* use the keystore because its more secure

* fix up linter compat again

* flaky test

* get the tests all passing

* get the example working with groups

* create a group with two addresses

* more tweaks to the example app to get groups working

* add streaming messages to groups

* a few example UI tweaks

* fix the lowercasing issue in the example app

* dump the schema again

* implement all the conversation functionality

* add new codec for membership changes

* write tests for it

* fix up the tests a bit'

* add more tests and group streaming

* get the new codec working as expected

* add pagination to messages

* fix up the library linting issues

* fix up flaky test

* fix up min sdk version issue again

* update the example app

* remove the saved wallet stuff from the demo

* get groups working again with signer improvements and membership changes

* fix linter

* remove syncs so the client will need to manage

* add pagination to group listing

* dont return self for peers and add erroring to new group creation

* update the syncing in the tests

* remove all the streams work and move to another PR

* Revert "remove all the streams work and move to another PR"

This reverts commit 76bfcc7.

* update test

* undo all the bad merge items

* attempt at a few different stream techiniques

* get message streaming working and start on group streaming

* remove unneeded class

* get group streaming working

* fix streaming for both

* fix the linter

* throw error if client doesnt support groups

---------

Co-authored-by: Nicholas Molnar <65710+neekolas@users.noreply.github.com>

Author: nplasterer

example/src/main/java/org/xmtp/android/example/MainViewModel.kt

@@ -77,7 +77,7 @@ class MainViewModel : ViewModel() {
     val stream: StateFlow<MainListItem?> =
         stateFlow(viewModelScope, null) { subscriptionCount ->
             if (ClientManager.clientState.value is ClientManager.ClientState.Ready) {
-                ClientManager.client.conversations.stream()
+                ClientManager.client.conversations.streamAll()
                     .flowWhileShared(
                         subscriptionCount,
                         SharingStarted.WhileSubscribed(1000L)

library/src/androidTest/java/org/xmtp/android/library/GroupTest.kt

@@ -2,6 +2,8 @@ package org.xmtp.android.library
 
 import androidx.test.ext.junit.runners.AndroidJUnit4
 import androidx.test.platform.app.InstrumentationRegistry
+import app.cash.turbine.test
+import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.runBlocking
 import org.junit.Assert.assertEquals
 import org.junit.Assert.assertThrows
@@ -197,4 +199,56 @@ class GroupTest {
         assertEquals(ReactionAction.Added, content?.action)
         assertEquals(ReactionSchema.Unicode, content?.schema)
     }
+
+    @OptIn(ExperimentalCoroutinesApi::class)
+    @Test
+    fun testCanStreamGroupMessages() = kotlinx.coroutines.test.runTest {
+        val group = boClient.conversations.newGroup(listOf(alix.walletAddress.lowercase()))
+
+        group.streamMessages().test {
+            group.send("hi")
+            assertEquals("hi", awaitItem().body)
+            group.send("hi again")
+            assertEquals("hi again", awaitItem().body)
+        }
+    }
+
+    @OptIn(ExperimentalCoroutinesApi::class)
+    @Test
+    fun testCanStreamDecryptedGroupMessages() = kotlinx.coroutines.test.runTest {
+        val group = boClient.conversations.newGroup(listOf(alix.walletAddress))
+
+        group.streamDecryptedMessages().test {
+            group.send("hi")
+            assertEquals("hi", awaitItem().encodedContent.content.toStringUtf8())
+            group.send("hi again")
+            assertEquals("hi again", awaitItem().encodedContent.content.toStringUtf8())
+        }
+    }
+
+    @OptIn(ExperimentalCoroutinesApi::class)
+    @Test
+    fun testCanStreamGroups() = kotlinx.coroutines.test.runTest {
+        boClient.conversations.streamGroups().test {
+            val group =
+                alixClient.conversations.newGroup(listOf(bo.walletAddress))
+            assertEquals(group.id.toHex(), awaitItem().topic)
+            val group2 =
+                caroClient.conversations.newGroup(listOf(bo.walletAddress))
+            assertEquals(group2.id.toHex(), awaitItem().topic)
+        }
+    }
+
+    @OptIn(ExperimentalCoroutinesApi::class)
+    @Test
+    fun testCanStreamGroupsAndConversations() = kotlinx.coroutines.test.runTest {
+        boClient.conversations.streamAll().test {
+            val group =
+                caroClient.conversations.newGroup(listOf(bo.walletAddress))
+            assertEquals(group.id.toHex(), awaitItem().topic)
+            val conversation =
+                boClient.conversations.newConversation(alix.walletAddress)
+            assertEquals(conversation.topic, awaitItem().topic)
+        }
+    }
 }

library/src/main/java/org/xmtp/android/library/Conversation.kt

@@ -318,15 +318,15 @@ sealed class Conversation {
         return when (this) {
             is V1 -> conversationV1.streamMessages()
             is V2 -> conversationV2.streamMessages()
-            is Group -> throw XMTPException("Coming follow up PR")
+            is Group -> group.streamMessages()
         }
     }
 
     fun streamDecryptedMessages(): Flow<DecryptedMessage> {
         return when (this) {
             is V1 -> conversationV1.streamDecryptedMessages()
             is V2 -> conversationV2.streamDecryptedMessages()
-            is Group -> throw XMTPException("Coming follow up PR")
+            is Group -> group.streamDecryptedMessages()
         }
     }
 

library/src/main/java/org/xmtp/android/library/Conversations.kt

@@ -3,9 +3,12 @@ package org.xmtp.android.library
 import android.util.Log
 import io.grpc.StatusException
 import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.channels.awaitClose
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.callbackFlow
 import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.merge
 import kotlinx.coroutines.runBlocking
 import org.xmtp.android.library.GRPCApiClient.Companion.makeQueryRequest
 import org.xmtp.android.library.GRPCApiClient.Companion.makeSubscribeRequest
@@ -32,7 +35,9 @@ import org.xmtp.android.library.messages.walletAddress
 import org.xmtp.proto.keystore.api.v1.Keystore.TopicMap.TopicData
 import org.xmtp.proto.message.contents.Contact
 import org.xmtp.proto.message.contents.Invitation
+import uniffi.xmtpv3.FfiConversationCallback
 import uniffi.xmtpv3.FfiConversations
+import uniffi.xmtpv3.FfiGroup
 import uniffi.xmtpv3.FfiListConversationsOptions
 import java.util.Date
 import kotlin.time.Duration.Companion.nanoseconds
@@ -477,7 +482,6 @@ data class Conversations(
         client.subscribeTopic(
             listOf(Topic.userIntro(client.address), Topic.userInvite(client.address)),
         ).collect { envelope ->
-
             if (envelope.contentTopic == Topic.userIntro(client.address).description) {
                 val conversationV1 = fromIntro(envelope = envelope)
                 if (!streamedConversationTopics.contains(conversationV1.topic)) {
@@ -496,6 +500,21 @@ data class Conversations(
         }
     }
 
+    fun streamAll(): Flow<Conversation> {
+        return merge(streamGroups(), stream())
+    }
+
+    fun streamGroups(): Flow<Conversation> = callbackFlow {
+        val groupCallback = object : FfiConversationCallback {
+            override fun onConversation(conversation: FfiGroup) {
+                trySend(Conversation.Group(Group(client, conversation)))
+            }
+        }
+        val stream = libXMTPConversations?.stream(groupCallback)
+            ?: throw XMTPException("Client does not support Groups")
+        awaitClose { stream.end() }
+    }
+
     /**
      * Get the stream of all messages of the current [Client]
      * @return Flow object of [DecodedMessage] that represents all the messages of the

library/src/main/java/org/xmtp/android/library/Group.kt

@@ -1,5 +1,8 @@
 package org.xmtp.android.library
 
+import kotlinx.coroutines.channels.awaitClose
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.callbackFlow
 import kotlinx.coroutines.runBlocking
 import org.xmtp.android.library.codecs.ContentCodec
 import org.xmtp.android.library.codecs.EncodedContent
@@ -10,6 +13,8 @@ import org.xmtp.android.library.messages.PagingInfoSortDirection
 import org.xmtp.proto.message.api.v1.MessageApiOuterClass
 import uniffi.xmtpv3.FfiGroup
 import uniffi.xmtpv3.FfiListMessagesOptions
+import uniffi.xmtpv3.FfiMessage
+import uniffi.xmtpv3.FfiMessageCallback
 import java.util.Date
 import kotlin.time.Duration.Companion.nanoseconds
 import kotlin.time.DurationUnit
@@ -34,7 +39,7 @@ class Group(val client: Client, private val libXMTPGroup: FfiGroup) {
         runBlocking {
             libXMTPGroup.send(contentBytes = encodedContent.toByteArray())
         }
-        return id.toString()
+        return id.toHex()
     }
 
     fun <T> prepareMessage(content: T, options: SendOptions?): EncodedContent {
@@ -136,4 +141,26 @@ class Group(val client: Client, private val libXMTPGroup: FfiGroup) {
             libXMTPGroup.listMembers().map { it.accountAddress }
         }
     }
+
+    fun streamMessages(): Flow<DecodedMessage> = callbackFlow {
+        val messageCallback = object : FfiMessageCallback {
+            override fun onMessage(message: FfiMessage) {
+                trySend(Message(client, message).decode())
+            }
+        }
+
+        val stream = libXMTPGroup.stream(messageCallback)
+        awaitClose { stream.end() }
+    }
+
+    fun streamDecryptedMessages(): Flow<DecryptedMessage> = callbackFlow {
+        val messageCallback = object : FfiMessageCallback {
+            override fun onMessage(message: FfiMessage) {
+                trySend(decrypt(Message(client, message)))
+            }
+        }
+
+        val stream = libXMTPGroup.stream(messageCallback)
+        awaitClose { stream.end() }
+    }
 }