remove all the streams work and move to another PR

Author: nplasterer

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

@@ -2,8 +2,6 @@ package org.xmtp.android.library
 
 import androidx.test.ext.junit.runners.AndroidJUnit4
 import androidx.test.platform.app.InstrumentationRegistry
-import app.cash.turbine.test
-import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.runBlocking
 import org.junit.Assert.assertEquals
 import org.junit.Before
@@ -18,7 +16,6 @@ import org.xmtp.android.library.messages.PrivateKey
 import org.xmtp.android.library.messages.PrivateKeyBuilder
 import org.xmtp.android.library.messages.walletAddress
 
-@OptIn(ExperimentalCoroutinesApi::class)
 @RunWith(AndroidJUnit4::class)
 class GroupTest {
     lateinit var fakeApiClient: FakeApiClient
@@ -177,38 +174,4 @@ class GroupTest {
         assertEquals(ReactionAction.Added, content?.action)
         assertEquals(ReactionSchema.Unicode, content?.schema)
     }
-
-    @Test
-    fun testCanStreamGroupMessages() = kotlinx.coroutines.test.runTest {
-        val group = boClient.conversations.newGroup(listOf(alix.walletAddress.lowercase()))
-
-        group.streamMessages().test {
-            group.send("hi")
-            assertEquals("hi", awaitItem().body)
-            awaitComplete()
-        }
-    }
-
-    @Test
-    fun testCanStreamGroups() = kotlinx.coroutines.test.runTest {
-        boClient.conversations.streamGroups().test {
-            val conversation =
-                boClient.conversations.newGroup(listOf(alix.walletAddress.lowercase()))
-            conversation.send(content = "hi")
-            assertEquals("hi", awaitItem().messages().first().body)
-            awaitComplete()
-        }
-    }
-
-    @Test
-    fun testCanStreamGroupsAndConversations() = kotlinx.coroutines.test.runTest {
-        boClient.conversations.stream(includeGroups = true).test {
-            val group =
-                boClient.conversations.newGroup(listOf(alix.walletAddress.lowercase()))
-            val conversation =
-                boClient.conversations.newConversation(alix.walletAddress.lowercase())
-            assertEquals("hi", awaitItem().messages().first().body)
-            awaitComplete()
-        }
-    }
 }

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

@@ -318,15 +318,15 @@ sealed class Conversation {
         return when (this) {
             is V1 -> conversationV1.streamMessages()
             is V2 -> conversationV2.streamMessages()
-            is Group -> group.streamMessages()
+            is Group -> throw XMTPException("Coming follow up PR")
         }
     }
 
     fun streamDecryptedMessages(): Flow<DecryptedMessage> {
         return when (this) {
             is V1 -> conversationV1.streamDecryptedMessages()
             is V2 -> conversationV2.streamDecryptedMessages()
-            is Group -> group.streamDecryptedMessages()
+            is Group -> throw XMTPException("Coming follow up PR")
         }
     }
 

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

@@ -3,11 +3,9 @@ package org.xmtp.android.library
 import android.util.Log
 import io.grpc.StatusException
 import kotlinx.coroutines.CancellationException
-import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.flow
-import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
 import org.xmtp.android.library.GRPCApiClient.Companion.makeQueryRequest
 import org.xmtp.android.library.GRPCApiClient.Companion.makeSubscribeRequest
@@ -36,7 +34,6 @@ import org.xmtp.proto.message.contents.Contact
 import org.xmtp.proto.message.contents.Invitation
 import uniffi.xmtpv3.FfiConversations
 import uniffi.xmtpv3.FfiListConversationsOptions
-import uniffi.xmtpv3.org.xmtp.android.library.libxmtp.GroupEmitter
 import java.util.Date
 import kotlin.time.Duration.Companion.nanoseconds
 import kotlin.time.DurationUnit
@@ -93,8 +90,8 @@ data class Conversations(
         if (accountAddresses.isEmpty()) {
             throw XMTPException("Cannot start an empty group chat.")
         }
-        if (accountAddresses.size == 1 && accountAddresses.first()
-                .lowercase() == client.address.lowercase()
+        if (accountAddresses.size == 1 &&
+            accountAddresses.first().lowercase() == client.address.lowercase()
         ) {
             throw XMTPException("Recipient is sender")
         }
@@ -476,21 +473,7 @@ data class Conversations(
      * of the information of those conversations according to the topics
      * @return Stream of data information for the conversations
      */
-    fun stream(includeGroups: Boolean = false): Flow<Conversation> = flow {
-        if (includeGroups) {
-            val groupEmitter = GroupEmitter()
-
-            coroutineScope {
-                launch {
-                    groupEmitter.groups.collect { group ->
-                        emit(Conversation.Group(Group(client, group)))
-                    }
-                }
-            }
-
-            libXMTPConversations?.stream(groupEmitter.callback)
-        }
-
+    fun stream(): Flow<Conversation> = flow {
         val streamedConversationTopics: MutableSet<String> = mutableSetOf()
         client.subscribeTopic(
             listOf(Topic.userIntro(client.address), Topic.userInvite(client.address)),
@@ -514,20 +497,6 @@ data class Conversations(
         }
     }
 
-    fun streamGroups(): Flow<Group> = flow {
-        val groupEmitter = GroupEmitter()
-
-        coroutineScope {
-            launch {
-                groupEmitter.groups.collect { group ->
-                    emit(Group(client, group))
-                }
-            }
-        }
-
-        libXMTPConversations?.stream(groupEmitter.callback)
-    }
-
     /**
      * Get the stream of all messages of the current [Client]
      * @return Flow object of [DecodedMessage] that represents all the messages of the

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

@@ -1,15 +1,10 @@
 package org.xmtp.android.library
 
-import kotlinx.coroutines.coroutineScope
-import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.flow
-import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
 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.libxmtp.Message
-import org.xmtp.android.library.libxmtp.MessageEmitter
 import org.xmtp.android.library.messages.DecryptedMessage
 import org.xmtp.android.library.messages.PagingInfoSortDirection
 import org.xmtp.proto.message.api.v1.MessageApiOuterClass
@@ -128,34 +123,6 @@ class Group(val client: Client, private val libXMTPGroup: FfiGroup) {
         )
     }
 
-    fun streamMessages(): Flow<DecodedMessage> = flow {
-        val messageEmitter = MessageEmitter()
-
-        coroutineScope {
-            launch {
-                messageEmitter.messages.collect { message ->
-                    emit(Message(client, message).decode())
-                }
-            }
-        }
-
-        libXMTPGroup.stream(messageEmitter.callback)
-    }
-
-    fun streamDecryptedMessages(): Flow<DecryptedMessage> = flow {
-        val messageEmitter = MessageEmitter()
-
-        coroutineScope {
-            launch {
-                messageEmitter.messages.collect { message ->
-                    emit(decrypt(Message(client, message)))
-                }
-            }
-        }
-
-        libXMTPGroup.stream(messageEmitter.callback)
-    }
-
     fun addMembers(addresses: List<String>) {
         runBlocking { libXMTPGroup.addMembers(addresses) }
     }