Close negentropy sync
This commit is contained in:
@@ -34,6 +34,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.CloseCmd
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.ReqCmd
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.displayUrl
|
||||
import com.vitorpamplona.quartz.nip77Negentropy.NegCloseCmd
|
||||
import com.vitorpamplona.quartz.nip77Negentropy.NegOpenCmd
|
||||
import kotlinx.coroutines.coroutineScope
|
||||
import kotlinx.coroutines.flow.Flow
|
||||
@@ -210,7 +211,7 @@ class RelayPool(
|
||||
}
|
||||
}
|
||||
|
||||
suspend fun sync(negOpenCmd: NegOpenCmd, relayUrl: String): Flow<NostrIncomingMessage> {
|
||||
suspend fun negentropySync(negOpenCmd: NegOpenCmd, relayUrl: String): Flow<NostrIncomingMessage> {
|
||||
addRelaysIfMissing(
|
||||
setOf(
|
||||
NormalizedRelayUrl(relayUrl).url.toRelayDTO()
|
||||
@@ -256,6 +257,28 @@ class RelayPool(
|
||||
}
|
||||
}
|
||||
|
||||
suspend fun closeNegentropySync(negCloseCmd: NegCloseCmd, relayUrl: String) {
|
||||
addRelaysIfMissing(
|
||||
setOf(
|
||||
NormalizedRelayUrl(relayUrl).url.toRelayDTO()
|
||||
)
|
||||
)
|
||||
|
||||
logger.d("socketClients: ${socketClients.map { it.socketUrl }}")
|
||||
val nostrSocketClient = socketClients.find { NormalizedRelayUrl(it.socketUrl).displayUrl() == NormalizedRelayUrl(relayUrl).displayUrl() }
|
||||
|
||||
val closeNegentropySubscription = OptimizedJsonMapper.toJson(negCloseCmd)
|
||||
|
||||
if (nostrSocketClient == null) {
|
||||
throw NetworkException("$relayUrl is not connected")
|
||||
}
|
||||
coroutineScope {
|
||||
with(nostrSocketClient) {
|
||||
sendMESSAGE(closeNegentropySubscription)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private fun Flow<NostrIncomingMessage>.transformWhileEventsAreIncoming() =
|
||||
transformWhile {
|
||||
emit(it)
|
||||
|
||||
@@ -21,6 +21,7 @@ import ac.cord.auxiliary.compose.managers.toHex
|
||||
import ac.cord.auxiliary.compose.network.sockets.NostrIncomingMessage
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.CloseCmd
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.ReqCmd
|
||||
import com.vitorpamplona.quartz.nip77Negentropy.NegCloseCmd
|
||||
import com.vitorpamplona.quartz.nip77Negentropy.NegOpenCmd
|
||||
import kotlinx.coroutines.flow.Flow
|
||||
|
||||
@@ -144,12 +145,6 @@ class RelaysSocketManager constructor(
|
||||
)
|
||||
}
|
||||
|
||||
suspend fun sync(negOpenCmd: NegOpenCmd, relayUrl: String): Flow<NostrIncomingMessage> {
|
||||
return relayPool.sync(
|
||||
negOpenCmd = negOpenCmd,
|
||||
relayUrl = relayUrl
|
||||
)
|
||||
}
|
||||
|
||||
suspend fun closeQuery(closeCmd: CloseCmd, relayUrl: String) {
|
||||
return relayPool.closeQuery(
|
||||
@@ -157,4 +152,18 @@ class RelaysSocketManager constructor(
|
||||
relayUrl = relayUrl
|
||||
)
|
||||
}
|
||||
|
||||
suspend fun negentropySync(negOpenCmd: NegOpenCmd, relayUrl: String): Flow<NostrIncomingMessage> {
|
||||
return relayPool.negentropySync(
|
||||
negOpenCmd = negOpenCmd,
|
||||
relayUrl = relayUrl
|
||||
)
|
||||
}
|
||||
|
||||
suspend fun closeNegentropySync(negCloseCmd: NegCloseCmd, relayUrl: String) {
|
||||
return relayPool.closeNegentropySync(
|
||||
negCloseCmd = negCloseCmd,
|
||||
relayUrl = relayUrl
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -28,6 +28,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.ReqCmd
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
|
||||
import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerSync
|
||||
import com.vitorpamplona.quartz.nip51Lists.encryption.PrivateTagsInContent
|
||||
import com.vitorpamplona.quartz.nip77Negentropy.NegCloseCmd
|
||||
import com.vitorpamplona.quartz.nip77Negentropy.NegOpenCmd
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
@@ -41,7 +42,6 @@ import kotlinx.coroutines.flow.distinctUntilChanged
|
||||
import kotlinx.coroutines.flow.getAndUpdate
|
||||
import kotlinx.coroutines.flow.timeout
|
||||
import kotlinx.coroutines.launch
|
||||
import kotlin.time.Clock
|
||||
import kotlin.time.Duration.Companion.milliseconds
|
||||
import kotlin.time.Instant
|
||||
|
||||
@@ -263,7 +263,11 @@ class NavigationViewModel(
|
||||
|
||||
scope.launch(Dispatchers.IO) {
|
||||
try {
|
||||
relaysSocketManager.sync(
|
||||
val negCloseCmd = NegCloseCmd(
|
||||
subId = negentropySynchronizeRequest.id,
|
||||
)
|
||||
|
||||
relaysSocketManager.negentropySync(
|
||||
negOpenCmd,
|
||||
negentropySynchronizeRequest.relayURL
|
||||
).collect { nostrIncomingMessage ->
|
||||
@@ -293,17 +297,14 @@ class NavigationViewModel(
|
||||
}
|
||||
is NostrIncomingMessage.EoseMessage -> {
|
||||
logger.d("Sync request has been successfully processed (${negentropySynchronizeRequest.relayURL}): $nostrIncomingMessage")
|
||||
val closeCommand = CloseCmd(
|
||||
subId = negentropySynchronizeRequest.id,
|
||||
)
|
||||
|
||||
relaysSocketManager.closeQuery(
|
||||
closeCommand,
|
||||
relaysSocketManager.closeNegentropySync(
|
||||
negCloseCmd,
|
||||
negentropySynchronizeRequest.relayURL
|
||||
)
|
||||
}
|
||||
is NostrIncomingMessage.NegentropyError -> {
|
||||
logger.d("Negentropy Erroy (need to synchronize like normal): ${nostrIncomingMessage.negentropyReason}")
|
||||
logger.d("Negentropy Error (need to synchronize like normal): ${nostrIncomingMessage.negentropyReason}")
|
||||
|
||||
val since = negentropySynchronizeRequest.synchronizationFilter.since // TODO: Update since -> until based on what we have in the local db...
|
||||
val until = negentropySynchronizeRequest.synchronizationFilter.until
|
||||
@@ -346,6 +347,11 @@ class NavigationViewModel(
|
||||
nostrRepository.scheduleBroadcastNostrEventRequests(
|
||||
broadcastNostrEventRequests
|
||||
)
|
||||
|
||||
relaysSocketManager.closeNegentropySync(
|
||||
negCloseCmd,
|
||||
negentropySynchronizeRequest.relayURL
|
||||
)
|
||||
}
|
||||
else -> {
|
||||
logger.d("Unhandled message ${negentropySynchronizeRequest.relayURL}: $nostrIncomingMessage")
|
||||
|
||||
Reference in New Issue
Block a user