Fix stream timeouts (#228)

* add a test and don't kill the stream prematurely

* get it working even up to 10 minutes

Author: nplasterer

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

@@ -6,7 +6,6 @@ import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
 import org.junit.Assert.assertEquals
-import org.junit.Ignore
 import org.junit.Test
 import org.junit.runner.RunWith
 import org.xmtp.android.library.codecs.TextCodec
@@ -79,15 +78,15 @@ class ConversationsTest {
     }
 
     @Test
-    @Ignore("CI Issues")
     fun testStreamAllMessages() {
         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 = runBlocking { boClient.conversations.newConversation(alixClient.address) }
+        val boConversation =
+            runBlocking { boClient.conversations.newConversation(alixClient.address) }
 
         // Record message stream across all conversations
         val allMessages = mutableListOf<DecodedMessage>()
@@ -97,7 +96,8 @@ class ConversationsTest {
                 alixClient.conversations.streamAllMessages().collect { message ->
                     allMessages.add(message)
                 }
-            } catch (e: Exception) {}
+            } catch (e: Exception) {
+            }
         }
         sleep(2500)
 
@@ -109,7 +109,8 @@ class ConversationsTest {
 
         val caro = PrivateKeyBuilder()
         val caroClient = Client().create(caro, clientOptions)
-        val caroConversation = runBlocking { caroClient.conversations.newConversation(alixClient.address) }
+        val caroConversation =
+            runBlocking { caroClient.conversations.newConversation(alixClient.address) }
         sleep(2500)
 
         for (i in 0 until 5) {
@@ -138,4 +139,37 @@ class ConversationsTest {
 
         assertEquals(allMessages.size, 15)
     }
+
+    @Test
+    fun testStreamTimeOutsAllMessages() {
+        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 =
+            runBlocking { 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)
+
+        runBlocking { boConversation.send(text = "first message") }
+        sleep(2000)
+        assertEquals(allMessages.size, 1)
+        sleep(121000)
+        runBlocking { boConversation.send(text = "second message") }
+        sleep(2000)
+        assertEquals(allMessages.size, 2)
+    }
 }

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

@@ -95,7 +95,6 @@ data class GRPCApiClient(
                     "backoffMultiplier" to 2.0,
                     "retryableStatusCodes" to listOf(
                         "UNAVAILABLE",
-                        "CANCELLED",
                     )
                 )
             )
@@ -107,7 +106,7 @@ data class GRPCApiClient(
             environment.getValue(),
             if (environment == XMTPEnvironment.LOCAL) 5556 else 443
         ).apply {
-            keepAliveTime(30L, TimeUnit.SECONDS)
+            keepAliveTime(3, TimeUnit.MINUTES)
             keepAliveTimeout(20L, TimeUnit.SECONDS)
             keepAliveWithoutCalls(true)
             if (environment != XMTPEnvironment.LOCAL) {