fix lots of the threading issues

nplasterer · nplasterer · commit 5bed13605fcc · 2024-03-12T20:25:33.000-07:00
diff --git a/library/src/androidTest/java/org/xmtp/android/library/ConversationTest.kt b/library/src/androidTest/java/org/xmtp/android/library/ConversationTest.kt
@@ -109,28 +109,30 @@ class ConversationTest {
         // Overwrite contact as legacy
         bobClient.publishUserContact(legacy = true)
         aliceClient.publishUserContact(legacy = true)
-        bobClient.publish(
-            envelopes = listOf(
-                EnvelopeBuilder.buildFromTopic(
-                    topic = Topic.userIntro(bob.walletAddress),
-                    timestamp = someTimeAgo,
-                    message = MessageBuilder.buildFromMessageV1(v1 = messageV1).toByteArray(),
-                ),
-                EnvelopeBuilder.buildFromTopic(
-                    topic = Topic.userIntro(alice.walletAddress),
-                    timestamp = someTimeAgo,
-                    message = MessageBuilder.buildFromMessageV1(v1 = messageV1).toByteArray(),
-                ),
-                EnvelopeBuilder.buildFromTopic(
-                    topic = Topic.directMessageV1(
-                        bob.walletAddress,
-                        alice.walletAddress,
+        runBlocking {
+            bobClient.publish(
+                envelopes = listOf(
+                    EnvelopeBuilder.buildFromTopic(
+                        topic = Topic.userIntro(bob.walletAddress),
+                        timestamp = someTimeAgo,
+                        message = MessageBuilder.buildFromMessageV1(v1 = messageV1).toByteArray(),
+                    ),
+                    EnvelopeBuilder.buildFromTopic(
+                        topic = Topic.userIntro(alice.walletAddress),
+                        timestamp = someTimeAgo,
+                        message = MessageBuilder.buildFromMessageV1(v1 = messageV1).toByteArray(),
+                    ),
+                    EnvelopeBuilder.buildFromTopic(
+                        topic = Topic.directMessageV1(
+                            bob.walletAddress,
+                            alice.walletAddress,
+                        ),
+                        timestamp = someTimeAgo,
+                        message = MessageBuilder.buildFromMessageV1(v1 = messageV1).toByteArray(),
                     ),
-                    timestamp = someTimeAgo,
-                    message = MessageBuilder.buildFromMessageV1(v1 = messageV1).toByteArray(),
                 ),
-            ),
-        )
+            )
+        }
         var conversation = aliceClient.conversations.newConversation(bob.walletAddress)
         assertEquals(conversation.peerAddress, bob.walletAddress)
         assertEquals(conversation.createdAt, someTimeAgo)
@@ -251,7 +253,7 @@ class ConversationTest {
             timestamp = Date(),
             message = MessageBuilder.buildFromMessageV2(v2 = tamperedMessage.messageV2).toByteArray(),
         )
-        aliceClient.publish(envelopes = listOf(tamperedEnvelope))
+        runBlocking { aliceClient.publish(envelopes = listOf(tamperedEnvelope)) }
         val bobConversation = bobClient.conversations.newConversation(
             aliceWallet.address,
             InvitationV1ContextBuilder.buildFromConversation("hi"),
@@ -370,7 +372,7 @@ class ConversationTest {
             ConversationV2.create(client = client, invitation = invitationv1, header = header)
         assertEquals(fakeContactWallet.address, conversation.peerAddress)
 
-        conversation.send(content = "hello world")
+        runBlocking { conversation.send(content = "hello world") }
 
         val conversationList = client.conversations.list()
         val recipientConversation = conversationList.lastOrNull()
@@ -646,7 +648,7 @@ class ConversationTest {
         assertEquals(conversation.version, Conversation.Version.V1)
         val preparedMessage = conversation.prepareMessage(content = "hi")
         val messageID = preparedMessage.messageId
-        conversation.send(prepared = preparedMessage)
+        runBlocking { conversation.send(prepared = preparedMessage) }
         val messages = conversation.messages()
         val message = messages[0]
         assertEquals("hi", message.body)
@@ -658,7 +660,7 @@ class ConversationTest {
         val conversation = aliceClient.conversations.newConversation(bob.walletAddress)
         val preparedMessage = conversation.prepareMessage(content = "hi")
         val messageID = preparedMessage.messageId
-        conversation.send(prepared = preparedMessage)
+        runBlocking { conversation.send(prepared = preparedMessage) }
         val messages = conversation.messages()
         val message = messages[0]
         assertEquals("hi", message.body)
@@ -673,7 +675,7 @@ class ConversationTest {
 
         // This does not need the `conversation` to `.publish` the message.
         // This simulates a background task publishing all pending messages upon connection.
-        aliceClient.publish(envelopes = preparedMessage.envelopes)
+        runBlocking { aliceClient.publish(envelopes = preparedMessage.envelopes) }
 
         val messages = conversation.messages()
         val message = messages[0]
diff --git a/library/src/androidTest/java/org/xmtp/android/library/GroupTest.kt b/library/src/androidTest/java/org/xmtp/android/library/GroupTest.kt
@@ -319,7 +319,12 @@ class GroupTest {
             schema = ReactionSchema.Unicode
         )
 
-        runBlocking { group.send(content = reaction, options = SendOptions(contentType = ContentTypeReaction)) }
+        runBlocking {
+            group.send(
+                content = reaction,
+                options = SendOptions(contentType = ContentTypeReaction)
+            )
+        }
         runBlocking { group.sync() }
 
         val messages = group.messages()
@@ -348,25 +353,47 @@ class GroupTest {
     fun testCanStreamAllGroupMessages() = kotlinx.coroutines.test.runTest {
         val group = caroClient.conversations.newGroup(listOf(alix.walletAddress))
         alixClient.conversations.syncGroups()
-        alixClient.conversations.streamAllGroupMessages().test {
-            group.send("hi")
-            assertEquals("hi", awaitItem().encodedContent.content.toStringUtf8())
-            group.send("hi again")
-            assertEquals("hi again", awaitItem().encodedContent.content.toStringUtf8())
+        val flow = alixClient.conversations.streamAllGroupMessages()
+        var counter = 0
+        val job = launch {
+            flow.catch { e ->
+                throw Exception("Error collecting flow: $e")
+            }.collect { message ->
+                counter++
+                assertEquals("hi $counter", message.encodedContent.content.toStringUtf8())
+                if (counter == 2) this.cancel()
+            }
         }
+
+        group.send("hi 1")
+        group.send("hi 2")
+
+        job.join()
     }
 
     @Test
     fun testCanStreamAllMessages() = kotlinx.coroutines.test.runTest {
         val group = caroClient.conversations.newGroup(listOf(alix.walletAddress))
         val conversation = boClient.conversations.newConversation(alix.walletAddress)
         alixClient.conversations.syncGroups()
-        alixClient.conversations.streamAllMessages(includeGroups = true).test {
-            group.send("hi")
-            assertEquals("hi", awaitItem().encodedContent.content.toStringUtf8())
-            conversation.send("hi again")
-            assertEquals("hi again", awaitItem().encodedContent.content.toStringUtf8())
+
+        val flow = alixClient.conversations.streamAllMessages(includeGroups = true)
+        var counter = 0
+        val job = launch {
+            flow.catch { e ->
+                throw Exception("Error collecting flow: $e")
+            }.collect { message ->
+                counter++
+                assertEquals("hi $counter", message.encodedContent.content.toStringUtf8())
+                if (counter == 2) this.cancel()
+            }
         }
+
+        group.send("hi 1")
+        Thread.sleep(1000)
+        conversation.send("hi 2")
+
+        job.join()
     }
 
     @Test
@@ -410,12 +437,24 @@ class GroupTest {
         val group = caroClient.conversations.newGroup(listOf(alix.walletAddress))
         val conversation = boClient.conversations.newConversation(alix.walletAddress)
         alixClient.conversations.syncGroups()
-        alixClient.conversations.streamAllDecryptedMessages(includeGroups = true).test {
-            group.send("hi")
-            assertEquals("hi", awaitItem().encodedContent.content.toStringUtf8())
-            conversation.send("hi again")
-            assertEquals("hi again", awaitItem().encodedContent.content.toStringUtf8())
+
+        val flow = alixClient.conversations.streamAllDecryptedMessages(includeGroups = true)
+        var counter = 0
+        val job = launch {
+            flow.catch { e ->
+                throw Exception("Error collecting flow: $e")
+            }.collect { message ->
+                counter++
+                assertEquals("hi $counter", message.encodedContent.content.toStringUtf8())
+                if (counter == 2) this.cancel()
+            }
         }
+
+        group.send("hi 1")
+        Thread.sleep(1000)
+        conversation.send("hi 2")
+
+        job.join()
     }
 
     @Test
diff --git a/library/src/androidTest/java/org/xmtp/android/library/MessageTest.kt b/library/src/androidTest/java/org/xmtp/android/library/MessageTest.kt
@@ -245,7 +245,7 @@ class MessageTest {
             peerAddress = "0xf4BF19Ed562651837bc11ff975472ABd239D35B5",
             sentAt = Date(),
         )
-        convo.send(text = "hello from kotlin")
+        runBlocking { convo.send(text = "hello from kotlin") }
         val messages = convo.messages()
         assertEquals(1, messages.size)
         assertEquals("hello from kotlin", messages[0].body)
diff --git a/library/src/androidTest/java/org/xmtp/android/library/TestHelpers.kt b/library/src/androidTest/java/org/xmtp/android/library/TestHelpers.kt
@@ -4,6 +4,7 @@ import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.flowOf
+import kotlinx.coroutines.runBlocking
 import org.junit.Assert.assertEquals
 import org.xmtp.android.library.codecs.Fetcher
 import org.xmtp.android.library.messages.ContactBundle
@@ -242,7 +243,7 @@ data class Fixtures(
             message = contactBundle.toByteString()
         }.build()
 
-        client.publish(envelopes = listOf(envelope))
+        runBlocking { client.publish(envelopes = listOf(envelope)) }
     }
 }
 
diff --git a/library/src/main/java/org/xmtp/android/library/Client.kt b/library/src/main/java/org/xmtp/android/library/Client.kt
@@ -497,7 +497,7 @@ class Client() {
         }
     }
 
-    fun publish(envelopes: List<Envelope>): PublishResponse {
+    suspend fun publish(envelopes: List<Envelope>): PublishResponse {
         val authorized = AuthorizedIdentity(
             address = address,
             authorized = privateKeyBundleV1.identityKey.publicKey,
@@ -506,7 +506,7 @@ class Client() {
         val authToken = authorized.createAuthToken()
         apiClient.setAuthToken(authToken)
 
-        return runBlocking { apiClient.publish(envelopes = envelopes) }
+        return apiClient.publish(envelopes = envelopes)
     }
 
     fun ensureUserContactPublished() {
diff --git a/library/src/main/java/org/xmtp/android/library/Contacts.kt b/library/src/main/java/org/xmtp/android/library/Contacts.kt
@@ -152,7 +152,7 @@ class ConsentList(val client: Client) {
                 ByteArray(message.size) { message[it] },
             )
 
-        client.publish(listOf(envelope))
+        runBlocking { client.publish(listOf(envelope)) }
     }
 
     fun allow(address: String): ConsentListEntry {
diff --git a/library/src/main/java/org/xmtp/android/library/Conversation.kt b/library/src/main/java/org/xmtp/android/library/Conversation.kt
@@ -169,7 +169,7 @@ sealed class Conversation {
         }
     }
 
-    fun send(prepared: PreparedMessage): String {
+    suspend fun send(prepared: PreparedMessage): String {
         return when (this) {
             is V1 -> conversationV1.send(prepared = prepared)
             is V2 -> conversationV2.send(prepared = prepared)
diff --git a/library/src/main/java/org/xmtp/android/library/ConversationV1.kt b/library/src/main/java/org/xmtp/android/library/ConversationV1.kt
@@ -164,11 +164,11 @@ data class ConversationV1(
         }
     }
 
-    fun send(text: String, options: SendOptions? = null): String {
+    suspend fun send(text: String, options: SendOptions? = null): String {
         return send(text = text, sendOptions = options, sentAt = null)
     }
 
-    internal fun send(
+    internal suspend fun send(
         text: String,
         sendOptions: SendOptions? = null,
         sentAt: Date? = null,
@@ -177,17 +177,17 @@ data class ConversationV1(
         return send(preparedMessage)
     }
 
-    fun <T> send(content: T, options: SendOptions? = null): String {
+    suspend fun <T> send(content: T, options: SendOptions? = null): String {
         val preparedMessage = prepareMessage(content = content, options = options)
         return send(preparedMessage)
     }
 
-    fun send(encodedContent: EncodedContent, options: SendOptions? = null): String {
+    suspend fun send(encodedContent: EncodedContent, options: SendOptions? = null): String {
         val preparedMessage = prepareMessage(encodedContent = encodedContent, options = options)
         return send(preparedMessage)
     }
 
-    fun send(prepared: PreparedMessage): String {
+    suspend fun send(prepared: PreparedMessage): String {
         client.publish(envelopes = prepared.envelopes)
         if (client.contacts.consentList.state(address = peerAddress) == ConsentState.UNKNOWN) {
             client.contacts.allow(addresses = listOf(peerAddress))
diff --git a/library/src/main/java/org/xmtp/android/library/ConversationV2.kt b/library/src/main/java/org/xmtp/android/library/ConversationV2.kt
@@ -173,22 +173,22 @@ data class ConversationV2(
         }
     }
 
-    fun <T> send(content: T, options: SendOptions? = null): String {
+    suspend fun <T> send(content: T, options: SendOptions? = null): String {
         val preparedMessage = prepareMessage(content = content, options = options)
         return send(preparedMessage)
     }
 
-    fun send(text: String, options: SendOptions? = null, sentAt: Date? = null): String {
+    suspend fun send(text: String, options: SendOptions? = null, sentAt: Date? = null): String {
         val preparedMessage = prepareMessage(content = text, options = options)
         return send(preparedMessage)
     }
 
-    fun send(encodedContent: EncodedContent, options: SendOptions?): String {
+    suspend fun send(encodedContent: EncodedContent, options: SendOptions?): String {
         val preparedMessage = prepareMessage(encodedContent = encodedContent, options = options)
         return send(preparedMessage)
     }
 
-    fun send(prepared: PreparedMessage): String {
+    suspend fun send(prepared: PreparedMessage): String {
         client.publish(envelopes = prepared.envelopes)
         if (client.contacts.consentList.state(address = peerAddress) == ConsentState.UNKNOWN) {
             client.contacts.allow(addresses = listOf(peerAddress))

Original file line number	Diff line number	Diff line change
`@@ -245,7 +245,7 @@ class MessageTest {`
`245`	`245`	`peerAddress = "0xf4BF19Ed562651837bc11ff975472ABd239D35B5",`
`246`	`246`	`sentAt = Date(),`
`247`	`247`	`)`
`248`		`- convo.send(text = "hello from kotlin")`
	`248`	`+ runBlocking { convo.send(text = "hello from kotlin") }`
`249`	`249`	`val messages = convo.messages()`
`250`	`250`	`assertEquals(1, messages.size)`
`251`	`251`	`assertEquals("hello from kotlin", messages[0].body)`
Original file line number	Diff line number	Diff line change
`@@ -4,6 +4,7 @@ import kotlinx.coroutines.flow.Flow`
`4`	`4`	`import kotlinx.coroutines.flow.MutableSharedFlow`
`5`	`5`	`import kotlinx.coroutines.flow.first`
`6`	`6`	`import kotlinx.coroutines.flow.flowOf`
	`7`	`+import kotlinx.coroutines.runBlocking`
`7`	`8`	`import org.junit.Assert.assertEquals`
`8`	`9`	`import org.xmtp.android.library.codecs.Fetcher`
`9`	`10`	`import org.xmtp.android.library.messages.ContactBundle`
`@@ -242,7 +243,7 @@ data class Fixtures(`
`242`	`243`	`message = contactBundle.toByteString()`
`243`	`244`	`}.build()`
`244`	`245`
`245`		`- client.publish(envelopes = listOf(envelope))`
	`246`	`+ runBlocking { client.publish(envelopes = listOf(envelope)) }`
`246`	`247`	`}`
`247`	`248`	`}`
`248`	`249`
Original file line number	Diff line number	Diff line change
`@@ -497,7 +497,7 @@ class Client() {`
`497`	`497`	`}`
`498`	`498`	`}`
`499`	`499`
`500`		`- fun publish(envelopes: List<Envelope>): PublishResponse {`
	`500`	`+ suspend fun publish(envelopes: List<Envelope>): PublishResponse {`
`501`	`501`	`val authorized = AuthorizedIdentity(`
`502`	`502`	`address = address,`
`503`	`503`	`authorized = privateKeyBundleV1.identityKey.publicKey,`
`@@ -506,7 +506,7 @@ class Client() {`
`506`	`506`	`val authToken = authorized.createAuthToken()`
`507`	`507`	`apiClient.setAuthToken(authToken)`
`508`	`508`
`509`		`- return runBlocking { apiClient.publish(envelopes = envelopes) }`
	`509`	`+ return apiClient.publish(envelopes = envelopes)`
`510`	`510`	`}`
`511`	`511`
`512`	`512`	`fun ensureUserContactPublished() {`
Original file line number	Diff line number	Diff line change
`@@ -152,7 +152,7 @@ class ConsentList(val client: Client) {`
`152`	`152`	`ByteArray(message.size) { message[it] },`
`153`	`153`	`)`
`154`	`154`
`155`		`- client.publish(listOf(envelope))`
	`155`	`+ runBlocking { client.publish(listOf(envelope)) }`
`156`	`156`	`}`
`157`	`157`
`158`	`158`	`fun allow(address: String): ConsentListEntry {`
Original file line number	Diff line number	Diff line change
`@@ -169,7 +169,7 @@ sealed class Conversation {`
`169`	`169`	`}`
`170`	`170`	`}`
`171`	`171`
`172`		`- fun send(prepared: PreparedMessage): String {`
	`172`	`+ suspend fun send(prepared: PreparedMessage): String {`
`173`	`173`	`return when (this) {`
`174`	`174`	`is V1 -> conversationV1.send(prepared = prepared)`
`175`	`175`	`is V2 -> conversationV2.send(prepared = prepared)`
Original file line number	Diff line number	Diff line change
`@@ -164,11 +164,11 @@ data class ConversationV1(`
`164`	`164`	`}`
`165`	`165`	`}`
`166`	`166`
`167`		`- fun send(text: String, options: SendOptions? = null): String {`
	`167`	`+ suspend fun send(text: String, options: SendOptions? = null): String {`
`168`	`168`	`return send(text = text, sendOptions = options, sentAt = null)`
`169`	`169`	`}`
`170`	`170`
`171`		`- internal fun send(`
	`171`	`+ internal suspend fun send(`
`172`	`172`	`text: String,`
`173`	`173`	`sendOptions: SendOptions? = null,`
`174`	`174`	`sentAt: Date? = null,`
`@@ -177,17 +177,17 @@ data class ConversationV1(`
`177`	`177`	`return send(preparedMessage)`
`178`	`178`	`}`
`179`	`179`
`180`		`- fun <T> send(content: T, options: SendOptions? = null): String {`
	`180`	`+ suspend fun <T> send(content: T, options: SendOptions? = null): String {`
`181`	`181`	`val preparedMessage = prepareMessage(content = content, options = options)`
`182`	`182`	`return send(preparedMessage)`
`183`	`183`	`}`
`184`	`184`
`185`		`- fun send(encodedContent: EncodedContent, options: SendOptions? = null): String {`
	`185`	`+ suspend fun send(encodedContent: EncodedContent, options: SendOptions? = null): String {`
`186`	`186`	`val preparedMessage = prepareMessage(encodedContent = encodedContent, options = options)`
`187`	`187`	`return send(preparedMessage)`
`188`	`188`	`}`
`189`	`189`
`190`		`- fun send(prepared: PreparedMessage): String {`
	`190`	`+ suspend fun send(prepared: PreparedMessage): String {`
`191`	`191`	`client.publish(envelopes = prepared.envelopes)`
`192`	`192`	`if (client.contacts.consentList.state(address = peerAddress) == ConsentState.UNKNOWN) {`
`193`	`193`	`client.contacts.allow(addresses = listOf(peerAddress))`
Original file line number	Diff line number	Diff line change
`@@ -173,22 +173,22 @@ data class ConversationV2(`
`173`	`173`	`}`
`174`	`174`	`}`
`175`	`175`
`176`		`- fun <T> send(content: T, options: SendOptions? = null): String {`
	`176`	`+ suspend fun <T> send(content: T, options: SendOptions? = null): String {`
`177`	`177`	`val preparedMessage = prepareMessage(content = content, options = options)`
`178`	`178`	`return send(preparedMessage)`
`179`	`179`	`}`
`180`	`180`
`181`		`- fun send(text: String, options: SendOptions? = null, sentAt: Date? = null): String {`
	`181`	`+ suspend fun send(text: String, options: SendOptions? = null, sentAt: Date? = null): String {`
`182`	`182`	`val preparedMessage = prepareMessage(content = text, options = options)`
`183`	`183`	`return send(preparedMessage)`
`184`	`184`	`}`
`185`	`185`
`186`		`- fun send(encodedContent: EncodedContent, options: SendOptions?): String {`
	`186`	`+ suspend fun send(encodedContent: EncodedContent, options: SendOptions?): String {`
`187`	`187`	`val preparedMessage = prepareMessage(encodedContent = encodedContent, options = options)`
`188`	`188`	`return send(preparedMessage)`
`189`	`189`	`}`
`190`	`190`
`191`		`- fun send(prepared: PreparedMessage): String {`
	`191`	`+ suspend fun send(prepared: PreparedMessage): String {`
`192`	`192`	`client.publish(envelopes = prepared.envelopes)`
`193`	`193`	`if (client.contacts.consentList.state(address = peerAddress) == ConsentState.UNKNOWN) {`
`194`	`194`	`client.contacts.allow(addresses = listOf(peerAddress))`