diff --git a/example/src/main/java/org/xmtp/android/example/MainActivity.kt b/example/src/main/java/org/xmtp/android/example/MainActivity.kt
index e85b5025a..239a36a89 100644
--- a/example/src/main/java/org/xmtp/android/example/MainActivity.kt
+++ b/example/src/main/java/org/xmtp/android/example/MainActivity.kt
@@ -8,6 +8,7 @@ import android.content.Context
 import android.content.Intent
 import android.content.pm.PackageManager
 import android.os.Bundle
+import android.util.Log
 import android.view.Menu
 import android.view.MenuItem
 import android.view.View
@@ -182,6 +183,7 @@ class MainActivity : AppCompatActivity(),
 
     private fun showError(message: String) {
         val error = message.ifBlank { resources.getString(R.string.error) }
+        Log.e("MainActivity", message)
         Toast.makeText(this, error, Toast.LENGTH_SHORT).show()
     }
 
diff --git a/library/build.gradle b/library/build.gradle
index ecb4db931..5504833b2 100644
--- a/library/build.gradle
+++ b/library/build.gradle
@@ -85,15 +85,15 @@ dependencies {
     implementation 'io.grpc:grpc-kotlin-stub:1.4.1'
     implementation 'io.grpc:grpc-okhttp:1.62.2'
     implementation 'io.grpc:grpc-protobuf-lite:1.62.2'
-    implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-core:1.7.3'
+    implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-core:1.8.0'
     implementation 'org.web3j:crypto:5.0.0'
     implementation "net.java.dev.jna:jna:5.14.0@aar"
     api 'com.google.protobuf:protobuf-kotlin-lite:3.22.3'
     api 'org.xmtp:proto-kotlin:3.61.1'
 
     testImplementation 'junit:junit:4.13.2'
-    androidTestImplementation 'app.cash.turbine:turbine:0.12.1'
-    androidTestImplementation 'org.jetbrains.kotlinx:kotlinx-coroutines-test:1.7.3'
+    androidTestImplementation 'app.cash.turbine:turbine:1.1.0'
+    androidTestImplementation 'org.jetbrains.kotlinx:kotlinx-coroutines-test:1.8.0'
     androidTestImplementation 'androidx.test.ext:junit:1.1.5'
     androidTestImplementation 'androidx.test.espresso:espresso-core:3.5.1'
 }
diff --git a/library/src/androidTest/java/org/xmtp/android/library/ClientTest.kt b/library/src/androidTest/java/org/xmtp/android/library/ClientTest.kt
index 702451e68..29970caa5 100644
--- a/library/src/androidTest/java/org/xmtp/android/library/ClientTest.kt
+++ b/library/src/androidTest/java/org/xmtp/android/library/ClientTest.kt
@@ -234,8 +234,8 @@ class ClientTest {
     fun testCanMessage() {
         val fixtures = fixtures()
         val notOnNetwork = PrivateKeyBuilder()
-        val canMessage = fixtures.aliceClient.canMessage(fixtures.bobClient.address)
-        val cannotMessage = fixtures.aliceClient.canMessage(notOnNetwork.address)
+        val canMessage = runBlocking { fixtures.aliceClient.canMessage(fixtures.bobClient.address) }
+        val cannotMessage = runBlocking { fixtures.aliceClient.canMessage(notOnNetwork.address) }
         assert(canMessage)
         assert(!cannotMessage)
     }
@@ -248,8 +248,8 @@ class ClientTest {
         val aliceClient = Client().create(aliceWallet, opts)
         runBlocking { aliceClient.ensureUserContactPublished() }
 
-        val canMessage = Client.canMessage(aliceWallet.address, opts)
-        val cannotMessage = Client.canMessage(notOnNetwork.address, opts)
+        val canMessage = runBlocking { Client.canMessage(aliceWallet.address, opts) }
+        val cannotMessage = runBlocking { Client.canMessage(notOnNetwork.address, opts) }
 
         assert(canMessage)
         assert(!cannotMessage)
diff --git a/library/src/androidTest/java/org/xmtp/android/library/ContactsTest.kt b/library/src/androidTest/java/org/xmtp/android/library/ContactsTest.kt
index dcbc0d9c1..49266715c 100644
--- a/library/src/androidTest/java/org/xmtp/android/library/ContactsTest.kt
+++ b/library/src/androidTest/java/org/xmtp/android/library/ContactsTest.kt
@@ -27,19 +27,6 @@ class ContactsTest {
         assertEquals(contactBundle?.walletAddress, fixtures.bob.walletAddress)
     }
 
-    @Test
-    fun testCachesContacts() {
-        val fixtures = fixtures()
-        runBlocking { fixtures.bobClient.ensureUserContactPublished() }
-        // Look up the first time
-        fixtures.aliceClient.contacts.find(fixtures.bob.walletAddress)
-        fixtures.fakeApiClient.assertNoQuery {
-            val contactBundle = fixtures.aliceClient.contacts.find(fixtures.bob.walletAddress)
-            assertEquals(contactBundle?.walletAddress, fixtures.bob.walletAddress)
-        }
-        assert(fixtures.aliceClient.contacts.has(fixtures.bob.walletAddress))
-    }
-
     @Test
     fun testAllowAddress() {
         val fixtures = fixtures()
diff --git a/library/src/androidTest/java/org/xmtp/android/library/ConversationTest.kt b/library/src/androidTest/java/org/xmtp/android/library/ConversationTest.kt
index 2e6005303..6dcde9dd1 100644
--- a/library/src/androidTest/java/org/xmtp/android/library/ConversationTest.kt
+++ b/library/src/androidTest/java/org/xmtp/android/library/ConversationTest.kt
@@ -1,9 +1,11 @@
 package org.xmtp.android.library
 
 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.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
 import org.junit.Assert.assertEquals
 import org.junit.Assert.assertFalse
@@ -46,7 +48,6 @@ import java.util.Date
 
 @RunWith(AndroidJUnit4::class)
 class ConversationTest {
-    lateinit var fakeApiClient: FakeApiClient
     lateinit var aliceWallet: PrivateKeyBuilder
     lateinit var bobWallet: PrivateKeyBuilder
     lateinit var alice: PrivateKey
@@ -62,7 +63,6 @@ class ConversationTest {
         alice = fixtures.alice
         bobWallet = fixtures.bobAccount
         bob = fixtures.bob
-        fakeApiClient = fixtures.fakeApiClient
         aliceClient = fixtures.aliceClient
         bobClient = fixtures.bobClient
     }
@@ -120,42 +120,11 @@ class ConversationTest {
             runBlocking { aliceClient.conversations.newConversation(bob.walletAddress) }
         assertEquals(conversation.peerAddress, bob.walletAddress)
         assertEquals(conversation.createdAt, someTimeAgo)
-        val existingMessages = fakeApiClient.published.size
         conversation = runBlocking { bobClient.conversations.newConversation(alice.walletAddress) }
-
-        assertEquals(
-            "published more messages when we shouldn't have",
-            existingMessages,
-            fakeApiClient.published.size,
-        )
         assertEquals(conversation.peerAddress, alice.walletAddress)
         assertEquals(conversation.createdAt, someTimeAgo)
     }
 
-    @Test
-    fun testCanFindExistingV2Conversation() {
-        val existingConversation = runBlocking {
-            bobClient.conversations.newConversation(
-                alice.walletAddress,
-                context = InvitationV1ContextBuilder.buildFromConversation("http://example.com/2"),
-            )
-        }
-        var conversation: Conversation? = null
-        fakeApiClient.assertNoPublish {
-            runBlocking {
-                conversation = bobClient.conversations.newConversation(
-                    alice.walletAddress,
-                    context = InvitationV1ContextBuilder.buildFromConversation("http://example.com/2"),
-                )
-            }
-        }
-        assertEquals(
-            "made new conversation instead of using existing one",
-            conversation!!.topic,
-            existingConversation.topic,
-        )
-    }
-
     @Test
     fun testCanLoadV1Messages() {
         // Overwrite contact as legacy so we can get v1
@@ -405,15 +374,6 @@ class ConversationTest {
         }
     }
 
-    @Test
-    fun testCanUseCachedConversation() {
-        runBlocking { bobClient.conversations.newConversation(alice.walletAddress) }
-
-        fakeApiClient.assertNoQuery {
-            runBlocking { bobClient.conversations.newConversation(alice.walletAddress) }
-        }
-    }
-
     @Test
     @Ignore("Rust seems to be Flaky with V1")
     fun testCanPaginateV1Messages() {
@@ -493,9 +453,18 @@ class ConversationTest {
             (topic.equals(steveConversation.topic) || topic.equals(bobConversation.topic))
         }
         assertEquals(3, messages.size)
-        assertTrue("isSteveOrBobConversation message 0", isSteveOrBobConversation(messages[0].topic))
-        assertTrue("isSteveOrBobConversation message 1", isSteveOrBobConversation(messages[1].topic))
-        assertTrue("isSteveOrBobConversation message 2", isSteveOrBobConversation(messages[2].topic))
+        assertTrue(
+            "isSteveOrBobConversation message 0",
+            isSteveOrBobConversation(messages[0].topic)
+        )
+        assertTrue(
+            "isSteveOrBobConversation message 1",
+            isSteveOrBobConversation(messages[1].topic)
+        )
+        assertTrue(
+            "isSteveOrBobConversation message 2",
+            isSteveOrBobConversation(messages[2].topic)
+        )
     }
 
     @Test
@@ -587,33 +556,82 @@ 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)
+    fun testCanStreamConversationsV2() {
+        val allMessages = mutableListOf<String>()
+
+        val job = CoroutineScope(Dispatchers.IO).launch {
+            try {
+                bobClient.conversations.stream()
+                    .collect { message ->
+                        allMessages.add(message.topic)
+                    }
+            } catch (e: Exception) {
+            }
         }
+        Thread.sleep(2500)
+
+        runBlocking {
+            bobClient.conversations.newConversation(alice.walletAddress)
+        }
+
+        Thread.sleep(1000)
+
+        assertEquals(1, allMessages.size)
+
+        job.cancel()
     }
 
     @Test
-    fun testStreamingMessagesFromV1Conversation() = kotlinx.coroutines.test.runTest {
+    fun testStreamingMessagesFromV1Conversation() {
         // Overwrite contact as legacy
         fixtures.publishLegacyContact(client = bobClient)
         fixtures.publishLegacyContact(client = aliceClient)
-        val conversation = aliceClient.conversations.newConversation(bob.walletAddress)
-        conversation.streamMessages().test {
-            conversation.send("hi alice")
-            assertEquals("hi alice", awaitItem().encodedContent.content.toStringUtf8())
+        val conversation =
+            runBlocking { aliceClient.conversations.newConversation(bob.walletAddress) }
+        val allMessages = mutableListOf<DecodedMessage>()
+
+        val job = CoroutineScope(Dispatchers.IO).launch {
+            try {
+                conversation.streamMessages().collect { message ->
+                    allMessages.add(message)
+                }
+            } catch (e: Exception) {
+            }
+        }
+        Thread.sleep(2500)
+
+        for (i in 0 until 5) {
+            runBlocking { conversation.send(text = "Message $i") }
+            Thread.sleep(1000)
         }
+
+        assertEquals(allMessages.size, 5)
+        job.cancel()
     }
 
     @Test
-    fun testStreamingMessagesFromV2Conversations() = kotlinx.coroutines.test.runTest {
-        val conversation = aliceClient.conversations.newConversation(bob.walletAddress)
-        conversation.streamMessages().test {
-            conversation.send("hi alice")
-            assertEquals("hi alice", awaitItem().encodedContent.content.toStringUtf8())
+    fun testStreamingMessagesFromV2Conversations() {
+        val conversation =
+            runBlocking { aliceClient.conversations.newConversation(bob.walletAddress) }
+        val allMessages = mutableListOf<DecodedMessage>()
+
+        val job = CoroutineScope(Dispatchers.IO).launch {
+            try {
+                conversation.streamMessages().collect { message ->
+                    allMessages.add(message)
+                }
+            } catch (e: Exception) {
+            }
+        }
+        Thread.sleep(2500)
+
+        for (i in 0 until 5) {
+            runBlocking { conversation.send(text = "Message $i") }
+            Thread.sleep(1000)
         }
+
+        assertEquals(allMessages.size, 5)
+        job.cancel()
     }
 
     @Test
@@ -743,7 +761,6 @@ class ConversationTest {
         }.build()
 
         val client = Client().create(account = PrivateKeyBuilder(key))
-        assertEquals(client.apiClient.environment, XMTPEnvironment.DEV)
         runBlocking {
             val conversations = client.conversations.list()
             assertEquals(1, conversations.size)
@@ -788,11 +805,14 @@ class ConversationTest {
     fun testCanHaveConsentState() {
         val bobConversation =
             runBlocking { bobClient.conversations.newConversation(alice.walletAddress, null) }
+        Thread.sleep(1000)
         val isAllowed = bobConversation.consentState() == ConsentState.ALLOWED
-
         // Conversations you start should start as allowed
         assertTrue("Bob convo should be allowed", isAllowed)
-        assertTrue("Bob contacts should be allowed", bobClient.contacts.isAllowed(alice.walletAddress))
+        assertTrue(
+            "Bob contacts should be allowed",
+            bobClient.contacts.isAllowed(alice.walletAddress)
+        )
 
         runBlocking {
             bobClient.contacts.deny(listOf(alice.walletAddress))
@@ -829,6 +849,7 @@ class ConversationTest {
     fun testCanHaveImplicitConsentOnMessageSend() {
         val bobConversation =
             runBlocking { bobClient.conversations.newConversation(alice.walletAddress, null) }
+        Thread.sleep(1000)
         val isAllowed = bobConversation.consentState() == ConsentState.ALLOWED
 
         // Conversations you start should start as allowed
@@ -860,12 +881,24 @@ class ConversationTest {
             bobClient.contacts.refreshConsentList()
             Thread.sleep(1000)
             assertEquals(bobClient.contacts.consentList.entries.size, 2)
-            assertTrue("Bob convo should be allowed", bobConversation.consentState() == ConsentState.ALLOWED)
-            assertTrue("Caro convo should be allowed", caroConversation.consentState() == ConsentState.ALLOWED)
+            assertTrue(
+                "Bob convo should be allowed",
+                bobConversation.consentState() == ConsentState.ALLOWED
+            )
+            assertTrue(
+                "Caro convo should be allowed",
+                caroConversation.consentState() == ConsentState.ALLOWED
+            )
             bobClient.contacts.deny(listOf(alice.walletAddress, fixtures.caro.walletAddress))
             assertEquals(bobClient.contacts.consentList.entries.size, 2)
-            assertTrue("Bob convo should be denied", bobConversation.consentState() == ConsentState.DENIED)
-            assertTrue("Caro convo should be denied", caroConversation.consentState() == ConsentState.DENIED)
+            assertTrue(
+                "Bob convo should be denied",
+                bobConversation.consentState() == ConsentState.DENIED
+            )
+            assertTrue(
+                "Caro convo should be denied",
+                caroConversation.consentState() == ConsentState.DENIED
+            )
         }
     }
 
diff --git a/library/src/androidTest/java/org/xmtp/android/library/ConversationsTest.kt b/library/src/androidTest/java/org/xmtp/android/library/ConversationsTest.kt
index 65cf3fb03..205dfb87c 100644
--- a/library/src/androidTest/java/org/xmtp/android/library/ConversationsTest.kt
+++ b/library/src/androidTest/java/org/xmtp/android/library/ConversationsTest.kt
@@ -36,7 +36,6 @@ import java.util.Date
 
 @RunWith(AndroidJUnit4::class)
 class ConversationsTest {
-    lateinit var fakeApiClient: FakeApiClient
     lateinit var alixWallet: PrivateKeyBuilder
     lateinit var boWallet: PrivateKeyBuilder
     lateinit var alix: PrivateKey
@@ -53,7 +52,6 @@ class ConversationsTest {
         alix = fixtures.alice
         boWallet = fixtures.bobAccount
         bo = fixtures.bob
-        fakeApiClient = fixtures.fakeApiClient
         alixClient = fixtures.aliceClient
         boClient = fixtures.bobClient
         caroClient = fixtures.caroClient
@@ -128,7 +126,7 @@ class ConversationsTest {
             runBlocking { boConversation.send(text = "Message $i") }
             sleep(1000)
         }
-        assertEquals(allMessages.size, 5)
+        assertEquals(5, allMessages.size)
 
         val caroConversation =
             runBlocking { caroClient.conversations.newConversation(alixClient.address) }
@@ -139,7 +137,7 @@ class ConversationsTest {
             sleep(1000)
         }
 
-        assertEquals(allMessages.size, 10)
+        assertEquals(10, allMessages.size)
 
         job.cancel()
 
@@ -158,7 +156,7 @@ class ConversationsTest {
             sleep(1000)
         }
 
-        assertEquals(allMessages.size, 15)
+        assertEquals(15, allMessages.size)
     }
 
     @Test
diff --git a/library/src/androidTest/java/org/xmtp/android/library/GroupTest.kt b/library/src/androidTest/java/org/xmtp/android/library/GroupTest.kt
index b0da68078..7ee39acbb 100644
--- a/library/src/androidTest/java/org/xmtp/android/library/GroupTest.kt
+++ b/library/src/androidTest/java/org/xmtp/android/library/GroupTest.kt
@@ -5,19 +5,13 @@ import androidx.test.platform.app.InstrumentationRegistry
 import app.cash.turbine.test
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.cancel
-import kotlinx.coroutines.flow.catch
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
-import kotlinx.coroutines.withContext
-import kotlinx.coroutines.withTimeout
 import org.junit.Assert.assertEquals
 import org.junit.Assert.assertThrows
 import org.junit.Before
-import org.junit.Ignore
 import org.junit.Test
 import org.junit.runner.RunWith
-import org.web3j.utils.Numeric
 import org.xmtp.android.library.codecs.ContentTypeGroupUpdated
 import org.xmtp.android.library.codecs.ContentTypeReaction
 import org.xmtp.android.library.codecs.GroupUpdatedCodec
@@ -25,6 +19,7 @@ import org.xmtp.android.library.codecs.Reaction
 import org.xmtp.android.library.codecs.ReactionAction
 import org.xmtp.android.library.codecs.ReactionCodec
 import org.xmtp.android.library.codecs.ReactionSchema
+import org.xmtp.android.library.messages.DecryptedMessage
 import org.xmtp.android.library.messages.MessageDeliveryStatus
 import org.xmtp.android.library.messages.PrivateKey
 import org.xmtp.android.library.messages.PrivateKeyBuilder
@@ -447,6 +442,7 @@ class GroupTest {
             } catch (e: Exception) {
             }
         }
+        Thread.sleep(1000)
 
         val alixGroup = runBlocking { alixClient.conversations.newGroup(listOf(bo.walletAddress)) }
 
@@ -581,56 +577,72 @@ class GroupTest {
     }
 
     @Test
-    fun testCanStreamAllGroupMessages() = kotlinx.coroutines.test.runTest {
-        val group = caroClient.conversations.newGroup(listOf(alix.walletAddress))
-        alixClient.conversations.syncGroups()
-        val flow = alixClient.conversations.streamAllGroupMessages()
-
-        withContext(Dispatchers.Default.limitedParallelism(1)) {
-            withTimeout(5000) { // Set a timeout to avoid long running tests
-                val job = launch {
-                    flow.catch { e ->
-                        throw Exception("Error collecting flow: $e")
-                    }.collect { message ->
-                        assertEquals("hi", message.encodedContent.content.toStringUtf8())
-                        this.cancel() // Cancel the collection after the assertion
-                    }
-                }
+    fun testCanStreamAllGroupMessages() {
+        val group = runBlocking { caroClient.conversations.newGroup(listOf(alix.walletAddress)) }
+        runBlocking { alixClient.conversations.syncGroups() }
 
-                group.send("hi")
+        val allMessages = mutableListOf<DecodedMessage>()
 
-                job.join()
+        val job = CoroutineScope(Dispatchers.IO).launch {
+            try {
+                alixClient.conversations.streamAllGroupMessages().collect { message ->
+                    allMessages.add(message)
+                }
+            } catch (e: Exception) {
             }
         }
+        Thread.sleep(2500)
+
+        for (i in 0 until 2) {
+            runBlocking { group.send(text = "Message $i") }
+            Thread.sleep(100)
+        }
+        assertEquals(2, allMessages.size)
+
+        val caroGroup =
+            runBlocking { caroClient.conversations.newGroup(listOf(alixClient.address)) }
+        Thread.sleep(2500)
+
+        for (i in 0 until 2) {
+            runBlocking { caroGroup.send(text = "Message $i") }
+            Thread.sleep(100)
+        }
+
+        assertEquals(4, allMessages.size)
+
+        job.cancel()
     }
 
     @Test
-    fun testCanStreamAllMessages() = kotlinx.coroutines.test.runTest {
-        val group = caroClient.conversations.newGroup(listOf(alix.walletAddress))
-        val conversation = boClient.conversations.newConversation(alix.walletAddress)
-        alixClient.conversations.syncGroups()
-
-        val flow = alixClient.conversations.streamAllMessages(includeGroups = true)
-        var counter = 0
-
-        withContext(Dispatchers.Default.limitedParallelism(1)) {
-            withTimeout(5000) { // Set a timeout to avoid long running tests
-                val job = launch {
-                    flow.catch { e ->
-                        throw Exception("Error collecting flow: $e")
-                    }.collect { message ->
-                        counter++
-                        assertEquals("hi", message.encodedContent.content.toStringUtf8())
-                        if (counter == 2) this.cancel() // Cancel the collection after receiving the second "hi"
-                    }
-                }
+    fun testCanStreamAllMessages() {
+        val group = runBlocking { caroClient.conversations.newGroup(listOf(alix.walletAddress)) }
+        val conversation =
+            runBlocking { boClient.conversations.newConversation(alix.walletAddress) }
+        runBlocking { alixClient.conversations.syncGroups() }
 
-                group.send("hi")
-                conversation.send("hi")
+        val allMessages = mutableListOf<DecodedMessage>()
 
-                job.join()
+        val job = CoroutineScope(Dispatchers.IO).launch {
+            try {
+                alixClient.conversations.streamAllMessages(includeGroups = true)
+                    .collect { message ->
+                        allMessages.add(message)
+                    }
+            } catch (e: Exception) {
             }
         }
+        Thread.sleep(2500)
+
+        runBlocking {
+            group.send("hi")
+            conversation.send("hi")
+        }
+
+        Thread.sleep(1000)
+
+        assertEquals(2, allMessages.size)
+
+        job.cancel()
     }
 
     @Test
@@ -647,66 +659,72 @@ class GroupTest {
     }
 
     @Test
-    fun testCanStreamAllDecryptedGroupMessages() = kotlinx.coroutines.test.runTest {
-        Client.register(codec = GroupUpdatedCodec())
-        val membershipChange = TranscriptMessages.GroupUpdated.newBuilder().build()
-        val group = caroClient.conversations.newGroup(listOf(alix.walletAddress))
-        alixClient.conversations.syncGroups()
+    fun testCanStreamAllDecryptedGroupMessages() {
+        val group = runBlocking { caroClient.conversations.newGroup(listOf(alix.walletAddress)) }
+        runBlocking { alixClient.conversations.syncGroups() }
 
-        val flow = alixClient.conversations.streamAllGroupDecryptedMessages()
-        var counter = 0
-
-        withContext(Dispatchers.Default.limitedParallelism(1)) {
-            withTimeout(5000) { // Set a timeout to avoid long running tests
-                val job = launch {
-                    flow.catch { e ->
-                        throw Exception("Error collecting flow: $e")
-                    }.collect { message ->
-                        counter++
-                        assertEquals("hi", message.encodedContent.content.toStringUtf8())
-                        if (counter == 2) this.cancel() // Cancel the collection after receiving the second "hi"
-                    }
+        val allMessages = mutableListOf<DecryptedMessage>()
+
+        val job = CoroutineScope(Dispatchers.IO).launch {
+            try {
+                alixClient.conversations.streamAllGroupDecryptedMessages().collect { message ->
+                    allMessages.add(message)
                 }
+            } catch (e: Exception) {
+            }
+        }
+        Thread.sleep(2500)
 
-                group.send("hi")
-                group.send(
-                    content = membershipChange,
-                    options = SendOptions(contentType = ContentTypeGroupUpdated),
-                )
-                group.send("hi")
+        for (i in 0 until 2) {
+            runBlocking { group.send(text = "Message $i") }
+            Thread.sleep(100)
+        }
+        assertEquals(2, allMessages.size)
 
-                job.join()
-            }
+        val caroGroup =
+            runBlocking { caroClient.conversations.newGroup(listOf(alixClient.address)) }
+        Thread.sleep(2500)
+
+        for (i in 0 until 2) {
+            runBlocking { caroGroup.send(text = "Message $i") }
+            Thread.sleep(100)
         }
+
+        assertEquals(4, allMessages.size)
+
+        job.cancel()
     }
 
     @Test
-    fun testCanStreamAllDecryptedMessages() = kotlinx.coroutines.test.runTest {
-        val group = caroClient.conversations.newGroup(listOf(alix.walletAddress))
-        val conversation = boClient.conversations.newConversation(alix.walletAddress)
-        alixClient.conversations.syncGroups()
-
-        val flow = alixClient.conversations.streamAllDecryptedMessages(includeGroups = true)
-        var counter = 0
-
-        withContext(Dispatchers.Default.limitedParallelism(1)) {
-            withTimeout(5000) { // Set a timeout to avoid long running tests
-                val job = launch {
-                    flow.catch { e ->
-                        throw Exception("Error collecting flow: $e")
-                    }.collect { message ->
-                        counter++
-                        assertEquals("hi", message.encodedContent.content.toStringUtf8())
-                        if (counter == 2) this.cancel() // Cancel the collection after receiving the second "hi"
-                    }
-                }
+    fun testCanStreamAllDecryptedMessages() {
+        val group = runBlocking { caroClient.conversations.newGroup(listOf(alix.walletAddress)) }
+        val conversation =
+            runBlocking { boClient.conversations.newConversation(alix.walletAddress) }
+        runBlocking { alixClient.conversations.syncGroups() }
 
-                group.send("hi")
-                conversation.send("hi")
+        val allMessages = mutableListOf<DecryptedMessage>()
 
-                job.join()
+        val job = CoroutineScope(Dispatchers.IO).launch {
+            try {
+                alixClient.conversations.streamAllDecryptedMessages(includeGroups = true)
+                    .collect { message ->
+                        allMessages.add(message)
+                    }
+            } catch (e: Exception) {
             }
         }
+        Thread.sleep(2500)
+
+        runBlocking {
+            group.send("hi")
+            conversation.send("hi")
+        }
+
+        Thread.sleep(1000)
+
+        assertEquals(2, allMessages.size)
+
+        job.cancel()
     }
 
     @Test
@@ -722,16 +740,30 @@ class GroupTest {
     }
 
     @Test
-    @Ignore("Flaky: CI")
-    fun testCanStreamGroupsAndConversations() = kotlinx.coroutines.test.runTest {
-        boClient.conversations.streamAll().test {
-            val group =
-                caroClient.conversations.newGroup(listOf(bo.walletAddress))
-            assertEquals(group.topic, awaitItem().topic)
-            val conversation =
-                alixClient.conversations.newConversation(bo.walletAddress)
-            assertEquals(conversation.topic, awaitItem().topic)
+    fun testCanStreamGroupsAndConversations() {
+        val allMessages = mutableListOf<String>()
+
+        val job = CoroutineScope(Dispatchers.IO).launch {
+            try {
+                alixClient.conversations.streamAll()
+                    .collect { message ->
+                        allMessages.add(message.topic)
+                    }
+            } catch (e: Exception) {}
         }
+        Thread.sleep(2500)
+
+        runBlocking {
+            alixClient.conversations.newConversation(bo.walletAddress)
+            Thread.sleep(2500)
+            caroClient.conversations.newGroup(listOf(alix.walletAddress))
+        }
+
+        Thread.sleep(2500)
+
+        assertEquals(2, allMessages.size)
+
+        job.cancel()
     }
 
     @Test
@@ -819,8 +851,23 @@ class GroupTest {
         runBlocking { alixClient.conversations.syncGroups() }
         val alixGroup = alixClient.findGroup(boGroup.id)
         runBlocking { alixGroup?.sync() }
-        val alixMessage = alixClient.findMessage(Numeric.hexStringToByteArray(boMessageId))
+        val alixMessage = alixClient.findMessage(boMessageId.hexToByteArray())
 
         assertEquals(alixMessage?.id?.toHex(), boMessageId)
     }
+
+    @Test
+    fun testTranslatingIds() {
+        val boGroup = runBlocking {
+            boClient.conversations.newGroup(
+                listOf(
+                    alix.walletAddress,
+                    caro.walletAddress
+                )
+            )
+        }
+        val hex = boGroup.id.toHex()
+
+        assert(hex.hexToByteArray().contentEquals(boGroup.id))
+    }
 }
diff --git a/library/src/androidTest/java/org/xmtp/android/library/InvitationTest.kt b/library/src/androidTest/java/org/xmtp/android/library/InvitationTest.kt
index 9ab706b7a..44e2572bc 100644
--- a/library/src/androidTest/java/org/xmtp/android/library/InvitationTest.kt
+++ b/library/src/androidTest/java/org/xmtp/android/library/InvitationTest.kt
@@ -3,7 +3,6 @@ package org.xmtp.android.library
 import androidx.test.ext.junit.runners.AndroidJUnit4
 import com.google.protobuf.kotlin.toByteString
 import kotlinx.coroutines.runBlocking
-import org.junit.Assert
 import org.junit.Assert.assertEquals
 import org.junit.Assert.assertNotEquals
 import org.junit.Test
@@ -52,7 +51,6 @@ class InvitationTest {
         }.build()
 
         val client = Client().create(account = PrivateKeyBuilder(key))
-        Assert.assertEquals(client.apiClient.environment, XMTPEnvironment.DEV)
         val conversations = runBlocking { client.conversations.list() }
         assertEquals(1, conversations.size)
         val message = runBlocking { conversations[0].messages().firstOrNull() }
diff --git a/library/src/androidTest/java/org/xmtp/android/library/LocalInstrumentedTest.kt b/library/src/androidTest/java/org/xmtp/android/library/LocalInstrumentedTest.kt
index ad840ea8e..325ccf6b0 100644
--- a/library/src/androidTest/java/org/xmtp/android/library/LocalInstrumentedTest.kt
+++ b/library/src/androidTest/java/org/xmtp/android/library/LocalInstrumentedTest.kt
@@ -27,6 +27,7 @@ import org.xmtp.proto.message.contents.Contact
 import org.xmtp.proto.message.contents.InvitationV1Kt.context
 import org.xmtp.proto.message.contents.PrivateKeyOuterClass
 import org.xmtp.proto.message.contents.PrivateKeyOuterClass.PrivateKeyBundle
+import uniffi.xmtpv3.createV2Client
 import java.util.Date
 
 @RunWith(AndroidJUnit4::class)
@@ -44,7 +45,6 @@ class LocalInstrumentedTest {
                 )
             )
         val client = Client().create(aliceWallet, clientOptions)
-        assertEquals(XMTPEnvironment.LOCAL, client.apiClient.environment)
         runBlocking {
             client.publishUserContact()
         }
@@ -63,9 +63,15 @@ class LocalInstrumentedTest {
         val identity = PrivateKey.newBuilder().build().generate()
         val authorized = alice.createIdentity(identity)
         val authToken = authorized.createAuthToken()
+        val v2Client = runBlocking {
+            createV2Client(
+                host = XMTPEnvironment.LOCAL.getUrl(),
+                isSecure = false
+            )
+        }
         val api = GRPCApiClient(
             environment = XMTPEnvironment.LOCAL,
-            secure = false,
+            rustV2Client = v2Client
         )
         api.setAuthToken(authToken)
         val encryptedBundle = authorized.toBundle.encrypted(alice)
@@ -93,9 +99,15 @@ class LocalInstrumentedTest {
         val identity = PrivateKeyBuilder().getPrivateKey()
         val authorized = aliceWallet.createIdentity(identity)
         val authToken = authorized.createAuthToken()
+        val v2Client = runBlocking {
+            createV2Client(
+                host = XMTPEnvironment.LOCAL.getUrl(),
+                isSecure = false
+            )
+        }
         val api = GRPCApiClient(
             environment = XMTPEnvironment.LOCAL,
-            secure = false,
+            rustV2Client = v2Client
         )
         api.setAuthToken(authToken)
         val encryptedBundle =
@@ -113,7 +125,6 @@ class LocalInstrumentedTest {
         val clientOptions =
             ClientOptions(api = ClientOptions.Api(env = XMTPEnvironment.LOCAL, isSecure = false))
         val client = Client().create(account = aliceWallet, options = clientOptions)
-        assertEquals(XMTPEnvironment.LOCAL, client.apiClient.environment)
         val contact = client.getUserContact(peerAddress = aliceWallet.address)
         assertEquals(
             contact?.v2?.keyBundle?.identityKey?.secp256K1Uncompressed,
@@ -754,9 +765,15 @@ class LocalInstrumentedTest {
         val identity = PrivateKey.newBuilder().build().generate()
         val authorized = alice.createIdentity(identity)
         val authToken = authorized.createAuthToken()
+        val v2Client = runBlocking {
+            createV2Client(
+                host = XMTPEnvironment.LOCAL.getUrl(),
+                isSecure = false
+            )
+        }
         val api = GRPCApiClient(
             environment = XMTPEnvironment.LOCAL,
-            secure = false,
+            rustV2Client = v2Client
         )
         api.setAuthToken(authToken)
         val encryptedBundle = authorized.toBundle.encrypted(alice)
diff --git a/library/src/androidTest/java/org/xmtp/android/library/MessageTest.kt b/library/src/androidTest/java/org/xmtp/android/library/MessageTest.kt
index 2b4d97209..341f83496 100644
--- a/library/src/androidTest/java/org/xmtp/android/library/MessageTest.kt
+++ b/library/src/androidTest/java/org/xmtp/android/library/MessageTest.kt
@@ -161,7 +161,6 @@ class MessageTest {
         }.build()
 
         val client = Client().create(account = PrivateKeyBuilder(key))
-        assertEquals(client.apiClient.environment, XMTPEnvironment.DEV)
         runBlocking {
             val convo = client.conversations.list()[0]
             val message = convo.messages()[0]
@@ -196,7 +195,6 @@ class MessageTest {
         }.build()
 
         val client = Client().create(account = PrivateKeyBuilder(key))
-        assertEquals(client.apiClient.environment, XMTPEnvironment.DEV)
         runBlocking {
             val convo = client.conversations.list()[0]
             convo.send(
@@ -209,6 +207,7 @@ class MessageTest {
     }
 
     @Test
+    @Ignore("Dev network flaky should be moved to local")
     fun testCanLoadAllConversations() {
         val ints = arrayOf(
             105, 207, 193, 11, 240, 115, 115, 204,
diff --git a/library/src/androidTest/java/org/xmtp/android/library/TestHelpers.kt b/library/src/androidTest/java/org/xmtp/android/library/TestHelpers.kt
index cdc307665..e3b65ee96 100644
--- a/library/src/androidTest/java/org/xmtp/android/library/TestHelpers.kt
+++ b/library/src/androidTest/java/org/xmtp/android/library/TestHelpers.kt
@@ -1,32 +1,16 @@
 package org.xmtp.android.library
 
-import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.MutableSharedFlow
-import kotlinx.coroutines.flow.first
-import kotlinx.coroutines.flow.flowOf
 import kotlinx.coroutines.runBlocking
-import org.junit.Assert.assertEquals
-import org.xmtp.android.library.codecs.Fetcher
 import org.xmtp.android.library.messages.ContactBundle
 import org.xmtp.android.library.messages.Envelope
-import org.xmtp.android.library.messages.Pagination
 import org.xmtp.android.library.messages.PrivateKey
 import org.xmtp.android.library.messages.PrivateKeyBuilder
 import org.xmtp.android.library.messages.Signature
 import org.xmtp.android.library.messages.Topic
 import org.xmtp.android.library.messages.toPublicKeyBundle
 import org.xmtp.android.library.messages.walletAddress
-import org.xmtp.proto.message.api.v1.MessageApiOuterClass
-import java.io.File
-import java.net.URL
 import java.util.Date
 
-class TestFetcher : Fetcher {
-    override fun fetch(url: URL): ByteArray {
-        return File(url.toString().replace("https://", "")).readBytes()
-    }
-}
-
 class FakeWallet : SigningKey {
     private var privateKey: PrivateKey
     private var privateKeyBuilder: PrivateKeyBuilder
@@ -57,152 +41,6 @@ class FakeWallet : SigningKey {
         get() = privateKey.walletAddress
 }
 
-class FakeStreamHolder {
-    private val flow = MutableSharedFlow<Envelope>()
-    suspend fun emit(value: Envelope) = flow.emit(value)
-    fun counts(): Flow<Envelope> = flow
-}
-
-class FakeApiClient : ApiClient {
-    override val environment: XMTPEnvironment = XMTPEnvironment.LOCAL
-    private var authToken: String? = null
-    private val responses: MutableMap<String, List<Envelope>> = mutableMapOf()
-    val published: MutableList<Envelope> = mutableListOf()
-    var forbiddingQueries = false
-    private var stream = FakeStreamHolder()
-
-    fun assertNoPublish(callback: () -> Unit) {
-        val oldCount = published.size
-        callback()
-        assertEquals(oldCount, published.size)
-    }
-
-    fun assertNoQuery(callback: () -> Unit) {
-        forbiddingQueries = true
-        callback()
-        forbiddingQueries = false
-    }
-
-    fun findPublishedEnvelope(topic: Topic): Envelope? =
-        findPublishedEnvelope(topic.description)
-
-    fun findPublishedEnvelope(topic: String): Envelope? {
-        for (envelope in published.reversed()) {
-            if (envelope.contentTopic == topic) {
-                return envelope
-            }
-        }
-        return null
-    }
-
-    override fun setAuthToken(token: String) {
-        authToken = token
-    }
-
-    override suspend fun queryTopic(
-        topic: Topic,
-        pagination: Pagination?,
-    ): MessageApiOuterClass.QueryResponse {
-        return query(topic = topic.description, pagination)
-    }
-
-    suspend fun send(envelope: Envelope) {
-        stream.emit(envelope)
-    }
-
-    override suspend fun envelopes(
-        topic: String,
-        pagination: Pagination?,
-    ): List<MessageApiOuterClass.Envelope> {
-        return query(topic = topic, pagination = pagination).envelopesList
-    }
-
-    override suspend fun batchQuery(requests: List<MessageApiOuterClass.QueryRequest>): MessageApiOuterClass.BatchQueryResponse {
-        val responses = requests.map {
-            query(it.getContentTopics(0), Pagination(after = Date(it.startTimeNs)))
-        }
-
-        return MessageApiOuterClass.BatchQueryResponse.newBuilder().also {
-            it.addAllResponses(responses)
-        }.build()
-    }
-
-    override suspend fun query(
-        topic: String,
-        pagination: Pagination?,
-        cursor: MessageApiOuterClass.Cursor?,
-    ): MessageApiOuterClass.QueryResponse {
-        var result: MutableList<Envelope> = mutableListOf()
-        val response = responses.toMutableMap().remove(topic)
-        if (response != null) {
-            result.addAll(response)
-        }
-        result.addAll(
-            published.filter {
-                it.contentTopic == topic
-            }.reversed()
-        )
-
-        val startAt = pagination?.before
-        if (startAt != null) {
-            result = result.filter { it.timestampNs < startAt.time }
-                .sortedBy { it.timestampNs }.toMutableList()
-        }
-        val endAt = pagination?.after
-        if (endAt != null) {
-            result = result.filter {
-                it.timestampNs > endAt.time
-            }
-                .sortedBy { it.timestampNs }.toMutableList()
-        }
-        val limit = pagination?.limit
-        if (limit != null) {
-            if (limit == 1) {
-                val first = result.firstOrNull()
-                if (first != null) {
-                    result = mutableListOf(first)
-                } else {
-                    result = mutableListOf()
-                }
-            } else {
-                result = result.take(limit - 1).toMutableList()
-            }
-        }
-
-        val direction = pagination?.direction
-        if (direction != null) {
-            when (direction) {
-                MessageApiOuterClass.SortDirection.SORT_DIRECTION_ASCENDING -> {
-                    result = result.reversed().toMutableList()
-                }
-
-                else -> Unit
-            }
-        }
-
-        return QueryResponse.newBuilder().also {
-            it.addAllEnvelopes(result)
-        }.build()
-    }
-
-    override suspend fun publish(envelopes: List<MessageApiOuterClass.Envelope>): MessageApiOuterClass.PublishResponse {
-        for (envelope in envelopes) {
-            send(envelope)
-        }
-        published.addAll(envelopes)
-        return PublishResponse.newBuilder().build()
-    }
-
-    override suspend fun subscribe(request: Flow<MessageApiOuterClass.SubscribeRequest>): Flow<MessageApiOuterClass.Envelope> {
-        val env = stream.counts().first()
-
-        if (request.first().contentTopicsList.contains(env.contentTopic)) {
-            return flowOf(env)
-        }
-        return flowOf()
-    }
-}
-
 data class Fixtures(
     val aliceAccount: PrivateKeyBuilder,
     val bobAccount: PrivateKeyBuilder,
@@ -211,7 +49,6 @@ data class Fixtures(
         ClientOptions.Api(XMTPEnvironment.LOCAL, isSecure = false)
     ),
 ) {
-    var fakeApiClient: FakeApiClient = FakeApiClient()
     var alice: PrivateKey = aliceAccount.getPrivateKey()
     var aliceClient: Client = Client().create(account = aliceAccount, options = clientOptions)
     var bob: PrivateKey = bobAccount.getPrivateKey()
diff --git a/library/src/main/java/README.md b/library/src/main/java/README.md
index 87dc47153..6020e2fa5 100644
--- a/library/src/main/java/README.md
+++ b/library/src/main/java/README.md
@@ -2,14 +2,15 @@
 
 Kotlin code emitted by the `bindings_ffi` crate in [libxmtp](https://github.com/xmtp/libxmtp) including how to get jni libraries
 
-## Process for updating from a [libxmtp](https://github.com/xmtp/libxmtp) Kotlin Binding Release (work in progress!)
-
-1. From repo [libxmtp](https://github.com/xmtp/libxmtp) checkout the branch you would like to make a release from
-2. Navigate to the `bindings_ffi` folder
-3. Follow the instructions for "Rebuilding this crate in the `bindings_ffi` [README](https://github.com/xmtp/libxmtp/tree/main/bindings_ffi#rebuilding-this-crate)"
-4. Copy the contents of `libxmtp/bindings_ffi/src/uniffi/xmtpv3/xmtpv3.kt` to `xmtp-android/library/src/main/java/xmtpv3.kt`
-5. Run format (cmd + opt + l) function to keep the code format consistent and diff small
-6. All instances of `value.forEach` should be changed to `value.iterator().forEach` to be compatible with API 23
-7. Copy the jniLibs from `libxmtp/bindings_ffi/jniLibs` to `xmtp-android/library/src/main/jniLibs`
+## Process for updating from a [libxmtp](https://github.com/xmtp/libxmtp) Kotlin Binding Release
+
+1. From repo [libxmtp](https://github.com/xmtp/libxmtp) run the [kotlin release action](https://github.com/xmtp/libxmtp/actions/workflows/release-kotlin-bindings.yml) for the branch you desire 
+2. Create a new branch in the `xmtp-android` repo
+   With `libxmtp` repo and `xmtp-android` (this repo) cloned locally in sibling directories, and `libxmtp` checked out to the correct release commit, run the script:
+   `./bindings_ffi/gen_kotlin.sh`
+3. Run format (cmd + opt + l) function to keep the code format consistent and diff small for `xmtp-android/library/src/main/java/xmtpv3.kt`
+4. Navigate to the [latest release](https://github.com/xmtp/libxmtp/releases) once the action completes
+5. Download the `LibXMTPKotlinFFI.zip` assets
+6. Unzip and then copy the jniLibs to `xmtp-android/library/src/main/jniLibs`
 
 You should now be on the latest libxmtp. Tests will fail if the jniLibs do not match the version of xmtpv3.
\ No newline at end of file
diff --git a/library/src/main/java/libxmtp-version.txt b/library/src/main/java/libxmtp-version.txt
index ed9660a73..8976d923a 100644
--- a/library/src/main/java/libxmtp-version.txt
+++ b/library/src/main/java/libxmtp-version.txt
@@ -1,3 +1,3 @@
-Version: bbf05eac
+Version: feb48641
 Branch: main
-Date: 2024-06-25 01:50:26 +0000
+Date: 2024-06-26 22:31:21 +0000
diff --git a/library/src/main/java/org/xmtp/android/library/ApiClient.kt b/library/src/main/java/org/xmtp/android/library/ApiClient.kt
index 19f28544b..61e817d3d 100644
--- a/library/src/main/java/org/xmtp/android/library/ApiClient.kt
+++ b/library/src/main/java/org/xmtp/android/library/ApiClient.kt
@@ -1,23 +1,30 @@
 package org.xmtp.android.library
 
-import io.grpc.ManagedChannel
-import io.grpc.ManagedChannelBuilder
-import io.grpc.Metadata
-import kotlinx.coroutines.flow.Flow
+import com.google.protobuf.kotlin.toByteString
+import org.xmtp.android.library.Util.Companion.envelopeFromFFi
 import org.xmtp.android.library.messages.Pagination
 import org.xmtp.android.library.messages.Topic
-import org.xmtp.proto.message.api.v1.MessageApiGrpcKt
-import org.xmtp.proto.message.api.v1.MessageApiOuterClass.BatchQueryRequest
 import org.xmtp.proto.message.api.v1.MessageApiOuterClass.BatchQueryResponse
 import org.xmtp.proto.message.api.v1.MessageApiOuterClass.Cursor
 import org.xmtp.proto.message.api.v1.MessageApiOuterClass.Envelope
-import org.xmtp.proto.message.api.v1.MessageApiOuterClass.PublishRequest
-import org.xmtp.proto.message.api.v1.MessageApiOuterClass.PublishResponse
+import org.xmtp.proto.message.api.v1.MessageApiOuterClass.PagingInfo
 import org.xmtp.proto.message.api.v1.MessageApiOuterClass.QueryRequest
 import org.xmtp.proto.message.api.v1.MessageApiOuterClass.QueryResponse
-import org.xmtp.proto.message.api.v1.MessageApiOuterClass.SubscribeRequest
+import org.xmtp.proto.message.api.v1.MessageApiOuterClass.SortDirection
+import uniffi.xmtpv3.FfiCursor
+import uniffi.xmtpv3.FfiEnvelope
+import uniffi.xmtpv3.FfiPagingInfo
+import uniffi.xmtpv3.FfiPublishRequest
+import uniffi.xmtpv3.FfiSortDirection
+import uniffi.xmtpv3.FfiV2ApiClient
+import uniffi.xmtpv3.FfiV2BatchQueryRequest
+import uniffi.xmtpv3.FfiV2BatchQueryResponse
+import uniffi.xmtpv3.FfiV2QueryRequest
+import uniffi.xmtpv3.FfiV2QueryResponse
+import uniffi.xmtpv3.FfiV2SubscribeRequest
+import uniffi.xmtpv3.FfiV2Subscription
+import uniffi.xmtpv3.FfiV2SubscriptionCallback
 import java.io.Closeable
-import java.util.concurrent.TimeUnit
 
 interface ApiClient {
     val environment: XMTPEnvironment
@@ -31,25 +38,19 @@ interface ApiClient {
     suspend fun queryTopic(topic: Topic, pagination: Pagination? = null): QueryResponse
     suspend fun batchQuery(requests: List<QueryRequest>): BatchQueryResponse
     suspend fun envelopes(topic: String, pagination: Pagination? = null): List<Envelope>
-    suspend fun publish(envelopes: List<Envelope>): PublishResponse
-    suspend fun subscribe(request: Flow<SubscribeRequest>): Flow<Envelope>
+    suspend fun publish(envelopes: List<Envelope>)
+    suspend fun subscribe(
+        request: FfiV2SubscribeRequest,
+        callback: FfiV2SubscriptionCallback,
+    ): FfiV2Subscription
 }
 
 data class GRPCApiClient(
     override val environment: XMTPEnvironment,
-    val secure: Boolean = true,
-    val appVersion: String? = null,
+    val rustV2Client: FfiV2ApiClient,
 ) :
     ApiClient, Closeable {
     companion object {
-        val AUTHORIZATION_HEADER_KEY: Metadata.Key<String> =
-            Metadata.Key.of("authorization", Metadata.ASCII_STRING_MARSHALLER)
-
-        val CLIENT_VERSION_HEADER_KEY: Metadata.Key<String> =
-            Metadata.Key.of("X-Client-Version", Metadata.ASCII_STRING_MARSHALLER)
-
-        val APP_VERSION_HEADER_KEY: Metadata.Key<String> =
-            Metadata.Key.of("X-App-Version", Metadata.ASCII_STRING_MARSHALLER)
 
         fun makeQueryRequest(
             topic: String,
@@ -79,44 +80,8 @@ data class GRPCApiClient(
                         }.build()
                     }
                 }.build()
+    }
 
-        fun makeSubscribeRequest(
-            topics: List<String>,
-        ): SubscribeRequest = SubscribeRequest.newBuilder().addAllContentTopics(topics).build()
-    }
-
-    private val retryPolicy = mapOf(
-        "methodConfig" to listOf(
-            mapOf(
-                "retryPolicy" to mapOf(
-                    "maxAttempts" to 4.0,
-                    "initialBackoff" to "0.5s",
-                    "maxBackoff" to "30s",
-                    "backoffMultiplier" to 2.0,
-                    "retryableStatusCodes" to listOf(
-                        "UNAVAILABLE",
-                    )
-                )
-            )
-        )
-    )
-
-    private val channel: ManagedChannel =
-        ManagedChannelBuilder.forAddress(
-            environment.getValue(),
-            if (environment == XMTPEnvironment.LOCAL) 5556 else 443
-        ).apply {
-            if (environment != XMTPEnvironment.LOCAL) {
-                useTransportSecurity()
-            } else {
-                usePlaintext()
-            }
-            defaultServiceConfig(retryPolicy)
-            enableRetry()
-        }.build()
-
-    private val client: MessageApiGrpcKt.MessageApiCoroutineStub =
-        MessageApiGrpcKt.MessageApiCoroutineStub(channel)
     private var authToken: String? = null
 
     override fun setAuthToken(token: String) {
@@ -129,16 +94,7 @@ data class GRPCApiClient(
         cursor: Cursor?,
     ): QueryResponse {
         val request = makeQueryRequest(topic, pagination, cursor)
-        val headers = Metadata()
-
-        authToken?.let { token ->
-            headers.put(AUTHORIZATION_HEADER_KEY, "Bearer $token")
-        }
-        headers.put(CLIENT_VERSION_HEADER_KEY, Constants.VERSION)
-        if (appVersion != null) {
-            headers.put(APP_VERSION_HEADER_KEY, appVersion)
-        }
-        return client.query(request, headers = headers)
+        return queryResponseFromFFi(rustV2Client.query(queryRequestToFFi(request)))
     }
 
     /**
@@ -175,47 +131,107 @@ data class GRPCApiClient(
     override suspend fun batchQuery(
         requests: List<QueryRequest>,
     ): BatchQueryResponse {
-        val batchRequest = BatchQueryRequest.newBuilder().addAllRequests(requests).build()
-        val headers = Metadata()
+        val batchRequest = requests.map { queryRequestToFFi(it) }
+        return batchQueryResponseFromFFi(rustV2Client.batchQuery(FfiV2BatchQueryRequest(requests = batchRequest)))
+    }
 
-        authToken?.let { token ->
-            headers.put(AUTHORIZATION_HEADER_KEY, "Bearer $token")
-        }
-        headers.put(CLIENT_VERSION_HEADER_KEY, Constants.VERSION)
-        if (appVersion != null) {
-            headers.put(APP_VERSION_HEADER_KEY, appVersion)
-        }
-        return client.batchQuery(batchRequest, headers = headers)
+    override suspend fun publish(envelopes: List<Envelope>) {
+        val ffiEnvelopes = envelopes.map { envelopeToFFi(it) }
+        val request = FfiPublishRequest(envelopes = ffiEnvelopes)
+
+        rustV2Client.publish(request = request, authToken = authToken ?: "")
     }
 
-    override suspend fun publish(envelopes: List<Envelope>): PublishResponse {
-        val request = PublishRequest.newBuilder().addAllEnvelopes(envelopes).build()
-        val headers = Metadata()
+    override suspend fun subscribe(
+        request: FfiV2SubscribeRequest,
+        callback: FfiV2SubscriptionCallback,
+    ): FfiV2Subscription {
+        return rustV2Client.subscribe(request, callback)
+    }
 
-        authToken?.let { token ->
-            headers.put(AUTHORIZATION_HEADER_KEY, "Bearer $token")
-        }
+    override fun close() {
+        rustV2Client.close()
+    }
 
-        headers.put(CLIENT_VERSION_HEADER_KEY, Constants.VERSION)
-        if (appVersion != null) {
-            headers.put(APP_VERSION_HEADER_KEY, appVersion)
-        }
+    private fun envelopeToFFi(envelope: Envelope): FfiEnvelope {
+        return FfiEnvelope(
+            contentTopic = envelope.contentTopic,
+            timestampNs = envelope.timestampNs.toULong(),
+            message = envelope.message.toByteArray()
+        )
+    }
 
-        return client.publish(request, headers)
+    private fun queryRequestToFFi(request: QueryRequest): FfiV2QueryRequest {
+        return FfiV2QueryRequest(
+            contentTopics = request.contentTopicsList,
+            startTimeNs = request.startTimeNs.toULong(),
+            endTimeNs = request.endTimeNs.toULong(),
+            pagingInfo = pagingInfoToFFi(request.pagingInfo)
+        )
+    }
+
+    private fun queryResponseFromFFi(response: FfiV2QueryResponse): QueryResponse {
+        return QueryResponse.newBuilder().also { queryResponse ->
+            queryResponse.addAllEnvelopes(response.envelopes.map { envelopeFromFFi(it) })
+            response.pagingInfo?.let {
+                queryResponse.pagingInfo = pagingInfoFromFFi(it)
+            }
+        }.build()
     }
 
-    override suspend fun subscribe(request: Flow<SubscribeRequest>): Flow<Envelope> {
-        val headers = Metadata()
+    private fun batchQueryResponseFromFFi(response: FfiV2BatchQueryResponse): BatchQueryResponse {
+        return BatchQueryResponse.newBuilder().also { queryResponse ->
+            queryResponse.addAllResponses(response.responses.map { queryResponseFromFFi(it) })
+        }.build()
+    }
+
+    private fun pagingInfoFromFFi(info: FfiPagingInfo): PagingInfo {
+        return PagingInfo.newBuilder().also {
+            it.limit = info.limit.toInt()
+            info.cursor?.let { cursor ->
+                it.cursor = cursorFromFFi(cursor)
+            }
+            it.direction = directionFromFfi(info.direction)
+        }.build()
+    }
 
-        headers.put(CLIENT_VERSION_HEADER_KEY, Constants.VERSION)
-        if (appVersion != null) {
-            headers.put(APP_VERSION_HEADER_KEY, appVersion)
+    private fun pagingInfoToFFi(info: PagingInfo): FfiPagingInfo {
+        return FfiPagingInfo(
+            limit = info.limit.toUInt(),
+            cursor = cursorToFFi(info.cursor),
+            direction = directionToFfi(info.direction)
+        )
+    }
+
+    private fun directionToFfi(direction: SortDirection): FfiSortDirection {
+        return when (direction) {
+            SortDirection.SORT_DIRECTION_ASCENDING -> FfiSortDirection.ASCENDING
+            SortDirection.SORT_DIRECTION_DESCENDING -> FfiSortDirection.DESCENDING
+            else -> FfiSortDirection.UNSPECIFIED
+        }
+    }
+
+    private fun directionFromFfi(direction: FfiSortDirection): SortDirection {
+        return when (direction) {
+            FfiSortDirection.ASCENDING -> SortDirection.SORT_DIRECTION_ASCENDING
+            FfiSortDirection.DESCENDING -> SortDirection.SORT_DIRECTION_DESCENDING
+            else -> SortDirection.SORT_DIRECTION_UNSPECIFIED
         }
+    }
 
-        return client.subscribe2(request, headers)
+    private fun cursorToFFi(cursor: Cursor): FfiCursor {
+        return FfiCursor(
+            digest = cursor.index.digest.toByteArray(),
+            senderTimeNs = cursor.index.senderTimeNs.toULong()
+        )
     }
 
-    override fun close() {
-        channel.shutdown().awaitTermination(5, TimeUnit.SECONDS)
+    private fun cursorFromFFi(cursor: FfiCursor): Cursor {
+        return Cursor.newBuilder().also {
+            it.index.toBuilder().also { index ->
+                index.digest = cursor.digest.toByteString()
+                index.senderTimeNs = cursor.senderTimeNs.toLong()
+            }.build()
+        }.build()
     }
 }
diff --git a/library/src/main/java/org/xmtp/android/library/Client.kt b/library/src/main/java/org/xmtp/android/library/Client.kt
index 099603023..d1d855d1e 100644
--- a/library/src/main/java/org/xmtp/android/library/Client.kt
+++ b/library/src/main/java/org/xmtp/android/library/Client.kt
@@ -8,13 +8,10 @@ import android.util.Log
 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.MessageV3
@@ -44,8 +41,12 @@ import org.xmtp.android.library.messages.walletAddress
 import org.xmtp.proto.message.api.v1.MessageApiOuterClass
 import org.xmtp.proto.message.api.v1.MessageApiOuterClass.BatchQueryResponse
 import org.xmtp.proto.message.api.v1.MessageApiOuterClass.QueryRequest
+import uniffi.xmtpv3.FfiV2SubscribeRequest
+import uniffi.xmtpv3.FfiV2Subscription
+import uniffi.xmtpv3.FfiV2SubscriptionCallback
 import uniffi.xmtpv3.FfiXmtpClient
 import uniffi.xmtpv3.createClient
+import uniffi.xmtpv3.createV2Client
 import uniffi.xmtpv3.generateInboxId
 import uniffi.xmtpv3.getInboxIdForAddress
 import uniffi.xmtpv3.getVersionInfo
@@ -150,16 +151,17 @@ class Client() {
             )
         }
 
-        fun canMessage(peerAddress: String, options: ClientOptions? = null): Boolean {
+        suspend fun canMessage(peerAddress: String, options: ClientOptions? = null): Boolean {
             val clientOptions = options ?: ClientOptions()
-            val api = GRPCApiClient(
-                environment = clientOptions.api.env,
-                secure = clientOptions.api.isSecure,
-            )
-            return runBlocking {
-                val topics = api.queryTopic(Topic.contact(peerAddress)).envelopesList
-                topics.isNotEmpty()
-            }
+            val v2Client =
+                createV2Client(
+                    host = clientOptions.api.env.getUrl(),
+                    isSecure = clientOptions.api.isSecure
+                )
+            clientOptions.api.appVersion?.let { v2Client.setAppVersion(it) }
+            val api = GRPCApiClient(environment = clientOptions.api.env, rustV2Client = v2Client)
+            val topics = api.queryTopic(Topic.contact(peerAddress)).envelopesList
+            return topics.isNotEmpty()
         }
     }
 
@@ -197,11 +199,14 @@ class Client() {
         options: ClientOptions? = null,
     ): Client {
         val clientOptions = options ?: ClientOptions()
-        val apiClient =
-            GRPCApiClient(
-                environment = clientOptions.api.env,
-                secure = clientOptions.api.isSecure,
+        val v2Client = runBlocking {
+            createV2Client(
+                host = clientOptions.api.env.getUrl(),
+                isSecure = clientOptions.api.isSecure
             )
+        }
+        clientOptions.api.appVersion?.let { v2Client.setAppVersion(it) }
+        val apiClient = GRPCApiClient(environment = clientOptions.api.env, rustV2Client = v2Client)
         return create(
             account = account,
             apiClient = apiClient,
@@ -264,11 +269,14 @@ class Client() {
     ): Client {
         val address = v1Bundle.identityKey.publicKey.recoverWalletSignerPublicKey().walletAddress
         val newOptions = options ?: ClientOptions()
-        val apiClient =
-            GRPCApiClient(
-                environment = newOptions.api.env,
-                secure = newOptions.api.isSecure,
+        val v2Client = runBlocking {
+            createV2Client(
+                host = newOptions.api.env.getUrl(),
+                isSecure = newOptions.api.isSecure
             )
+        }
+        newOptions.api.appVersion?.let { v2Client.setAppVersion(it) }
+        val apiClient = GRPCApiClient(environment = newOptions.api.env, rustV2Client = v2Client)
         val (v3Client, dbPath) = if (isV3Enabled(options)) {
             runBlocking {
                 ffiXmtpClient(
@@ -500,12 +508,18 @@ class Client() {
         return apiClient.batchQuery(requests)
     }
 
-    suspend fun subscribe(topics: List<String>): Flow<Envelope> {
-        return subscribe2(flowOf(makeSubscribeRequest(topics)))
+    suspend fun subscribe(
+        topics: List<String>,
+        callback: FfiV2SubscriptionCallback,
+    ): FfiV2Subscription {
+        return subscribe2(FfiV2SubscribeRequest(topics), callback)
     }
 
-    suspend fun subscribe2(request: Flow<MessageApiOuterClass.SubscribeRequest>): Flow<Envelope> {
-        return apiClient.subscribe(request = request)
+    suspend fun subscribe2(
+        request: FfiV2SubscribeRequest,
+        callback: FfiV2SubscriptionCallback,
+    ): FfiV2Subscription {
+        return apiClient.subscribe(request, callback)
     }
 
     suspend fun fetchConversation(
@@ -540,7 +554,7 @@ class Client() {
         throw XMTPException("Error no V3 client initialized")
     }
 
-    suspend fun publish(envelopes: List<Envelope>): PublishResponse {
+    suspend fun publish(envelopes: List<Envelope>) {
         val authorized = AuthorizedIdentity(
             address = address,
             authorized = privateKeyBundleV1.identityKey.publicKey,
@@ -549,7 +563,7 @@ class Client() {
         val authToken = authorized.createAuthToken()
         apiClient.setAuthToken(authToken)
 
-        return apiClient.publish(envelopes = envelopes)
+        apiClient.publish(envelopes = envelopes)
     }
 
     suspend fun ensureUserContactPublished() {
@@ -623,8 +637,8 @@ class Client() {
      * @return false when [peerAddress] has never signed up for XMTP
      * or when the message is addressed to the sender (no self-messaging).
      */
-    fun canMessage(peerAddress: String): Boolean {
-        return runBlocking { query(Topic.contact(peerAddress)).envelopesList.size > 0 }
+    suspend fun canMessage(peerAddress: String): Boolean {
+        return query(Topic.contact(peerAddress)).envelopesList.size > 0
     }
 
     suspend fun canMessageV3(addresses: List<String>): Map<String, Boolean> {
diff --git a/library/src/main/java/org/xmtp/android/library/Contacts.kt b/library/src/main/java/org/xmtp/android/library/Contacts.kt
index a4e5576fe..69122745e 100644
--- a/library/src/main/java/org/xmtp/android/library/Contacts.kt
+++ b/library/src/main/java/org/xmtp/android/library/Contacts.kt
@@ -127,7 +127,7 @@ class ConsentList(
 
     suspend fun publish(entries: List<ConsentListEntry>) {
         val payload = PrivatePreferencesAction.newBuilder().also {
-            entries.forEach { entry ->
+            entries.iterator().forEach { entry ->
                 when (entry.entryType) {
                     ConsentListEntry.EntryType.ADDRESS -> {
                         when (entry.consentType) {
diff --git a/library/src/main/java/org/xmtp/android/library/ConversationV1.kt b/library/src/main/java/org/xmtp/android/library/ConversationV1.kt
index 196336975..3ce2b7a1a 100644
--- a/library/src/main/java/org/xmtp/android/library/ConversationV1.kt
+++ b/library/src/main/java/org/xmtp/android/library/ConversationV1.kt
@@ -1,9 +1,12 @@
 package org.xmtp.android.library
 
 import android.util.Log
+import kotlinx.coroutines.channels.awaitClose
 import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.callbackFlow
+import kotlinx.coroutines.launch
 import org.web3j.crypto.Hash
+import org.xmtp.android.library.Util.Companion.envelopeFromFFi
 import org.xmtp.android.library.codecs.ContentCodec
 import org.xmtp.android.library.codecs.EncodedContent
 import org.xmtp.android.library.codecs.compress
@@ -22,6 +25,8 @@ 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 uniffi.xmtpv3.FfiEnvelope
+import uniffi.xmtpv3.FfiV2SubscriptionCallback
 import java.util.Date
 
 data class ConversationV1(
@@ -39,10 +44,14 @@ data class ConversationV1(
      * current [Client] as userInvite and userIntro
      * @see Conversations.streamAllMessages
      */
-    fun streamMessages(): Flow<DecodedMessage> = flow {
-        client.subscribe(listOf(topic.description)).collect {
-            emit(decode(envelope = it))
+    fun streamMessages(): Flow<DecodedMessage> = callbackFlow {
+        val streamCallback = object : FfiV2SubscriptionCallback {
+            override fun onMessage(message: FfiEnvelope) {
+                trySend(decode(envelope = envelopeFromFFi(message)))
+            }
         }
+        val stream = client.subscribe(listOf(topic.description), streamCallback)
+        awaitClose { launch { stream.end() } }
     }
 
     /**
@@ -267,15 +276,23 @@ data class ConversationV1(
     val ephemeralTopic: String
         get() = topic.description.replace("/xmtp/0/dm-", "/xmtp/0/dmE-")
 
-    fun streamEphemeral(): Flow<Envelope> = flow {
-        client.subscribe(listOf(ephemeralTopic)).collect {
-            emit(it)
+    fun streamEphemeral(): Flow<Envelope> = callbackFlow {
+        val streamCallback = object : FfiV2SubscriptionCallback {
+            override fun onMessage(message: FfiEnvelope) {
+                trySend(envelopeFromFFi(message))
+            }
         }
+        val stream = client.subscribe(listOf(ephemeralTopic), streamCallback)
+        awaitClose { launch { stream.end() } }
     }
 
-    fun streamDecryptedMessages(): Flow<DecryptedMessage> = flow {
-        client.subscribe(listOf(topic.description)).collect {
-            emit(decrypt(envelope = it))
+    fun streamDecryptedMessages(): Flow<DecryptedMessage> = callbackFlow {
+        val streamCallback = object : FfiV2SubscriptionCallback {
+            override fun onMessage(message: FfiEnvelope) {
+                trySend(decrypt(envelope = envelopeFromFFi(message)))
+            }
         }
+        val stream = client.subscribe(listOf(topic.description), streamCallback)
+        awaitClose { launch { stream.end() } }
     }
 }
diff --git a/library/src/main/java/org/xmtp/android/library/ConversationV2.kt b/library/src/main/java/org/xmtp/android/library/ConversationV2.kt
index e11fda4f0..e021226ef 100644
--- a/library/src/main/java/org/xmtp/android/library/ConversationV2.kt
+++ b/library/src/main/java/org/xmtp/android/library/ConversationV2.kt
@@ -1,9 +1,10 @@
 package org.xmtp.android.library
 
 import android.util.Log
+import kotlinx.coroutines.channels.awaitClose
 import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.flow
-import kotlinx.coroutines.flow.mapNotNull
+import kotlinx.coroutines.flow.callbackFlow
+import kotlinx.coroutines.launch
 import org.web3j.crypto.Hash
 import org.xmtp.android.library.codecs.ContentCodec
 import org.xmtp.android.library.codecs.EncodedContent
@@ -21,6 +22,8 @@ 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 uniffi.xmtpv3.FfiEnvelope
+import uniffi.xmtpv3.FfiV2SubscriptionCallback
 import java.util.Date
 
 data class ConversationV2(
@@ -137,10 +140,16 @@ data class ConversationV2(
         )
     }
 
-    fun streamMessages(): Flow<DecodedMessage> = flow {
-        client.subscribe(listOf(topic)).mapNotNull { decodeEnvelopeOrNull(envelope = it) }.collect {
-            emit(it)
+    fun streamMessages(): Flow<DecodedMessage> = callbackFlow {
+        val streamCallback = object : FfiV2SubscriptionCallback {
+            override fun onMessage(message: FfiEnvelope) {
+                decodeEnvelopeOrNull(envelope = Util.envelopeFromFFi(message))?.let {
+                    trySend(it)
+                }
+            }
         }
+        val stream = client.subscribe(listOf(topic), streamCallback)
+        awaitClose { launch { stream.end() } }
     }
 
     /**
@@ -268,15 +277,23 @@ data class ConversationV2(
     val ephemeralTopic: String
         get() = topic.replace("/xmtp/0/m", "/xmtp/0/mE")
 
-    fun streamEphemeral(): Flow<Envelope> = flow {
-        client.subscribe(listOf(ephemeralTopic)).collect {
-            emit(it)
+    fun streamEphemeral(): Flow<Envelope> = callbackFlow {
+        val streamCallback = object : FfiV2SubscriptionCallback {
+            override fun onMessage(message: FfiEnvelope) {
+                trySend(Util.envelopeFromFFi(message))
+            }
         }
+        val stream = client.subscribe(listOf(ephemeralTopic), streamCallback)
+        awaitClose { launch { stream.end() } }
     }
 
-    fun streamDecryptedMessages(): Flow<DecryptedMessage> = flow {
-        client.subscribe(listOf(topic)).collect {
-            emit(decrypt(envelope = it))
+    fun streamDecryptedMessages(): Flow<DecryptedMessage> = callbackFlow {
+        val streamCallback = object : FfiV2SubscriptionCallback {
+            override fun onMessage(message: FfiEnvelope) {
+                trySend(decrypt(envelope = Util.envelopeFromFFi(message)))
+            }
         }
+        val stream = client.subscribe(listOf(topic), streamCallback)
+        awaitClose { launch { stream.end() } }
     }
 }
diff --git a/library/src/main/java/org/xmtp/android/library/Conversations.kt b/library/src/main/java/org/xmtp/android/library/Conversations.kt
index 7733b5205..928c8a560 100644
--- a/library/src/main/java/org/xmtp/android/library/Conversations.kt
+++ b/library/src/main/java/org/xmtp/android/library/Conversations.kt
@@ -2,16 +2,13 @@ package org.xmtp.android.library
 
 import android.util.Log
 import com.google.protobuf.kotlin.toByteString
-import io.grpc.StatusException
-import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.channels.awaitClose
 import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.callbackFlow
-import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.flow.merge
+import kotlinx.coroutines.launch
 import org.xmtp.android.library.GRPCApiClient.Companion.makeQueryRequest
-import org.xmtp.android.library.GRPCApiClient.Companion.makeSubscribeRequest
+import org.xmtp.android.library.Util.Companion.envelopeFromFFi
 import org.xmtp.android.library.libxmtp.MessageV3
 import org.xmtp.android.library.messages.DecryptedMessage
 import org.xmtp.android.library.messages.Envelope
@@ -42,11 +39,16 @@ import org.xmtp.proto.message.contents.Invitation
 import uniffi.xmtpv3.FfiConversationCallback
 import uniffi.xmtpv3.FfiConversations
 import uniffi.xmtpv3.FfiCreateGroupOptions
+import uniffi.xmtpv3.FfiEnvelope
 import uniffi.xmtpv3.FfiGroup
 import uniffi.xmtpv3.FfiListConversationsOptions
 import uniffi.xmtpv3.FfiMessage
 import uniffi.xmtpv3.FfiMessageCallback
+import uniffi.xmtpv3.FfiV2SubscribeRequest
+import uniffi.xmtpv3.FfiV2Subscription
+import uniffi.xmtpv3.FfiV2SubscriptionCallback
 import uniffi.xmtpv3.GroupPermissions
+import uniffi.xmtpv3.NoPointer
 import java.util.Date
 import kotlin.time.Duration.Companion.nanoseconds
 import kotlin.time.DurationUnit
@@ -354,23 +356,24 @@ data class Conversations(
             }.toMutableMap()
         }
 
-        topics.forEach {
+        topics.iterator().forEach {
             val conversation = it.value
             val hmacKeys = HmacKeys.newBuilder()
             if (conversation.keyMaterial != null) {
-                (thirtyDayPeriodsSinceEpoch - 1..thirtyDayPeriodsSinceEpoch + 1).forEach { value ->
-                    val info = "$value-${client.address}"
-                    val hmacKey =
-                        Crypto.deriveKey(
-                            conversation.keyMaterial!!,
-                            ByteArray(0),
-                            info.toByteArray(Charsets.UTF_8),
-                        )
-                    val hmacKeyData = HmacKeyData.newBuilder()
-                    hmacKeyData.hmacKey = hmacKey.toByteString()
-                    hmacKeyData.thirtyDayPeriodsSinceEpoch = value
-                    hmacKeys.addValues(hmacKeyData)
-                }
+                (thirtyDayPeriodsSinceEpoch - 1..thirtyDayPeriodsSinceEpoch + 1).iterator()
+                    .forEach { value ->
+                        val info = "$value-${client.address}"
+                        val hmacKey =
+                            Crypto.deriveKey(
+                                conversation.keyMaterial!!,
+                                ByteArray(0),
+                                info.toByteArray(Charsets.UTF_8),
+                            )
+                        val hmacKeyData = HmacKeyData.newBuilder()
+                        hmacKeyData.hmacKey = hmacKey.toByteString()
+                        hmacKeyData.thirtyDayPeriodsSinceEpoch = value
+                        hmacKeys.addValues(hmacKeyData)
+                    }
                 hmacKeysResponse.putHmacKeys(conversation.topic, hmacKeys.build())
             }
         }
@@ -555,30 +558,40 @@ data class Conversations(
      * of the information of those conversations according to the topics
      * @return Stream of data information for the conversations
      */
-    fun stream(): Flow<Conversation> = flow {
+    fun stream(): Flow<Conversation> = callbackFlow {
         val streamedConversationTopics: MutableSet<String> = mutableSetOf()
-        client.subscribe(
-            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)
-                if (!streamedConversationTopics.contains(conversationV1.topic)) {
-                    streamedConversationTopics.add(conversationV1.topic)
-                    emit(conversationV1)
+        val subscriptionCallback = object : FfiV2SubscriptionCallback {
+            override fun onMessage(message: FfiEnvelope) {
+                val envelope = envelopeFromFFi(message)
+                if (envelope.contentTopic == Topic.userIntro(client.address).description) {
+                    val conversationV1 = fromIntro(envelope = envelope)
+                    if (!streamedConversationTopics.contains(conversationV1.topic)) {
+                        streamedConversationTopics.add(conversationV1.topic)
+                        trySend(conversationV1)
+                    }
                 }
-            }
 
-            if (envelope.contentTopic == Topic.userInvite(client.address).description) {
-                val conversationV2 = fromInvite(envelope = envelope)
-                if (!streamedConversationTopics.contains(conversationV2.topic)) {
-                    streamedConversationTopics.add(conversationV2.topic)
-                    emit(conversationV2)
+                if (envelope.contentTopic == Topic.userInvite(client.address).description) {
+                    val conversationV2 = fromInvite(envelope = envelope)
+                    if (!streamedConversationTopics.contains(conversationV2.topic)) {
+                        streamedConversationTopics.add(conversationV2.topic)
+                        trySend(conversationV2)
+                    }
                 }
             }
         }
+
+        val stream = client.subscribe2(
+            FfiV2SubscribeRequest(
+                listOf(
+                    Topic.userIntro(client.address).description,
+                    Topic.userInvite(client.address).description
+                )
+            ),
+            subscriptionCallback
+        )
+
+        awaitClose { launch { stream.end() } }
     }
 
     fun streamAll(): Flow<Conversation> {
@@ -640,7 +653,7 @@ data class Conversations(
      * @return Flow object of [DecodedMessage] that represents all the messages of the
      * current [Client] as userInvite and userIntro
      */
-    private fun streamAllV2Messages(): Flow<DecodedMessage> = flow {
+    private fun streamAllV2Messages(): Flow<DecodedMessage> = callbackFlow {
         val topics = mutableListOf(
             Topic.userInvite(client.address).description,
             Topic.userIntro(client.address).description,
@@ -650,49 +663,44 @@ data class Conversations(
             topics.add(conversation.topic)
         }
 
-        val subscribeFlow = MutableStateFlow(makeSubscribeRequest(topics))
-
-        while (true) {
-            try {
-                client.subscribe2(request = subscribeFlow).collect { envelope ->
-                    when {
-                        conversationsByTopic.containsKey(envelope.contentTopic) -> {
-                            val conversation = conversationsByTopic[envelope.contentTopic]
-                            val decoded = conversation?.decode(envelope)
-                            decoded?.let { emit(it) }
-                        }
+        val subscriptionRequest = FfiV2SubscribeRequest(topics)
+        var stream = FfiV2Subscription(NoPointer)
 
-                        envelope.contentTopic.startsWith("/xmtp/0/invite-") -> {
-                            val conversation = fromInvite(envelope = envelope)
-                            conversationsByTopic[conversation.topic] = conversation
-                            topics.add(conversation.topic)
-                            subscribeFlow.value = makeSubscribeRequest(topics)
-                        }
+        val subscriptionCallback = object : FfiV2SubscriptionCallback {
+            override fun onMessage(message: FfiEnvelope) {
+                when {
+                    conversationsByTopic.containsKey(message.contentTopic) -> {
+                        val conversation = conversationsByTopic[message.contentTopic]
+                        val decoded = conversation?.decode(envelopeFromFFi(message))
+                        decoded?.let { trySend(it) }
+                    }
 
-                        envelope.contentTopic.startsWith("/xmtp/0/intro-") -> {
-                            val conversation = fromIntro(envelope = envelope)
-                            conversationsByTopic[conversation.topic] = conversation
-                            val decoded = conversation.decode(envelope)
-                            emit(decoded)
-                            topics.add(conversation.topic)
-                            subscribeFlow.value = makeSubscribeRequest(topics)
-                        }
+                    message.contentTopic.startsWith("/xmtp/0/invite-") -> {
+                        val conversation = fromInvite(envelope = envelopeFromFFi(message))
+                        conversationsByTopic[conversation.topic] = conversation
+                        topics.add(conversation.topic)
+                        subscriptionRequest.contentTopics = topics
+                        launch { stream.update(subscriptionRequest) }
+                    }
 
-                        else -> {}
+                    message.contentTopic.startsWith("/xmtp/0/intro-") -> {
+                        val conversation = fromIntro(envelope = envelopeFromFFi(message))
+                        conversationsByTopic[conversation.topic] = conversation
+                        val decoded = conversation.decode(envelopeFromFFi(message))
+                        trySend(decoded)
+                        topics.add(conversation.topic)
+                        subscriptionRequest.contentTopics = topics
+                        launch { stream.update(subscriptionRequest) }
                     }
+
+                    else -> {}
                 }
-            } catch (error: CancellationException) {
-                break
-            } catch (error: StatusException) {
-                if (error.status.code == io.grpc.Status.Code.UNAVAILABLE) {
-                    continue
-                } else {
-                    break
-                }
-            } catch (error: Exception) {
-                continue
             }
         }
+
+        stream = client.subscribe2(subscriptionRequest, subscriptionCallback)
+
+        awaitClose { launch { stream.end() } }
     }
 
     fun streamAllMessages(includeGroups: Boolean = false): Flow<DecodedMessage> {
@@ -711,7 +719,7 @@ data class Conversations(
         }
     }
 
-    private fun streamAllV2DecryptedMessages(): Flow<DecryptedMessage> = flow {
+    private fun streamAllV2DecryptedMessages(): Flow<DecryptedMessage> = callbackFlow {
         val topics = mutableListOf(
             Topic.userInvite(client.address).description,
             Topic.userIntro(client.address).description,
@@ -721,48 +729,43 @@ data class Conversations(
             topics.add(conversation.topic)
         }
 
-        val subscribeFlow = MutableStateFlow(makeSubscribeRequest(topics))
+        val subscriptionRequest = FfiV2SubscribeRequest(topics)
+        var stream = FfiV2Subscription(NoPointer)
 
-        while (true) {
-            try {
-                client.subscribe2(request = subscribeFlow).collect { envelope ->
-                    when {
-                        conversationsByTopic.containsKey(envelope.contentTopic) -> {
-                            val conversation = conversationsByTopic[envelope.contentTopic]
-                            val decrypted = conversation?.decrypt(envelope)
-                            decrypted?.let { emit(it) }
-                        }
-
-                        envelope.contentTopic.startsWith("/xmtp/0/invite-") -> {
-                            val conversation = fromInvite(envelope = envelope)
-                            conversationsByTopic[conversation.topic] = conversation
-                            topics.add(conversation.topic)
-                            subscribeFlow.value = makeSubscribeRequest(topics)
-                        }
+        val subscriptionCallback = object : FfiV2SubscriptionCallback {
+            override fun onMessage(message: FfiEnvelope) {
+                when {
+                    conversationsByTopic.containsKey(message.contentTopic) -> {
+                        val conversation = conversationsByTopic[message.contentTopic]
+                        val decrypted = conversation?.decrypt(envelopeFromFFi(message))
+                        decrypted?.let { trySend(it) }
+                    }
 
-                        envelope.contentTopic.startsWith("/xmtp/0/intro-") -> {
-                            val conversation = fromIntro(envelope = envelope)
-                            conversationsByTopic[conversation.topic] = conversation
-                            val decrypted = conversation.decrypt(envelope)
-                            emit(decrypted)
-                            topics.add(conversation.topic)
-                            subscribeFlow.value = makeSubscribeRequest(topics)
-                        }
+                    message.contentTopic.startsWith("/xmtp/0/invite-") -> {
+                        val conversation = fromInvite(envelope = envelopeFromFFi(message))
+                        conversationsByTopic[conversation.topic] = conversation
+                        topics.add(conversation.topic)
+                        subscriptionRequest.contentTopics = topics
+                        launch { stream.update(subscriptionRequest) }
+                    }
 
-                        else -> {}
+                    message.contentTopic.startsWith("/xmtp/0/intro-") -> {
+                        val conversation = fromIntro(envelope = envelopeFromFFi(message))
+                        conversationsByTopic[conversation.topic] = conversation
+                        val decrypted = conversation.decrypt(envelopeFromFFi(message))
+                        trySend(decrypted)
+                        topics.add(conversation.topic)
+                        subscriptionRequest.contentTopics = topics
+                        launch { stream.update(subscriptionRequest) }
                     }
+
+                    else -> {}
                 }
-            } catch (error: CancellationException) {
-                break
-            } catch (error: StatusException) {
-                if (error.status.code == io.grpc.Status.Code.UNAVAILABLE) {
-                    continue
-                } else {
-                    break
-                }
-            } catch (error: Exception) {
-                continue
             }
         }
+
+        stream = client.subscribe2(subscriptionRequest, subscriptionCallback)
+
+        awaitClose { launch { stream.end() } }
     }
 }
diff --git a/library/src/main/java/org/xmtp/android/library/Util.kt b/library/src/main/java/org/xmtp/android/library/Util.kt
index 4b1420d69..c902fdcaa 100644
--- a/library/src/main/java/org/xmtp/android/library/Util.kt
+++ b/library/src/main/java/org/xmtp/android/library/Util.kt
@@ -1,6 +1,10 @@
 package org.xmtp.android.library
 
+import com.google.protobuf.kotlin.toByteString
 import org.bouncycastle.jcajce.provider.digest.Keccak
+import org.web3j.utils.Numeric
+import org.xmtp.proto.message.api.v1.MessageApiOuterClass
+import uniffi.xmtpv3.FfiEnvelope
 
 class Util {
     companion object {
@@ -8,7 +12,17 @@ class Util {
             val digest256 = Keccak.Digest256()
             return digest256.digest(data)
         }
+
+        fun envelopeFromFFi(envelope: FfiEnvelope): MessageApiOuterClass.Envelope {
+            return MessageApiOuterClass.Envelope.newBuilder().also {
+                it.contentTopic = envelope.contentTopic
+                it.timestampNs = envelope.timestampNs.toLong()
+                it.message = envelope.message.toByteString()
+            }.build()
+        }
     }
 }
 
 fun ByteArray.toHex(): String = joinToString(separator = "") { eachByte -> "%02x".format(eachByte) }
+
+fun String.hexToByteArray(): ByteArray = Numeric.hexStringToByteArray(this)
diff --git a/library/src/main/java/xmtpv3.kt b/library/src/main/java/xmtpv3.kt
index 1bac1275a..f7de6734f 100644
--- a/library/src/main/java/xmtpv3.kt
+++ b/library/src/main/java/xmtpv3.kt
@@ -49,10 +49,8 @@ open class RustBuffer : Structure() {
     // When dealing with these fields, make sure to call `toULong()`.
     @JvmField
     var capacity: Long = 0
-
     @JvmField
     var len: Long = 0
-
     @JvmField
     var data: Pointer? = null
 
@@ -137,7 +135,6 @@ class RustBufferByReference : ByReference(16) {
 open class ForeignBytes : Structure() {
     @JvmField
     var len: Int = 0
-
     @JvmField
     var data: Pointer? = null
 
@@ -226,7 +223,6 @@ internal const val UNIFFI_CALL_UNEXPECTED_ERROR = 2.toByte()
 internal open class UniffiRustCallStatus : Structure() {
     @JvmField
     var code: Byte = 0
-
     @JvmField
     var error_buf: RustBuffer.ByValue = RustBuffer.ByValue()
 
@@ -268,7 +264,7 @@ interface UniffiRustCallStatusErrorHandler<E> {
 // Call a rust function that returns a Result<>.  Pass in the Error class companion that corresponds to the Err
 private inline fun <U, E : Exception> uniffiRustCallWithError(
     errorHandler: UniffiRustCallStatusErrorHandler<E>,
-    callback: (UniffiRustCallStatus) -> U
+    callback: (UniffiRustCallStatus) -> U,
 ): U {
     var status = UniffiRustCallStatus();
     val return_value = callback(status)
@@ -279,7 +275,7 @@ private inline fun <U, E : Exception> uniffiRustCallWithError(
 // Check UniffiRustCallStatus and throw an error if the call wasn't successful
 private fun <E : Exception> uniffiCheckCallStatus(
     errorHandler: UniffiRustCallStatusErrorHandler<E>,
-    status: UniffiRustCallStatus
+    status: UniffiRustCallStatus,
 ) {
     if (status.isSuccess()) {
         return
@@ -329,7 +325,7 @@ internal inline fun <T, reified E : Throwable> uniffiTraitInterfaceCallWithError
     callStatus: UniffiRustCallStatus,
     makeCall: () -> T,
     writeReturn: (T) -> Unit,
-    lowerError: (E) -> RustBuffer.ByValue
+    lowerError: (E) -> RustBuffer.ByValue,
 ) {
     try {
         writeReturn(makeCall())
@@ -384,7 +380,7 @@ private fun findLibraryName(componentName: String): String {
 }
 
 private inline fun <reified Lib : Library> loadIndirect(
-    componentName: String
+    componentName: String,
 ): Lib {
     return Native.load<Lib>(findLibraryName(componentName), Lib::class.java)
 }
@@ -738,6 +734,15 @@ internal interface UniffiCallbackInterfaceFfiMessageCallbackMethod0 : com.sun.jn
     )
 }
 
+internal interface UniffiCallbackInterfaceFfiV2SubscriptionCallbackMethod0 : com.sun.jna.Callback {
+    fun callback(
+        `uniffiHandle`: Long,
+        `message`: RustBuffer.ByValue,
+        `uniffiOutReturn`: Pointer,
+        uniffiCallStatus: UniffiRustCallStatus,
+    )
+}
+
 @Structure.FieldOrder("getAddress", "sign", "uniffiFree")
 internal open class UniffiVTableCallbackInterfaceFfiInboxOwner(
     @JvmField internal var `getAddress`: UniffiCallbackInterfaceFfiInboxOwnerMethod0? = null,
@@ -812,6 +817,24 @@ internal open class UniffiVTableCallbackInterfaceFfiMessageCallback(
 
 }
 
+@Structure.FieldOrder("onMessage", "uniffiFree")
+internal open class UniffiVTableCallbackInterfaceFfiV2SubscriptionCallback(
+    @JvmField internal var `onMessage`: UniffiCallbackInterfaceFfiV2SubscriptionCallbackMethod0? = null,
+    @JvmField internal var `uniffiFree`: UniffiCallbackInterfaceFree? = null,
+) : Structure() {
+    class UniffiByValue(
+        `onMessage`: UniffiCallbackInterfaceFfiV2SubscriptionCallbackMethod0? = null,
+        `uniffiFree`: UniffiCallbackInterfaceFree? = null,
+    ) : UniffiVTableCallbackInterfaceFfiV2SubscriptionCallback(`onMessage`, `uniffiFree`),
+        Structure.ByValue
+
+    internal fun uniffiSetValue(other: UniffiVTableCallbackInterfaceFfiV2SubscriptionCallback) {
+        `onMessage` = other.`onMessage`
+        `uniffiFree` = other.`uniffiFree`
+    }
+
+}
+
 
 // A JNA Library to expose the extern-C FFI definitions.
 // This is an implementation detail which will be called internally by the public API.
@@ -827,6 +850,7 @@ internal interface UniffiLib : Library {
                     uniffiCallbackInterfaceFfiInboxOwner.register(lib)
                     uniffiCallbackInterfaceFfiLogger.register(lib)
                     uniffiCallbackInterfaceFfiMessageCallback.register(lib)
+                    uniffiCallbackInterfaceFfiV2SubscriptionCallback.register(lib)
                 }
         }
 
@@ -1088,7 +1112,7 @@ internal interface UniffiLib : Library {
     ): Unit
 
     fun uniffi_xmtpv3_fn_method_ffiv2apiclient_subscribe(
-        `ptr`: Pointer, `request`: RustBuffer.ByValue,
+        `ptr`: Pointer, `request`: RustBuffer.ByValue, `callback`: Long,
     ): Long
 
     fun uniffi_xmtpv3_fn_clone_ffiv2subscription(
@@ -1103,9 +1127,9 @@ internal interface UniffiLib : Library {
         `ptr`: Pointer,
     ): Long
 
-    fun uniffi_xmtpv3_fn_method_ffiv2subscription_next(
-        `ptr`: Pointer,
-    ): Long
+    fun uniffi_xmtpv3_fn_method_ffiv2subscription_is_closed(
+        `ptr`: Pointer, uniffi_out_err: UniffiRustCallStatus,
+    ): Byte
 
     fun uniffi_xmtpv3_fn_method_ffiv2subscription_update(
         `ptr`: Pointer, `req`: RustBuffer.ByValue,
@@ -1183,6 +1207,10 @@ internal interface UniffiLib : Library {
         `vtable`: UniffiVTableCallbackInterfaceFfiMessageCallback,
     ): Unit
 
+    fun uniffi_xmtpv3_fn_init_callback_vtable_ffiv2subscriptioncallback(
+        `vtable`: UniffiVTableCallbackInterfaceFfiV2SubscriptionCallback,
+    ): Unit
+
     fun uniffi_xmtpv3_fn_func_create_client(
         `logger`: Long,
         `host`: RustBuffer.ByValue,
@@ -1699,7 +1727,7 @@ internal interface UniffiLib : Library {
     fun uniffi_xmtpv3_checksum_method_ffiv2subscription_end(
     ): Short
 
-    fun uniffi_xmtpv3_checksum_method_ffiv2subscription_next(
+    fun uniffi_xmtpv3_checksum_method_ffiv2subscription_is_closed(
     ): Short
 
     fun uniffi_xmtpv3_checksum_method_ffiv2subscription_update(
@@ -1756,6 +1784,9 @@ internal interface UniffiLib : Library {
     fun uniffi_xmtpv3_checksum_method_ffimessagecallback_on_message(
     ): Short
 
+    fun uniffi_xmtpv3_checksum_method_ffiv2subscriptioncallback_on_message(
+    ): Short
+
     fun ffi_xmtpv3_uniffi_contract_version(
     ): Int
 
@@ -1965,16 +1996,16 @@ private fun uniffiCheckApiChecksums(lib: UniffiLib) {
     if (lib.uniffi_xmtpv3_checksum_method_ffiv2apiclient_set_app_version() != 28472.toShort()) {
         throw RuntimeException("UniFFI API checksum mismatch: try cleaning and rebuilding your project")
     }
-    if (lib.uniffi_xmtpv3_checksum_method_ffiv2apiclient_subscribe() != 31004.toShort()) {
+    if (lib.uniffi_xmtpv3_checksum_method_ffiv2apiclient_subscribe() != 48530.toShort()) {
         throw RuntimeException("UniFFI API checksum mismatch: try cleaning and rebuilding your project")
     }
-    if (lib.uniffi_xmtpv3_checksum_method_ffiv2subscription_end() != 54394.toShort()) {
+    if (lib.uniffi_xmtpv3_checksum_method_ffiv2subscription_end() != 38721.toShort()) {
         throw RuntimeException("UniFFI API checksum mismatch: try cleaning and rebuilding your project")
     }
-    if (lib.uniffi_xmtpv3_checksum_method_ffiv2subscription_next() != 27536.toShort()) {
+    if (lib.uniffi_xmtpv3_checksum_method_ffiv2subscription_is_closed() != 4358.toShort()) {
         throw RuntimeException("UniFFI API checksum mismatch: try cleaning and rebuilding your project")
     }
-    if (lib.uniffi_xmtpv3_checksum_method_ffiv2subscription_update() != 16562.toShort()) {
+    if (lib.uniffi_xmtpv3_checksum_method_ffiv2subscription_update() != 24211.toShort()) {
         throw RuntimeException("UniFFI API checksum mismatch: try cleaning and rebuilding your project")
     }
     if (lib.uniffi_xmtpv3_checksum_method_ffixmtpclient_can_message() != 53502.toShort()) {
@@ -2028,6 +2059,9 @@ private fun uniffiCheckApiChecksums(lib: UniffiLib) {
     if (lib.uniffi_xmtpv3_checksum_method_ffimessagecallback_on_message() != 5286.toShort()) {
         throw RuntimeException("UniFFI API checksum mismatch: try cleaning and rebuilding your project")
     }
+    if (lib.uniffi_xmtpv3_checksum_method_ffiv2subscriptioncallback_on_message() != 30049.toShort()) {
+        throw RuntimeException("UniFFI API checksum mismatch: try cleaning and rebuilding your project")
+    }
 }
 
 // Async support
@@ -2051,7 +2085,7 @@ internal suspend fun <T, F, E : Exception> uniffiRustCallAsync(
     completeFunc: (Long, UniffiRustCallStatus) -> F,
     freeFunc: (Long) -> Unit,
     liftFunc: (F) -> T,
-    errorHandler: UniffiRustCallStatusErrorHandler<E>
+    errorHandler: UniffiRustCallStatusErrorHandler<E>,
 ): T {
     try {
         do {
@@ -2435,7 +2469,7 @@ private class JavaLangRefCleaner : UniffiCleaner {
 }
 
 private class JavaLangRefCleanable(
-    val cleanable: java.lang.ref.Cleaner.Cleanable
+    val cleanable: java.lang.ref.Cleaner.Cleanable,
 ) : UniffiCleaner.Cleanable {
     override fun clean() = cleanable.clean()
 }
@@ -2444,7 +2478,7 @@ public interface FfiConversationsInterface {
 
     suspend fun `createGroup`(
         `accountAddresses`: List<kotlin.String>,
-        `opts`: FfiCreateGroupOptions
+        `opts`: FfiCreateGroupOptions,
     ): FfiGroup
 
     suspend fun `list`(`opts`: FfiListConversationsOptions): List<FfiGroup>
@@ -2546,7 +2580,7 @@ open class FfiConversations : Disposable, AutoCloseable, FfiConversationsInterfa
     @Suppress("ASSIGNED_BUT_NEVER_ACCESSED_VARIABLE")
     override suspend fun `createGroup`(
         `accountAddresses`: List<kotlin.String>,
-        `opts`: FfiCreateGroupOptions
+        `opts`: FfiCreateGroupOptions,
     ): FfiGroup {
         return uniffiRustCallAsync(
             callWithPointer { thisPtr ->
@@ -4273,7 +4307,7 @@ public interface FfiSignatureRequestInterface {
     suspend fun `addScwSignature`(
         `signatureBytes`: kotlin.ByteArray,
         `address`: kotlin.String,
-        `chainRpcUrl`: kotlin.String
+        `chainRpcUrl`: kotlin.String,
     )
 
     suspend fun `isReady`(): kotlin.Boolean
@@ -4408,7 +4442,7 @@ open class FfiSignatureRequest : Disposable, AutoCloseable, FfiSignatureRequestI
     override suspend fun `addScwSignature`(
         `signatureBytes`: kotlin.ByteArray,
         `address`: kotlin.String,
-        `chainRpcUrl`: kotlin.String
+        `chainRpcUrl`: kotlin.String,
     ) {
         return uniffiRustCallAsync(
             callWithPointer { thisPtr ->
@@ -4919,7 +4953,10 @@ public interface FfiV2ApiClientInterface {
 
     fun `setAppVersion`(`version`: kotlin.String)
 
-    suspend fun `subscribe`(`request`: FfiV2SubscribeRequest): FfiV2Subscription
+    suspend fun `subscribe`(
+        `request`: FfiV2SubscribeRequest,
+        `callback`: FfiV2SubscriptionCallback,
+    ): FfiV2Subscription
 
     companion object
 }
@@ -5115,12 +5152,16 @@ open class FfiV2ApiClient : Disposable, AutoCloseable, FfiV2ApiClientInterface {
 
     @Throws(GenericException::class)
     @Suppress("ASSIGNED_BUT_NEVER_ACCESSED_VARIABLE")
-    override suspend fun `subscribe`(`request`: FfiV2SubscribeRequest): FfiV2Subscription {
+    override suspend fun `subscribe`(
+        `request`: FfiV2SubscribeRequest,
+        `callback`: FfiV2SubscriptionCallback,
+    ): FfiV2Subscription {
         return uniffiRustCallAsync(
             callWithPointer { thisPtr ->
                 UniffiLib.INSTANCE.uniffi_xmtpv3_fn_method_ffiv2apiclient_subscribe(
                     thisPtr,
                     FfiConverterTypeFfiV2SubscribeRequest.lower(`request`),
+                    FfiConverterTypeFfiV2SubscriptionCallback.lower(`callback`),
                 )
             },
             { future, callback, continuation ->
@@ -5273,17 +5314,34 @@ public object FfiConverterTypeFfiV2ApiClient : FfiConverter<FfiV2ApiClient, Poin
 //
 
 
+/**
+ * Subscription to a stream of V2 Messages
+ */
 public interface FfiV2SubscriptionInterface {
 
+    /**
+     * End the subscription, waiting for the subscription to close entirely.
+     * # Errors
+     * * Errors if subscription event task encounters join error
+     */
     suspend fun `end`()
 
-    suspend fun `next`(): FfiEnvelope
+    /**
+     * Check if the subscription is closed
+     */
+    fun `isClosed`(): kotlin.Boolean
 
+    /**
+     * Update subscription with new topics
+     */
     suspend fun `update`(`req`: FfiV2SubscribeRequest)
 
     companion object
 }
 
+/**
+ * Subscription to a stream of V2 Messages
+ */
 open class FfiV2Subscription : Disposable, AutoCloseable, FfiV2SubscriptionInterface {
 
     constructor(pointer: Pointer) {
@@ -5366,6 +5424,12 @@ open class FfiV2Subscription : Disposable, AutoCloseable, FfiV2SubscriptionInter
     }
 
 
+    /**
+     * End the subscription, waiting for the subscription to close entirely.
+     * # Errors
+     * * Errors if subscription event task encounters join error
+     */
+    @Throws(GenericException::class)
     @Suppress("ASSIGNED_BUT_NEVER_ACCESSED_VARIABLE")
     override suspend fun `end`() {
         return uniffiRustCallAsync(
@@ -5393,43 +5457,30 @@ open class FfiV2Subscription : Disposable, AutoCloseable, FfiV2SubscriptionInter
             { Unit },
 
             // Error FFI converter
-            UniffiNullRustCallStatusErrorHandler,
+            GenericException.ErrorHandler,
         )
     }
 
 
-    @Throws(GenericException::class)
-    @Suppress("ASSIGNED_BUT_NEVER_ACCESSED_VARIABLE")
-    override suspend fun `next`(): FfiEnvelope {
-        return uniffiRustCallAsync(
-            callWithPointer { thisPtr ->
-                UniffiLib.INSTANCE.uniffi_xmtpv3_fn_method_ffiv2subscription_next(
-                    thisPtr,
-
+    /**
+     * Check if the subscription is closed
+     */
+    override fun `isClosed`(): kotlin.Boolean {
+        return FfiConverterBoolean.lift(
+            callWithPointer {
+                uniffiRustCall() { _status ->
+                    UniffiLib.INSTANCE.uniffi_xmtpv3_fn_method_ffiv2subscription_is_closed(
+                        it, _status
                     )
-            },
-            { future, callback, continuation ->
-                UniffiLib.INSTANCE.ffi_xmtpv3_rust_future_poll_rust_buffer(
-                    future,
-                    callback,
-                    continuation
-                )
-            },
-            { future, continuation ->
-                UniffiLib.INSTANCE.ffi_xmtpv3_rust_future_complete_rust_buffer(
-                    future,
-                    continuation
-                )
-            },
-            { future -> UniffiLib.INSTANCE.ffi_xmtpv3_rust_future_free_rust_buffer(future) },
-            // lift function
-            { FfiConverterTypeFfiEnvelope.lift(it) },
-            // Error FFI converter
-            GenericException.ErrorHandler,
+                }
+            }
         )
     }
 
 
+    /**
+     * Update subscription with new topics
+     */
     @Throws(GenericException::class)
     @Suppress("ASSIGNED_BUT_NEVER_ACCESSED_VARIABLE")
     override suspend fun `update`(`req`: FfiV2SubscribeRequest) {
@@ -5987,7 +6038,7 @@ public object FfiConverterTypeFfiXmtpClient : FfiConverter<FfiXmtpClient, Pointe
 data class FfiCreateGroupOptions(
     var `permissions`: GroupPermissions?,
     var `groupName`: kotlin.String?,
-    var `groupImageUrlSquare`: kotlin.String?
+    var `groupImageUrlSquare`: kotlin.String?,
 ) {
 
     companion object
@@ -6019,7 +6070,7 @@ public object FfiConverterTypeFfiCreateGroupOptions :
 
 data class FfiCursor(
     var `digest`: kotlin.ByteArray,
-    var `senderTimeNs`: kotlin.ULong
+    var `senderTimeNs`: kotlin.ULong,
 ) {
 
     companion object
@@ -6048,7 +6099,7 @@ public object FfiConverterTypeFfiCursor : FfiConverterRustBuffer<FfiCursor> {
 data class FfiEnvelope(
     var `contentTopic`: kotlin.String,
     var `timestampNs`: kotlin.ULong,
-    var `message`: kotlin.ByteArray
+    var `message`: kotlin.ByteArray,
 ) {
 
     companion object
@@ -6081,7 +6132,7 @@ data class FfiGroupMember(
     var `inboxId`: kotlin.String,
     var `accountAddresses`: List<kotlin.String>,
     var `installationIds`: List<kotlin.ByteArray>,
-    var `permissionLevel`: FfiPermissionLevel
+    var `permissionLevel`: FfiPermissionLevel,
 ) {
 
     companion object
@@ -6116,7 +6167,7 @@ public object FfiConverterTypeFfiGroupMember : FfiConverterRustBuffer<FfiGroupMe
 data class FfiListConversationsOptions(
     var `createdAfterNs`: kotlin.Long?,
     var `createdBeforeNs`: kotlin.Long?,
-    var `limit`: kotlin.Long?
+    var `limit`: kotlin.Long?,
 ) {
 
     companion object
@@ -6150,7 +6201,7 @@ data class FfiListMessagesOptions(
     var `sentBeforeNs`: kotlin.Long?,
     var `sentAfterNs`: kotlin.Long?,
     var `limit`: kotlin.Long?,
-    var `deliveryStatus`: FfiDeliveryStatus?
+    var `deliveryStatus`: FfiDeliveryStatus?,
 ) {
 
     companion object
@@ -6190,7 +6241,7 @@ data class FfiMessage(
     var `senderInboxId`: kotlin.String,
     var `content`: kotlin.ByteArray,
     var `kind`: FfiGroupMessageKind,
-    var `deliveryStatus`: FfiDeliveryStatus
+    var `deliveryStatus`: FfiDeliveryStatus,
 ) {
 
     companion object
@@ -6234,7 +6285,7 @@ public object FfiConverterTypeFfiMessage : FfiConverterRustBuffer<FfiMessage> {
 data class FfiPagingInfo(
     var `limit`: kotlin.UInt,
     var `cursor`: FfiCursor?,
-    var `direction`: FfiSortDirection
+    var `direction`: FfiSortDirection,
 ) {
 
     companion object
@@ -6264,7 +6315,7 @@ public object FfiConverterTypeFfiPagingInfo : FfiConverterRustBuffer<FfiPagingIn
 
 
 data class FfiPublishRequest(
-    var `envelopes`: List<FfiEnvelope>
+    var `envelopes`: List<FfiEnvelope>,
 ) {
 
     companion object
@@ -6288,7 +6339,7 @@ public object FfiConverterTypeFfiPublishRequest : FfiConverterRustBuffer<FfiPubl
 
 
 data class FfiV2BatchQueryRequest(
-    var `requests`: List<FfiV2QueryRequest>
+    var `requests`: List<FfiV2QueryRequest>,
 ) {
 
     companion object
@@ -6313,7 +6364,7 @@ public object FfiConverterTypeFfiV2BatchQueryRequest :
 
 
 data class FfiV2BatchQueryResponse(
-    var `responses`: List<FfiV2QueryResponse>
+    var `responses`: List<FfiV2QueryResponse>,
 ) {
 
     companion object
@@ -6341,7 +6392,7 @@ data class FfiV2QueryRequest(
     var `contentTopics`: List<kotlin.String>,
     var `startTimeNs`: kotlin.ULong,
     var `endTimeNs`: kotlin.ULong,
-    var `pagingInfo`: FfiPagingInfo?
+    var `pagingInfo`: FfiPagingInfo?,
 ) {
 
     companion object
@@ -6375,7 +6426,7 @@ public object FfiConverterTypeFfiV2QueryRequest : FfiConverterRustBuffer<FfiV2Qu
 
 data class FfiV2QueryResponse(
     var `envelopes`: List<FfiEnvelope>,
-    var `pagingInfo`: FfiPagingInfo?
+    var `pagingInfo`: FfiPagingInfo?,
 ) {
 
     companion object
@@ -6402,7 +6453,7 @@ public object FfiConverterTypeFfiV2QueryResponse : FfiConverterRustBuffer<FfiV2Q
 
 
 data class FfiV2SubscribeRequest(
-    var `contentTopics`: List<kotlin.String>
+    var `contentTopics`: List<kotlin.String>,
 ) {
 
     companion object
@@ -6970,6 +7021,57 @@ public object FfiConverterTypeFfiMessageCallback :
     FfiConverterCallbackInterface<FfiMessageCallback>()
 
 
+public interface FfiV2SubscriptionCallback {
+
+    fun `onMessage`(`message`: FfiEnvelope)
+
+    companion object
+}
+
+
+// Put the implementation in an object so we don't pollute the top-level namespace
+internal object uniffiCallbackInterfaceFfiV2SubscriptionCallback {
+    internal object `onMessage` : UniffiCallbackInterfaceFfiV2SubscriptionCallbackMethod0 {
+        override fun callback(
+            `uniffiHandle`: Long,
+            `message`: RustBuffer.ByValue,
+            `uniffiOutReturn`: Pointer,
+            uniffiCallStatus: UniffiRustCallStatus,
+        ) {
+            val uniffiObj = FfiConverterTypeFfiV2SubscriptionCallback.handleMap.get(uniffiHandle)
+            val makeCall = { ->
+                uniffiObj.`onMessage`(
+                    FfiConverterTypeFfiEnvelope.lift(`message`),
+                )
+            }
+            val writeReturn = { _: Unit -> Unit }
+            uniffiTraitInterfaceCall(uniffiCallStatus, makeCall, writeReturn)
+        }
+    }
+
+    internal object uniffiFree : UniffiCallbackInterfaceFree {
+        override fun callback(handle: Long) {
+            FfiConverterTypeFfiV2SubscriptionCallback.handleMap.remove(handle)
+        }
+    }
+
+    internal var vtable = UniffiVTableCallbackInterfaceFfiV2SubscriptionCallback.UniffiByValue(
+        `onMessage`,
+        uniffiFree,
+    )
+
+    // Registers the foreign callback with the Rust side.
+    // This method is generated for each callback interface.
+    internal fun register(lib: UniffiLib) {
+        lib.uniffi_xmtpv3_fn_init_callback_vtable_ffiv2subscriptioncallback(vtable)
+    }
+}
+
+// The ffiConverter which transforms the Callbacks in to handles to pass to Rust.
+public object FfiConverterTypeFfiV2SubscriptionCallback :
+    FfiConverterCallbackInterface<FfiV2SubscriptionCallback>()
+
+
 public object FfiConverterOptionalLong : FfiConverterRustBuffer<kotlin.Long?> {
     override fun read(buf: ByteBuffer): kotlin.Long? {
         if (buf.get().toInt() == 0) {
@@ -7402,7 +7504,7 @@ public object FfiConverterMapStringBoolean :
         // The parens on `(k, v)` here ensure we're calling the right method,
         // which is important for compatibility with older android devices.
         // Ref https://blog.danlew.net/2017/03/16/kotlin-puzzler-whose-line-is-it-anyways/
-        value.forEach { (k, v) ->
+        value.iterator().forEach { (k, v) ->
             FfiConverterString.write(k, buf)
             FfiConverterBoolean.write(v, buf)
         }
@@ -7443,7 +7545,7 @@ suspend fun `createClient`(
     `accountAddress`: kotlin.String,
     `nonce`: kotlin.ULong,
     `legacySignedPrivateKeyProto`: kotlin.ByteArray?,
-    `historySyncUrl`: kotlin.String?
+    `historySyncUrl`: kotlin.String?,
 ): FfiXmtpClient {
     return uniffiRustCallAsync(
         UniffiLib.INSTANCE.uniffi_xmtpv3_fn_func_create_client(
@@ -7513,7 +7615,7 @@ suspend fun `createV2Client`(`host`: kotlin.String, `isSecure`: kotlin.Boolean):
 @Throws(GenericException::class)
 fun `diffieHellmanK256`(
     `privateKeyBytes`: kotlin.ByteArray,
-    `publicKeyBytes`: kotlin.ByteArray
+    `publicKeyBytes`: kotlin.ByteArray,
 ): kotlin.ByteArray {
     return FfiConverterByteArray.lift(
         uniffiRustCallWithError(GenericException) { _status ->
@@ -7557,7 +7659,7 @@ suspend fun `getInboxIdForAddress`(
     `logger`: FfiLogger,
     `host`: kotlin.String,
     `isSecure`: kotlin.Boolean,
-    `accountAddress`: kotlin.String
+    `accountAddress`: kotlin.String,
 ): kotlin.String? {
     return uniffiRustCallAsync(
         UniffiLib.INSTANCE.uniffi_xmtpv3_fn_func_get_inbox_id_for_address(
@@ -7625,7 +7727,7 @@ fun `publicKeyFromPrivateKeyK256`(`privateKeyBytes`: kotlin.ByteArray): kotlin.B
 @Throws(GenericException::class)
 fun `recoverAddress`(
     `signatureBytes`: kotlin.ByteArray,
-    `predigestMessage`: kotlin.String
+    `predigestMessage`: kotlin.String,
 ): kotlin.String {
     return FfiConverterString.lift(
         uniffiRustCallWithError(GenericException) { _status ->
@@ -7642,7 +7744,7 @@ fun `recoverAddress`(
 @Throws(GenericException::class)
 fun `recoverPublicKeyK256Keccak256`(
     `message`: kotlin.ByteArray,
-    `signature`: kotlin.ByteArray
+    `signature`: kotlin.ByteArray,
 ): kotlin.ByteArray {
     return FfiConverterByteArray.lift(
         uniffiRustCallWithError(GenericException) { _status ->
@@ -7659,7 +7761,7 @@ fun `recoverPublicKeyK256Keccak256`(
 @Throws(GenericException::class)
 fun `recoverPublicKeyK256Sha256`(
     `message`: kotlin.ByteArray,
-    `signature`: kotlin.ByteArray
+    `signature`: kotlin.ByteArray,
 ): kotlin.ByteArray {
     return FfiConverterByteArray.lift(
         uniffiRustCallWithError(GenericException) { _status ->
@@ -7687,7 +7789,7 @@ fun `sha256`(`input`: kotlin.ByteArray): kotlin.ByteArray {
 fun `userPreferencesDecrypt`(
     `publicKey`: kotlin.ByteArray,
     `privateKey`: kotlin.ByteArray,
-    `message`: kotlin.ByteArray
+    `message`: kotlin.ByteArray,
 ): kotlin.ByteArray {
     return FfiConverterByteArray.lift(
         uniffiRustCallWithError(GenericException) { _status ->
@@ -7706,7 +7808,7 @@ fun `userPreferencesDecrypt`(
 fun `userPreferencesEncrypt`(
     `publicKey`: kotlin.ByteArray,
     `privateKey`: kotlin.ByteArray,
-    `message`: kotlin.ByteArray
+    `message`: kotlin.ByteArray,
 ): kotlin.ByteArray {
     return FfiConverterByteArray.lift(
         uniffiRustCallWithError(GenericException) { _status ->
@@ -7726,7 +7828,7 @@ fun `verifyK256Sha256`(
     `signedBy`: kotlin.ByteArray,
     `message`: kotlin.ByteArray,
     `signature`: kotlin.ByteArray,
-    `recoveryId`: kotlin.UByte
+    `recoveryId`: kotlin.UByte,
 ): kotlin.Boolean {
     return FfiConverterBoolean.lift(
         uniffiRustCallWithError(GenericException) { _status ->
diff --git a/library/src/main/jniLibs/arm64-v8a/libuniffi_xmtpv3.so b/library/src/main/jniLibs/arm64-v8a/libuniffi_xmtpv3.so
old mode 100755
new mode 100644
index b4c0a2ab5..bbb3c93de
Binary files a/library/src/main/jniLibs/arm64-v8a/libuniffi_xmtpv3.so and b/library/src/main/jniLibs/arm64-v8a/libuniffi_xmtpv3.so differ
diff --git a/library/src/main/jniLibs/armeabi-v7a/libuniffi_xmtpv3.so b/library/src/main/jniLibs/armeabi-v7a/libuniffi_xmtpv3.so
old mode 100755
new mode 100644
index 7efd1eac3..fff785cd7
Binary files a/library/src/main/jniLibs/armeabi-v7a/libuniffi_xmtpv3.so and b/library/src/main/jniLibs/armeabi-v7a/libuniffi_xmtpv3.so differ
diff --git a/library/src/main/jniLibs/x86/libuniffi_xmtpv3.so b/library/src/main/jniLibs/x86/libuniffi_xmtpv3.so
old mode 100755
new mode 100644
index b64b5deb0..e3540ee0e
Binary files a/library/src/main/jniLibs/x86/libuniffi_xmtpv3.so and b/library/src/main/jniLibs/x86/libuniffi_xmtpv3.so differ
diff --git a/library/src/main/jniLibs/x86_64/libuniffi_xmtpv3.so b/library/src/main/jniLibs/x86_64/libuniffi_xmtpv3.so
old mode 100755
new mode 100644
index 02b73415b..c855e1316
Binary files a/library/src/main/jniLibs/x86_64/libuniffi_xmtpv3.so and b/library/src/main/jniLibs/x86_64/libuniffi_xmtpv3.so differ
diff --git a/library/src/test/java/org/xmtp/android/library/TestHelpers.kt b/library/src/test/java/org/xmtp/android/library/TestHelpers.kt
index 329a80f6f..bab270e74 100644
--- a/library/src/test/java/org/xmtp/android/library/TestHelpers.kt
+++ b/library/src/test/java/org/xmtp/android/library/TestHelpers.kt
@@ -1,25 +1,10 @@
 package org.xmtp.android.library
 
-import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.MutableSharedFlow
-import kotlinx.coroutines.flow.first
-import kotlinx.coroutines.flow.flowOf
-import kotlinx.coroutines.runBlocking
-import org.junit.Assert.assertEquals
 import org.xmtp.android.library.codecs.Fetcher
-import org.xmtp.android.library.messages.ContactBundle
-import org.xmtp.android.library.messages.Envelope
-import org.xmtp.android.library.messages.Pagination
 import org.xmtp.android.library.messages.PrivateKey
 import org.xmtp.android.library.messages.PrivateKeyBuilder
-import org.xmtp.android.library.messages.Signature
-import org.xmtp.android.library.messages.Topic
-import org.xmtp.android.library.messages.toPublicKeyBundle
-import org.xmtp.android.library.messages.walletAddress
-import org.xmtp.proto.message.api.v1.MessageApiOuterClass
 import java.io.File
 import java.net.URL
-import java.util.Date
 
 class TestFetcher : Fetcher {
     override fun fetch(url: URL): ByteArray {
@@ -27,200 +12,24 @@ class TestFetcher : Fetcher {
     }
 }
 
-class FakeWallet : SigningKey {
-    private var privateKey: PrivateKey
-    private var privateKeyBuilder: PrivateKeyBuilder
-
-    constructor(key: PrivateKey, builder: PrivateKeyBuilder) {
-        privateKey = key
-        privateKeyBuilder = builder
-    }
-
-    companion object {
-        fun generate(): FakeWallet {
-            val key = PrivateKeyBuilder()
-            return FakeWallet(key.getPrivateKey(), key)
-        }
-    }
-
-    override suspend fun sign(data: ByteArray): Signature {
-        return privateKeyBuilder.sign(data)
-    }
-
-    override suspend fun sign(message: String): Signature {
-        return privateKeyBuilder.sign(message)
-    }
-
-    override val address: String
-        get() = privateKey.walletAddress
-}
-
-class FakeStreamHolder {
-    private val flow = MutableSharedFlow<Envelope>()
-    suspend fun emit(value: Envelope) = flow.emit(value)
-    fun counts(): Flow<Envelope> = flow
-}
-
-class FakeApiClient : ApiClient {
-    override val environment: XMTPEnvironment = XMTPEnvironment.LOCAL
-    private var authToken: String? = null
-    private val responses: MutableMap<String, List<Envelope>> = mutableMapOf()
-    private val published: MutableList<Envelope> = mutableListOf()
-    private var forbiddingQueries = false
-    private var stream = FakeStreamHolder()
-
-    fun assertNoPublish(callback: () -> Unit) {
-        val oldCount = published.size
-        callback()
-        assertEquals(oldCount, published.size)
-    }
-
-    fun assertNoQuery(callback: () -> Unit) {
-        forbiddingQueries = true
-        callback()
-        forbiddingQueries = false
-    }
-
-    fun findPublishedEnvelope(topic: Topic): Envelope? =
-        findPublishedEnvelope(topic.description)
-
-    fun findPublishedEnvelope(topic: String): Envelope? {
-        for (envelope in published.reversed()) {
-            if (envelope.contentTopic == topic) {
-                return envelope
-            }
-        }
-        return null
-    }
-
-    override fun setAuthToken(token: String) {
-        authToken = token
-    }
-
-    override suspend fun queryTopic(
-        topic: Topic,
-        pagination: Pagination?,
-    ): MessageApiOuterClass.QueryResponse {
-        return query(topic = topic.description, pagination)
-    }
-
-    override suspend fun batchQuery(requests: List<MessageApiOuterClass.QueryRequest>): MessageApiOuterClass.BatchQueryResponse {
-        val response = query(requests.first().getContentTopics(0))
-
-        return MessageApiOuterClass.BatchQueryResponse.newBuilder().also {
-            it.addResponses(response)
-        }.build()
-    }
-
-    suspend fun send(envelope: Envelope) {
-        stream.emit(envelope)
-    }
-
-    override suspend fun envelopes(
-        topic: String,
-        pagination: Pagination?,
-    ): List<MessageApiOuterClass.Envelope> {
-        return query(topic = topic, pagination = pagination).envelopesList
-    }
-
-    override suspend fun query(
-        topic: String,
-        pagination: Pagination?,
-        cursor: MessageApiOuterClass.Cursor?,
-    ): MessageApiOuterClass.QueryResponse {
-        var result: MutableList<Envelope> = mutableListOf()
-        val response = responses.toMutableMap().remove(topic)
-        if (response != null) {
-            result.addAll(response)
-        }
-        result.addAll(
-            published.filter {
-                it.contentTopic == topic
-            }.reversed()
-        )
-
-        val startAt = pagination?.before
-        if (startAt != null) {
-            result = result.filter { it.timestampNs < startAt.time * 1_000_000 }
-                .sortedBy { it.timestampNs }.toMutableList()
-        }
-        val endAt = pagination?.after
-        if (endAt != null) {
-            result = result.filter { it.timestampNs > endAt.time * 1_000_000 }
-                .sortedBy { it.timestampNs }.toMutableList()
-        }
-        val limit = pagination?.limit
-        if (limit != null) {
-            if (limit == 1) {
-                val first = result.firstOrNull()
-                if (first != null) {
-                    result = mutableListOf(first)
-                } else {
-                    result = mutableListOf()
-                }
-            } else {
-                result = result.take(limit - 1).toMutableList()
-            }
-        }
-
-        val direction = pagination?.direction
-        if (direction != null) {
-            when (direction) {
-                MessageApiOuterClass.SortDirection.SORT_DIRECTION_ASCENDING -> {
-                    result = result.reversed().toMutableList()
-                }
-
-                else -> Unit
-            }
-        }
-
-        return QueryResponse.newBuilder().also {
-            it.addAllEnvelopes(result)
-        }.build()
-    }
-
-    override suspend fun publish(envelopes: List<MessageApiOuterClass.Envelope>): MessageApiOuterClass.PublishResponse {
-        for (envelope in envelopes) {
-            send(envelope)
-        }
-        published.addAll(envelopes)
-        return PublishResponse.newBuilder().build()
-    }
-
-    override suspend fun subscribe(request: Flow<MessageApiOuterClass.SubscribeRequest>): Flow<MessageApiOuterClass.Envelope> {
-        val env = stream.counts().first()
-
-        if (request.first().contentTopicsList.contains(env.contentTopic)) {
-            return flowOf(env)
-        }
-        return flowOf()
-    }
-}
-
-data class Fixtures(val aliceAccount: PrivateKeyBuilder, val bobAccount: PrivateKeyBuilder) {
-    var fakeApiClient: FakeApiClient = FakeApiClient()
+data class Fixtures(
+    val aliceAccount: PrivateKeyBuilder,
+    val bobAccount: PrivateKeyBuilder,
+    val clientOptions: ClientOptions? = ClientOptions(
+        ClientOptions.Api(XMTPEnvironment.LOCAL, isSecure = false)
+    ),
+) {
+    var aliceClient: Client = Client().create(account = aliceAccount, options = clientOptions)
     var alice: PrivateKey = aliceAccount.getPrivateKey()
-    var aliceClient: Client = Client().create(account = aliceAccount, apiClient = fakeApiClient)
     var bob: PrivateKey = bobAccount.getPrivateKey()
-    var bobClient: Client = Client().create(account = bobAccount, apiClient = fakeApiClient)
-
-    constructor() : this(aliceAccount = PrivateKeyBuilder(), bobAccount = PrivateKeyBuilder())
+    var bobClient: Client = Client().create(account = bobAccount, options = clientOptions)
 
-    fun publishLegacyContact(client: Client) {
-        val contactBundle = ContactBundle.newBuilder().also { builder ->
-            builder.v1 = builder.v1.toBuilder().also {
-                it.keyBundle = client.privateKeyBundleV1.toPublicKeyBundle()
-            }.build()
-        }.build()
-        val envelope = Envelope.newBuilder().apply {
-            contentTopic = Topic.contact(client.address).description
-            timestampNs = (Date().time * 1_000_000)
-            message = contactBundle.toByteString()
-        }.build()
-
-        runBlocking { client.publish(envelopes = listOf(envelope)) }
-    }
+    constructor(clientOptions: ClientOptions?) : this(
+        aliceAccount = PrivateKeyBuilder(),
+        bobAccount = PrivateKeyBuilder(),
+        clientOptions = clientOptions
+    )
 }
 
-fun fixtures(): Fixtures =
-    Fixtures()
+fun fixtures(clientOptions: ClientOptions? = null): Fixtures =
+    Fixtures(clientOptions)