Makes sure subscriptions are closed on the NostrClient utility methods

2026-03-22 15:18:48 -04:00
parent 439b392aa0
commit 37f6a7b133
3 changed files with 44 additions and 36 deletions
@@ -65,15 +65,18 @@ suspend fun INostrClient.queryCountSuspend(

    subscribe(listener)

-    queryCount(subId = subId, filters = mapOf(relay to listOf(filter)))
-
    val result =
-        withTimeoutOrNull(timeoutMs) {
-            resultChannel.receive()
+        try {
+            queryCount(subId = subId, filters = mapOf(relay to listOf(filter)))
+
+            withTimeoutOrNull(timeoutMs) {
+                resultChannel.receive()
+            }
+        } finally {
+            close(subId)
+            unsubscribe(listener)
        }

-    close(subId)
-    unsubscribe(listener)
    resultChannel.close()

    return result
@@ -98,38 +98,41 @@ suspend fun INostrClient.sendAndWaitForResponseDetailed(
            }
        }

-    subscribe(subscription)
+    val receivedResults =
+        try {
+            subscribe(subscription)

-    // subscribe before sending the result.
-    val resultSubscription =
-        coroutineScope {
-            val result =
-                async {
-                    val receivedResults = mutableMapOf<NormalizedRelayUrl, Boolean>()
-                    // The withTimeout block will cancel the coroutine if the loop takes too long
-                    withTimeoutOrNull(timeoutInSeconds * 1000) {
-                        while (receivedResults.size < relayList.size) {
-                            val result = resultChannel.receive()
+            // subscribe before sending the result.
+            val resultSubscription =
+                coroutineScope {
+                    val result =
+                        async {
+                            val receivedResults = mutableMapOf<NormalizedRelayUrl, Boolean>()
+                            // The withTimeout block will cancel the coroutine if the loop takes too long
+                            withTimeoutOrNull(timeoutInSeconds * 1000) {
+                                while (receivedResults.size < relayList.size) {
+                                    val result = resultChannel.receive()

-                            val currentResult = receivedResults[result.relay]
-                            // do not override a successful result.
-                            if (currentResult == null || !currentResult) {
-                                receivedResults[result.relay] = result.success
+                                    val currentResult = receivedResults[result.relay]
+                                    // do not override a successful result.
+                                    if (currentResult == null || !currentResult) {
+                                        receivedResults[result.relay] = result.success
+                                    }
+                                }
                            }
+                            receivedResults
                        }
-                    }
-                    receivedResults
+
+                    send(event, relayList)
+
+                    result
                }

-            send(event, relayList)
-
-            result
+            resultSubscription.await()
+        } finally {
+            unsubscribe(subscription)
        }

-    val receivedResults = resultSubscription.await()
-
-    unsubscribe(subscription)
-
    // Clean up the channel
    resultChannel.close()

@@ -104,14 +104,16 @@ suspend fun INostrClient.downloadFirstEvent(
            }
        }

-    openReqSubscription(subscriptionId, filters, listener)
-
    val result =
-        withTimeoutOrNull(30000) {
-            resultChannel.receive()
-        }
+        try {
+            openReqSubscription(subscriptionId, filters, listener)

-    close(subscriptionId)
+            withTimeoutOrNull(30000) {
+                resultChannel.receive()
+            }
+        } finally {
+            close(subscriptionId)
+        }

    resultChannel.close()