diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/RelayProxyClientConnector.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/RelayProxyClientConnector.kt index 423dbb931..1cb1148e1 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/RelayProxyClientConnector.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/RelayProxyClientConnector.kt @@ -95,7 +95,10 @@ class RelayProxyClientConnector( client.connect() } else { Log.d("ManageRelayServices", "Relay Services have changed, reconnecting relays that need to") - client.reconnect(true) + client.reconnect( + onlyIfChanged = true, + ignoreRetryDelays = true, + ) } }.onStart { Log.d("ManageRelayServices", "Resuming Relay Services") diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/INostrClient.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/INostrClient.kt index 79b6a595d..6e86149b5 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/INostrClient.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/INostrClient.kt @@ -36,7 +36,10 @@ interface INostrClient { fun disconnect() - fun reconnect(onlyIfChanged: Boolean = false) + fun reconnect( + onlyIfChanged: Boolean = false, + ignoreRetryDelays: Boolean = false, + ) fun isActive(): Boolean @@ -78,7 +81,10 @@ object EmptyNostrClient : INostrClient { override fun disconnect() { } - override fun reconnect(onlyIfChanged: Boolean) { } + override fun reconnect( + onlyIfChanged: Boolean, + ignoreRetryDelays: Boolean, + ) { } override fun isActive() = false 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 a2fe4b705..2db300300 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 @@ -37,14 +37,15 @@ import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.DelicateCoroutinesApi import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.FlowPreview +import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.SharingStarted import kotlinx.coroutines.flow.combine +import kotlinx.coroutines.flow.debounce import kotlinx.coroutines.flow.flowOn import kotlinx.coroutines.flow.onEach import kotlinx.coroutines.flow.sample import kotlinx.coroutines.flow.stateIn import kotlinx.coroutines.launch -import kotlinx.coroutines.sync.Mutex /** * The NostrClient manages Nostr relay operations, subscriptions, and event delivery. It maintains: @@ -140,21 +141,35 @@ class NostrClient( override fun isActive() = isActive - val myReconnectMutex = Mutex() + class Reconnect( + val onlyIfChanged: Boolean, + val ignoreRetryDelays: Boolean, + ) - override fun reconnect(onlyIfChanged: Boolean) { - if (myReconnectMutex.tryLock()) { - try { - if (onlyIfChanged) { - relayPool.reconnectIfNeedsToORIfItIsTime() + val refreshConnection = MutableStateFlow(Reconnect(false, false)) + + @OptIn(FlowPreview::class) + val debouncingConnection = + refreshConnection + .debounce(200) + .onEach { + if (it.onlyIfChanged) { + relayPool.reconnectIfNeedsTo(it.ignoreRetryDelays) } else { relayPool.disconnect() relayPool.connect() } - } finally { - myReconnectMutex.unlock() - } - } + }.stateIn( + scope, + SharingStarted.Eagerly, + false, + ) + + override fun reconnect( + onlyIfChanged: Boolean, + ignoreRetryDelays: Boolean, + ) { + refreshConnection.tryEmit(Reconnect(onlyIfChanged, ignoreRetryDelays)) } fun needsToResendRequest( @@ -250,7 +265,7 @@ class NostrClient( } // wakes up all the other relays - relayPool.reconnectIfNeedsToORIfItIsTime() + reconnect(true) } } @@ -289,7 +304,7 @@ class NostrClient( } // wakes up all the other relays - relayPool.reconnectIfNeedsToORIfItIsTime() + reconnect(true) } } @@ -301,7 +316,7 @@ class NostrClient( relayPool.getRelay(connectedRelay)?.send(event) // wakes up all the other relays - relayPool.reconnectIfNeedsToORIfItIsTime() + reconnect(true) } } @@ -315,7 +330,7 @@ class NostrClient( relayPool.send(event, relayList) // wakes up all the other relays - relayPool.reconnectIfNeedsToORIfItIsTime() + reconnect(true) } } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/RelayPool.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/RelayPool.kt index 7c969c6ab..d91cb08d9 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/RelayPool.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/RelayPool.kt @@ -28,7 +28,6 @@ import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayState import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl -import com.vitorpamplona.quartz.utils.TimeUtils import com.vitorpamplona.quartz.utils.cache.LargeCache import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow @@ -71,17 +70,7 @@ class RelayPool( fun getRelay(url: NormalizedRelayUrl): IRelayClient? = relays.get(url) - var lastReconnectCall = 0L - - fun reconnectIfNeedsToORIfItIsTime() { - if (lastReconnectCall < TimeUtils.oneMinuteAgo()) { - reconnectIfNeedsTo() - - lastReconnectCall = TimeUtils.now() - } - } - - fun reconnectIfNeedsTo() { + fun reconnectIfNeedsTo(ignoreRetryDelays: Boolean = false) { relays.forEach { url, relay -> if (relay.isConnected()) { if (relay.needsToReconnect()) { @@ -91,7 +80,7 @@ class RelayPool( } } else { // relay is not connected. Connect if it is time - relay.connectAndSyncFiltersIfDisconnected() + relay.connectAndSyncFiltersIfDisconnected(ignoreRetryDelays) } } updateStatus() diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/single/IRelayClient.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/single/IRelayClient.kt index 76b562722..dc2f6e268 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/single/IRelayClient.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/single/IRelayClient.kt @@ -34,7 +34,7 @@ interface IRelayClient { fun connectAndRunAfterSync(onConnected: () -> Unit) - fun connectAndSyncFiltersIfDisconnected() + fun connectAndSyncFiltersIfDisconnected(ignoreRetryDelays: Boolean = false) fun isConnected(): Boolean 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 03ebbcf1e..ed95cf216 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 @@ -387,10 +387,10 @@ open class BasicRelayClient( } } - override fun connectAndSyncFiltersIfDisconnected() { + override fun connectAndSyncFiltersIfDisconnected(ignoreRetryDelays: Boolean) { if (!isConnectionStarted() && !connectingMutex.load()) { // waits 60 seconds to reconnect after disconnected. - if (TimeUtils.now() > lastConnectTentativeInSeconds + delayToConnectInSeconds) { + if (ignoreRetryDelays || TimeUtils.now() > lastConnectTentativeInSeconds + delayToConnectInSeconds) { upRelayDelayToConnect() connect() }