Creates an option to ignore reconnection relays that should be used when the network settings change.

This commit is contained in:
Vitor Pamplona
2025-10-03 18:03:04 -04:00
parent 26462a150e
commit cf8c8e47a2
6 changed files with 47 additions and 34 deletions
@@ -95,7 +95,10 @@ class RelayProxyClientConnector(
client.connect() client.connect()
} else { } else {
Log.d("ManageRelayServices", "Relay Services have changed, reconnecting relays that need to") Log.d("ManageRelayServices", "Relay Services have changed, reconnecting relays that need to")
client.reconnect(true) client.reconnect(
onlyIfChanged = true,
ignoreRetryDelays = true,
)
} }
}.onStart { }.onStart {
Log.d("ManageRelayServices", "Resuming Relay Services") Log.d("ManageRelayServices", "Resuming Relay Services")
@@ -36,7 +36,10 @@ interface INostrClient {
fun disconnect() fun disconnect()
fun reconnect(onlyIfChanged: Boolean = false) fun reconnect(
onlyIfChanged: Boolean = false,
ignoreRetryDelays: Boolean = false,
)
fun isActive(): Boolean fun isActive(): Boolean
@@ -78,7 +81,10 @@ object EmptyNostrClient : INostrClient {
override fun disconnect() { } override fun disconnect() { }
override fun reconnect(onlyIfChanged: Boolean) { } override fun reconnect(
onlyIfChanged: Boolean,
ignoreRetryDelays: Boolean,
) { }
override fun isActive() = false override fun isActive() = false
@@ -37,14 +37,15 @@ import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.DelicateCoroutinesApi import kotlinx.coroutines.DelicateCoroutinesApi
import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.FlowPreview import kotlinx.coroutines.FlowPreview
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.SharingStarted import kotlinx.coroutines.flow.SharingStarted
import kotlinx.coroutines.flow.combine import kotlinx.coroutines.flow.combine
import kotlinx.coroutines.flow.debounce
import kotlinx.coroutines.flow.flowOn import kotlinx.coroutines.flow.flowOn
import kotlinx.coroutines.flow.onEach import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.flow.sample import kotlinx.coroutines.flow.sample
import kotlinx.coroutines.flow.stateIn import kotlinx.coroutines.flow.stateIn
import kotlinx.coroutines.launch import kotlinx.coroutines.launch
import kotlinx.coroutines.sync.Mutex
/** /**
* The NostrClient manages Nostr relay operations, subscriptions, and event delivery. It maintains: * The NostrClient manages Nostr relay operations, subscriptions, and event delivery. It maintains:
@@ -140,21 +141,35 @@ class NostrClient(
override fun isActive() = isActive override fun isActive() = isActive
val myReconnectMutex = Mutex() class Reconnect(
val onlyIfChanged: Boolean,
val ignoreRetryDelays: Boolean,
)
override fun reconnect(onlyIfChanged: Boolean) { val refreshConnection = MutableStateFlow(Reconnect(false, false))
if (myReconnectMutex.tryLock()) {
try { @OptIn(FlowPreview::class)
if (onlyIfChanged) { val debouncingConnection =
relayPool.reconnectIfNeedsToORIfItIsTime() refreshConnection
.debounce(200)
.onEach {
if (it.onlyIfChanged) {
relayPool.reconnectIfNeedsTo(it.ignoreRetryDelays)
} else { } else {
relayPool.disconnect() relayPool.disconnect()
relayPool.connect() relayPool.connect()
} }
} finally { }.stateIn(
myReconnectMutex.unlock() scope,
} SharingStarted.Eagerly,
} false,
)
override fun reconnect(
onlyIfChanged: Boolean,
ignoreRetryDelays: Boolean,
) {
refreshConnection.tryEmit(Reconnect(onlyIfChanged, ignoreRetryDelays))
} }
fun needsToResendRequest( fun needsToResendRequest(
@@ -250,7 +265,7 @@ class NostrClient(
} }
// wakes up all the other relays // wakes up all the other relays
relayPool.reconnectIfNeedsToORIfItIsTime() reconnect(true)
} }
} }
@@ -289,7 +304,7 @@ class NostrClient(
} }
// wakes up all the other relays // wakes up all the other relays
relayPool.reconnectIfNeedsToORIfItIsTime() reconnect(true)
} }
} }
@@ -301,7 +316,7 @@ class NostrClient(
relayPool.getRelay(connectedRelay)?.send(event) relayPool.getRelay(connectedRelay)?.send(event)
// wakes up all the other relays // wakes up all the other relays
relayPool.reconnectIfNeedsToORIfItIsTime() reconnect(true)
} }
} }
@@ -315,7 +330,7 @@ class NostrClient(
relayPool.send(event, relayList) relayPool.send(event, relayList)
// wakes up all the other relays // wakes up all the other relays
relayPool.reconnectIfNeedsToORIfItIsTime() reconnect(true)
} }
} }
@@ -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.client.single.IRelayClient
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
import com.vitorpamplona.quartz.utils.TimeUtils
import com.vitorpamplona.quartz.utils.cache.LargeCache import com.vitorpamplona.quartz.utils.cache.LargeCache
import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.StateFlow
@@ -71,17 +70,7 @@ class RelayPool(
fun getRelay(url: NormalizedRelayUrl): IRelayClient? = relays.get(url) fun getRelay(url: NormalizedRelayUrl): IRelayClient? = relays.get(url)
var lastReconnectCall = 0L fun reconnectIfNeedsTo(ignoreRetryDelays: Boolean = false) {
fun reconnectIfNeedsToORIfItIsTime() {
if (lastReconnectCall < TimeUtils.oneMinuteAgo()) {
reconnectIfNeedsTo()
lastReconnectCall = TimeUtils.now()
}
}
fun reconnectIfNeedsTo() {
relays.forEach { url, relay -> relays.forEach { url, relay ->
if (relay.isConnected()) { if (relay.isConnected()) {
if (relay.needsToReconnect()) { if (relay.needsToReconnect()) {
@@ -91,7 +80,7 @@ class RelayPool(
} }
} else { } else {
// relay is not connected. Connect if it is time // relay is not connected. Connect if it is time
relay.connectAndSyncFiltersIfDisconnected() relay.connectAndSyncFiltersIfDisconnected(ignoreRetryDelays)
} }
} }
updateStatus() updateStatus()
@@ -34,7 +34,7 @@ interface IRelayClient {
fun connectAndRunAfterSync(onConnected: () -> Unit) fun connectAndRunAfterSync(onConnected: () -> Unit)
fun connectAndSyncFiltersIfDisconnected() fun connectAndSyncFiltersIfDisconnected(ignoreRetryDelays: Boolean = false)
fun isConnected(): Boolean fun isConnected(): Boolean
@@ -387,10 +387,10 @@ open class BasicRelayClient(
} }
} }
override fun connectAndSyncFiltersIfDisconnected() { override fun connectAndSyncFiltersIfDisconnected(ignoreRetryDelays: Boolean) {
if (!isConnectionStarted() && !connectingMutex.load()) { if (!isConnectionStarted() && !connectingMutex.load()) {
// waits 60 seconds to reconnect after disconnected. // waits 60 seconds to reconnect after disconnected.
if (TimeUtils.now() > lastConnectTentativeInSeconds + delayToConnectInSeconds) { if (ignoreRetryDelays || TimeUtils.now() > lastConnectTentativeInSeconds + delayToConnectInSeconds) {
upRelayDelayToConnect() upRelayDelayToConnect()
connect() connect()
} }