From e70f1fedc68b97a0f2b13b50713e0195ff26165a Mon Sep 17 00:00:00 2001 From: Vitor Pamplona Date: Wed, 30 Jul 2025 17:33:12 -0400 Subject: [PATCH] Adds a coroutine to the relay class so that we can defer processing events --- .../com/vitorpamplona/amethyst/Amethyst.kt | 2 +- .../relayClient/CacheClientConnector.kt | 3 + .../nip01Core/relay/client/NostrClient.kt | 1 + .../client/single/basic/BasicRelayClient.kt | 61 +++++++++++++------ .../client/single/simple/SimpleRelayClient.kt | 3 + 5 files changed, 49 insertions(+), 21 deletions(-) diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/Amethyst.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/Amethyst.kt index 1129c7773..d115ad178 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/Amethyst.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/Amethyst.kt @@ -112,7 +112,7 @@ class Amethyst : Application() { val relayProxyClientConnector = RelayProxyClientConnector(torProxySettingsAnchor, okHttpClients, connManager, client, applicationIOScope) // Verifies and inserts in the cache from all relays, all subscriptions - val cacheClientConnector = CacheClientConnector(client, cache) + val cacheClientConnector = CacheClientConnector(client, cache, applicationIOScope) // Show messages from the Relay and controls their dismissal val notifyCoordinator = NotifyCoordinator(client) diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/CacheClientConnector.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/CacheClientConnector.kt index f03417ace..0371fc4fe 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/CacheClientConnector.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/CacheClientConnector.kt @@ -21,15 +21,18 @@ package com.vitorpamplona.amethyst.service.relayClient import com.vitorpamplona.amethyst.model.LocalCache +import com.vitorpamplona.amethyst.model.LocalCache.markAsSeen import com.vitorpamplona.quartz.nip01Core.core.HexKey import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.EventCollector import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.RelayInsertConfirmationCollector import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import kotlinx.coroutines.CoroutineScope class CacheClientConnector( val client: NostrClient, val cache: LocalCache, + val scope: CoroutineScope, ) { val receiver = EventCollector(client) { event, relay -> diff --git a/quartz/src/main/java/com/vitorpamplona/quartz/nip01Core/relay/client/NostrClient.kt b/quartz/src/main/java/com/vitorpamplona/quartz/nip01Core/relay/client/NostrClient.kt index 1b629750c..fa5fbf395 100644 --- a/quartz/src/main/java/com/vitorpamplona/quartz/nip01Core/relay/client/NostrClient.kt +++ b/quartz/src/main/java/com/vitorpamplona/quartz/nip01Core/relay/client/NostrClient.kt @@ -114,6 +114,7 @@ class NostrClient( socketBuilder = websocketBuilder, listener = relayPool, stats = RelayStats.get(relay), + scope = scope, ) { liveRelay -> activeRequests.forEachSub(relay, liveRelay::sendRequest) activeCounts.forEachSub(relay, liveRelay::sendCount) diff --git a/quartz/src/main/java/com/vitorpamplona/quartz/nip01Core/relay/client/single/basic/BasicRelayClient.kt b/quartz/src/main/java/com/vitorpamplona/quartz/nip01Core/relay/client/single/basic/BasicRelayClient.kt index 031362f8b..7555915d9 100644 --- a/quartz/src/main/java/com/vitorpamplona/quartz/nip01Core/relay/client/single/basic/BasicRelayClient.kt +++ b/quartz/src/main/java/com/vitorpamplona/quartz/nip01Core/relay/client/single/basic/BasicRelayClient.kt @@ -50,6 +50,9 @@ import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebsocketBuilder import com.vitorpamplona.quartz.nip42RelayAuth.RelayAuthEvent import com.vitorpamplona.quartz.utils.TimeUtils import com.vitorpamplona.quartz.utils.bytesUsedInMemory +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.launch import java.util.concurrent.atomic.AtomicBoolean import kotlin.coroutines.cancellation.CancellationException @@ -76,11 +79,13 @@ open class BasicRelayClient( val socketBuilder: WebsocketBuilder, val listener: IRelayClientListener, val stats: RelayStat = RelayStat(), + val scope: CoroutineScope, val defaultOnConnect: (BasicRelayClient) -> Unit = { }, ) : IRelayClient { companion object { // waits 3 minutes to reconnect once things fail const val DELAY_TO_RECONNECT_IN_MSECS = 500L + const val EVENT_MESSAGE_PREFIX = "[\"${EventMessage.LABEL}\"" } private val logTag = "Relay ${url.displayUrl()}" @@ -160,31 +165,21 @@ open class BasicRelayClient( markConnectionAsReady(pingMillis, compression) - onConnected() + scope.launch(Dispatchers.Default) { + onConnected() + } listener.onRelayStateChange(this@BasicRelayClient, RelayState.CONNECTED) } override fun onMessage(text: String) { - // Log.d(logTag, "Receiving: $text") - stats.addBytesReceived(text.bytesUsedInMemory()) - - try { - when (val msg = parser.parse(text)) { - is EventMessage -> processEvent(msg) - is EoseMessage -> processEose(msg) - is NoticeMessage -> processNotice(msg) - is OkMessage -> processOk(msg, onConnected) - is AuthMessage -> processAuth(msg) - is NotifyMessage -> processNotify(msg) - is ClosedMessage -> processClosed(msg) - else -> processUnkownMessage(text) + if (text.startsWith(EVENT_MESSAGE_PREFIX)) { + // defers the parsing of ["EVENTS" to avoid blocking the HTTP thread + scope.launch(Dispatchers.Default) { + consumeIncomingCommand(text, onConnected) } - } catch (e: Throwable) { - if (e is CancellationException) throw e - stats.newError("Error processing: $text") - Log.e(logTag, "Error processing: $text") - listener.onError(this@BasicRelayClient, "", Error("Error processing $text")) + } else { + consumeIncomingCommand(text, onConnected) } } @@ -231,6 +226,32 @@ open class BasicRelayClient( } } + fun consumeIncomingCommand( + text: String, + onConnected: () -> Unit, + ) { + // Log.d(logTag, "Receiving: $text") + stats.addBytesReceived(text.bytesUsedInMemory()) + + try { + when (val msg = parser.parse(text)) { + is EventMessage -> processEvent(msg) + is EoseMessage -> processEose(msg) + is NoticeMessage -> processNotice(msg) + is OkMessage -> processOk(msg, onConnected) + is AuthMessage -> processAuth(msg) + is NotifyMessage -> processNotify(msg) + is ClosedMessage -> processClosed(msg) + else -> processUnknownMessage(text) + } + } catch (e: Throwable) { + if (e is CancellationException) throw e + stats.newError("Error processing: $text") + Log.e(logTag, "Error processing: $text") + listener.onError(this@BasicRelayClient, "", Error("Error processing $text")) + } + } + fun markConnectionAsReady( pingInMs: Long, usingCompression: Boolean, @@ -315,7 +336,7 @@ open class BasicRelayClient( listener.onClosed(this@BasicRelayClient, msg.subscriptionId, msg.message) } - private fun processUnkownMessage(newMessage: String) { + private fun processUnknownMessage(newMessage: String) { stats.newError("Unsupported message: $newMessage") Log.w(logTag, "Unsupported message: $newMessage") listener.onError(this, "", Error("Unsupported message: $newMessage")) diff --git a/quartz/src/main/java/com/vitorpamplona/quartz/nip01Core/relay/client/single/simple/SimpleRelayClient.kt b/quartz/src/main/java/com/vitorpamplona/quartz/nip01Core/relay/client/single/simple/SimpleRelayClient.kt index 6064dbc2a..0619ffb41 100644 --- a/quartz/src/main/java/com/vitorpamplona/quartz/nip01Core/relay/client/single/simple/SimpleRelayClient.kt +++ b/quartz/src/main/java/com/vitorpamplona/quartz/nip01Core/relay/client/single/simple/SimpleRelayClient.kt @@ -26,6 +26,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.client.single.basic.BasicRelayCl import com.vitorpamplona.quartz.nip01Core.relay.client.stats.RelayStat import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebsocketBuilder +import kotlinx.coroutines.CoroutineScope /** * This relay client saves any event that will be sent in an outbox @@ -36,11 +37,13 @@ class SimpleRelayClient( socketBuilder: WebsocketBuilder, listener: IRelayClientListener, stats: RelayStat = RelayStat(), + scopeToParseEvents: CoroutineScope, defaultOnConnect: (BasicRelayClient) -> Unit = { }, ) : IRelayClient by BasicRelayClient( url, socketBuilder, OutboxCache(listener), stats, + scopeToParseEvents, defaultOnConnect, )