Adds a coroutine to the relay class so that we can defer processing events

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)

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 ->

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)

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")
+            if (text.startsWith(EVENT_MESSAGE_PREFIX)) {
-            stats.addBytesReceived(text.bytesUsedInMemory())
+                // defers the parsing of ["EVENTS" to avoid blocking the HTTP thread
-
+                scope.launch(Dispatchers.Default) {
-            try {
+                    consumeIncomingCommand(text, onConnected)
-                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)
                 }
-            } catch (e: Throwable) {
+            } else {
-                if (e is CancellationException) throw e
+                consumeIncomingCommand(text, onConnected)
-                stats.newError("Error processing: $text")
-                Log.e(logTag, "Error processing: $text")
-                listener.onError(this@BasicRelayClient, "", Error("Error processing $text"))
             }
         }
 
@@ -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"))

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,
     )