Merge pull request #102 from xmtp/np/stop-streams

feat: Ability to cancel streams

android/src/main/java/expo/modules/xmtpreactnativesdk/XMTPModule.kt

@@ -420,6 +420,16 @@ class XMTPModule : Module() {
             )
         }
 
+        Function("unsubscribeFromConversations") {
+            logV("unsubscribeFromConversations")
+            subscriptions["conversations"]?.cancel()
+        }
+
+        Function("unsubscribeFromAllMessages") {
+            logV("unsubscribeFromAllMessages")
+            subscriptions["messages"]?.cancel()
+        }
+
         AsyncFunction("unsubscribeFromMessages") { clientAddress: String, topic: String ->
             logV("unsubscribeFromMessages")
             unsubscribeFromMessages(

example/dev/local/docker-compose.yml

@@ -21,31 +21,17 @@ services:
     depends_on:
       - db
     healthcheck:
-      test: ["CMD", "lsof", "-i", ":5556"]
+      test: [ "CMD", "lsof", "-i", ":5556" ]
       interval: 3s
       timeout: 10s
       retries: 5
   db:
     image: postgres:13
     environment:
       POSTGRES_PASSWORD: xmtp
-
-  upload-service:
-    build: ./upload-service
-    ports:
-      - 4567:4567
-
-  caddy:
-    image: caddy:latest
-    ports:
-      - "443:443"
-    volumes:
-      - ./upload-service/Caddyfile:/etc/caddy/Caddyfile
-      - ./upload-service/data/data:/data
-      - ./upload-service/data/config:/config
-#  js:
-#    restart: always
-#    depends_on:
-#      wakunode:
-#        condition: service_healthy
-#    build: ./test
+  js:
+    restart: always
+    depends_on:
+      wakunode:
+        condition: service_healthy
+    build: ./test

example/src/tests.ts

@@ -359,6 +359,9 @@ test("can stream messages", async () => {
       throw Error("Unexpected convo message topic " + convoMessages[i].topic);
     }
   }
+  alice.conversations.cancelStream();
+  alice.conversations.cancelStreamAllMessages();
+
   return true;
 });
 

ios/XMTPModule.swift

@@ -393,6 +393,14 @@ public class XMTPModule: Module {
             try await subscribeToMessages(clientAddress: clientAddress, topic: topic)
         }
 
+        Function("unsubscribeFromConversations") {
+            subscriptions["conversations"]?.cancel()
+        }
+
+        Function("unsubscribeFromAllMessages") {
+            subscriptions["messages"]?.cancel()
+        }
+
         AsyncFunction("unsubscribeFromMessages") { (clientAddress: String, topic: String) in
             try await unsubscribeFromMessages(clientAddress: clientAddress, topic: topic)
         }

src/index.ts

@@ -225,6 +225,14 @@ export async function subscribeToMessages(
   return await XMTPModule.subscribeToMessages(clientAddress, topic);
 }
 
+export function unsubscribeFromConversations() {
+  return XMTPModule.unsubscribeFromConversations();
+}
+
+export function unsubscribeFromAllMessages() {
+  return XMTPModule.unsubscribeFromAllMessages();
+}
+
 export async function unsubscribeFromMessages(
   clientAddress: string,
   topic: string,

src/lib/Conversations.ts

@@ -90,4 +90,12 @@ export default class Conversations {
       },
     );
   }
+
+  cancelStream() {
+    XMTPModule.unsubscribeFromConversations();
+  }
+
+  cancelStreamAllMessages() {
+    XMTPModule.unsubscribeFromAllMessages();
+  }
 }