attempt at a few different stream techiniques

Author: nplasterer

dev/local/docker-compose.yml

@@ -23,6 +23,11 @@ services:
     image: xmtp/mls-validation-service:latest
     platform: linux/amd64
 
+  db:
+    image: postgres:13
+    environment:
+      POSTGRES_PASSWORD: xmtp
+
   mlsdb:
     image: postgres:13
     environment:

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

@@ -4,6 +4,8 @@ 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.flow.collect
+import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
 import org.junit.Assert.assertEquals
 import org.junit.Assert.assertThrows
@@ -18,6 +20,7 @@ import org.xmtp.android.library.codecs.ReactionSchema
 import org.xmtp.android.library.messages.PrivateKey
 import org.xmtp.android.library.messages.PrivateKeyBuilder
 import org.xmtp.android.library.messages.walletAddress
+import kotlin.time.Duration
 
 @RunWith(AndroidJUnit4::class)
 class GroupTest {
@@ -212,6 +215,7 @@ class GroupTest {
         }
     }
 
+    @OptIn(ExperimentalCoroutinesApi::class)
     @Test
     fun testCanStreamGroups() = kotlinx.coroutines.test.runTest {
         boClient.conversations.streamGroups().test {

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

@@ -24,17 +24,22 @@ import org.xmtp.android.library.messages.Topic
 import org.xmtp.android.library.messages.createDeterministic
 import org.xmtp.android.library.messages.decrypt
 import org.xmtp.android.library.messages.getInvitation
+import org.xmtp.android.library.messages.getPublicKeyBundle
 import org.xmtp.android.library.messages.header
 import org.xmtp.android.library.messages.involves
 import org.xmtp.android.library.messages.recipientAddress
 import org.xmtp.android.library.messages.senderAddress
 import org.xmtp.android.library.messages.sentAt
+import org.xmtp.android.library.messages.toPublicKeyBundle
 import org.xmtp.android.library.messages.toSignedPublicKeyBundle
+import org.xmtp.android.library.messages.toV2
 import org.xmtp.android.library.messages.walletAddress
 import org.xmtp.proto.keystore.api.v1.Keystore.TopicMap.TopicData
 import org.xmtp.proto.message.contents.Contact
 import org.xmtp.proto.message.contents.Invitation
+import uniffi.xmtpv3.FfiConversationCallback
 import uniffi.xmtpv3.FfiConversations
+import uniffi.xmtpv3.FfiGroup
 import uniffi.xmtpv3.FfiListConversationsOptions
 import uniffi.xmtpv3.org.xmtp.android.library.libxmtp.GroupEmitter
 import java.util.Date

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

@@ -1,20 +1,21 @@
 package org.xmtp.android.library
 
-import kotlinx.coroutines.coroutineScope
+import android.util.Log
+import kotlinx.coroutines.channels.awaitClose
 import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.flow
-import kotlinx.coroutines.launch
+import kotlinx.coroutines.flow.callbackFlow
 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
 import uniffi.xmtpv3.FfiGroup
 import uniffi.xmtpv3.FfiListMessagesOptions
+import uniffi.xmtpv3.FfiMessage
+import uniffi.xmtpv3.FfiMessageCallback
 import java.util.Date
 import kotlin.time.Duration.Companion.nanoseconds
 import kotlin.time.DurationUnit
@@ -142,31 +143,33 @@ 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())
-                }
+    fun streamMessages(): Flow<DecodedMessage> = callbackFlow {
+        val messageCallback = object : FfiMessageCallback {
+            override fun onMessage(message: FfiMessage) {
+                Log.e(
+                    "LOPI",
+                    "INFO Callback - Message callback with ID: " + message.id.toHex() + ", members: " + message.addrFrom
+                )
+                trySend(Message(client, message).decode())
             }
         }
 
-        libXMTPGroup.stream(messageEmitter.callback)
+        val stream = libXMTPGroup.stream(messageCallback)
+        awaitClose { stream.end() }
     }
 
-    fun streamDecryptedMessages(): Flow<DecryptedMessage> = flow {
-        val messageEmitter = MessageEmitter()
-
-        coroutineScope {
-            launch {
-                messageEmitter.messages.collect { message ->
-                    emit(decrypt(Message(client, message)))
-                }
+    fun streamDecryptedMessages(): Flow<DecryptedMessage> = callbackFlow {
+        val messageCallback = object : FfiMessageCallback {
+            override fun onMessage(message: FfiMessage) {
+                Log.e(
+                    "LOPI",
+                    "INFO Callback - Message callback with ID: " + message.id.toHex() + ", members: " + message.addrFrom
+                )
+                trySend(decrypt(Message(client, message)))
             }
         }
 
-        libXMTPGroup.stream(messageEmitter.callback)
+        val stream = libXMTPGroup.stream(messageCallback)
+        awaitClose { stream.end() }
     }
 }