Moves OKHttp relay implementations to use a Channel in order to guarantee incoming message order.

amethyst/src/main/java/com/vitorpamplona/amethyst/service/okhttp/OkHttpWebSocket.kt

@@ -24,6 +24,13 @@ import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
 import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebSocket
 import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebSocketListener
 import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebsocketBuilder
+import com.vitorpamplona.quartz.nip01Core.relay.sockets.okhttp.BasicOkHttpWebSocket.Companion.exceptionHandler
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.channels.trySendBlocking
+import kotlinx.coroutines.launch
 import okhttp3.OkHttpClient
 import okhttp3.Request
 import okhttp3.Response
@@ -33,7 +40,6 @@ class OkHttpWebSocket(
     val httpClient: (url: NormalizedRelayUrl) -> OkHttpClient,
     val out: WebSocketListener,
 ) : WebSocket {
-    private val listener = OkHttpWebsocketListener()
     private var usingOkHttp: OkHttpClient? = null
     private var socket: okhttp3.WebSocket? = null
 
@@ -64,10 +70,21 @@ class OkHttpWebSocket(
 
     override fun connect() {
         usingOkHttp = httpClient(url)
-        socket = usingOkHttp?.newWebSocket(buildRequest(), listener)
+        socket = usingOkHttp?.newWebSocket(buildRequest(), OkHttpWebsocketListener(out))
     }
 
-    inner class OkHttpWebsocketListener : okhttp3.WebSocketListener() {
+    inner class OkHttpWebsocketListener(
+        val out: WebSocketListener,
+    ) : okhttp3.WebSocketListener() {
+        val scope = CoroutineScope(Dispatchers.Default + exceptionHandler)
+        val incomingMessages: Channel<String> = Channel(Channel.UNLIMITED)
+        val job = // Launch a coroutine to process messages from the channel.
+            scope.launch {
+                for (message in incomingMessages) {
+                    out.onMessage(message)
+                }
+            }
+
         override fun onOpen(
             webSocket: okhttp3.WebSocket,
             response: Response,
@@ -79,7 +96,12 @@ class OkHttpWebSocket(
         override fun onMessage(
             webSocket: okhttp3.WebSocket,
             text: String,
-        ) = out.onMessage(text)
+        ) {
+            // Asynchronously send the received message to the channel.
+            // `trySendBlocking` is used here for simplicity within the callback,
+            // but it's important to understand potential thread blocking if the buffer is full.
+            incomingMessages.trySendBlocking(text)
+        }
 
         override fun onClosing(
             webSocket: okhttp3.WebSocket,
@@ -92,6 +114,11 @@ class OkHttpWebSocket(
             code: Int,
             reason: String,
         ) {
+            // Close the channel on failure, and propagate the error.
+            incomingMessages.close()
+            job.cancel()
+            scope.cancel()
+
             socket = null
             out.onClosed(code, reason)
         }
@@ -101,6 +128,11 @@ class OkHttpWebSocket(
             t: Throwable,
             response: Response?,
         ) {
+            // Close the channel on failure, and propagate the error.
+            incomingMessages.close()
+            job.cancel()
+            scope.cancel()
+
             socket = null
             out.onFailure(t, response?.code, response?.message)
         }

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/NostrClient.kt

@@ -115,12 +115,13 @@ class NostrClient(
             socketBuilder = websocketBuilder,
             listener = relayPool,
             stats = RelayStats.get(relay),
-            scope = scope,
         ) { liveRelay ->
             if (isActive) {
-                activeRequests.forEachSub(relay, liveRelay::sendRequest)
+                scope.launch(Dispatchers.Default) {
-                activeCounts.forEachSub(relay, liveRelay::sendCount)
+                    activeRequests.forEachSub(relay, liveRelay::sendRequest)
-                eventOutbox.forEachUnsentEvent(relay, liveRelay::send)
+                    activeCounts.forEachSub(relay, liveRelay::sendCount)
+                    eventOutbox.forEachUnsentEvent(relay, liveRelay::send)
+                }
             }
         }
 

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/single/basic/BasicRelayClient.kt

@@ -51,9 +51,6 @@ import com.vitorpamplona.quartz.nip42RelayAuth.RelayAuthEvent
 import com.vitorpamplona.quartz.utils.Log
 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 kotlin.concurrent.atomics.AtomicBoolean
 import kotlin.concurrent.atomics.ExperimentalAtomicApi
 import kotlin.coroutines.cancellation.CancellationException
@@ -82,13 +79,11 @@ open class BasicRelayClient(
     val socketBuilder: WebsocketBuilder,
     val listener: IRelayClientListener,
     val stats: RelayStat = RelayStat(),
-    val scope: CoroutineScope,
     val defaultOnConnect: (BasicRelayClient) -> Unit = { },
 ) : IRelayClient {
     companion object {
         // minimum wait time to reconnect: 1 second
         const val DELAY_TO_RECONNECT_IN_SECS = 1
-        const val EVENT_MESSAGE_PREFIX = "[\"${EventMessage.LABEL}\""
     }
 
     private val logTag = "Relay ${url.displayUrl()}"
@@ -166,24 +161,13 @@ open class BasicRelayClient(
 
             markConnectionAsReady(pingMillis, compression)
 
-            scope.launch(Dispatchers.Default) {
+            onConnected()
-                onConnected()
-            }
 
             listener.onRelayStateChange(this@BasicRelayClient, RelayState.CONNECTED)
         }
 
         override fun onMessage(text: String) {
-            // Log.d(logTag, "Receiving: $text")
+            consumeIncomingMessage(text, onConnected)
-
-            if (text.startsWith(EVENT_MESSAGE_PREFIX)) {
-                // defers the parsing of ["EVENTS" to avoid blocking the HTTP thread
-                scope.launch(Dispatchers.Default) {
-                    consumeIncomingCommand(text, onConnected)
-                }
-            } else {
-                consumeIncomingCommand(text, onConnected)
-            }
         }
 
         override fun onClosing(
@@ -237,7 +221,7 @@ open class BasicRelayClient(
         }
     }
 
-    fun consumeIncomingCommand(
+    fun consumeIncomingMessage(
         text: String,
         onConnected: () -> Unit,
     ) {

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/single/simple/SimpleRelayClient.kt

@@ -26,7 +26,6 @@ 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
@@ -37,13 +36,11 @@ class SimpleRelayClient(
     socketBuilder: WebsocketBuilder,
     listener: IRelayClientListener,
     stats: RelayStat = RelayStat(),
-    scopeToParseEvents: CoroutineScope,
     defaultOnConnect: (BasicRelayClient) -> Unit = { },
 ) : IRelayClient by BasicRelayClient(
         url,
         socketBuilder,
         OutboxCache(listener),
         stats,
-        scopeToParseEvents,
         defaultOnConnect,
     )

quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/relay/sockets/okhttp/BasicOkHttpWebSocket.kt

@@ -24,6 +24,14 @@ import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
 import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebSocket
 import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebSocketListener
 import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebsocketBuilder
+import com.vitorpamplona.quartz.utils.Log
+import kotlinx.coroutines.CoroutineExceptionHandler
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.channels.trySendBlocking
+import kotlinx.coroutines.launch
 import okhttp3.OkHttpClient
 import okhttp3.Request
 import okhttp3.Response
@@ -35,6 +43,14 @@ class BasicOkHttpWebSocket(
     val httpClient: (NormalizedRelayUrl) -> OkHttpClient,
     val out: WebSocketListener,
 ) : WebSocket {
+    companion object {
+        // Exists to avoid exceptions stopping the coroutine
+        val exceptionHandler =
+            CoroutineExceptionHandler { _, throwable ->
+                Log.e("BasicOkHttpWebSocket", "WebsocketListener Caught exception: ${throwable.message}", throwable)
+            }
+    }
+
     private var socket: OkHttpWebSocket? = null
 
     override fun needsReconnect() = socket == null
@@ -44,6 +60,15 @@ class BasicOkHttpWebSocket(
 
         val listener =
             object : OkHttpWebSocketListener() {
+                val scope = CoroutineScope(Dispatchers.Default + exceptionHandler)
+                val incomingMessages: Channel<String> = Channel(Channel.UNLIMITED)
+                val job = // Launch a coroutine to process messages from the channel.
+                    scope.launch {
+                        for (message in incomingMessages) {
+                            out.onMessage(message)
+                        }
+                    }
+
                 override fun onOpen(
                     webSocket: OkHttpWebSocket,
                     response: Response,
@@ -55,7 +80,13 @@ class BasicOkHttpWebSocket(
                 override fun onMessage(
                     webSocket: OkHttpWebSocket,
                     text: String,
-                ) = out.onMessage(text)
+                ) {
+                    Log.d("OkHttpWebsocketListener", "Processing: $text")
+                    // Asynchronously send the received message to the channel.
+                    // `trySendBlocking` is used here for simplicity within the callback,
+                    // but it's important to understand potential thread blocking if the buffer is full.
+                    incomingMessages.trySendBlocking(text)
+                }
 
                 override fun onClosing(
                     webSocket: OkHttpWebSocket,
@@ -67,13 +98,27 @@ class BasicOkHttpWebSocket(
                     webSocket: OkHttpWebSocket,
                     code: Int,
                     reason: String,
-                ) = out.onClosed(code, reason)
+                ) {
+                    // Close the channel when the WebSocket connection is closed.
+                    incomingMessages.close()
+                    job.cancel()
+                    scope.cancel()
+
+                    out.onClosed(code, reason)
+                }
 
                 override fun onFailure(
                     webSocket: OkHttpWebSocket,
                     t: Throwable,
                     response: Response?,
-                ) = out.onFailure(t, response?.code, response?.message)
+                ) {
+                    // Close the channel on failure, and propagate the error.
+                    incomingMessages.close()
+                    job.cancel()
+                    scope.cancel()
+
+                    out.onFailure(t, response?.code, response?.message)
+                }
             }
 
         socket = httpClient(url).newWebSocket(request, listener)