feat: Rust V2

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

@@ -1,9 +1,11 @@
 package org.xmtp.android.library
 
 import androidx.test.ext.junit.runners.AndroidJUnit4
-import app.cash.turbine.test
 import com.google.protobuf.kotlin.toByteString
 import com.google.protobuf.kotlin.toByteStringUtf8
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.flow.catch
+import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
 import org.junit.Assert.assertEquals
 import org.junit.Assert.assertFalse
@@ -588,11 +590,19 @@ class ConversationTest {
 
     @Test
     fun testCanStreamConversationsV2() = kotlinx.coroutines.test.runTest {
-        bobClient.conversations.stream().test {
-            val conversation = bobClient.conversations.newConversation(alice.walletAddress)
-            conversation.send(content = "hi")
-            assertEquals("hi", awaitItem().messages(limit = 1).first().body)
+        val flow = bobClient.conversations.stream()
+        val job = launch {
+            flow.catch { e ->
+                throw Exception("Error collecting flow: $e")
+            }.collect { convo ->
+                assert(convo.topic.isNotEmpty())
+                this.cancel()
+            }
         }
+
+        bobClient.conversations.newConversation(alice.walletAddress)
+
+        job.join()
     }
 
     @Test
@@ -601,19 +611,37 @@ class ConversationTest {
         fixtures.publishLegacyContact(client = bobClient)
         fixtures.publishLegacyContact(client = aliceClient)
         val conversation = aliceClient.conversations.newConversation(bob.walletAddress)
-        conversation.streamMessages().test {
-            conversation.send("hi alice")
-            assertEquals("hi alice", awaitItem().encodedContent.content.toStringUtf8())
+
+        val flow = conversation.streamMessages()
+        val job = launch {
+            flow.catch { e ->
+                throw Exception("Error collecting flow: $e")
+            }.collect { message ->
+                assertEquals("hi alice", message.encodedContent.content.toStringUtf8())
+                this.cancel()
+            }
         }
-    }
 
+        conversation.send("hi alice")
+
+        job.join()
+    }
     @Test
     fun testStreamingMessagesFromV2Conversations() = kotlinx.coroutines.test.runTest {
         val conversation = aliceClient.conversations.newConversation(bob.walletAddress)
-        conversation.streamMessages().test {
-            conversation.send("hi alice")
-            assertEquals("hi alice", awaitItem().encodedContent.content.toStringUtf8())
+        val flow = conversation.streamMessages()
+        val job = launch {
+            flow.catch { e ->
+                throw Exception("Error collecting flow: $e")
+            }.collect { message ->
+                assertEquals("hi alice", message.encodedContent.content.toStringUtf8())
+                this.cancel()
+            }
         }
+
+        conversation.send("hi alice")
+
+        job.join()
     }
 
     @Test

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

@@ -507,14 +507,32 @@ class GroupTest {
 
     @Test
     fun testCanStreamGroupsAndConversations() = kotlinx.coroutines.test.runTest {
-        boClient.conversations.streamAll().test {
-            val group =
-                caroClient.conversations.newGroup(listOf(bo.walletAddress))
-            assertEquals(group.topic, awaitItem().topic)
-            val conversation =
-                alixClient.conversations.newConversation(bo.walletAddress)
-            assertEquals(conversation.topic, awaitItem().topic)
+        val flow = boClient.conversations.streamAll()
+        var counter = 0
+        val job = launch {
+            flow.catch { e ->
+                throw Exception("Error collecting flow: $e")
+            }.collect { convo ->
+                counter++
+                when (convo) {
+                    is Conversation.Group -> {
+                        assert(true)
+                    }
+                    is Conversation.V2 -> {
+                        assert(true)
+                    }
+                    else -> {
+                        assert(false)
+                    }
+                }
+                if (counter == 2) this.cancel()
+            }
         }
+
+        caroClient.conversations.newGroup(listOf(bo.walletAddress))
+        alixClient.conversations.newConversation(bo.walletAddress)
+
+        job.join()
     }
 
     @Test

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

@@ -27,6 +27,7 @@ import org.xmtp.proto.message.contents.Contact
 import org.xmtp.proto.message.contents.InvitationV1Kt.context
 import org.xmtp.proto.message.contents.PrivateKeyOuterClass
 import org.xmtp.proto.message.contents.PrivateKeyOuterClass.PrivateKeyBundle
+import uniffi.xmtpv3.createV2Client
 import java.util.Date
 
 @RunWith(AndroidJUnit4::class)
@@ -63,9 +64,15 @@ class LocalInstrumentedTest {
         val identity = PrivateKey.newBuilder().build().generate()
         val authorized = alice.createIdentity(identity)
         val authToken = authorized.createAuthToken()
+        val v2Client = runBlocking {
+            createV2Client(
+                host = XMTPEnvironment.LOCAL.getUrl(),
+                isSecure = false
+            )
+        }
         val api = GRPCApiClient(
             environment = XMTPEnvironment.LOCAL,
-            secure = false,
+            rustV2Client = v2Client
         )
         api.setAuthToken(authToken)
         val encryptedBundle = authorized.toBundle.encrypted(alice)
@@ -93,9 +100,15 @@ class LocalInstrumentedTest {
         val identity = PrivateKeyBuilder().getPrivateKey()
         val authorized = aliceWallet.createIdentity(identity)
         val authToken = authorized.createAuthToken()
+        val v2Client = runBlocking {
+            createV2Client(
+                host = XMTPEnvironment.LOCAL.getUrl(),
+                isSecure = false
+            )
+        }
         val api = GRPCApiClient(
             environment = XMTPEnvironment.LOCAL,
-            secure = false,
+            rustV2Client = v2Client
         )
         api.setAuthToken(authToken)
         val encryptedBundle =
@@ -754,9 +767,15 @@ class LocalInstrumentedTest {
         val identity = PrivateKey.newBuilder().build().generate()
         val authorized = alice.createIdentity(identity)
         val authToken = authorized.createAuthToken()
+        val v2Client = runBlocking {
+            createV2Client(
+                host = XMTPEnvironment.LOCAL.getUrl(),
+                isSecure = false
+            )
+        }
         val api = GRPCApiClient(
             environment = XMTPEnvironment.LOCAL,
-            secure = false,
+            rustV2Client = v2Client
         )
         api.setAuthToken(authToken)
         val encryptedBundle = authorized.toBundle.encrypted(alice)

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

@@ -185,18 +185,18 @@ class FakeApiClient : ApiClient {
         }.build()
     }
 
-    override suspend fun publish(envelopes: List<MessageApiOuterClass.Envelope>): MessageApiOuterClass.PublishResponse {
+    override suspend fun publish(envelopes: List<MessageApiOuterClass.Envelope>) {
         for (envelope in envelopes) {
             send(envelope)
         }
         published.addAll(envelopes)
-        return PublishResponse.newBuilder().build()
+        PublishResponse.newBuilder().build()
     }
 
-    override suspend fun subscribe(request: Flow<MessageApiOuterClass.SubscribeRequest>): Flow<MessageApiOuterClass.Envelope> {
+    override suspend fun subscribe(topics: List<String>): Flow<MessageApiOuterClass.Envelope> {
         val env = stream.counts().first()
 
-        if (request.first().contentTopicsList.contains(env.contentTopic)) {
+        if (topics.contains(env.contentTopic)) {
             return flowOf(env)
         }
         return flowOf()