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

refactor(geode): test fixtures + collectUntilEose helper, move to testFixtures sourceSet

Browse Source 
The in-process relay was usable but every consumer test paid a 10–15
line tax for scope/client setup, manual cleanup inside the test body
(leaks on assertion failure), hardcoded "ws://127.0.0.1:7770/" magic
strings, and an inline collectUntilEose pattern reinvented per file.

Net: ~250 LOC removed, every test gets correct @After cleanup, and
the synthetic event builders no longer ship in geode's production jar.

- Move geode.fixtures (synthetic event builders) and the new
  geode.testing package from src/main to a new src/testFixtures
  source set via the java-test-fixtures plugin. Production geode jar
  no longer includes them; consumers wire them with
  testImplementation(testFixtures(project(":geode"))).
- Add geode.testing.RelayClientTest open class — owns hub, scope,
  client, defaultRelay, defaultRelayUrl with @After-driven cleanup.
- Add geode.testing.collectUntilEose / collectUntilEoseMulti
  NostrClient extensions with a shared subId counter and timeout knob.
  Replaces hand-rolled subscriber loops in 4+ files.
- Add RelayHub.DEFAULT_URL constant so tests stop typing
  "ws://127.0.0.1:7770/" inline.
- Migrate the 8 quartz NostrClient*Test files + geode's
  Nip01ComplianceTest to extend RelayClientTest and (where REQ/EOSE
  is the pattern) call collectUntilEose. Delete the redundant
  BaseNostrClientTest wrapper.
This commit is contained in:
Claude
2026-05-07 13:42:20 +00:00
parent c2f24a5213
commit c19bd4e92e
18 changed files with 342 additions and 493 deletions
Download Patch File Download Diff File Expand all files Collapse all files
geode/src/main/kotlin/com/vitorpamplona/geode/RelayHub.kt
+12
View File
@@ -86,4 +86,16 @@ class RelayHub(
relays.values.forEach { runCatching { it.close() } }
relays.clear()
}
companion object {
/**
* Default URL for tests that only need one relay. The URL itself
* has no semantic meaning — it's just a stable key into the hub
* — but it normalises through [RelayUrlNormalizer] (loopback) so
* the production [com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer]
* accepts it. Prefer this over typing `"ws://127.0.0.1:7770/"`
* everywhere.
*/
val DEFAULT_URL: NormalizedRelayUrl = RelayUrlNormalizer.normalize("ws://127.0.0.1:7770/")
}
}
geode/src/test/kotlin/com/vitorpamplona/geode/Nip01ComplianceTest.kt
+32 -140
View File
@@ -21,26 +21,22 @@
package com.vitorpamplona.geode
import com.vitorpamplona.geode.fixtures.SyntheticEvents
import com.vitorpamplona.geode.testing.RelayClientTest
import com.vitorpamplona.geode.testing.collectUntilEose
import com.vitorpamplona.geode.testing.collectUntilEoseMulti
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener
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.nip01Core.signers.NostrSignerSync
import com.vitorpamplona.quartz.nip23LongContent.LongTextNoteEvent
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.cancel
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withTimeout
import kotlinx.coroutines.withTimeoutOrNull
import kotlin.test.AfterTest
import kotlin.test.BeforeTest
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertNull
@@ -62,29 +58,9 @@ import kotlin.test.assertTrue
* spec-compliant relay (nostr-rs-relay, strfry, khatru, …) — only the
* `socketBuilder` and the relay URL would change.
*/
class Nip01ComplianceTest {
private lateinit var hub: RelayHub
private lateinit var scope: CoroutineScope
private lateinit var client: NostrClient
private val relayUrl: NormalizedRelayUrl = RelayUrlNormalizer.normalize("ws://127.0.0.1:7770/")
@BeforeTest
fun setup() {
hub = RelayHub()
scope = CoroutineScope(Dispatchers.Default + SupervisorJob())
client = NostrClient(hub, scope)
}
@AfterTest
fun teardown() {
client.disconnect()
scope.cancel()
hub.close()
}
class Nip01ComplianceTest : RelayClientTest() {
private suspend fun preload(vararg events: Event) {
hub.getOrCreate(relayUrl).preload(*events)
defaultRelay.preload(*events)
}
private fun fakeEvent(
@@ -108,7 +84,7 @@ class Nip01ComplianceTest {
fakeEvent(3, kind = 1),
)
val (events, eose) = collectUntilEose(Filter(kinds = listOf(1)))
val (events, eose) = client.collectUntilEose(defaultRelayUrl, Filter(kinds = listOf(1)))
assertEquals(2, events.size)
assertEquals(setOf(SyntheticEvents.hexId(1), SyntheticEvents.hexId(3)), events.map { it.id }.toSet())
@@ -126,7 +102,7 @@ class Nip01ComplianceTest {
fakeEvent(4, kind = 1, createdAt = 400),
)
val (events, _) = collectUntilEose(Filter(kinds = listOf(1), limit = 2))
val (events, _) = client.collectUntilEose(defaultRelayUrl, Filter(kinds = listOf(1), limit = 2))
assertEquals(2, events.size)
assertEquals(400L, events[0].createdAt)
@@ -145,7 +121,7 @@ class Nip01ComplianceTest {
fakeEvent(3, kind = 1, pubKey = alice),
)
val (events, _) = collectUntilEose(Filter(authors = listOf(alice)))
val (events, _) = client.collectUntilEose(defaultRelayUrl, Filter(authors = listOf(alice)))
assertEquals(2, events.size)
assertTrue(events.all { it.pubKey == alice })
@@ -157,7 +133,7 @@ class Nip01ComplianceTest {
runBlocking {
preload(fakeEvent(1), fakeEvent(2), fakeEvent(3))
val (events, _) = collectUntilEose(Filter(ids = listOf(SyntheticEvents.hexId(2))))
val (events, _) = client.collectUntilEose(defaultRelayUrl, Filter(ids = listOf(SyntheticEvents.hexId(2))))
assertEquals(1, events.size)
assertEquals(SyntheticEvents.hexId(2), events[0].id)
@@ -174,7 +150,7 @@ class Nip01ComplianceTest {
fakeEvent(4, createdAt = 400),
)
val (events, _) = collectUntilEose(Filter(since = 150L, until = 350L))
val (events, _) = client.collectUntilEose(defaultRelayUrl, Filter(since = 150L, until = 350L))
assertEquals(setOf(200L, 300L), events.map { it.createdAt }.toSet())
}
@@ -190,7 +166,7 @@ class Nip01ComplianceTest {
fakeEvent(3, tags = arrayOf(arrayOf("e", target), arrayOf("p", SyntheticEvents.hexId(8)))),
)
val (events, _) = collectUntilEose(Filter(tags = mapOf("e" to listOf(target))))
val (events, _) = client.collectUntilEose(defaultRelayUrl, Filter(tags = mapOf("e" to listOf(target))))
assertEquals(setOf(SyntheticEvents.hexId(1), SyntheticEvents.hexId(3)), events.map { it.id }.toSet())
}
@@ -206,7 +182,7 @@ class Nip01ComplianceTest {
fakeEvent(3, tags = arrayOf(arrayOf("p", targetPubkey), arrayOf("e", SyntheticEvents.hexId(99)))),
)
val (events, _) = collectUntilEose(Filter(tags = mapOf("p" to listOf(targetPubkey))))
val (events, _) = client.collectUntilEose(defaultRelayUrl, Filter(tags = mapOf("p" to listOf(targetPubkey))))
assertEquals(setOf(SyntheticEvents.hexId(1), SyntheticEvents.hexId(3)), events.map { it.id }.toSet())
}
@@ -221,7 +197,7 @@ class Nip01ComplianceTest {
fakeEvent(3, tags = arrayOf(arrayOf("t", "nostr"), arrayOf("t", "kotlin"))),
)
val (events, _) = collectUntilEose(Filter(tags = mapOf("t" to listOf("nostr"))))
val (events, _) = client.collectUntilEose(defaultRelayUrl, Filter(tags = mapOf("t" to listOf("nostr"))))
assertEquals(setOf(SyntheticEvents.hexId(1), SyntheticEvents.hexId(3)), events.map { it.id }.toSet())
}
@@ -241,7 +217,7 @@ class Nip01ComplianceTest {
fakeEvent(3, tags = arrayOf(arrayOf("e", SyntheticEvents.hexId(999)))),
)
val (events, _) = collectUntilEose(Filter(tags = mapOf("e" to listOf(a, b))))
val (events, _) = client.collectUntilEose(defaultRelayUrl, Filter(tags = mapOf("e" to listOf(a, b))))
assertEquals(setOf(SyntheticEvents.hexId(1), SyntheticEvents.hexId(2)), events.map { it.id }.toSet())
}
@@ -262,7 +238,8 @@ class Nip01ComplianceTest {
)
val (events, _) =
collectUntilEoseMulti(
client.collectUntilEoseMulti(
defaultRelayUrl,
listOf(
Filter(kinds = listOf(1)),
Filter(kinds = listOf(7)),
@@ -294,7 +271,7 @@ class Nip01ComplianceTest {
client.subscribe(
"sub-A",
mapOf(relayUrl to listOf(Filter(kinds = listOf(1)))),
mapOf(defaultRelayUrl to listOf(Filter(kinds = listOf(1)))),
object : SubscriptionListener {
override fun onEvent(
event: Event,
@@ -315,7 +292,7 @@ class Nip01ComplianceTest {
)
client.subscribe(
"sub-B",
mapOf(relayUrl to listOf(Filter(kinds = listOf(4)))),
mapOf(defaultRelayUrl to listOf(Filter(kinds = listOf(4)))),
object : SubscriptionListener {
override fun onEvent(
event: Event,
@@ -362,7 +339,7 @@ class Nip01ComplianceTest {
fakeEvent(2, kind = 0, pubKey = pubkey, createdAt = 200, content = "new"),
)
val (events, _) = collectUntilEose(Filter(kinds = listOf(0), authors = listOf(pubkey)))
val (events, _) = client.collectUntilEose(defaultRelayUrl, Filter(kinds = listOf(0), authors = listOf(pubkey)))
assertEquals(1, events.size)
assertEquals("new", events[0].content)
@@ -383,7 +360,11 @@ class Nip01ComplianceTest {
val v3 = signer.sign(LongTextNoteEvent.build("list-b", "title", dTag = "list-b", createdAt = 100))
preload(v1, v2, v3)
val (events, _) = collectUntilEose(Filter(kinds = listOf(LongTextNoteEvent.KIND), authors = listOf(signer.pubKey)))
val (events, _) =
client.collectUntilEose(
defaultRelayUrl,
Filter(kinds = listOf(LongTextNoteEvent.KIND), authors = listOf(signer.pubKey)),
)
assertEquals(2, events.size)
assertEquals(setOf("new", "list-b"), events.map { it.content }.toSet())
@@ -399,7 +380,7 @@ class Nip01ComplianceTest {
val gotEose = Channel<Unit>(UNLIMITED)
client.subscribe(
"live-1",
mapOf(relayUrl to listOf(Filter(kinds = listOf(1)))),
mapOf(defaultRelayUrl to listOf(Filter(kinds = listOf(1)))),
object : SubscriptionListener {
override fun onEvent(
event: Event,
@@ -423,7 +404,7 @@ class Nip01ComplianceTest {
// Inject an event through the wire path (not preload — that bypasses
// the live broadcast that subscriptions feed off of).
hub.getOrCreate(relayUrl).publish(fakeEvent(99, kind = 1, content = "live"))
defaultRelay.publish(fakeEvent(99, kind = 1, content = "live"))
val received = withTimeout(5000) { ch.receive() }
assertEquals("live", received.content)
@@ -438,7 +419,7 @@ class Nip01ComplianceTest {
val gotEose = Channel<Unit>(UNLIMITED)
client.subscribe(
"live-2",
mapOf(relayUrl to listOf(Filter(kinds = listOf(1)))),
mapOf(defaultRelayUrl to listOf(Filter(kinds = listOf(1)))),
object : SubscriptionListener {
override fun onEvent(
event: Event,
@@ -459,7 +440,7 @@ class Nip01ComplianceTest {
)
withTimeout(5000) { gotEose.receive() }
hub.getOrCreate(relayUrl).publish(fakeEvent(98, kind = 4, content = "off-topic"))
defaultRelay.publish(fakeEvent(98, kind = 4, content = "off-topic"))
val seen = withTimeoutOrNull(500) { ch.receive() }
assertNull(seen, "kind 4 should not match a kind-1 subscription")
@@ -479,7 +460,7 @@ class Nip01ComplianceTest {
val gotEose = Channel<Unit>(UNLIMITED)
client.subscribe(
"eph-1",
mapOf(relayUrl to listOf(Filter(kinds = listOf(20_001)))),
mapOf(defaultRelayUrl to listOf(Filter(kinds = listOf(20_001)))),
object : SubscriptionListener {
override fun onEvent(
event: Event,
@@ -500,7 +481,7 @@ class Nip01ComplianceTest {
)
withTimeout(5000) { gotEose.receive() }
hub.getOrCreate(relayUrl).publish(fakeEvent(70, kind = 20_001, content = "ephemeral-payload"))
defaultRelay.publish(fakeEvent(70, kind = 20_001, content = "ephemeral-payload"))
val received = withTimeout(5000) { ch.receive() }
assertEquals("ephemeral-payload", received.content)
@@ -516,10 +497,10 @@ class Nip01ComplianceTest {
fun ephemeralEventIsNotStoredAndDoesNotShowOnFollowupReq() =
runBlocking {
// Publish ephemeral first — no live subscriber listening.
hub.getOrCreate(relayUrl).publish(fakeEvent(71, kind = 20_002, content = "vanish"))
defaultRelay.publish(fakeEvent(71, kind = 20_002, content = "vanish"))
// Late subscriber: should see EOSE with no events.
val (events, eose) = collectUntilEose(Filter(kinds = listOf(20_002)))
val (events, eose) = client.collectUntilEose(defaultRelayUrl, Filter(kinds = listOf(20_002)))
assertTrue(eose, "EOSE must fire for an ephemeral kind even if zero events match")
assertEquals(0, events.size, "Ephemeral events must not be persisted")
}
@@ -570,93 +551,4 @@ class Nip01ComplianceTest {
assertEquals("from-a", received[relayA])
assertEquals("from-b", received[relayB])
}
// -- Helpers ------------------------------------------------------------
/** Subscribes synchronously and returns the events received before EOSE. */
private suspend fun collectUntilEose(filter: Filter): Pair<List<Event>, Boolean> {
val ch = Channel<Either>(UNLIMITED)
val subId = "sub-${System.nanoTime()}"
client.subscribe(
subId,
mapOf(relayUrl to listOf(filter)),
object : SubscriptionListener {
override fun onEvent(
event: Event,
isLive: Boolean,
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
) {
ch.trySend(Either.Ev(event))
}
override fun onEose(
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
) {
ch.trySend(Either.Eose)
}
},
)
val events = mutableListOf<Event>()
var eose = false
withTimeout(5000) {
while (!eose) {
when (val msg = ch.receive()) {
is Either.Ev -> events += msg.event
Either.Eose -> eose = true
}
}
}
client.unsubscribe(subId)
return events to eose
}
/** Variant of [collectUntilEose] that subscribes with multiple filters. */
private suspend fun collectUntilEoseMulti(filters: List<Filter>): Pair<List<Event>, Boolean> {
val ch = Channel<Either>(UNLIMITED)
val subId = "sub-${System.nanoTime()}"
client.subscribe(
subId,
mapOf(relayUrl to filters),
object : SubscriptionListener {
override fun onEvent(
event: Event,
isLive: Boolean,
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
) {
ch.trySend(Either.Ev(event))
}
override fun onEose(
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
) {
ch.trySend(Either.Eose)
}
},
)
val events = mutableListOf<Event>()
var eose = false
withTimeout(5000) {
while (!eose) {
when (val msg = ch.receive()) {
is Either.Ev -> events += msg.event
Either.Eose -> eose = true
}
}
}
client.unsubscribe(subId)
return events to eose
}
private sealed interface Either {
data class Ev(
val event: Event,
) : Either
object Eose : Either
}
}
geode/src/main/kotlin/com/vitorpamplona/geode/fixtures/RelayFixtures.kt → geode/src/testFixtures/kotlin/com/vitorpamplona/geode/fixtures/RelayFixtures.kt
View File
geode/src/main/kotlin/com/vitorpamplona/geode/fixtures/SyntheticEvents.kt → geode/src/testFixtures/kotlin/com/vitorpamplona/geode/fixtures/SyntheticEvents.kt
View File
geode/src/testFixtures/kotlin/com/vitorpamplona/geode/testing/RelayClientTest.kt
+78
View File
@@ -0,0 +1,78 @@
/*
* 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.geode.testing
import com.vitorpamplona.geode.Relay
import com.vitorpamplona.geode.RelayHub
import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.cancel
import org.junit.After
/**
* Base class for tests that drive a real [NostrClient] against an
* in-process [RelayHub]. Owns the lifecycle of the four pieces every
* such test needs:
*
* - [hub] — the registry of in-process relays (also serves as
* `WebsocketBuilder` for [NostrClient]).
* - [scope] — application coroutine scope for the client.
* - [client] — a [NostrClient] wired to [hub] and [scope].
* - [defaultRelay] / [defaultRelayUrl] — convenience handles for the
* single-relay case (the most common in tests).
*
* Cleanup happens in [tearDownRelayClientTest], registered with
* JUnit's [@After][After], so an assertion failure does NOT leak the
* scope, the SQLite event store, or the WebSocket bridge — a recurring
* problem with the previous "clean up at the end of the test body"
* pattern.
*
* Subclasses that need their own setup/teardown should add their own
* `@Before` / `@After` methods; JUnit runs all of them.
*
* Multi-relay tests use [hub] directly:
* ```
* val relayA = RelayUrlNormalizer.normalize("ws://relay-a/")
* hub.getOrCreate(relayA).preload(eventA)
* hub.getOrCreate(relayB).preload(eventB)
* ```
*/
open class RelayClientTest {
val hub: RelayHub = RelayHub()
val scope: CoroutineScope = CoroutineScope(Dispatchers.Default + SupervisorJob())
val client: NostrClient = NostrClient(hub, scope)
/** Stable URL for the single-relay case — see [RelayHub.DEFAULT_URL]. */
val defaultRelayUrl: NormalizedRelayUrl get() = RelayHub.DEFAULT_URL
/** Lazy handle to the relay at [defaultRelayUrl]. Auto-created on first read. */
val defaultRelay: Relay get() = hub.getOrCreate(defaultRelayUrl)
@After
fun tearDownRelayClientTest() {
client.disconnect()
scope.cancel()
hub.close()
}
}
geode/src/testFixtures/kotlin/com/vitorpamplona/geode/testing/SubscriptionTesting.kt
+121
View File
@@ -0,0 +1,121 @@
/*
* 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.geode.testing
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED
import kotlinx.coroutines.withTimeout
/** Tagged result of [collectUntilEose]: stored events plus whether EOSE actually arrived. */
data class CollectResult(
val events: List<Event>,
val eoseReceived: Boolean,
)
/**
* Subscribe with [filter] on a single [relay], drain the
* historical-replay phase, and return when EOSE arrives. The
* subscription is closed before this returns. The pattern that 80% of
* REQ-style tests need.
*
* ```
* val (events, eose) = client.collectUntilEose(defaultRelayUrl, Filter(kinds = listOf(1)))
* assertEquals(20, events.size)
* assertTrue(eose)
* ```
*
* @param timeoutMillis time to wait for EOSE before failing the test.
* Default 5 s — generous for in-process; tighten if needed.
*/
suspend fun NostrClient.collectUntilEose(
relay: NormalizedRelayUrl,
filter: Filter,
timeoutMillis: Long = 5_000,
): CollectResult = collectUntilEoseMulti(relay, listOf(filter), timeoutMillis)
/**
* Multi-filter variant. NIP-01 allows a REQ to carry several filters
* that the relay OR's together. EOSE fires once after the union of all
* filters has been replayed.
*/
suspend fun NostrClient.collectUntilEoseMulti(
relay: NormalizedRelayUrl,
filters: List<Filter>,
timeoutMillis: Long = 5_000,
): CollectResult {
val ch = Channel<Signal>(UNLIMITED)
val subId = "test-sub-${nextSubId()}"
subscribe(
subId,
mapOf(relay to filters),
object : SubscriptionListener {
override fun onEvent(
event: Event,
isLive: Boolean,
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
) {
ch.trySend(Signal.Ev(event))
}
override fun onEose(
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
) {
ch.trySend(Signal.Eose)
}
},
)
val events = mutableListOf<Event>()
var eose = false
try {
withTimeout(timeoutMillis) {
while (!eose) {
when (val msg = ch.receive()) {
is Signal.Ev -> events += msg.event
Signal.Eose -> eose = true
}
}
}
} finally {
unsubscribe(subId)
}
return CollectResult(events, eose)
}
private sealed interface Signal {
data class Ev(
val event: Event,
) : Signal
object Eose : Signal
}
/** Monotonic counter for unique sub-ids inside a JVM. */
private var subIdSeq: Int = 0
private fun nextSubId(): Int = ++subIdSeq