clean up the subscribe2 under the hood

Author: nplasterer

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

@@ -9,10 +9,12 @@ import com.google.crypto.tink.subtle.Base64
 import com.google.gson.GsonBuilder
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.flowOf
 import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.withContext
 import org.web3j.crypto.Keys
 import org.web3j.crypto.Keys.toChecksumAddress
+import org.xmtp.android.library.GRPCApiClient.Companion.makeSubscribeRequest
 import org.xmtp.android.library.codecs.ContentCodec
 import org.xmtp.android.library.codecs.TextCodec
 import org.xmtp.android.library.libxmtp.XMTPLogger
@@ -497,7 +499,10 @@ class Client() {
         return apiClient.batchQuery(requests)
     }
 
-    suspend fun subscribe(request: Flow<MessageApiOuterClass.SubscribeRequest>): Flow<Envelope> {
+    suspend fun subscribe(topics: List<String>): Flow<Envelope> {
+        return subscribe2(flowOf(makeSubscribeRequest(topics)))
+    }
+    suspend fun subscribe2(request: Flow<MessageApiOuterClass.SubscribeRequest>): Flow<Envelope> {
         return apiClient.subscribe(request = request)
     }
 

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

@@ -2,13 +2,13 @@ package org.xmtp.android.library
 
 import android.util.Log
 import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.runBlocking
 import org.web3j.crypto.Hash
 import org.xmtp.android.library.codecs.ContentCodec
 import org.xmtp.android.library.codecs.EncodedContent
 import org.xmtp.android.library.codecs.compress
+import org.xmtp.android.library.messages.DecryptedMessage
 import org.xmtp.android.library.messages.Envelope
 import org.xmtp.android.library.messages.EnvelopeBuilder
 import org.xmtp.android.library.messages.Message
@@ -23,7 +23,6 @@ import org.xmtp.android.library.messages.sentAt
 import org.xmtp.android.library.messages.toPublicKeyBundle
 import org.xmtp.android.library.messages.walletAddress
 import org.xmtp.proto.message.api.v1.MessageApiOuterClass
-import org.xmtp.android.library.messages.DecryptedMessage
 import java.util.Date
 
 data class ConversationV1(
@@ -42,13 +41,7 @@ data class ConversationV1(
      * @see Conversations.streamAllMessages
      */
     fun streamMessages(): Flow<DecodedMessage> = flow {
-        client.subscribe(
-            MutableStateFlow(
-                GRPCApiClient.makeSubscribeRequest(
-                    listOf(topic.description)
-                )
-            )
-        ).collect {
+        client.subscribe(listOf(topic.description)).collect {
             emit(decode(envelope = it))
         }
     }
@@ -279,25 +272,13 @@ data class ConversationV1(
         get() = topic.description.replace("/xmtp/0/dm-", "/xmtp/0/dmE-")
 
     fun streamEphemeral(): Flow<Envelope> = flow {
-        client.subscribe(
-            MutableStateFlow(
-                GRPCApiClient.makeSubscribeRequest(
-                    listOf(ephemeralTopic)
-                )
-            )
-        ).collect {
+        client.subscribe(listOf(ephemeralTopic)).collect {
             emit(it)
         }
     }
 
     fun streamDecryptedMessages(): Flow<DecryptedMessage> = flow {
-        client.subscribe(
-            MutableStateFlow(
-                GRPCApiClient.makeSubscribeRequest(
-                    listOf(topic.description)
-                )
-            )
-        ).collect {
+        client.subscribe(listOf(topic.description)).collect {
             emit(decrypt(envelope = it))
         }
     }

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

@@ -2,14 +2,14 @@ package org.xmtp.android.library
 
 import android.util.Log
 import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.flow.mapNotNull
 import kotlinx.coroutines.runBlocking
 import org.web3j.crypto.Hash
 import org.xmtp.android.library.codecs.ContentCodec
 import org.xmtp.android.library.codecs.EncodedContent
 import org.xmtp.android.library.codecs.compress
+import org.xmtp.android.library.messages.DecryptedMessage
 import org.xmtp.android.library.messages.Envelope
 import org.xmtp.android.library.messages.EnvelopeBuilder
 import org.xmtp.android.library.messages.Message
@@ -22,7 +22,6 @@ import org.xmtp.android.library.messages.getPublicKeyBundle
 import org.xmtp.android.library.messages.walletAddress
 import org.xmtp.proto.message.api.v1.MessageApiOuterClass
 import org.xmtp.proto.message.contents.Invitation
-import org.xmtp.android.library.messages.DecryptedMessage
 import java.util.Date
 
 data class ConversationV2(
@@ -139,13 +138,7 @@ data class ConversationV2(
     }
 
     fun streamMessages(): Flow<DecodedMessage> = flow {
-        client.subscribe(
-            MutableStateFlow(
-                GRPCApiClient.makeSubscribeRequest(
-                    listOf(topic)
-                )
-            )
-        ).mapNotNull { decodeEnvelopeOrNull(envelope = it) }.collect {
+        client.subscribe(listOf(topic)).mapNotNull { decodeEnvelopeOrNull(envelope = it) }.collect {
             emit(it)
         }
     }
@@ -276,25 +269,13 @@ data class ConversationV2(
         get() = topic.replace("/xmtp/0/m", "/xmtp/0/mE")
 
     fun streamEphemeral(): Flow<Envelope> = flow {
-        client.subscribe(
-            MutableStateFlow(
-                GRPCApiClient.makeSubscribeRequest(
-                    listOf(ephemeralTopic)
-                )
-            )
-        ).collect {
+        client.subscribe(listOf(ephemeralTopic)).collect {
             emit(it)
         }
     }
 
     fun streamDecryptedMessages(): Flow<DecryptedMessage> = flow {
-        client.subscribe(
-            MutableStateFlow(
-                GRPCApiClient.makeSubscribeRequest(
-                    listOf(topic)
-                )
-            )
-        ).collect {
+        client.subscribe(listOf(topic)).collect {
             emit(decrypt(envelope = it))
         }
     }

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

@@ -527,14 +527,10 @@ data class Conversations(
     fun stream(): Flow<Conversation> = flow {
         val streamedConversationTopics: MutableSet<String> = mutableSetOf()
         client.subscribe(
-            MutableStateFlow(
-                makeSubscribeRequest(
-                    listOf(
-                        Topic.userIntro(client.address).description,
-                        Topic.userInvite(client.address).description
-                    )
-                )
-            ),
+            listOf(
+                Topic.userIntro(client.address).description,
+                Topic.userInvite(client.address).description
+            )
         ).collect { envelope ->
             if (envelope.contentTopic == Topic.userIntro(client.address).description) {
                 val conversationV1 = fromIntro(envelope = envelope)
@@ -621,7 +617,7 @@ data class Conversations(
 
         while (true) {
             try {
-                client.subscribe(request = subscribeFlow).collect { envelope ->
+                client.subscribe2(request = subscribeFlow).collect { envelope ->
                     when {
                         conversationsByTopic.containsKey(envelope.contentTopic) -> {
                             val conversation = conversationsByTopic[envelope.contentTopic]
@@ -692,7 +688,7 @@ data class Conversations(
 
         while (true) {
             try {
-                client.subscribe(request = subscribeFlow).collect { envelope ->
+                client.subscribe2(request = subscribeFlow).collect { envelope ->
                     when {
                         conversationsByTopic.containsKey(envelope.contentTopic) -> {
                             val conversation = conversationsByTopic[envelope.contentTopic]