nick/amethyst
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Improves the FetchFirst to not fail in the first eose from the first relay.

Browse Source
This commit is contained in:
Vitor Pamplona
2026-04-20 11:15:09 -04:00
parent d7cc99b196
commit ad624031bf
1 changed files with 28 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchFirstExt.kt
+24 -10
View File
@@ -29,6 +29,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer
import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED
import kotlinx.coroutines.selects.select
import kotlinx.coroutines.withTimeoutOrNull import kotlinx.coroutines.withTimeoutOrNull
suspend fun INostrClient.fetchFirst( suspend fun INostrClient.fetchFirst(
@@ -66,8 +67,11 @@ suspend fun INostrClient.fetchFirst(
suspend fun INostrClient.fetchFirst( suspend fun INostrClient.fetchFirst(
subscriptionId: String = newSubId(), subscriptionId: String = newSubId(),
filters: Map<NormalizedRelayUrl, List<Filter>>, filters: Map<NormalizedRelayUrl, List<Filter>>,
timeoutMs: Long = 30_000L,
): Event? { ): Event? {
val resultChannel = Channel<Event?>(UNLIMITED) val eventChannel = Channel<Event>(UNLIMITED)
val doneChannel = Channel<NormalizedRelayUrl>(UNLIMITED)
val remaining = filters.keys.toMutableSet()
val listener = val listener =
object : SubscriptionListener { object : SubscriptionListener {
@@ -77,7 +81,7 @@ suspend fun INostrClient.fetchFirst(
relay: NormalizedRelayUrl, relay: NormalizedRelayUrl,
forFilters: List<Filter>?, forFilters: List<Filter>?,
) { ) {
resultChannel.trySend(event) eventChannel.trySend(event)
} }
override fun onCannotConnect( override fun onCannotConnect(
@@ -85,7 +89,7 @@ suspend fun INostrClient.fetchFirst(
message: String, message: String,
forFilters: List<Filter>?, forFilters: List<Filter>?,
) { ) {
resultChannel.trySend(null) doneChannel.trySend(relay)
} }
override fun onClosed( override fun onClosed(
@@ -93,29 +97,39 @@ suspend fun INostrClient.fetchFirst(
relay: NormalizedRelayUrl, relay: NormalizedRelayUrl,
forFilters: List<Filter>?, forFilters: List<Filter>?,
) { ) {
resultChannel.trySend(null) doneChannel.trySend(relay)
} }
override fun onEose( override fun onEose(
relay: NormalizedRelayUrl, relay: NormalizedRelayUrl,
forFilters: List<Filter>?, forFilters: List<Filter>?,
) { ) {
resultChannel.trySend(null) doneChannel.trySend(relay)
} }
} }
val result = var result: Event? = null
try { try {
subscribe(subscriptionId, filters, listener) subscribe(subscriptionId, filters, listener)
withTimeoutOrNull(30000) { withTimeoutOrNull(timeoutMs) {
resultChannel.receive() while (remaining.isNotEmpty()) {
select {
eventChannel.onReceive { event ->
result = event
remaining.clear()
}
doneChannel.onReceive { relay ->
remaining.remove(relay)
}
}
}
} }
} finally { } finally {
unsubscribe(subscriptionId) unsubscribe(subscriptionId)
eventChannel.close()
doneChannel.close()
} }
resultChannel.close()
return result return result
} }