From b83ea6152207255aa854d04394f21048665e1881 Mon Sep 17 00:00:00 2001 From: Vitor Pamplona Date: Thu, 2 Oct 2025 17:45:58 -0400 Subject: [PATCH] Moves OKHttp relay implementations to use a Channel in order to guarantee incoming message order. --- .../service/okhttp/OkHttpWebSocket.kt | 40 +++++++++++++-- .../nip01Core/relay/client/NostrClient.kt | 9 ++-- .../client/single/basic/BasicRelayClient.kt | 22 ++------ .../client/single/simple/SimpleRelayClient.kt | 3 -- .../sockets/okhttp/BasicOkHttpWebSocket.kt | 51 +++++++++++++++++-- 5 files changed, 92 insertions(+), 33 deletions(-) diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/okhttp/OkHttpWebSocket.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/okhttp/OkHttpWebSocket.kt index f570d9b57..2cfa30326 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/okhttp/OkHttpWebSocket.kt +++ b/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 = 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) } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/NostrClient.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/NostrClient.kt index adae3512f..6a2b9a37f 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/NostrClient.kt +++ b/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) - activeCounts.forEachSub(relay, liveRelay::sendCount) - eventOutbox.forEachUnsentEvent(relay, liveRelay::send) + scope.launch(Dispatchers.Default) { + activeRequests.forEachSub(relay, liveRelay::sendRequest) + activeCounts.forEachSub(relay, liveRelay::sendCount) + eventOutbox.forEachUnsentEvent(relay, liveRelay::send) + } } } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/single/basic/BasicRelayClient.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/single/basic/BasicRelayClient.kt index b9d6553b7..03ebbcf1e 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/single/basic/BasicRelayClient.kt +++ b/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") - - 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) - } + consumeIncomingMessage(text, onConnected) } override fun onClosing( @@ -237,7 +221,7 @@ open class BasicRelayClient( } } - fun consumeIncomingCommand( + fun consumeIncomingMessage( text: String, onConnected: () -> Unit, ) { diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/single/simple/SimpleRelayClient.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/single/simple/SimpleRelayClient.kt index 0619ffb41..6064dbc2a 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/single/simple/SimpleRelayClient.kt +++ b/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, ) diff --git a/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/relay/sockets/okhttp/BasicOkHttpWebSocket.kt b/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/relay/sockets/okhttp/BasicOkHttpWebSocket.kt index 685c17d86..36e99b570 100644 --- a/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/relay/sockets/okhttp/BasicOkHttpWebSocket.kt +++ b/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 = 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)