some subscribe tweaks

Author: nplasterer

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

@@ -4,6 +4,9 @@ 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 +591,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

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

@@ -149,39 +149,23 @@ data class GRPCApiClient(
         rustV2Client.publish(request = request, authToken = authToken ?: "")
     }
 
-    override suspend fun subscribe(topics: List<String>): Flow<Envelope> = callbackFlow {
-        val request = SubscribeRequest.newBuilder().apply {
-            addAllContentTopics(topics)
-        }.build()
-        val subscription = rustV2Client.subscribe(
-            FfiV2SubscribeRequest(contentTopics = request.contentTopicsList)
-        )
+    override suspend fun subscribe(topics: List<String>): Flow<Envelope> = flow {
         try {
-            // Launch a separate coroutine for subscription
-            val job = launch {
-                try {
-                    // Continuously emit envelopes received from the subscription.
-                    while (true) {
-                        val nextEnvelope = envelopeFromFFi(subscription.next())
-                        trySend(nextEnvelope).isSuccess
-                    }
-                } catch (e: Exception) {
-                    if (e !is CancellationException) {
-                        subscription.close()
-                        throw XMTPException("ApiClientError.subscribeError: ${e.message}", e)
-                    }
+            val subscription = rustV2Client.subscribe(FfiV2SubscribeRequest(topics))
+            try {
+                while (true) {
+                    val nextEnvelope = subscription.next()
+                    emit(envelopeFromFFi(nextEnvelope))
                 }
-            }
-            awaitClose {
-                job.cancel()
-                subscription.close()
+            } catch (e: Exception) {
+                throw e
+            } finally {
+                subscription.end()
             }
         } catch (e: Exception) {
-            subscription.close()
             throw e
         }
     }
-
     override fun close() {
         rustV2Client.close()
     }