diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/reqs/NostrClientStaticReqUntilEoseAsFlow.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/reqs/NostrClientStaticReqUntilEoseAsFlow.kt new file mode 100644 index 000000000..371c48d29 --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/reqs/NostrClientStaticReqUntilEoseAsFlow.kt @@ -0,0 +1,92 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.quartz.nip01Core.relay.client.reqs + +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.core.HexKey +import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer +import com.vitorpamplona.quartz.utils.RandomInstance +import kotlinx.coroutines.channels.awaitClose +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.callbackFlow + +/** + * 1. Subscribes to a req while the flow is active + * 2. Emits accumulated events as a list on each new arrival + * 3. Closes the flow (completes) when the relay sends EOSE, + * cancelling the subscription automatically via [awaitClose]. + */ +fun INostrClient.reqUntilEoseAsFlow( + relay: String, + filters: List, +) = reqUntilEoseAsFlow(RelayUrlNormalizer.normalize(relay), filters) + +fun INostrClient.reqUntilEoseAsFlow( + relay: String, + filter: Filter, +) = reqUntilEoseAsFlow(RelayUrlNormalizer.normalize(relay), listOf(filter)) + +fun INostrClient.reqUntilEoseAsFlow( + relay: NormalizedRelayUrl, + filter: Filter, +) = reqUntilEoseAsFlow(relay, listOf(filter)) + +fun INostrClient.reqUntilEoseAsFlow( + relay: NormalizedRelayUrl, + filters: List, +): Flow> = + callbackFlow { + val subId = RandomInstance.randomChars(10) + val eventIds = mutableSetOf() + var currentEvents = listOf() + + val listener = + object : IRequestListener { + override fun onEvent( + event: Event, + isLive: Boolean, + relay: NormalizedRelayUrl, + forFilters: List?, + ) { + if (event.id !in eventIds) { + currentEvents = currentEvents + event + eventIds.add(event.id) + trySend(currentEvents) + } + } + + override fun onEose( + relay: NormalizedRelayUrl, + forFilters: List?, + ) { + close() + } + } + + openReqSubscription(subId, mapOf(relay to filters), listener) + + awaitClose { + close(subId) + } + } diff --git a/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientSubscriptionUntilEoseAsFlowTest.kt b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientSubscriptionUntilEoseAsFlowTest.kt new file mode 100644 index 000000000..ddcb9fde9 --- /dev/null +++ b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientSubscriptionUntilEoseAsFlowTest.kt @@ -0,0 +1,125 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.quartz.nip01Core.relay + +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.metadata.MetadataEvent +import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient +import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.reqUntilEoseAsFlow +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.utils.Log +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.FlowPreview +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.cancel +import kotlinx.coroutines.flow.debounce +import kotlinx.coroutines.launch +import kotlinx.coroutines.test.advanceUntilIdle +import kotlinx.coroutines.test.runTest +import kotlin.test.Test +import kotlin.test.assertEquals + +class NostrClientSubscriptionUntilEoseAsFlowTest : BaseNostrClientTest() { + fun List.printDates(): String { + val starting = this[0].createdAt + return joinToString { (it.createdAt - starting).toString() } + } + + @OptIn(ExperimentalCoroutinesApi::class) + @Test + fun testNostrClientSubscriptionUntilEoseAsFlow() = + runTest { + val appScope = CoroutineScope(Dispatchers.Default + SupervisorJob()) + val client = NostrClient(socketBuilder, appScope) + + val flow = + client.reqUntilEoseAsFlow( + relay = "wss://relay.damus.io", + filter = + Filter( + kinds = listOf(MetadataEvent.KIND), + limit = 10, + ), + ) + + var feedStates = listOf() + val job = + launch { + flow.collect { + Log.d("ZZ", "List timestamp deltas ${it.printDates()}") + feedStates = it + } + } + + // Advance the test dispatcher to ensure emissions are processed + while (feedStates.size < 10) { + advanceUntilIdle() + } + + job.cancel() // Cancel the collection job + + client.disconnect() + appScope.cancel() + + assertEquals(10, feedStates.size) + } + + @OptIn(FlowPreview::class, ExperimentalCoroutinesApi::class) + @Test + fun testNostrClientSubscriptionUntilEoseAsFlowDebouncing() = + runTest { + val appScope = CoroutineScope(Dispatchers.Default + SupervisorJob()) + val client = NostrClient(socketBuilder, appScope) + + val flow = + client.reqUntilEoseAsFlow( + relay = "wss://relay.damus.io", + filter = + Filter( + kinds = listOf(MetadataEvent.KIND), + limit = 10, + ), + ) + + var feedStates = listOf() + val job = + launch { + flow.debounce(100).collect { + Log.d("ZZ", "List timestamp deltas ${it.printDates()}") + feedStates = it + } + } + + // Advance the test dispatcher to ensure emissions are processed + while (feedStates.size < 10) { + advanceUntilIdle() + } + + job.cancel() // Cancel the collection job + + client.disconnect() + appScope.cancel() + + assertEquals(10, feedStates.size) + } +}