diff --git a/android/build.gradle b/android/build.gradle
index a27aa2d73..725acf348 100644
--- a/android/build.gradle
+++ b/android/build.gradle
@@ -101,7 +101,7 @@ dependencies {
   implementation 'com.google.code.gson:gson:2.10.1'
   implementation 'com.facebook.react:react-native:0.71.3'
   implementation "com.daveanthonythomas.moshipack:moshipack:1.0.1"
-  implementation "org.xmtp:android:0.8.4"
+    implementation "org.xmtp:android:0.9.0"
   // xmtp-android local testing setup below (comment org.xmtp:android above)
   // implementation files('<PATH TO XMTP-ANDROID>/xmtp-android/library/build/outputs/aar/library-debug.aar')
   // implementation 'com.google.crypto.tink:tink-android:1.7.0'
diff --git a/android/src/main/java/expo/modules/xmtpreactnativesdk/XMTPModule.kt b/android/src/main/java/expo/modules/xmtpreactnativesdk/XMTPModule.kt
index 200c55f0d..38d0b61a2 100644
--- a/android/src/main/java/expo/modules/xmtpreactnativesdk/XMTPModule.kt
+++ b/android/src/main/java/expo/modules/xmtpreactnativesdk/XMTPModule.kt
@@ -9,6 +9,7 @@ import androidx.core.net.toUri
 import com.google.gson.JsonParser
 import com.google.protobuf.kotlin.toByteString
 import expo.modules.kotlin.exception.Exceptions
+import expo.modules.kotlin.functions.Coroutine
 import expo.modules.kotlin.modules.Module
 import expo.modules.kotlin.modules.ModuleDefinition
 import expo.modules.xmtpreactnativesdk.wrappers.ConsentWrapper
@@ -28,6 +29,7 @@ import kotlinx.coroutines.Job
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.suspendCancellableCoroutine
+import kotlinx.coroutines.withContext
 import org.json.JSONObject
 import org.xmtp.android.library.Client
 import org.xmtp.android.library.ClientOptions
@@ -330,7 +332,7 @@ class XMTPModule : Module() {
         }
 
         // Export the conversation's serialized topic data.
-        AsyncFunction("exportConversationTopicData") { clientAddress: String, topic: String ->
+        AsyncFunction("exportConversationTopicData") Coroutine { clientAddress: String, topic: String ->
             logV("exportConversationTopicData")
             val conversation = findConversation(clientAddress, topic)
                 ?: throw XMTPException("no conversation found for $topic")
@@ -437,7 +439,7 @@ class XMTPModule : Module() {
             ).toJson()
         }
 
-        AsyncFunction("sendEncodedContent") { clientAddress: String, topic: String, encodedContentData: List<Int> ->
+        AsyncFunction("sendEncodedContent") Coroutine { clientAddress: String, topic: String, encodedContentData: List<Int> ->
             val conversation =
                 findConversation(
                     clientAddress = clientAddress,
@@ -458,7 +460,7 @@ class XMTPModule : Module() {
             conversation.send(encodedContent = encodedContent)
         }
 
-        AsyncFunction("listConversations") { clientAddress: String ->
+        AsyncFunction("listConversations") Coroutine { clientAddress: String ->
             logV("listConversations")
             val client = clients[clientAddress] ?: throw XMTPException("No client")
             val conversationList = client.conversations.list()
@@ -490,7 +492,7 @@ class XMTPModule : Module() {
             }
         }
 
-        AsyncFunction("loadMessages") { clientAddress: String, topic: String, limit: Int?, before: Long?, after: Long?, direction: String? ->
+        AsyncFunction("loadMessages") Coroutine { clientAddress: String, topic: String, limit: Int?, before: Long?, after: Long?, direction: String? ->
             logV("loadMessages")
             val conversation =
                 findConversation(
@@ -527,7 +529,7 @@ class XMTPModule : Module() {
             )?.map { DecodedMessageWrapper.encode(it) }
         }
 
-        AsyncFunction("loadBatchMessages") { clientAddress: String, topics: List<String> ->
+        AsyncFunction("loadBatchMessages") Coroutine { clientAddress: String, topics: List<String> ->
             logV("loadBatchMessages")
             val client = clients[clientAddress] ?: throw XMTPException("No client")
             val topicsList = mutableListOf<Pair<String, Pagination>>()
@@ -572,7 +574,7 @@ class XMTPModule : Module() {
                 .map { DecodedMessageWrapper.encode(it) }
         }
 
-        AsyncFunction("sendMessage") { clientAddress: String, conversationTopic: String, contentJson: String ->
+        AsyncFunction("sendMessage") Coroutine { clientAddress: String, conversationTopic: String, contentJson: String ->
             logV("sendMessage")
             val conversation =
                 findConversation(
@@ -602,7 +604,7 @@ class XMTPModule : Module() {
             )
         }
 
-        AsyncFunction("prepareMessage") { clientAddress: String, conversationTopic: String, contentJson: String ->
+        AsyncFunction("prepareMessage") Coroutine { clientAddress: String, conversationTopic: String, contentJson: String ->
             logV("prepareMessage")
             val conversation =
                 findConversation(
@@ -625,7 +627,7 @@ class XMTPModule : Module() {
             ).toJson()
         }
 
-        AsyncFunction("prepareEncodedMessage") { clientAddress: String, conversationTopic: String, encodedContentData: List<Int> ->
+        AsyncFunction("prepareEncodedMessage") Coroutine { clientAddress: String, conversationTopic: String, encodedContentData: List<Int> ->
             logV("prepareEncodedMessage")
             val conversation =
                 findConversation(
@@ -658,7 +660,7 @@ class XMTPModule : Module() {
             ).toJson()
         }
 
-        AsyncFunction("sendPreparedMessage") { clientAddress: String, preparedLocalMessageJson: String ->
+        AsyncFunction("sendPreparedMessage") Coroutine { clientAddress: String, preparedLocalMessageJson: String ->
             logV("sendPreparedMessage")
             val client = clients[clientAddress] ?: throw XMTPException("No client")
             val local = PreparedLocalMessage.fromJson(preparedLocalMessageJson)
@@ -676,7 +678,7 @@ class XMTPModule : Module() {
             prepared.messageId
         }
 
-        AsyncFunction("createConversation") { clientAddress: String, peerAddress: String, contextJson: String ->
+        AsyncFunction("createConversation") Coroutine { clientAddress: String, peerAddress: String, contextJson: String ->
             logV("createConversation: $contextJson")
             val client = clients[clientAddress] ?: throw XMTPException("No client")
             val context = JsonParser.parseString(contextJson).asJsonObject
@@ -791,7 +793,7 @@ class XMTPModule : Module() {
             subscribeToAllGroupMessages(clientAddress = clientAddress)
         }
 
-        AsyncFunction("subscribeToMessages") { clientAddress: String, topic: String ->
+        AsyncFunction("subscribeToMessages") Coroutine { clientAddress: String, topic: String ->
             logV("subscribeToMessages")
             subscribeToMessages(
                 clientAddress = clientAddress,
@@ -827,7 +829,7 @@ class XMTPModule : Module() {
             subscriptions[getGroupMessagesKey(clientAddress)]?.cancel()
         }
 
-        AsyncFunction("unsubscribeFromMessages") { clientAddress: String, topic: String ->
+        AsyncFunction("unsubscribeFromMessages") Coroutine { clientAddress: String, topic: String ->
             logV("unsubscribeFromMessages")
             unsubscribeFromMessages(
                 clientAddress = clientAddress,
@@ -859,7 +861,7 @@ class XMTPModule : Module() {
             }
         }
 
-        AsyncFunction("decodeMessage") { clientAddress: String, topic: String, encryptedMessage: String ->
+        AsyncFunction("decodeMessage") Coroutine { clientAddress: String, topic: String, encryptedMessage: String ->
             logV("decodeMessage")
             val encryptedMessageData = Base64.decode(encryptedMessage, NO_WRAP)
             val envelope = EnvelopeBuilder.buildFromString(topic, Date(), encryptedMessageData)
@@ -885,13 +887,13 @@ class XMTPModule : Module() {
             client.contacts.isDenied(address)
         }
 
-        AsyncFunction("denyContacts") { clientAddress: String, addresses: List<String> ->
+        AsyncFunction("denyContacts") Coroutine { clientAddress: String, addresses: List<String> ->
             logV("denyContacts")
             val client = clients[clientAddress] ?: throw XMTPException("No client")
             client.contacts.deny(addresses)
         }
 
-        AsyncFunction("allowContacts") { clientAddress: String, addresses: List<String> ->
+        AsyncFunction("allowContacts") Coroutine { clientAddress: String, addresses: List<String> ->
             val client = clients[clientAddress] ?: throw XMTPException("No client")
             client.contacts.allow(addresses)
         }
@@ -902,7 +904,7 @@ class XMTPModule : Module() {
             consentList.entries.map { ConsentWrapper.encode(it.value) }
         }
 
-        AsyncFunction("conversationConsentState") { clientAddress: String, conversationTopic: String ->
+        AsyncFunction("conversationConsentState") Coroutine { clientAddress: String, conversationTopic: String ->
             val conversation = findConversation(clientAddress, conversationTopic)
                 ?: throw XMTPException("no conversation found for $conversationTopic")
             consentStateToString(conversation.consentState())
@@ -954,7 +956,7 @@ class XMTPModule : Module() {
     // Helpers
     //
 
-    private fun findConversation(
+    private suspend fun findConversation(
         clientAddress: String,
         topic: String,
     ): Conversation? {
@@ -1118,7 +1120,7 @@ class XMTPModule : Module() {
         }
     }
 
-    private fun subscribeToMessages(clientAddress: String, topic: String) {
+    private suspend fun subscribeToMessages(clientAddress: String, topic: String) {
         val conversation =
             findConversation(
                 clientAddress = clientAddress,
@@ -1188,7 +1190,7 @@ class XMTPModule : Module() {
         return "groups:$clientAddress"
     }
 
-    private fun unsubscribeFromMessages(
+    private suspend fun unsubscribeFromMessages(
         clientAddress: String,
         topic: String,
     ) {
diff --git a/example/App.tsx b/example/App.tsx
index cf3d6828f..f6c70c434 100644
--- a/example/App.tsx
+++ b/example/App.tsx
@@ -18,6 +18,7 @@ import GroupScreen from './src/GroupScreen'
 import HomeScreen from './src/HomeScreen'
 import LaunchScreen from './src/LaunchScreen'
 import { Navigator } from './src/Navigation'
+import StreamScreen from './src/StreamScreen'
 import TestScreen from './src/TestScreen'
 
 const queryClient = new QueryClient()
@@ -100,6 +101,11 @@ export default function App() {
                 component={ConversationCreateScreen}
                 options={{ title: 'New Conversation' }}
               />
+              <Navigator.Screen
+                name="streamTest"
+                component={StreamScreen}
+                options={{ title: 'Stream Tests' }}
+              />
             </Navigator.Navigator>
           </NavigationContainer>
         </XmtpProvider>
diff --git a/example/ios/Podfile.lock b/example/ios/Podfile.lock
index 307a0fe61..6fd46c5dc 100644
--- a/example/ios/Podfile.lock
+++ b/example/ios/Podfile.lock
@@ -769,4 +769,4 @@ SPEC CHECKSUMS:
 
 PODFILE CHECKSUM: 95d6ace79946933ecf80684613842ee553dd76a2
 
-COCOAPODS: 1.14.3
+COCOAPODS: 1.15.2
diff --git a/example/src/LaunchScreen.tsx b/example/src/LaunchScreen.tsx
index acfce332c..86e00b16f 100644
--- a/example/src/LaunchScreen.tsx
+++ b/example/src/LaunchScreen.tsx
@@ -148,6 +148,14 @@ export default function LaunchScreen(
         <Text style={styles.label}>External Wallet:</Text>
         <ConnectWallet theme="dark" />
       </View>
+      <View key="stream-tests" style={{ margin: 16, marginTop: 16 }}>
+        <Button
+          color="green"
+          title="Create Stream Tests"
+          onPress={() => navigation.navigate('streamTest')}
+          accessibilityLabel="Unit-tests"
+        />
+      </View>
       {signer && (
         <>
           <View key="connected-dev" style={{ margin: 16 }}>
diff --git a/example/src/Navigation.tsx b/example/src/Navigation.tsx
index 5858cf2c5..080990e68 100644
--- a/example/src/Navigation.tsx
+++ b/example/src/Navigation.tsx
@@ -9,6 +9,7 @@ export type NavigationParamList = {
   group: { id: string }
   conversation: { topic: string }
   conversationCreate: undefined
+  streamTest: undefined
 }
 
 export const Navigator = createNativeStackNavigator<NavigationParamList>()
diff --git a/example/src/StreamScreen.tsx b/example/src/StreamScreen.tsx
new file mode 100644
index 000000000..0944d06fa
--- /dev/null
+++ b/example/src/StreamScreen.tsx
@@ -0,0 +1,329 @@
+/* eslint-disable @typescript-eslint/no-floating-promises */
+import React, { useCallback, useEffect, useState } from 'react'
+import { Button, FlatList, Text, View, Switch, Pressable } from 'react-native'
+
+import { Client, Conversation, DecodedMessage } from '../../src'
+
+let events: StreamEvent[] = []
+
+const useClient = () => {
+  const [client, setClient] = useState<Client<any> | null>(null)
+
+  useEffect(() => {
+    Client.createRandom().then((client) => {
+      setClient(client)
+    })
+  }, [])
+
+  return client
+}
+
+const useConversation = (client: Client<any>, recieve: boolean) => {
+  const [conversation, setConversation] = useState<Conversation<any> | null>(
+    null
+  )
+  const [otherClientConvo, setOtherClientConvo] =
+    useState<Conversation<any> | null>(null)
+  const conversationClient = useClient()
+
+  useEffect(() => {
+    if (!conversationClient) {
+      return
+    }
+    if (!client) {
+      return
+    }
+
+    const setupConversations = async () => {
+      const conversation = await client.conversations.newConversation(
+        conversationClient.address
+      )
+      events.push({ type: 'start_conv', timestamp: Date.now() })
+      setConversation(conversation)
+      const conversations = await conversationClient.conversations.list()
+      const otherConvo = conversations.find(
+        (convo) => convo.topic === conversation.topic
+      )
+      if (otherConvo) {
+        setOtherClientConvo(otherConvo)
+      }
+    }
+
+    const setupReceiveConvo = async () => {
+      const conversation =
+        await conversationClient.conversations.newConversation(
+          conversationClient.address
+        )
+      events.push({ type: 'receive_convo', timestamp: Date.now() })
+      setOtherClientConvo(conversation)
+      const conversations = await client.conversations.list()
+      const otherConvo = conversations.find(
+        (convo) => convo.topic === conversation.topic
+      )
+      if (otherConvo) {
+        setConversation(otherConvo)
+      }
+    }
+    if (recieve) {
+      setupReceiveConvo()
+    } else {
+      setupConversations()
+    }
+  }, [conversationClient, client])
+
+  return { conversation, otherClientConvo }
+}
+
+interface ConversationItemProps {
+  client: Client<any>
+  received: boolean
+  onSendMessage: () => void
+  onReceiveMessage: () => void
+}
+
+const ConversationItem = ({
+  client,
+  received,
+  onSendMessage,
+  onReceiveMessage,
+}: ConversationItemProps) => {
+  const { conversation, otherClientConvo } = useConversation(client, received)
+  const [messages, setMessages] = useState<DecodedMessage[]>([])
+  const [shouldStream, setShouldStream] = useState<boolean>(true)
+  const [show, setShow] = useState<boolean>(true)
+
+  useEffect(() => {
+    let sub: (() => void) | undefined = undefined
+
+    const setSub = async () => {
+      sub = await conversation?.streamMessages(async (message) => {
+        setMessages((prev) => [...prev, message])
+      })
+    }
+    if (shouldStream) {
+      setSub()
+    }
+    return () => {
+      sub?.()
+    }
+  }, [conversation, shouldStream])
+
+  const toggleStream = useCallback(() => {
+    setShouldStream((prev) => !prev)
+  }, [setShouldStream])
+
+  const toggleShow = useCallback(() => {
+    setShow((prev) => !prev)
+  }, [setShow])
+
+  const sendMessage = useCallback(() => {
+    conversation?.send('Hello!')
+    onSendMessage()
+  }, [conversation, onSendMessage])
+
+  const receiveMessage = useCallback(() => {
+    otherClientConvo?.send('Hi!')
+    onReceiveMessage()
+  }, [otherClientConvo, client.address, onReceiveMessage])
+
+  return (
+    <View>
+      <Pressable onPress={toggleShow}>
+        <Text style={{ color: show ? 'black' : 'grey' }}>
+          {conversation?.topic} {messages.length}
+        </Text>
+      </Pressable>
+      {show && (
+        <>
+          <View style={{ flexDirection: 'row', flexWrap: 'wrap' }}>
+            <Button
+              title="Send new message"
+              color="green"
+              onPress={sendMessage}
+            />
+            <Button
+              title="Receive message"
+              color="blue"
+              onPress={receiveMessage}
+            />
+            <View style={{ flexDirection: 'row' }}>
+              <Text>Stream</Text>
+              <Switch value={shouldStream} onValueChange={toggleStream} />
+            </View>
+          </View>
+          <FlatList
+            data={messages}
+            keyExtractor={(item) => item.id}
+            renderItem={({ item }) => <Text>{item.sent}</Text>}
+          />
+        </>
+      )}
+    </View>
+  )
+}
+
+interface StreamEvent {
+  type:
+    | 'start_conv'
+    | 'receive_convo'
+    | 'receive_message'
+    | 'send_message'
+    | 'conv_stream_start'
+    | 'conv_stream_cancel'
+    | 'all_messages_start'
+    | 'all_messages_cancel'
+    | 'all_messages_receive'
+    | 'all_start'
+    | 'all_cancel'
+    | 'all_receive'
+  timestamp: number
+}
+
+export default function StreamScreen() {
+  const client = useClient()
+  const [showEvents, setShowEvents] = useState<boolean>(false)
+  const [conversations, setConversations] = useState<boolean[]>([])
+  const [stream, setStream] = useState<boolean>(true)
+  const [streamCount, setStreamCount] = useState<number>(0)
+  const [streamMessages, setStreamMessages] = useState<boolean>(true)
+  const [streamMessagesCount, setStreamMessagesCount] = useState<number>(0)
+
+  useEffect(() => {
+    events = []
+  }, [])
+
+  const toggleStreamMessages = useCallback(() => {
+    setStreamMessages((prev) => !prev)
+  }, [setStreamMessages])
+  const toggleStreamAll = useCallback(() => {
+    setStream((prev) => !prev)
+  }, [setStream])
+  const toggleShowEvents = useCallback(() => {
+    setShowEvents((prev) => !prev)
+  }, [setShowEvents])
+
+  const addConversation = useCallback(() => {
+    setConversations((prev) => [...prev, false])
+  }, [setConversations])
+
+  const removeConversation = useCallback(() => {
+    setConversations((prev) => prev.slice(0, prev.length - 1))
+  }, [setConversations])
+
+  const onSendMessage = useCallback(() => {
+    events.push({ type: 'send_message', timestamp: Date.now() })
+  }, [])
+
+  const onReceiveMessage = useCallback(() => {
+    events.push({ type: 'receive_message', timestamp: Date.now() })
+  }, [])
+
+  const onStreamAllMessages = useCallback(() => {
+    events.push({ type: 'all_messages_start', timestamp: Date.now() })
+  }, [])
+
+  const onCancelStreamAllMessages = useCallback(() => {
+    events.push({ type: 'all_messages_cancel', timestamp: Date.now() })
+  }, [])
+
+  const onStreamAll = useCallback(() => {
+    events.push({ type: 'all_start', timestamp: Date.now() })
+  }, [])
+
+  const onCancelStreamAll = useCallback(() => {
+    events.push({ type: 'all_cancel', timestamp: Date.now() })
+  }, [])
+
+  useEffect(() => {
+    if (!client) {
+      return
+    }
+    const setupSub = async () => {
+      onStreamAllMessages()
+      client.conversations.streamAllMessages(async () => {
+        setStreamMessagesCount((prev) => prev + 1)
+      })
+    }
+    if (streamMessages) {
+      setupSub()
+    }
+
+    return () => {
+      onCancelStreamAllMessages()
+      client.conversations.cancelStreamAllMessages()
+    }
+  }, [client, stream, onStreamAllMessages, onCancelStreamAllMessages])
+
+  useEffect(() => {
+    if (!client) {
+      return
+    }
+    const setupSub = async () => {
+      onStreamAll()
+      client.conversations.stream(async () => {
+        setStreamCount((prev) => prev + 1)
+      })
+    }
+    if (streamMessages) {
+      setupSub()
+    }
+
+    return () => {
+      onCancelStreamAll()
+      client.conversations.cancelStream()
+    }
+  }, [client, stream, onStreamAllMessages, onCancelStreamAllMessages])
+
+  if (!client) {
+    return (
+      <View>
+        <Text>Loading</Text>
+      </View>
+    )
+  }
+
+  return (
+    <View style={{ flex: 1 }}>
+      {showEvents && <Text selectable>{JSON.stringify(events)}</Text>}
+      <Button title="Show events" onPress={toggleShowEvents} color="blue" />
+      <View>
+        <Button
+          title="Add new conversation"
+          color="green"
+          onPress={addConversation}
+        />
+        <Button
+          title="Remove latest conversation"
+          color="red"
+          onPress={removeConversation}
+        />
+        <View style={{ flexDirection: 'row' }}>
+          <View style={{ flexDirection: 'row' }}>
+            <Text>Stream All: {streamCount}</Text>
+            <Switch value={stream} onValueChange={toggleStreamAll} />
+          </View>
+          <View style={{ flexDirection: 'row' }}>
+            <Text>Stream All Messages: {streamMessagesCount}</Text>
+            <Switch
+              value={streamMessages}
+              onValueChange={toggleStreamMessages}
+            />
+          </View>
+        </View>
+      </View>
+      <FlatList
+        // style={{ flexGrow: 1 }}
+        data={conversations}
+        keyExtractor={(item, idx) => idx.toString()}
+        renderItem={({ item }) => (
+          <ConversationItem
+            client={client}
+            received={item}
+            onSendMessage={onSendMessage}
+            onReceiveMessage={onReceiveMessage}
+          />
+        )}
+      />
+    </View>
+  )
+}
diff --git a/example/src/tests/tests.ts b/example/src/tests/tests.ts
index 1472dce17..cb8c7c425 100644
--- a/example/src/tests/tests.ts
+++ b/example/src/tests/tests.ts
@@ -67,6 +67,8 @@ class NumberCodec implements JSContentCodec<NumberRef> {
   }
 }
 
+const LONG_STREAM_DELAY = 20000
+
 export const tests: Test[] = []
 
 function sleep(ms: number) {
@@ -652,7 +654,7 @@ test('can stream conversations with delay', async () => {
     'Unexpected all convos count ' + allConvos.length
   )
 
-  await sleep(15000)
+  await sleep(LONG_STREAM_DELAY)
 
   await bo.conversations.newConversation(alix.address, {
     conversationID: 'convo-3',
@@ -876,7 +878,7 @@ test('can stream all msgs with delay', async () => {
     'Unexpected all messages count ' + allMessages.length
   )
 
-  await sleep(15000)
+  await sleep(LONG_STREAM_DELAY)
   // Starts a new conversation.
   const caro = await Client.createRandom({ env: 'dev' })
   const caroConvo = await caro.conversations.newConversation(alix.address)
@@ -892,7 +894,7 @@ test('can stream all msgs with delay', async () => {
     'Unexpected all messages count ' + allMessages.length
   )
 
-  await sleep(15000)
+  await sleep(LONG_STREAM_DELAY)
 
   for (let i = 0; i < 5; i++) {
     await boConvo.send({ text: `Message ${i}` })
@@ -1506,3 +1508,221 @@ test('get all HMAC keys', async () => {
 
   return true
 })
+
+test('can handle complex streaming setup', async () => {
+  const bo = await Client.createRandom({ env: 'dev' })
+  await delayToPropogate()
+  const alix = await Client.createRandom({ env: 'dev' })
+  await delayToPropogate()
+
+  const allConvos: Conversation<any>[] = []
+  await alix.conversations.stream(async (convo) => {
+    allConvos.push(convo)
+  })
+  const allMessages: DecodedMessage[] = []
+  await alix.conversations.streamAllMessages(async (message) => {
+    allMessages.push(message)
+  })
+
+  const conv1 = await bo.conversations.newConversation(alix.address)
+  await delayToPropogate()
+
+  await bo.conversations.newConversation(alix.address, {
+    conversationID: 'convo-2',
+    metadata: {},
+  })
+  const allConvMessages: DecodedMessage[] = []
+  conv1.streamMessages(async (message) => {
+    allConvMessages.push(message)
+  })
+  await conv1.send({ text: 'Hello' })
+  await delayToPropogate()
+
+  assert(
+    allConvos.length === 2,
+    'Unexpected all convos count1 ' + allConvos.length
+  )
+
+  assert(
+    allMessages.length === 1,
+    'Unexpected all messages count2 ' + allMessages.length
+  )
+
+  assert(
+    allConvMessages.length === 1,
+    'Unexpected all conv messages count3 ' + allConvMessages.length
+  )
+
+  await sleep(LONG_STREAM_DELAY)
+  const conv3 = await bo.conversations.newConversation(alix.address, {
+    conversationID: 'convo-3',
+    metadata: {},
+  })
+  const allConv3Messages: DecodedMessage[] = []
+  conv3.streamMessages(async (message) => {
+    allConv3Messages.push(message)
+  })
+  await conv1.send({ text: 'Hello' })
+  await conv3.send({ text: 'Hello' })
+  await delayToPropogate()
+
+  assert(
+    allConvos.length === 3,
+    'Unexpected all convos count4 ' + allConvos.length
+  )
+
+  assert(
+    allMessages.length === 2, // TODO: should be 3
+    'Unexpected all messages count5 ' + allMessages.length
+  )
+
+  assert(
+    allConvMessages.length === 2,
+    'Unexpected all conv messages count6 ' + allConvMessages.length
+  )
+
+  assert(
+    allConv3Messages.length === 1,
+    'Unexpected all conv3 messages count7 ' + allConv3Messages.length
+  )
+
+  alix.conversations.cancelStream()
+  alix.conversations.cancelStreamAllMessages()
+
+  await bo.conversations.newConversation(alix.address, {
+    conversationID: 'convo-4',
+    metadata: {},
+  })
+  await conv3.send({ text: 'Hello' })
+
+  assert(
+    allConvos.length === 3,
+    'Unexpected all convos count8 ' + allConvos.length
+  )
+
+  assert(
+    allMessages.length === 3,
+    'Unexpected all messages count9 ' + allMessages.length
+  )
+
+  assert(
+    allConvMessages.length === 2,
+    'Unexpected all conv messages count10 ' + allConvMessages.length
+  )
+
+  assert(
+    allConv3Messages.length === 2,
+    'Unexpected all conv3 messages count11 ' + allConv3Messages.length
+  )
+
+  return true
+})
+
+test('can handle complex streaming setup with messages from self', async () => {
+  const bo = await Client.createRandom({ env: 'dev' })
+  await delayToPropogate()
+  const alix = await Client.createRandom({ env: 'dev' })
+  await delayToPropogate()
+
+  const allConvos: Conversation<any>[] = []
+  await alix.conversations.stream(async (convo) => {
+    allConvos.push(convo)
+  })
+  const allMessages: DecodedMessage[] = []
+  await alix.conversations.streamAllMessages(async (message) => {
+    allMessages.push(message)
+  })
+
+  const conv1 = await alix.conversations.newConversation(bo.address)
+  await delayToPropogate()
+
+  await alix.conversations.newConversation(bo.address, {
+    conversationID: 'convo-2',
+    metadata: {},
+  })
+  const allConvMessages: DecodedMessage[] = []
+  conv1.streamMessages(async (message) => {
+    allConvMessages.push(message)
+  })
+  await conv1.send({ text: 'Hello' })
+  await delayToPropogate()
+
+  assert(
+    allConvos.length === 2,
+    'Unexpected all convos count1 ' + allConvos.length
+  )
+
+  assert(
+    allMessages.length === 1,
+    'Unexpected all messages count2 ' + allMessages.length
+  )
+
+  assert(
+    allConvMessages.length === 1,
+    'Unexpected all conv messages count3 ' + allConvMessages.length
+  )
+
+  await sleep(LONG_STREAM_DELAY)
+  const conv3 = await alix.conversations.newConversation(bo.address, {
+    conversationID: 'convo-3',
+    metadata: {},
+  })
+  const allConv3Messages: DecodedMessage[] = []
+  conv3.streamMessages(async (message) => {
+    allConv3Messages.push(message)
+  })
+  await conv1.send({ text: 'Hello' })
+  await conv3.send({ text: 'Hello' })
+  await delayToPropogate()
+
+  assert(
+    allConvos.length === 3,
+    'Unexpected all convos count4 ' + allConvos.length
+  )
+
+  assert(
+    allMessages.length === 2, // TODO: should be 3
+    'Unexpected all messages count5 ' + allMessages.length
+  )
+
+  assert(
+    allConvMessages.length === 3,
+    'Unexpected all conv messages count6 ' + allConvMessages.length
+  )
+
+  assert(
+    allConv3Messages.length === 1,
+    'Unexpected all conv3 messages count7 ' + allConv3Messages.length
+  )
+
+  alix.conversations.cancelStream()
+  alix.conversations.cancelStreamAllMessages()
+
+  await bo.conversations.newConversation(alix.address, {
+    conversationID: 'convo-4',
+    metadata: {},
+  })
+  await conv3.send({ text: 'Hello' })
+
+  assert(
+    allConvos.length === 3,
+    'Unexpected all convos count8 ' + allConvos.length
+  )
+
+  assert(
+    allMessages.length === 3,
+    'Unexpected all messages count9 ' + allMessages.length
+  )
+
+  assert(
+    allConvMessages.length === 2,
+    'Unexpected all conv messages count10 ' + allConvMessages.length
+  )
+
+  assert(
+    allConv3Messages.length === 2,
+    'Unexpected all conv3 messages count11 ' + allConv3Messages.length
+  )
+
+  return true
+})
diff --git a/src/lib/DecodedMessage.ts b/src/lib/DecodedMessage.ts
index 43acd0c06..24321a412 100644
--- a/src/lib/DecodedMessage.ts
+++ b/src/lib/DecodedMessage.ts
@@ -9,6 +9,10 @@ import {
 import { TextCodec } from './NativeCodecs/TextCodec'
 import { DefaultContentTypes } from './types/DefaultContentType'
 
+const allowEmptyProperties: (keyof NativeMessageContent)[] = [
+  'text',
+  'readReceipt',
+]
 export class DecodedMessage<
   ContentTypes extends DefaultContentTypes = DefaultContentTypes,
 > {
@@ -106,7 +110,9 @@ export class DecodedMessage<
       for (const codec of Object.values(this.client.codecRegistry)) {
         if (
           ('contentKey' in codec && this.nativeContent[codec.contentKey]) ||
-          this.nativeContent.hasOwnProperty('text')
+          allowEmptyProperties.some((prop) =>
+            this.nativeContent.hasOwnProperty(prop)
+          )
         ) {
           return (
             codec as NativeContentCodec<