Fix: Stream All Messages not correctly continuing flow

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

@@ -1,6 +1,10 @@
 package org.xmtp.android.library
 
 import androidx.test.ext.junit.runners.AndroidJUnit4
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
 import org.junit.Assert.assertEquals
 import org.junit.Test
 import org.junit.runner.RunWith
@@ -16,6 +20,7 @@ import org.xmtp.android.library.messages.createDeterministic
 import org.xmtp.android.library.messages.getPublicKeyBundle
 import org.xmtp.android.library.messages.toPublicKeyBundle
 import org.xmtp.android.library.messages.walletAddress
+import java.lang.Thread.sleep
 import java.util.Date
 
 @RunWith(AndroidJUnit4::class)
@@ -71,4 +76,65 @@ class ConversationsTest {
         assertEquals(conversation.peerAddress, newWallet.address)
         assertEquals(conversation.createdAt.time, created.time)
     }
+
+    @Test
+    fun testStreamAllMessages() = runBlocking {
+        val bo = PrivateKeyBuilder()
+        val alix = PrivateKeyBuilder()
+        val clientOptions =
+            ClientOptions(api = ClientOptions.Api(env = XMTPEnvironment.LOCAL, isSecure = false))
+        val boClient = Client().create(bo, clientOptions)
+        val alixClient = Client().create(alix, clientOptions)
+        val boConversation = boClient.conversations.newConversation(alixClient.address)
+
+        // Record message stream across all conversations
+        val allMessages = mutableListOf<DecodedMessage>()
+
+        val job = CoroutineScope(Dispatchers.IO).launch {
+            try {
+                alixClient.conversations.streamAllMessages().collect { message ->
+                    allMessages.add(message)
+                }
+            } catch (e: Exception) {}
+        }
+        sleep(2500)
+
+        for (i in 0 until 5) {
+            boConversation.send(text = "Message $i")
+            sleep(1000)
+        }
+        assertEquals(allMessages.size, 5)
+
+        val caro = PrivateKeyBuilder()
+        val caroClient = Client().create(caro, clientOptions)
+        val caroConversation = caroClient.conversations.newConversation(alixClient.address)
+
+        sleep(2500)
+
+        for (i in 0 until 5) {
+            caroConversation.send(text = "Message $i")
+            sleep(1000)
+        }
+
+        assertEquals(allMessages.size, 10)
+
+        job.cancel()
+
+        CoroutineScope(Dispatchers.IO).launch {
+            try {
+                alixClient.conversations.streamAllMessages().collect { message ->
+                    allMessages.add(message)
+                }
+            } catch (e: Exception) {
+            }
+        }
+        sleep(2500)
+
+        for (i in 0 until 5) {
+            boConversation.send(text = "Message $i")
+            sleep(1000)
+        }
+
+        assertEquals(allMessages.size, 15)
+    }
 }

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

@@ -1,8 +1,10 @@
 package org.xmtp.android.library
 
 import android.util.Log
+import kotlinx.coroutines.currentCoroutineContext
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.job
 import kotlinx.coroutines.runBlocking
 import org.xmtp.android.library.GRPCApiClient.Companion.makeQueryRequest
 import org.xmtp.android.library.messages.Envelope
@@ -374,24 +376,26 @@ data class Conversations(
                         }
 
                         envelope.contentTopic.startsWith("/xmtp/0/invite-") -> {
-                            val conversation = fromInvite(envelope)
+                            val conversation = fromInvite(envelope = envelope)
                             conversationsByTopic[conversation.topic] = conversation
-                            // Break so we can resubscribe with the new conversation
-                            return@collect
+                            topics.add(conversation.topic)
+                            currentCoroutineContext().job.cancel()
                         }
 
                         envelope.contentTopic.startsWith("/xmtp/0/intro-") -> {
-                            val conversation = fromIntro(envelope)
+                            val conversation = fromIntro(envelope = envelope)
                             conversationsByTopic[conversation.topic] = conversation
                             val decoded = conversation.decode(envelope)
                             emit(decoded)
-                            // Break so we can resubscribe with the new conversation
-                            return@collect
+                            topics.add(conversation.topic)
+                            currentCoroutineContext().job.cancel()
                         }
+
+                        else -> {}
                     }
                 }
             } catch (error: Exception) {
-                throw error
+                continue
             }
         }
     }