* 'main' of https://github.com/vitorpamplona/amethyst:
  fix: keep screen on during PiP playback and survive screen lock
  feat: add RelayDiscoveryEvent renderer for feed display
  feat: add OpenGraph preview to WebBookmark cards
  revert: keep INostrClient/NostrClient naming and restore onEose/isLive
  refactor: simplify NostrClient API for beginner-friendliness
This commit is contained in:
Vitor Pamplona
2026-03-28 19:25:29 -04:00
90 changed files with 870 additions and 505 deletions
@@ -33,7 +33,6 @@ import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
import com.vitorpamplona.quartz.nip01Core.crypto.verify
import com.vitorpamplona.quartz.nip01Core.jackson.JacksonMapper
import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient
import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerInternal
import com.vitorpamplona.quartz.nip01Core.tags.aTag.ATag
import com.vitorpamplona.quartz.nip03Timestamp.EmptyOtsResolverBuilder
@@ -24,7 +24,6 @@ import androidx.test.ext.junit.runners.AndroidJUnit4
import com.vitorpamplona.amethyst.model.Constants
import com.vitorpamplona.amethyst.service.okhttp.DefaultContentTypeInterceptor
import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.RelayLogger
import com.vitorpamplona.quartz.nip01Core.relay.client.auth.RelayAuthenticator
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.normalizeRelayUrl
@@ -69,7 +69,6 @@ import com.vitorpamplona.amethyst.ui.screen.UiSettingsState
import com.vitorpamplona.amethyst.ui.tor.TorManager
import com.vitorpamplona.quartz.nip01Core.core.Address
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.RelayLogger
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.RelayOfflineTracker
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.stats.RelayReqStats
@@ -245,7 +244,7 @@ class AppModules(
scope = applicationIOScope,
)
// Connects the NostrClient class with okHttp
// Connects the INostrClient class with okHttp
val websocketBuilder =
OkHttpWebSocket.Builder { url ->
val useTor = torEvaluatorFlow.flow.value.useTor(url)
@@ -136,7 +136,7 @@ import com.vitorpamplona.quartz.nip01Core.hints.EventHintProvider
import com.vitorpamplona.quartz.nip01Core.hints.PubKeyHintProvider
import com.vitorpamplona.quartz.nip01Core.metadata.MetadataEvent
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.downloadFirstEvent
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchFirst
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
import com.vitorpamplona.quartz.nip01Core.signers.EventTemplate
@@ -606,7 +606,7 @@ class Account(
onResponse: (Response?) -> Unit,
) {
val (event, relay) = nip47SignerState.sendNwcRequest(request, onResponse)
client.send(event, setOf(relay))
client.publish(event, setOf(relay))
}
suspend fun sendZapPaymentRequestFor(
@@ -615,7 +615,7 @@ class Account(
onResponse: (Response?) -> Unit,
) {
val (event, relay) = nip47SignerState.sendZapPaymentRequestFor(bolt11, zappedNote, onResponse)
client.send(event, setOf(relay))
client.publish(event, setOf(relay))
}
suspend fun createZapRequestFor(
@@ -701,7 +701,7 @@ class Account(
suspend fun boost(note: Note) {
RepostAction.repost(note, signer)?.let { event ->
client.send(event, computeMyReactionToNote(note, event))
client.publish(event, computeMyReactionToNote(note, event))
cache.justConsumeMyOwnEvent(event)
}
}
@@ -723,7 +723,7 @@ class Account(
event: Event,
relays: Set<NormalizedRelayUrl>,
) {
client.send(event, relays)
client.publish(event, relays)
cache.justConsumeMyOwnEvent(event)
}
@@ -939,7 +939,7 @@ class Account(
// download the event and send it.
noteEvent.host?.let { host ->
client
.downloadFirstEvent(
.fetchFirst(
filters =
note.relays.associateWith { relay ->
listOf(
@@ -1012,7 +1012,7 @@ class Account(
fun sendAutomatic(event: Event?) {
if (event == null) return
cache.justConsumeMyOwnEvent(event)
client.send(event, computeRelayListToBroadcast(event))
client.publish(event, computeRelayListToBroadcast(event))
}
suspend fun sendWebBookmark(
@@ -1043,7 +1043,7 @@ class Account(
fun sendMyPublicAndPrivateOutbox(event: Event?) {
if (event == null) return
cache.justConsumeMyOwnEvent(event)
client.send(event, outboxRelays.flow.value)
client.publish(event, outboxRelays.flow.value)
}
fun sendMyPublicAndPrivateOutbox(events: List<Event>) {
@@ -1054,7 +1054,7 @@ class Account(
}
fun sendLiterallyEverywhere(event: Event) {
client.send(event, followPlusAllMineWithIndex.flow.value + client.availableRelaysFlow().value)
client.publish(event, followPlusAllMineWithIndex.flow.value + client.availableRelaysFlow().value)
cache.justConsumeMyOwnEvent(event)
}
@@ -1287,7 +1287,7 @@ class Account(
) {
val event = signer.sign(template)
cache.justConsumeMyOwnEvent(event)
client.send(event, relayList)
client.publish(event, relayList)
}
suspend fun <T : Event> signAndSendPrivatelyOrBroadcast(
@@ -1298,9 +1298,9 @@ class Account(
cache.justConsumeMyOwnEvent(event)
val relays = relayList(event)
if (!relays.isNullOrEmpty()) {
client.send(event, relays.toSet())
client.publish(event, relays.toSet())
} else {
client.send(event, computeRelayListToBroadcast(event))
client.publish(event, computeRelayListToBroadcast(event))
}
return event
}
@@ -1320,7 +1320,7 @@ class Account(
val relayList = computeRelayListToBroadcast(note)
client.send(event, relayList)
client.publish(event, relayList)
broadcast.forEach { client.send(it, relayList) }
@@ -1344,7 +1344,7 @@ class Account(
val relayList = computeRelayListToBroadcast(note)
client.send(event, relayList)
client.publish(event, relayList)
broadcast.forEach { client.send(it, relayList) }
@@ -1463,9 +1463,9 @@ class Account(
val relayList = privateStorageRelayList.flow.value + localRelayList.flow.value
if (relayList.isNotEmpty()) {
client.send(event, relayList + noteRelays)
client.publish(event, relayList + noteRelays)
} else {
client.send(event, outboxRelays.flow.value + noteRelays)
client.publish(event, outboxRelays.flow.value + noteRelays)
}
cache.justConsumeMyOwnEvent(event)
}
@@ -1489,9 +1489,9 @@ class Account(
val relayList = privateStorageRelayList.flow.value + localRelayList.flow.value
if (relayList.isNotEmpty()) {
client.send(event, relayList + noteRelays)
client.publish(event, relayList + noteRelays)
} else {
client.send(event, outboxRelays.flow.value + noteRelays)
client.publish(event, outboxRelays.flow.value + noteRelays)
}
cache.justConsumeMyOwnEvent(event)
}
@@ -1652,7 +1652,7 @@ class Account(
val note = cache.getOrCreateNote(event.id)
val relayList = computeRelayListToBroadcast(note)
client.send(event, relayList = relayList)
client.publish(event, relayList = relayList)
broadcast.forEach { client.send(it, relayList) }
}
@@ -24,7 +24,7 @@ import androidx.compose.runtime.Stable
import com.vitorpamplona.amethyst.model.LocalCache
import com.vitorpamplona.amethyst.model.LocalCache.notes
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.downloadFirstEvent
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchFirst
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
import com.vitorpamplona.quartz.nip01Core.signers.NostrSigner
@@ -105,7 +105,7 @@ class VanishRequestsState(
try {
val foundEvent =
withContext(Dispatchers.IO) {
client.downloadFirstEvent(
client.fetchFirst(
relay = relay,
filter =
Filter(
@@ -22,7 +22,7 @@ package com.vitorpamplona.amethyst.service.broadcast
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.IRelayClientListener
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.OkMessage
@@ -90,7 +90,7 @@ class BroadcastTracker {
val resultChannel = Channel<RelayResponse>(UNLIMITED)
val subscription =
object : IRelayClientListener {
object : RelayConnectionListener {
override fun onCannotConnect(
relay: IRelayClient,
errorMessage: String,
@@ -143,7 +143,7 @@ class BroadcastTracker {
}
try {
client.subscribe(subscription)
client.addConnectionListener(subscription)
val finalBroadcast =
coroutineScope {
@@ -178,7 +178,7 @@ class BroadcastTracker {
}
// Send after setting up listener
client.send(event, relays)
client.publish(event, relays)
resultCollector.await()
}
@@ -192,7 +192,7 @@ class BroadcastTracker {
Log.d(TAG, "Broadcast $trackingId complete: ${finalBroadcast.successCount}/${finalBroadcast.totalRelays} success")
} finally {
client.unsubscribe(subscription)
client.removeConnectionListener(subscription)
}
}
@@ -266,7 +266,7 @@ class BroadcastTracker {
val resultChannel = Channel<RelayResponse>(UNLIMITED)
val subscription =
object : IRelayClientListener {
object : RelayConnectionListener {
override fun onCannotConnect(
relay: IRelayClient,
errorMessage: String,
@@ -318,7 +318,7 @@ class BroadcastTracker {
}
}
client.subscribe(subscription)
client.addConnectionListener(subscription)
val finalBroadcast =
coroutineScope {
@@ -370,12 +370,12 @@ class BroadcastTracker {
currentBroadcast.copy(status = newStatus)
}
client.send(event, relaysToRetry)
client.publish(event, relaysToRetry)
resultCollector.await()
}
client.unsubscribe(subscription)
client.removeConnectionListener(subscription)
resultChannel.close()
// Update in active broadcasts
@@ -60,14 +60,22 @@ class PipVideoActivity : ComponentActivity() {
}
}
override fun onStop() {
super.onStop()
finishAndRemoveTask()
override fun onPictureInPictureModeChanged(isInPictureInPictureMode: Boolean) {
super.onPictureInPictureModeChanged(isInPictureInPictureMode)
if (!isInPictureInPictureMode) {
// User dismissed PiP (swiped away or expanded).
finishAndRemoveTask()
}
}
override fun finish() {
finishAndRemoveTask()
super.finish()
override fun onStop() {
super.onStop()
if (!isInPictureInPictureMode) {
// Only finish if we're not in PiP mode.
// When the screen locks while in PiP, we stay alive
// so the PlaybackService can continue audio playback.
finishAndRemoveTask()
}
}
companion object {
@@ -32,7 +32,9 @@ import androidx.compose.ui.Alignment
import androidx.compose.ui.Modifier
import androidx.compose.ui.layout.ContentScale
import androidx.compose.ui.platform.LocalContext
import androidx.compose.ui.platform.LocalView
import androidx.core.content.ContextCompat
import androidx.media3.common.Player
import androidx.media3.common.util.UnstableApi
import androidx.media3.ui.compose.ContentFrame
import androidx.media3.ui.compose.state.rememberMuteButtonState
@@ -51,6 +53,8 @@ fun RenderPipVideo(
controller: MediaControllerState,
waveformData: WaveformData?,
) {
KeepScreenOnWhilePlaying(controller)
val modifier =
remember {
val ratio =
@@ -78,6 +82,31 @@ fun RenderPipVideo(
}
}
@Composable
fun KeepScreenOnWhilePlaying(controller: MediaControllerState) {
val view = LocalView.current
DisposableEffect(controller.controller, view) {
val listener =
object : Player.Listener {
override fun onIsPlayingChanged(isPlaying: Boolean) {
if (view.keepScreenOn != isPlaying) {
view.keepScreenOn = isPlaying
}
}
}
// Set initial state
view.keepScreenOn = controller.controller.isPlaying
controller.controller.addListener(listener)
onDispose {
controller.controller.removeListener(listener)
view.keepScreenOn = false
}
}
}
@Composable
fun RegisterControllerReceiver(controllerState: MediaControllerState) {
val context = LocalContext.current
@@ -27,7 +27,7 @@ import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.pool.RelayBasedFilter
import com.vitorpamplona.quartz.nip01Core.relay.client.pool.groupByRelay
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.subscriptions.Subscription
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
@@ -68,7 +68,7 @@ abstract class PerUniqueIdEoseManager<T, U : Any>(
open fun newSub(key: T): Subscription =
requestNewSubscription(
object : IRequestListener {
object : SubscriptionListener {
override fun onEose(
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
@@ -28,7 +28,7 @@ import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.pool.RelayBasedFilter
import com.vitorpamplona.quartz.nip01Core.relay.client.pool.groupByRelay
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.subscriptions.Subscription
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
@@ -69,7 +69,7 @@ abstract class PerUserAndFollowListEoseManager<T, U : Any>(
open fun newSub(key: T): Subscription =
requestNewSubscription(
object : IRequestListener {
object : SubscriptionListener {
override fun onEose(
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
@@ -28,7 +28,7 @@ import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.pool.RelayBasedFilter
import com.vitorpamplona.quartz.nip01Core.relay.client.pool.groupByRelay
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.subscriptions.Subscription
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
@@ -67,7 +67,7 @@ abstract class PerUserEoseManager<T>(
open fun newSub(key: T): Subscription =
requestNewSubscription(
object : IRequestListener {
object : SubscriptionListener {
override fun onEose(
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
@@ -25,7 +25,7 @@ import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.pool.RelayBasedFilter
import com.vitorpamplona.quartz.nip01Core.relay.client.pool.groupByRelay
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener
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
@@ -41,7 +41,7 @@ abstract class SingleSubNoEoseCacheEoseManager<T>(
) : BaseEoseManager<T>(client, allKeys) {
val sub =
requestNewSubscription(
object : IRequestListener {
object : SubscriptionListener {
override fun onEose(
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
@@ -35,7 +35,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.RelayOfflineT
import com.vitorpamplona.quartz.nip01Core.relay.client.auth.IAuthStatus
import com.vitorpamplona.quartz.nip01Core.relay.client.pool.RelayBasedFilter
import com.vitorpamplona.quartz.nip01Core.relay.client.pool.groupByRelay
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.newSubId
import com.vitorpamplona.quartz.nip01Core.relay.client.subscriptions.SubscriptionController
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
@@ -107,7 +107,7 @@ class AccountFollowsLoaderSubAssembler(
val sub =
orchestrator.requestNewSubscription(
if (isDebug) logTag + newSubId() else newSubId(),
object : IRequestListener {
object : SubscriptionListener {
override fun onEose(
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
@@ -31,7 +31,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.RelayOfflineTracker
import com.vitorpamplona.quartz.nip01Core.relay.client.pool.RelayBasedFilter
import com.vitorpamplona.quartz.nip01Core.relay.client.pool.groupByRelay
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener
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.nip65RelayList.AdvertisedRelayListEvent
@@ -68,7 +68,7 @@ class UserOutboxFinderSubAssembler(
val sub =
requestNewSubscription(
object : IRequestListener {
object : SubscriptionListener {
override fun onEose(
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
@@ -29,7 +29,7 @@ import com.vitorpamplona.amethyst.service.relays.EOSEAccountFast
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.pool.groupByRelay
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener
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.utils.TimeUtils
@@ -61,7 +61,7 @@ class UserWatcherSubAssembler(
val sub =
requestNewSubscription(
object : IRequestListener {
object : SubscriptionListener {
override fun onEose(
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
@@ -21,7 +21,7 @@
package com.vitorpamplona.amethyst.service.relayClient.speedLogger
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.IRelayClientListener
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.EventMessage
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message
@@ -29,7 +29,7 @@ import com.vitorpamplona.quartz.utils.Log
import com.vitorpamplona.quartz.utils.bytesUsedInMemory
/**
* Listens to NostrClient's onNotify messages from the relay
* Listens to INostrClient's onNotify messages from the relay
*/
class RelaySpeedLogger(
val client: INostrClient,
@@ -41,7 +41,7 @@ class RelaySpeedLogger(
var current = FrameStat()
private val clientListener =
object : IRelayClientListener {
object : RelayConnectionListener {
override fun onIncomingMessage(
relay: IRelayClient,
msgStr: String,
@@ -55,7 +55,7 @@ class RelaySpeedLogger(
init {
Log.d(TAG, "Init, Subscribe")
client.subscribe(clientListener)
client.addConnectionListener(clientListener)
// OkHttpDebugLogging.enableHttp2()
// OkHttpDebugLogging.enableTaskRunner()
}
@@ -63,6 +63,6 @@ class RelaySpeedLogger(
fun destroy() {
// makes sure to run
Log.d(TAG, "Destroy, Unsubscribe")
client.unsubscribe(clientListener)
client.removeConnectionListener(clientListener)
}
}
@@ -144,6 +144,7 @@ import com.vitorpamplona.amethyst.ui.note.types.RenderPostApproval
import com.vitorpamplona.amethyst.ui.note.types.RenderPrivateMessage
import com.vitorpamplona.amethyst.ui.note.types.RenderPublicMessage
import com.vitorpamplona.amethyst.ui.note.types.RenderReaction
import com.vitorpamplona.amethyst.ui.note.types.RenderRelayDiscovery
import com.vitorpamplona.amethyst.ui.note.types.RenderReport
import com.vitorpamplona.amethyst.ui.note.types.RenderTextEvent
import com.vitorpamplona.amethyst.ui.note.types.RenderTextModificationEvent
@@ -244,6 +245,7 @@ import com.vitorpamplona.quartz.nip64Chess.challenge.offer.LiveChessGameChalleng
import com.vitorpamplona.quartz.nip64Chess.end.LiveChessGameEndEvent
import com.vitorpamplona.quartz.nip64Chess.game.ChessGameEvent
import com.vitorpamplona.quartz.nip65RelayList.AdvertisedRelayListEvent
import com.vitorpamplona.quartz.nip66RelayMonitor.discovery.RelayDiscoveryEvent
import com.vitorpamplona.quartz.nip68Picture.PictureEvent
import com.vitorpamplona.quartz.nip71Video.VideoHorizontalEvent
import com.vitorpamplona.quartz.nip71Video.VideoNormalEvent
@@ -911,6 +913,10 @@ private fun RenderNoteRow(
DisplayBroadcastRelayList(baseNote, backgroundColor, accountViewModel, nav)
}
is RelayDiscoveryEvent -> {
RenderRelayDiscovery(baseNote, accountViewModel, nav)
}
is PinListEvent -> {
RenderPinListEvent(baseNote, backgroundColor, accountViewModel, nav)
}
@@ -0,0 +1,297 @@
/*
* 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.amethyst.ui.note.types
import androidx.compose.foundation.layout.Arrangement
import androidx.compose.foundation.layout.Column
import androidx.compose.foundation.layout.ExperimentalLayoutApi
import androidx.compose.foundation.layout.FlowRow
import androidx.compose.foundation.layout.Row
import androidx.compose.foundation.layout.Spacer
import androidx.compose.foundation.layout.fillMaxWidth
import androidx.compose.foundation.layout.padding
import androidx.compose.foundation.layout.size
import androidx.compose.foundation.layout.width
import androidx.compose.foundation.shape.RoundedCornerShape
import androidx.compose.material.icons.Icons
import androidx.compose.material.icons.filled.Dns
import androidx.compose.material.icons.filled.Language
import androidx.compose.material.icons.filled.Lock
import androidx.compose.material.icons.filled.LockOpen
import androidx.compose.material.icons.filled.Numbers
import androidx.compose.material.icons.filled.Tag
import androidx.compose.material.icons.filled.TravelExplore
import androidx.compose.material3.Icon
import androidx.compose.material3.MaterialTheme
import androidx.compose.material3.Surface
import androidx.compose.material3.Text
import androidx.compose.runtime.Composable
import androidx.compose.runtime.remember
import androidx.compose.ui.Alignment
import androidx.compose.ui.Modifier
import androidx.compose.ui.graphics.Color
import androidx.compose.ui.graphics.vector.ImageVector
import androidx.compose.ui.platform.LocalContext
import androidx.compose.ui.text.font.FontWeight
import androidx.compose.ui.text.style.TextAlign
import androidx.compose.ui.text.style.TextOverflow
import androidx.compose.ui.unit.dp
import com.vitorpamplona.amethyst.R
import com.vitorpamplona.amethyst.model.Note
import com.vitorpamplona.amethyst.ui.navigation.navs.INav
import com.vitorpamplona.amethyst.ui.screen.loggedIn.AccountViewModel
import com.vitorpamplona.amethyst.ui.stringRes
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.displayUrl
import com.vitorpamplona.quartz.nip66RelayMonitor.discovery.RelayDiscoveryEvent
@OptIn(ExperimentalLayoutApi::class)
@Composable
fun RenderRelayDiscovery(
baseNote: Note,
accountViewModel: AccountViewModel,
nav: INav,
) {
val noteEvent = baseNote.event as? RelayDiscoveryEvent ?: return
val context = LocalContext.current
val relayUrl = remember(noteEvent) { noteEvent.relay() }
val rttOpen = remember(noteEvent) { noteEvent.rttOpen() }
val rttRead = remember(noteEvent) { noteEvent.rttRead() }
val rttWrite = remember(noteEvent) { noteEvent.rttWrite() }
val networkTypes = remember(noteEvent) { noteEvent.networkTypes() }
val relayTypes = remember(noteEvent) { noteEvent.relayTypes() }
val supportedNips = remember(noteEvent) { noteEvent.supportedNips() }
val requirements = remember(noteEvent) { noteEvent.requirements() }
val acceptedKinds = remember(noteEvent) { noteEvent.acceptedKinds() }
val topics = remember(noteEvent) { noteEvent.topics() }
val geohashes = remember(noteEvent) { noteEvent.geohashes() }
Column(
modifier = Modifier.fillMaxWidth(),
verticalArrangement = Arrangement.spacedBy(8.dp),
) {
// Relay URL header
if (relayUrl != null) {
Text(
text = relayUrl.displayUrl(),
style = MaterialTheme.typography.titleMedium,
fontWeight = FontWeight.Bold,
color = MaterialTheme.colorScheme.primary,
maxLines = 1,
overflow = TextOverflow.Ellipsis,
)
}
// RTT metrics
if (rttOpen != null || rttRead != null || rttWrite != null) {
Row(
modifier = Modifier.fillMaxWidth(),
horizontalArrangement = Arrangement.SpaceEvenly,
) {
rttOpen?.let {
RttMetricChip(stringRes(R.string.relay_monitor_rtt_open), it)
}
rttRead?.let {
RttMetricChip(stringRes(R.string.relay_monitor_rtt_read), it)
}
rttWrite?.let {
RttMetricChip(stringRes(R.string.relay_monitor_rtt_write), it)
}
}
}
// Network type
if (networkTypes.isNotEmpty()) {
DiscoveryInfoRow(
icon = Icons.Default.Language,
label = stringRes(R.string.relay_monitor_network),
value = networkTypes.joinToString { it.code },
)
}
// Relay type
if (relayTypes.isNotEmpty()) {
DiscoveryInfoRow(
icon = Icons.Default.Dns,
label = stringRes(R.string.relay_monitor_relay_type),
value = relayTypes.joinToString(),
)
}
// Requirements
if (requirements.isNotEmpty()) {
DiscoveryInfoRow(
icon = if (requirements.any { !it.negated }) Icons.Default.Lock else Icons.Default.LockOpen,
label = stringRes(R.string.relay_monitor_requirements),
value =
requirements.joinToString { req ->
if (req.negated) "!${req.value}" else req.value
},
)
}
// Supported NIPs
if (supportedNips.isNotEmpty()) {
DiscoveryInfoRow(
icon = Icons.Default.Numbers,
label = stringRes(R.string.relay_monitor_supported_nips),
value = supportedNips.sorted().joinToString(),
)
}
// Accepted kinds
if (acceptedKinds.isNotEmpty()) {
DiscoveryInfoRow(
icon = Icons.Default.Dns,
label = stringRes(R.string.relay_discovery_accepted_kinds),
value =
acceptedKinds.joinToString { kind ->
if (kind.negated) "!${kind.kind}" else "${kind.kind}"
},
)
}
// Topics
if (topics.isNotEmpty()) {
Row(
modifier = Modifier.padding(vertical = 2.dp),
verticalAlignment = Alignment.CenterVertically,
) {
Icon(
Icons.Default.Tag,
contentDescription = null,
modifier = Modifier.size(18.dp),
tint = MaterialTheme.colorScheme.secondary,
)
Spacer(Modifier.width(12.dp))
FlowRow(
horizontalArrangement = Arrangement.spacedBy(6.dp),
verticalArrangement = Arrangement.spacedBy(4.dp),
) {
topics.forEach { topic ->
TopicChip(topic)
}
}
}
}
// Geohashes
if (geohashes.isNotEmpty()) {
DiscoveryInfoRow(
icon = Icons.Default.TravelExplore,
label = stringRes(R.string.relay_discovery_geohash),
value = geohashes.joinToString(),
)
}
// Content description
if (noteEvent.content.isNotBlank()) {
Text(
text = noteEvent.content,
style = MaterialTheme.typography.bodyMedium,
color = MaterialTheme.colorScheme.onSurfaceVariant,
maxLines = 5,
overflow = TextOverflow.Ellipsis,
)
}
}
}
@Composable
private fun RttMetricChip(
label: String,
ms: Long,
) {
val color =
when {
ms < 200 -> Color(0xFF4CAF50)
ms < 500 -> Color(0xFFFFC107)
else -> MaterialTheme.colorScheme.error
}
Column(horizontalAlignment = Alignment.CenterHorizontally) {
Text(
text = label,
style = MaterialTheme.typography.labelSmall,
color = MaterialTheme.colorScheme.onSurfaceVariant,
)
Surface(
shape = RoundedCornerShape(50),
color = color.copy(alpha = 0.15f),
) {
Text(
text = stringRes(R.string.relay_monitor_ms, ms.toInt()),
modifier = Modifier.padding(horizontal = 10.dp, vertical = 4.dp),
style = MaterialTheme.typography.labelMedium,
fontWeight = FontWeight.Bold,
color = color,
)
}
}
}
@Composable
private fun TopicChip(topic: String) {
Surface(
shape = RoundedCornerShape(50),
color = MaterialTheme.colorScheme.secondaryContainer,
) {
Text(
text = "#$topic",
modifier = Modifier.padding(horizontal = 10.dp, vertical = 4.dp),
style = MaterialTheme.typography.labelMedium,
color = MaterialTheme.colorScheme.onSecondaryContainer,
)
}
}
@Composable
private fun DiscoveryInfoRow(
icon: ImageVector,
label: String,
value: String,
) {
Row(
modifier = Modifier.padding(vertical = 2.dp),
verticalAlignment = Alignment.CenterVertically,
) {
Icon(
icon,
contentDescription = null,
modifier = Modifier.size(18.dp),
tint = MaterialTheme.colorScheme.secondary,
)
Spacer(Modifier.width(12.dp))
Text(
text = label,
style = MaterialTheme.typography.labelLarge,
maxLines = 1,
)
Text(
text = value,
textAlign = TextAlign.End,
style = MaterialTheme.typography.bodyMedium,
fontWeight = FontWeight.SemiBold,
modifier = Modifier.weight(1f),
overflow = TextOverflow.Ellipsis,
maxLines = 1,
)
}
}
@@ -98,7 +98,6 @@ import com.vitorpamplona.quartz.nip01Core.core.toHexKey
import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
import com.vitorpamplona.quartz.nip01Core.hints.EventHintBundle
import com.vitorpamplona.quartz.nip01Core.relay.client.EmptyNostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.RelayOfflineTracker
import com.vitorpamplona.quartz.nip01Core.relay.client.auth.EmptyIAuthStatus
import com.vitorpamplona.quartz.nip01Core.relay.client.auth.RelayAuthenticator
@@ -26,7 +26,7 @@ import com.vitorpamplona.amethyst.Amethyst
import com.vitorpamplona.amethyst.model.Account
import com.vitorpamplona.amethyst.service.replace
import com.vitorpamplona.amethyst.ui.screen.loggedIn.AccountViewModel
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.queryCountSuspend
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.count
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.flow.MutableStateFlow
@@ -99,7 +99,7 @@ abstract class BasicRelaySetupInfoModel : ViewModel() {
filters.forEach { countFilter ->
viewModelScope.launch(Dispatchers.IO) {
val result = client.queryCountSuspend(item.relay, countFilter.filter)
val result = client.count(item.relay, countFilter.filter)
if (result != null) {
_countResults.update { currentMap ->
val current = currentMap[item.relay] ?: RelayCountResult()
@@ -28,8 +28,8 @@ import com.vitorpamplona.amethyst.ui.screen.loggedIn.relays.eventsync.EventSync.
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.client.accessories.reqBypassingRelayLimits
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.IRelayClientListener
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchAllPages
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.OkMessage
@@ -65,7 +65,7 @@ import kotlin.coroutines.cancellation.CancellationException
* Each relay is paginated individually: after EOSE the oldest [Event.createdAt] seen on that
* relay becomes the next `until` cursor, repeating until the relay returns no new events.
*
* OK (true) responses from destination relays are tracked via [IRelayClientListener] and
* OK (true) responses from destination relays are tracked via [RelayConnectionListener] and
* attributed back to the source relay that contributed each event.
*
* Live activity is emitted via [liveActivity] so the UI can show a per-relay log of events
@@ -406,7 +406,7 @@ class EventSync(
)
val okListener =
object : IRelayClientListener {
object : RelayConnectionListener {
override fun onCannotConnect(
relay: IRelayClient,
errorMessage: String,
@@ -500,7 +500,7 @@ class EventSync(
_syncState.emit(runningState)
clientBuilder().use { client ->
client.subscribe(okListener)
client.addConnectionListener(okListener)
try {
client.downloadFromPool(
relays = relaysToProcess,
@@ -523,21 +523,21 @@ class EventSync(
// Each routing rule is independent: an event can match more than one.
if (isMyEvent && outboxTargets.isNotEmpty()) {
if (outboxDedup.add(event.id)) {
client.send(event, outboxTargets)
client.publish(event, outboxTargets)
newEvent = true
}
matchesAtLeastOneFilter = true
}
if (mentionsMe && isDmKind && dmTargets.isNotEmpty()) {
if (dmDedup.add(event.id)) {
client.send(event, dmTargets)
client.publish(event, dmTargets)
newEvent = true
}
matchesAtLeastOneFilter = true
}
if (mentionsMe && !isDmKind && inboxTargets.isNotEmpty()) {
if (inboxDedup.add(event.id)) {
client.send(event, inboxTargets)
client.publish(event, inboxTargets)
newEvent = true
}
matchesAtLeastOneFilter = true
@@ -610,7 +610,7 @@ class EventSync(
if (e is CancellationException) throw e
_syncState.value = SyncState.Error(e.message ?: "Unknown error", filterSince, filterUntil)
} finally {
client.unsubscribe(okListener)
client.removeConnectionListener(okListener)
}
}
}
@@ -666,5 +666,5 @@ class EventSync(
filters: List<Filter>,
onNewPage: (Long) -> Unit,
onEvent: (Event) -> Unit,
): Int = reqBypassingRelayLimits(relay, filters, RELAY_TIMEOUT_MS, onNewPage, onEvent)
): Int = fetchAllPages(relay, filters, RELAY_TIMEOUT_MS, onNewPage, onEvent)
}
@@ -31,7 +31,7 @@ import com.vitorpamplona.amethyst.ui.screen.loggedIn.AccountViewModel
import com.vitorpamplona.amethyst.ui.screen.loggedIn.relays.common.BasicRelaySetupInfo
import com.vitorpamplona.amethyst.ui.screen.loggedIn.relays.common.RelayCountResult
import com.vitorpamplona.amethyst.ui.screen.loggedIn.relays.common.relaySetupInfoBuilder
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.queryCountSuspend
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.count
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
import com.vitorpamplona.quartz.nip65RelayList.tags.AdvertisedRelayInfo
@@ -131,7 +131,7 @@ class Nip65RelayListViewModel : ViewModel() {
_homeRelays.value.forEach { item ->
viewModelScope.launch(Dispatchers.IO) {
val result = client.queryCountSuspend(item.relay, Filter(authors = listOf(account.pubKey)))
val result = client.count(item.relay, Filter(authors = listOf(account.pubKey)))
if (result != null) {
val countResult =
RelayCountResult(
@@ -150,7 +150,7 @@ class Nip65RelayListViewModel : ViewModel() {
_notificationRelays.value.forEach { item ->
viewModelScope.launch(Dispatchers.IO) {
val result = client.queryCountSuspend(item.relay, Filter(tags = mapOf("p" to listOf(account.pubKey))))
val result = client.count(item.relay, Filter(tags = mapOf("p" to listOf(account.pubKey))))
if (result != null) {
val countResult =
RelayCountResult(
@@ -53,22 +53,28 @@ import androidx.compose.material3.TextButton
import androidx.compose.runtime.Composable
import androidx.compose.runtime.getValue
import androidx.compose.runtime.mutableStateOf
import androidx.compose.runtime.produceState
import androidx.compose.runtime.remember
import androidx.compose.runtime.rememberCoroutineScope
import androidx.compose.runtime.setValue
import androidx.compose.ui.Alignment
import androidx.compose.ui.Modifier
import androidx.compose.ui.draw.clip
import androidx.compose.ui.focus.onFocusChanged
import androidx.compose.ui.graphics.Color
import androidx.compose.ui.layout.ContentScale
import androidx.compose.ui.platform.LocalUriHandler
import androidx.compose.ui.res.stringResource
import androidx.compose.ui.text.font.FontWeight
import androidx.compose.ui.text.style.TextOverflow
import androidx.compose.ui.unit.dp
import androidx.lifecycle.compose.collectAsStateWithLifecycle
import coil3.compose.AsyncImage
import com.vitorpamplona.amethyst.R
import com.vitorpamplona.amethyst.commons.ui.feeds.FeedContentState
import com.vitorpamplona.amethyst.commons.ui.feeds.FeedState
import com.vitorpamplona.amethyst.model.Note
import com.vitorpamplona.amethyst.model.UrlCachedPreviewer
import com.vitorpamplona.amethyst.ui.components.UrlPreviewState
import com.vitorpamplona.amethyst.ui.feeds.RefresheableBox
import com.vitorpamplona.amethyst.ui.feeds.RenderFeedContentState
@@ -83,6 +89,7 @@ import com.vitorpamplona.amethyst.ui.screen.loggedIn.AccountViewModel
import com.vitorpamplona.amethyst.ui.stringRes
import com.vitorpamplona.amethyst.ui.theme.DividerThickness
import com.vitorpamplona.amethyst.ui.theme.FeedPadding
import com.vitorpamplona.amethyst.ui.theme.QuoteBorder
import com.vitorpamplona.amethyst.ui.theme.Size55Modifier
import com.vitorpamplona.quartz.nipB0WebBookmarks.WebBookmarkEvent
import kotlinx.coroutines.launch
@@ -238,6 +245,17 @@ private fun WebBookmarkCard(
)
}
@Suppress("ProduceStateDoesNotAssignValue")
val urlPreviewState by
produceState(
initialValue = UrlCachedPreviewer.cache.get(event.url()) ?: UrlPreviewState.Loading,
key1 = event.url(),
) {
if (value == UrlPreviewState.Loading) {
accountViewModel.urlPreview(event.url()) { value = it }
}
}
Column(
modifier =
Modifier
@@ -245,6 +263,23 @@ private fun WebBookmarkCard(
.clickable { uriHandler.openUri(event.url()) }
.padding(16.dp),
) {
val previewInfo = (urlPreviewState as? UrlPreviewState.Loaded)?.previewInfo
if (previewInfo?.imageUrlFullPath != null) {
AsyncImage(
model = previewInfo.imageUrlFullPath,
contentDescription = event.title() ?: previewInfo.title,
contentScale = ContentScale.Crop,
modifier =
Modifier
.fillMaxWidth()
.height(180.dp)
.clip(QuoteBorder),
)
Spacer(modifier = Modifier.height(8.dp))
}
Row(
modifier = Modifier.fillMaxWidth(),
horizontalArrangement = Arrangement.SpaceBetween,
@@ -252,7 +287,7 @@ private fun WebBookmarkCard(
) {
Column(modifier = Modifier.weight(1f)) {
Text(
text = event.title() ?: event.url(),
text = event.title() ?: previewInfo?.title?.ifBlank { null } ?: event.url(),
style = MaterialTheme.typography.titleMedium,
fontWeight = FontWeight.Bold,
maxLines = 1,
@@ -260,9 +295,9 @@ private fun WebBookmarkCard(
)
Text(
text = event.url(),
text = previewInfo?.verifiedUrl?.host ?: event.url(),
style = MaterialTheme.typography.bodySmall,
color = MaterialTheme.colorScheme.primary,
color = Color.Gray,
maxLines = 1,
overflow = TextOverflow.Ellipsis,
)
@@ -291,10 +326,11 @@ private fun WebBookmarkCard(
}
}
if (event.description().isNotBlank()) {
val description = event.description().ifBlank { previewInfo?.description ?: "" }
if (description.isNotBlank()) {
Spacer(modifier = Modifier.height(4.dp))
Text(
text = event.description(),
text = description,
style = MaterialTheme.typography.bodyMedium,
maxLines = 3,
overflow = TextOverflow.Ellipsis,
+2
View File
@@ -839,6 +839,8 @@
<string name="relay_monitor_requirements">Requirements</string>
<string name="relay_monitor_last_check">Last check</string>
<string name="relay_monitor_ms">%1$d ms</string>
<string name="relay_discovery_accepted_kinds">Accepted Kinds</string>
<string name="relay_discovery_geohash">Location</string>
<string name="relay_active_subscriptions">Active Subscriptions</string>
<string name="relay_active_outbox">Pending Outbox Events</string>
<string name="relay_req_subscriptions">REQ Subscriptions (%1$d)</string>
@@ -22,7 +22,7 @@ package com.vitorpamplona.amethyst.commons.chess
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.newSubId
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
@@ -49,7 +49,7 @@ enum class RelayFetchStatus {
/**
* One-shot relay fetch helper for chess events.
*
* Follows the existing INostrClient + IRequestListener + Channel pattern
* Follows the existing INostrClient + SubscriptionListener + Channel pattern
* from quartz (see NostrClientSingleDownloadExt.kt).
*
* Each fetch opens a subscription, collects events until EOSE from all relays,
@@ -62,7 +62,7 @@ class ChessRelayFetchHelper(
/**
* Fetch events matching filters from relays, waiting for EOSE.
*
* @param filters Map of relay filter list (same format as INostrClient.openReqSubscription)
* @param filters Map of relay filter list (same format as INostrClient.subscribe)
* @param timeoutMs Max time to wait for relays to respond (default from ChessConfig)
* @param onProgress Optional callback for progress updates per relay
* @return Deduplicated list of events received before timeout/EOSE
@@ -88,7 +88,7 @@ class ChessRelayFetchHelper(
}
val listener =
object : IRequestListener {
object : SubscriptionListener {
override fun onEvent(
event: Event,
isLive: Boolean,
@@ -114,7 +114,7 @@ class ChessRelayFetchHelper(
}
}
client.openReqSubscription(subId, filters, listener)
client.subscribe(subId, filters, listener)
val eoseResult = withTimeoutOrNull(timeoutMs) { allEose.await() }
// Mark timed-out relays
@@ -127,7 +127,7 @@ class ChessRelayFetchHelper(
}
}
client.close(subId)
client.unsubscribe(subId)
return events.values.toList()
}
@@ -25,7 +25,7 @@ import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper
import com.vitorpamplona.quartz.nip01Core.core.toHexKey
import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.req
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.StaticSubscription
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerInternal
@@ -122,14 +122,17 @@ object NostrConnectLoginUseCase {
val deferred = CompletableDeferred<NostrConnectEvent>()
val subscription =
client.req(
relays = relays.toList(),
filter =
Filter(
kinds = listOf(NostrConnectEvent.KIND),
tags = mapOf("p" to listOf(ephemeralPubKey)),
since = TimeUtils.now() - 60,
),
StaticSubscription(
client,
relays.associateWith {
listOf(
Filter(
kinds = listOf(NostrConnectEvent.KIND),
tags = mapOf("p" to listOf(ephemeralPubKey)),
since = TimeUtils.now() - 60,
),
)
},
) { event ->
if (event is NostrConnectEvent && !deferred.isCompleted) {
deferred.complete(event)
@@ -207,7 +210,7 @@ object NostrConnectLoginUseCase {
remoteKey = connectData.signerPubkey,
signer = ephemeralSigner,
)
client.send(ackEvent, relays)
client.publish(ackEvent, relays)
}
private fun generateSecret(): String {
@@ -28,7 +28,7 @@ import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.core.HexKey
import com.vitorpamplona.quartz.nip01Core.metadata.MetadataEvent
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.newSubId
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
@@ -79,7 +79,7 @@ class FeedMetadataCoordinator(
// Create listener to pass events to the callback
val listener =
if (onEvent != null) {
object : IRequestListener {
object : SubscriptionListener {
override fun onEvent(
event: Event,
isLive: Boolean,
@@ -93,7 +93,7 @@ class FeedMetadataCoordinator(
null
}
client.openReqSubscription(
client.subscribe(
subId = newSubId(),
filters = filterMap,
listener = listener,
@@ -22,7 +22,7 @@ package com.vitorpamplona.amethyst.commons.relayClient.eoseManagers
import com.vitorpamplona.amethyst.commons.service.BundledUpdate
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.newSubId
import com.vitorpamplona.quartz.nip01Core.relay.client.subscriptions.SubscriptionController
import kotlinx.coroutines.Dispatchers
@@ -38,7 +38,7 @@ abstract class BaseEoseManager<T>(
fun getSubscription(subId: String) = orchestrator.getSub(subId)
fun requestNewSubscription(listener: IRequestListener) = orchestrator.requestNewSubscription(newSubId(), listener)
fun requestNewSubscription(listener: SubscriptionListener) = orchestrator.requestNewSubscription(newSubId(), listener)
fun dismissSubscription(subId: String) = orchestrator.dismissSubscription(subId)
@@ -26,7 +26,7 @@ import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.pool.RelayBasedFilter
import com.vitorpamplona.quartz.nip01Core.relay.client.pool.groupByRelay
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.subscriptions.Subscription
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
@@ -82,7 +82,7 @@ abstract class PerKeyEoseManager<T, K : Any>(
*/
open fun newSub(queryState: T): Subscription =
requestNewSubscription(
object : IRequestListener {
object : SubscriptionListener {
override fun onEose(
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
@@ -26,7 +26,7 @@ import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.pool.RelayBasedFilter
import com.vitorpamplona.quartz.nip01Core.relay.client.pool.groupByRelay
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener
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.utils.TimeUtils
@@ -66,7 +66,7 @@ abstract class SingleSubEoseManager<T>(
val sub =
requestNewSubscription(
object : IRequestListener {
object : SubscriptionListener {
override fun onEose(
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
@@ -22,8 +22,8 @@ package com.vitorpamplona.amethyst.commons.chess
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.sendAndWaitForResponse
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.publishAndConfirm
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.newSubId
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
@@ -42,7 +42,7 @@ data class BroadcastResult(
/**
* Helper for broadcasting chess events to relays with reliable delivery.
*
* Uses sendAndWaitForResponse to get actual OK confirmations from relays,
* Uses publishAndConfirm to get actual OK confirmations from relays,
* ensuring the event was actually received and accepted.
*/
class ChessEventBroadcaster(
@@ -84,7 +84,7 @@ class ChessEventBroadcaster(
)
val listener =
object : IRequestListener {
object : SubscriptionListener {
override fun onEvent(
event: Event,
isLive: Boolean,
@@ -100,19 +100,19 @@ class ChessEventBroadcaster(
// Open subscription to all target relays (triggers connection)
val filterMap = targetRelays.associateWith { listOf(dummyFilter) }
client.openReqSubscription(subId, filterMap, listener)
client.subscribe(subId, filterMap, listener)
// Wait for relays to connect (poll with timeout)
waitForRelays(targetRelays, 5000L)
// Close the dummy subscription
client.close(subId)
client.unsubscribe(subId)
}
// Step 3: Send the event and wait for OK responses
val success = client.sendAndWaitForResponse(event, targetRelays, timeoutSeconds)
val success = client.publishAndConfirm(event, targetRelays, timeoutSeconds)
// Note: sendAndWaitForResponse only returns aggregate success (any relay accepted)
// Note: publishAndConfirm only returns aggregate success (any relay accepted)
// We don't have per-relay results, so relayResults is empty
return BroadcastResult(
success = success,
@@ -193,7 +193,7 @@ fun main() {
}
Window(
onCloseRequest = ::exitApplication,
onSubscriptionCloseduest = ::exitApplication,
state = windowState,
title = "Amethyst",
) {
@@ -30,8 +30,7 @@ import com.vitorpamplona.amethyst.desktop.network.DesktopHttpClient
import com.vitorpamplona.quartz.nip01Core.core.hexToByteArray
import com.vitorpamplona.quartz.nip01Core.core.toHexKey
import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.IRelayClientListener
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message
import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.Command
@@ -134,7 +133,7 @@ class AccountManager internal constructor(
private val nip46ClientMutex = Mutex()
private var nip46Client: NostrClient? = null
private suspend fun getOrCreateNip46Client(): NostrClient =
private suspend fun getOrCreateNip46Client(): INostrClient =
nip46ClientMutex.withLock {
nip46Client ?: NostrClient(
BasicOkHttpWebSocket.Builder(DesktopHttpClient::getHttpClient),
@@ -156,7 +155,7 @@ class AccountManager internal constructor(
* but we must wait for the websocket to be ready before sending requests.
*/
private suspend fun awaitNip46RelayConnection(
client: NostrClient,
client: INostrClient,
targetRelays: Set<NormalizedRelayUrl>,
) {
withTimeout(NIP46_RELAY_CONNECT_TIMEOUT_MS) {
@@ -180,8 +179,8 @@ class AccountManager internal constructor(
}
}
private fun createLoginRelayListener(): IRelayClientListener =
object : IRelayClientListener {
private fun createLoginRelayListener(): RelayConnectionListener =
object : RelayConnectionListener {
override fun onConnected(
relay: IRelayClient,
pingMillis: Int,
@@ -316,7 +315,7 @@ class AccountManager internal constructor(
LoginProgress.ConnectingToRelays(
relaysFromUri.associateWith { RelayLoginStatus.CONNECTING },
)
nip46Client.subscribe(listener)
nip46Client.addConnectionListener(listener)
_loginProgress.value =
LoginProgress.WaitingForSigner(
@@ -356,7 +355,7 @@ class AccountManager internal constructor(
return Result.failure(Exception("Connection failed: ${e.message}"))
} finally {
_loginProgress.value = null
client?.unsubscribe(listener)
client?.removeConnectionListener(listener)
}
}
@@ -376,7 +375,7 @@ class AccountManager internal constructor(
LoginProgress.ConnectingToRelays(
uriData.relays.associateWith { RelayLoginStatus.CONNECTING },
)
nip46Client.subscribe(listener)
nip46Client.addConnectionListener(listener)
onUriGenerated(uriData.uri)
@@ -415,7 +414,7 @@ class AccountManager internal constructor(
return Result.failure(Exception("Connection failed: ${e.message}"))
} finally {
_loginProgress.value = null
client?.unsubscribe(listener)
client?.removeConnectionListener(listener)
}
}
@@ -21,9 +21,9 @@
package com.vitorpamplona.amethyst.desktop.network
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.IRelayClientListener
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message
import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.Command
@@ -43,7 +43,7 @@ import kotlinx.coroutines.flow.asStateFlow
*/
open class RelayConnectionManager(
websocketBuilder: WebsocketBuilder,
) : IRelayClientListener {
) : RelayConnectionListener {
private val _client = NostrClient(websocketBuilder)
/** Exposes the underlying INostrClient for subscription coordinators */
@@ -56,7 +56,7 @@ open class RelayConnectionManager(
val availableRelays: StateFlow<Set<NormalizedRelayUrl>> = _client.availableRelaysFlow()
init {
_client.subscribe(this)
_client.addConnectionListener(this)
}
fun connect() {
@@ -85,21 +85,21 @@ open class RelayConnectionManager(
subId: String,
filters: List<Filter>,
relays: Set<NormalizedRelayUrl> = availableRelays.value,
listener: IRequestListener? = null,
listener: SubscriptionListener? = null,
) {
val filterMap = relays.associateWith { filters }
_client.openReqSubscription(subId, filterMap, listener)
_client.subscribe(subId, filterMap, listener)
}
fun unsubscribe(subId: String) {
_client.close(subId)
_client.unsubscribe(subId)
}
fun send(
fun publish(
event: Event,
relays: Set<NormalizedRelayUrl> = connectedRelays.value,
) {
_client.send(event, relays)
_client.publish(event, relays)
}
/**
@@ -107,21 +107,21 @@ open class RelayConnectionManager(
*/
fun broadcastToAll(event: Event) {
val connected = connectedRelays.value
send(event, connected)
publish(event, connected)
}
/**
* Sends an event to a specific relay (for NWC).
* Adds the relay if not already in the list.
*/
fun sendToRelay(
fun publishToRelay(
relay: NormalizedRelayUrl,
event: Event,
) {
if (relay !in availableRelays.value) {
updateRelayStatus(relay) { it.copy(connected = false, error = null) }
}
_client.send(event, setOf(relay))
_client.publish(event, setOf(relay))
}
/**
@@ -138,11 +138,11 @@ open class RelayConnectionManager(
updateRelayStatus(relay) { it.copy(connected = false, error = null) }
}
val filterMap = mapOf(relay to filters)
_client.openReqSubscription(
_client.subscribe(
subId = subId,
filters = filterMap,
listener =
object : IRequestListener {
object : SubscriptionListener {
override fun onEvent(
event: Event,
isLive: Boolean,
@@ -162,7 +162,7 @@ open class RelayConnectionManager(
relay: NormalizedRelayUrl,
subId: String,
) {
_client.close(subId)
_client.unsubscribe(subId)
}
private fun updateRelayStatus(
@@ -103,7 +103,7 @@ class NwcPaymentHandler(
zappedNote?.addZapPayment(requestNote, null)
// Send request to wallet's relay
relayManager.sendToRelay(nwcConnection.relayUri, requestEvent)
relayManager.publishToRelay(nwcConnection.relayUri, requestEvent)
// Subscribe and wait for response with timeout
return withTimeoutOrNull(timeoutMs) {
@@ -27,7 +27,7 @@ import com.vitorpamplona.amethyst.desktop.chess.DesktopChessEventCache
import com.vitorpamplona.amethyst.desktop.network.RelayConnectionManager
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.relay.client.pool.groupByRelay
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener
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.nip64Chess.challenge.accept.LiveChessGameAcceptEvent
@@ -95,7 +95,7 @@ class DesktopChessSubscriptionController(
filters = allFilters,
relays = state.relays,
listener =
object : IRequestListener {
object : SubscriptionListener {
override fun onEvent(
event: Event,
isLive: Boolean,
@@ -31,7 +31,7 @@ import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.core.HexKey
import com.vitorpamplona.quartz.nip01Core.metadata.MetadataEvent
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener
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.CoroutineScope
@@ -153,7 +153,7 @@ class DesktopRelaySubscriptionsCoordinator(
// Cancel any existing subscription with this ID
screenSubscriptions.remove(subId)?.cancel()
client.close(subId)
client.unsubscribe(subId)
if (noteIds.isEmpty() || relays.isEmpty()) return subId
@@ -177,7 +177,7 @@ class DesktopRelaySubscriptionsCoordinator(
)
val listener =
object : IRequestListener {
object : SubscriptionListener {
override fun onEvent(
event: Event,
isLive: Boolean,
@@ -190,7 +190,7 @@ class DesktopRelaySubscriptionsCoordinator(
val job =
scope.launch {
client.openReqSubscription(
client.subscribe(
subId = subId,
filters = relays.associateWith { filters },
listener = listener,
@@ -206,7 +206,7 @@ class DesktopRelaySubscriptionsCoordinator(
*/
fun releaseInteractions(subId: String) {
screenSubscriptions.remove(subId)?.cancel()
client.close(subId)
client.unsubscribe(subId)
}
/**
@@ -217,7 +217,7 @@ class DesktopRelaySubscriptionsCoordinator(
// Start rate limiter to process queued metadata requests
rateLimiter.start { pubkey ->
// When rate limiter dequeues a pubkey, subscribe to its metadata
client.openReqSubscription(
client.subscribe(
filters =
indexRelays.associateWith {
listOf(
@@ -281,7 +281,7 @@ class DesktopRelaySubscriptionsCoordinator(
if (inboxRelays.isEmpty() && outboxRelays.isEmpty()) return
val listener =
object : IRequestListener {
object : SubscriptionListener {
override fun onEvent(
event: Event,
isLive: Boolean,
@@ -296,7 +296,7 @@ class DesktopRelaySubscriptionsCoordinator(
if (inboxRelays.isNotEmpty()) {
val inboxSubId = generateSubId("dm-inbox-${userPubKeyHex.take(8)}")
activeDmSubIds.add(inboxSubId)
client.openReqSubscription(
client.subscribe(
subId = inboxSubId,
filters =
inboxRelays.associateWith {
@@ -310,7 +310,7 @@ class DesktopRelaySubscriptionsCoordinator(
if (outboxRelays.isNotEmpty()) {
val outboxSubId = generateSubId("dm-outbox-${userPubKeyHex.take(8)}")
activeDmSubIds.add(outboxSubId)
client.openReqSubscription(
client.subscribe(
subId = outboxSubId,
filters =
outboxRelays.associateWith {
@@ -324,7 +324,7 @@ class DesktopRelaySubscriptionsCoordinator(
if (inboxRelays.isNotEmpty()) {
val giftWrapSubId = generateSubId("giftwrap-${userPubKeyHex.take(8)}")
activeDmSubIds.add(giftWrapSubId)
client.openReqSubscription(
client.subscribe(
subId = giftWrapSubId,
filters =
inboxRelays.associateWith {
@@ -340,7 +340,7 @@ class DesktopRelaySubscriptionsCoordinator(
*/
fun unsubscribeFromDms() {
activeDmSubIds.forEach { subId ->
client.close(subId)
client.unsubscribe(subId)
}
activeDmSubIds.clear()
}
@@ -353,7 +353,7 @@ class DesktopRelaySubscriptionsCoordinator(
// Clean up screen-triggered subscriptions
screenSubscriptions.forEach { (subId, job) ->
job.cancel()
client.close(subId)
client.unsubscribe(subId)
}
screenSubscriptions.clear()
_lastEventAt.value = null
@@ -25,7 +25,7 @@ import androidx.compose.runtime.DisposableEffect
import androidx.compose.runtime.remember
import com.vitorpamplona.amethyst.desktop.network.RelayConnectionManager
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener
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
@@ -80,7 +80,7 @@ fun rememberSubscription(
filters = cfg.filters,
relays = cfg.relays,
listener =
object : IRequestListener {
object : SubscriptionListener {
override fun onEvent(
event: Event,
isLive: Boolean,
@@ -175,7 +175,7 @@ fun ArticleEditorScreen(
)
// TODO: send() is fire-and-forget; markPublished runs before relay ack.
// Consider waiting for relay OK response before marking as published.
relayManager.send(event)
relayManager.publish(event)
draftStore.markPublished(slug)
onPublished()
} catch (e: Exception) {
@@ -170,7 +170,7 @@ fun FeedScreen(
val connectedRelays by relayManager.connectedRelays.collectAsState()
val followedUsers by localCache.followedUsers.collectAsState()
// Available relay URLs — openReqSubscription triggers connection on-demand
// Available relay URLs — subscribe triggers connection on-demand
val allRelayUrls = remember(relayStatuses) { relayStatuses.keys }
var replyToEvent by remember { mutableStateOf<Event?>(null) }
@@ -72,7 +72,7 @@ import com.vitorpamplona.amethyst.desktop.nwc.NwcPaymentHandler
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.hints.EventHintBundle
import com.vitorpamplona.quartz.nip01Core.metadata.MetadataEvent
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener
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.nip19Bech32.entities.NEvent
@@ -436,7 +436,7 @@ private suspend fun fetchMetadataForUsers(
filters = filters,
relays = relays,
listener =
object : IRequestListener {
object : SubscriptionListener {
override fun onEvent(
event: Event,
isLive: Boolean,
@@ -1058,7 +1058,7 @@ private suspend fun fetchUserLightningAddress(
filters = filters,
relays = relays,
listener =
object : IRequestListener {
object : SubscriptionListener {
override fun onEvent(
event: Event,
isLive: Boolean,
@@ -171,7 +171,7 @@ fun SearchScreen(
}
}
// NIP-50 people search subscription (use allRelayUrls — openReqSubscription will connect)
// NIP-50 people search subscription (use allRelayUrls — subscribe will connect)
rememberSubscription(connectedRelays, debouncedQuery, relayManager = relayManager) {
if (allRelayUrls.isEmpty() || debouncedQuery.isEmpty) {
return@rememberSubscription null
@@ -29,7 +29,7 @@ import com.vitorpamplona.amethyst.desktop.cache.DesktopLocalCache
import com.vitorpamplona.amethyst.desktop.network.RelayConnectionManager
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.metadata.MetadataEvent
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener
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.nip04Dm.messages.PrivateDmEvent
@@ -161,7 +161,7 @@ class ChatroomListState(
)
val listener =
object : IRequestListener {
object : SubscriptionListener {
override fun onEvent(
event: Event,
isLive: Boolean,
@@ -23,7 +23,7 @@ package com.vitorpamplona.amethyst.desktop.ui.chats
import com.vitorpamplona.amethyst.commons.ui.chat.DmBroadcastStatus
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.sendAndWaitForResponseDetailed
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.publishAndConfirmDetailed
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.MutableStateFlow
@@ -56,7 +56,7 @@ class DmSendTracker(
val allSuccessful = mutableSetOf<NormalizedRelayUrl>()
for ((event, relays) in events) {
val results = client.sendAndWaitForResponseDetailed(event, relays, 10)
val results = client.publishAndConfirmDetailed(event, relays, 10)
allSuccessful.addAll(results.filter { it.value }.keys)
_status.value = DmBroadcastStatus.Sending(allSuccessful.size, totalRelays)
}
@@ -28,8 +28,8 @@ import com.vitorpamplona.amethyst.desktop.viewmodels.DesktopFeedViewModel
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.client.listeners.IRelayClientListener
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
@@ -69,7 +69,7 @@ class CoordinatorPipelineTest {
* This lets us test the coordinator's event routing without network dependencies.
*/
private class StubNostrClient : INostrClient {
val openedSubs = mutableMapOf<String, Pair<Map<NormalizedRelayUrl, List<Filter>>, IRequestListener?>>()
val openedSubs = mutableMapOf<String, Pair<Map<NormalizedRelayUrl, List<Filter>>, SubscriptionListener?>>()
override fun connectedRelaysFlow(): StateFlow<Set<NormalizedRelayUrl>> = MutableStateFlow(emptySet())
@@ -88,17 +88,17 @@ class CoordinatorPipelineTest {
override fun isActive() = false
override fun renewFilters(relay: IRelayClient) {}
override fun syncFilters(relay: IRelayClient) {}
override fun openReqSubscription(
override fun subscribe(
subId: String,
filters: Map<NormalizedRelayUrl, List<Filter>>,
listener: IRequestListener?,
listener: SubscriptionListener?,
) {
openedSubs[subId] = filters to listener
}
override fun queryCount(
override fun count(
subId: String,
filters: Map<NormalizedRelayUrl, List<Filter>>,
) {}
@@ -112,9 +112,9 @@ class CoordinatorPipelineTest {
relayList: Set<NormalizedRelayUrl>,
) {}
override fun subscribe(listener: IRelayClientListener) {}
override fun addConnectionListener(listener: RelayConnectionListener) {}
override fun unsubscribe(listener: IRelayClientListener) {}
override fun removeConnectionListener(listener: RelayConnectionListener) {}
override fun getReqFiltersOrNull(subId: String): Map<NormalizedRelayUrl, List<Filter>>? = null
@@ -376,7 +376,7 @@ class CoordinatorPipelineTest {
val noteIds = listOf("n1".padEnd(64, '0'))
val subId = coordinator.requestInteractions(noteIds, setOf(relayUrl))
// openReqSubscription is launched in scope — wait for it
// subscribe is launched in scope — wait for it
delay(200)
assertTrue(client.openedSubs.containsKey(subId), "Interaction subscription should be opened")
@@ -22,8 +22,8 @@ package com.vitorpamplona.quartz.nip01Core.relay.client
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.core.HexKey
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.IRelayClientListener
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
import com.vitorpamplona.quartz.nip01Core.relay.client.single.newSubId
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
@@ -52,29 +52,29 @@ interface INostrClient : AutoCloseable {
* This is called every time the relay connects
* and when auth is successful
*/
fun renewFilters(relay: IRelayClient)
fun syncFilters(relay: IRelayClient)
fun openReqSubscription(
fun subscribe(
subId: String = newSubId(),
filters: Map<NormalizedRelayUrl, List<Filter>>,
listener: IRequestListener? = null,
listener: SubscriptionListener? = null,
)
fun queryCount(
fun count(
subId: String = newSubId(),
filters: Map<NormalizedRelayUrl, List<Filter>>,
)
fun close(subId: String)
fun unsubscribe(subId: String)
fun send(
fun publish(
event: Event,
relayList: Set<NormalizedRelayUrl>,
)
fun subscribe(listener: IRelayClientListener)
fun addConnectionListener(listener: RelayConnectionListener)
fun unsubscribe(listener: IRelayClientListener)
fun removeConnectionListener(listener: RelayConnectionListener)
fun getReqFiltersOrNull(subId: String): Map<NormalizedRelayUrl, List<Filter>>?
@@ -103,29 +103,29 @@ class EmptyNostrClient : INostrClient {
override fun isActive() = false
override fun renewFilters(relay: IRelayClient) { }
override fun syncFilters(relay: IRelayClient) { }
override fun openReqSubscription(
override fun subscribe(
subId: String,
filters: Map<NormalizedRelayUrl, List<Filter>>,
listener: IRequestListener?,
listener: SubscriptionListener?,
) { }
override fun queryCount(
override fun count(
subId: String,
filters: Map<NormalizedRelayUrl, List<Filter>>,
) { }
override fun close(subId: String) { }
override fun unsubscribe(subId: String) { }
override fun send(
override fun publish(
event: Event,
relayList: Set<NormalizedRelayUrl>,
) { }
override fun subscribe(listener: IRelayClientListener) {}
override fun addConnectionListener(listener: RelayConnectionListener) {}
override fun unsubscribe(listener: IRelayClientListener) {}
override fun removeConnectionListener(listener: RelayConnectionListener) {}
override fun getReqFiltersOrNull(subId: String): Map<NormalizedRelayUrl, List<Filter>>? = null
@@ -22,12 +22,12 @@ package com.vitorpamplona.quartz.nip01Core.relay.client
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.core.HexKey
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.IRelayClientListener
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.pool.PoolCounts
import com.vitorpamplona.quartz.nip01Core.relay.client.pool.PoolEventOutbox
import com.vitorpamplona.quartz.nip01Core.relay.client.pool.PoolRequests
import com.vitorpamplona.quartz.nip01Core.relay.client.pool.RelayPool
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message
import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.AuthCmd
@@ -52,7 +52,7 @@ import kotlinx.coroutines.flow.stateIn
import kotlinx.coroutines.launch
/**
* The NostrClient manages Nostr relay operations, subscriptions, and event delivery. It maintains:
* The INostrClient manages Nostr relay operations, subscriptions, and event delivery. It maintains:
* - A RelayPool for managing connections to a collection of Nostr relays
* - Active subscriptions tracking through PoolSubscriptionRepository
* - An event outbox for managing unsent events and retry logic
@@ -80,7 +80,7 @@ class NostrClient(
private val websocketBuilder: WebsocketBuilder,
private val parentScope: CoroutineScope = CoroutineScope(Dispatchers.IO + SupervisorJob()),
) : INostrClient,
IRelayClientListener,
RelayConnectionListener,
AutoCloseable {
private val relayPool: RelayPool = RelayPool(websocketBuilder, this)
@@ -91,7 +91,7 @@ class NostrClient(
private val activeCounts: PoolCounts = PoolCounts()
private val eventOutbox: PoolEventOutbox = PoolEventOutbox()
private var listeners = setOf<IRelayClientListener>()
private var listeners = setOf<RelayConnectionListener>()
// controls the state of the client in such a way that if it is active
// new filters will be sent to the relays and a potential reconnect can
@@ -164,10 +164,10 @@ class NostrClient(
refreshConnection.tryEmit(Reconnect(onlyIfChanged, ignoreRetryDelays))
}
override fun openReqSubscription(
override fun subscribe(
subId: String,
filters: Map<NormalizedRelayUrl, List<Filter>>,
listener: IRequestListener?,
listener: SubscriptionListener?,
) {
val relaysToUpdate = activeRequests.addOrUpdate(subId, filters, listener)
@@ -188,7 +188,7 @@ class NostrClient(
}
}
override fun queryCount(
override fun count(
subId: String,
filters: Map<NormalizedRelayUrl, List<Filter>>,
) {
@@ -208,7 +208,7 @@ class NostrClient(
}
}
override fun send(
override fun publish(
event: Event,
relayList: Set<NormalizedRelayUrl>,
) {
@@ -222,7 +222,7 @@ class NostrClient(
}
}
override fun close(subId: String) {
override fun unsubscribe(subId: String) {
val relaysToUpdateReqs = activeRequests.remove(subId)
val relaysToUpdateCounts = activeCounts.remove(subId)
@@ -232,7 +232,7 @@ class NostrClient(
}
}
override fun renewFilters(relay: IRelayClient) {
override fun syncFilters(relay: IRelayClient) {
if (isActive) {
scope.launch {
activeRequests.syncState(relay.url, relay::sendOrConnectAndSync)
@@ -259,7 +259,7 @@ class NostrClient(
pingMillis: Int,
compressed: Boolean,
) {
renewFilters(relay)
syncFilters(relay)
listeners.forEach { it.onConnected(relay, pingMillis, compressed) }
}
@@ -308,13 +308,13 @@ class NostrClient(
listeners.forEach { it.onCannotConnect(relay, errorMessage) }
}
override fun subscribe(listener: IRelayClientListener) {
override fun addConnectionListener(listener: RelayConnectionListener) {
listeners = listeners.plus(listener)
}
fun isSubscribed(listener: IRelayClientListener): Boolean = listeners.contains(listener)
fun hasConnectionListener(listener: RelayConnectionListener): Boolean = listeners.contains(listener)
override fun unsubscribe(listener: IRelayClientListener) {
override fun removeConnectionListener(listener: RelayConnectionListener) {
listeners = listeners.minus(listener)
}
@@ -22,21 +22,21 @@ package com.vitorpamplona.quartz.nip01Core.relay.client.accessories
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.IRelayClientListener
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.EventMessage
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message
import com.vitorpamplona.quartz.utils.Log
/**
* Listens to NostrClient's onEvent messages for caching purposes.
* Listens to INostrClient's onEvent messages for caching purposes.
*/
class EventCollector(
val client: INostrClient,
val onEvent: (event: Event, relay: IRelayClient) -> Unit,
) {
private val clientListener =
object : IRelayClientListener {
object : RelayConnectionListener {
override fun onIncomingMessage(
relay: IRelayClient,
msgStr: String,
@@ -50,12 +50,12 @@ class EventCollector(
init {
Log.d("EventCollector", "Init, Subscribe")
client.subscribe(clientListener)
client.addConnectionListener(clientListener)
}
fun destroy() {
// makes sure to run
Log.d("EventCollector", "Destroy, Unsubscribe")
client.unsubscribe(clientListener)
client.removeConnectionListener(clientListener)
}
}
@@ -21,7 +21,7 @@
package com.vitorpamplona.quartz.nip01Core.relay.client.accessories
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.IRelayClientListener
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
import com.vitorpamplona.quartz.nip01Core.relay.client.single.newSubId
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.CountMessage
@@ -43,7 +43,7 @@ import kotlinx.coroutines.withTimeoutOrNull
* @param timeoutMs How long to wait for a response (default 15 s).
* @return The [CountResult], or `null` on timeout.
*/
suspend fun INostrClient.queryCountSuspend(
suspend fun INostrClient.count(
relay: NormalizedRelayUrl,
filter: Filter,
timeoutMs: Long = 15_000,
@@ -52,7 +52,7 @@ suspend fun INostrClient.queryCountSuspend(
val resultChannel = Channel<CountResult>(UNLIMITED)
val listener =
object : IRelayClientListener {
object : RelayConnectionListener {
override fun onIncomingMessage(
relay: IRelayClient,
msgStr: String,
@@ -64,18 +64,18 @@ suspend fun INostrClient.queryCountSuspend(
}
}
subscribe(listener)
addConnectionListener(listener)
val result =
try {
queryCount(subId = subId, filters = mapOf(relay to listOf(filter)))
count(subId = subId, filters = mapOf(relay to listOf(filter)))
withTimeoutOrNull(timeoutMs) {
resultChannel.receive()
}
} finally {
close(subId)
unsubscribe(listener)
unsubscribe(subId)
removeConnectionListener(listener)
}
resultChannel.close()
@@ -92,7 +92,7 @@ suspend fun INostrClient.queryCountSuspend(
* @param timeoutMs How long to wait for all responses (default 15 s).
* @return Map of relay -> [CountResult] for every relay that responded in time.
*/
suspend fun INostrClient.queryCountSuspend(
suspend fun INostrClient.count(
filters: Map<NormalizedRelayUrl, List<Filter>>,
timeoutMs: Long = 15_000,
): Map<NormalizedRelayUrl, CountResult> {
@@ -102,7 +102,7 @@ suspend fun INostrClient.queryCountSuspend(
val resultChannel = Channel<Pair<NormalizedRelayUrl, CountResult>>(UNLIMITED)
val listener =
object : IRelayClientListener {
object : RelayConnectionListener {
override fun onIncomingMessage(
relay: IRelayClient,
msgStr: String,
@@ -115,12 +115,12 @@ suspend fun INostrClient.queryCountSuspend(
}
}
subscribe(listener)
addConnectionListener(listener)
filters.forEach { (relay, filterList) ->
val subId = newSubId()
subIdToRelay[subId] = relay
queryCount(subId = subId, filters = mapOf(relay to filterList))
count(subId = subId, filters = mapOf(relay to filterList))
}
val results = mutableMapOf<NormalizedRelayUrl, CountResult>()
@@ -132,8 +132,8 @@ suspend fun INostrClient.queryCountSuspend(
}
}
subIdToRelay.keys.forEach { close(it) }
unsubscribe(listener)
subIdToRelay.keys.forEach { unsubscribe(it) }
removeConnectionListener(listener)
resultChannel.close()
return results
@@ -155,7 +155,7 @@ suspend fun INostrClient.queryCountSuspend(
* @param timeoutMs How long to wait for all responses (default 15 s).
* @return A merged [CountResult], or `null` if no relay responded.
*/
suspend fun INostrClient.queryCountMergedHll(
suspend fun INostrClient.countMerged(
relays: List<NormalizedRelayUrl>,
filter: Filter,
timeoutMs: Long = 15_000,
@@ -163,7 +163,7 @@ suspend fun INostrClient.queryCountMergedHll(
if (relays.isEmpty()) return null
val results =
queryCountSuspend(
count(
filters = relays.associateWith { listOf(filter) },
timeoutMs = timeoutMs,
)
@@ -23,7 +23,7 @@ package com.vitorpamplona.quartz.nip01Core.relay.client.accessories
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.client.reqs.IRequestListener
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.newSubId
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
@@ -31,45 +31,45 @@ import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.withTimeoutOrNull
suspend fun INostrClient.query(
suspend fun INostrClient.fetchAll(
relay: String,
filter: Filter,
timeoutMs: Long = 30_000L,
) = query(newSubId(), mapOf(RelayUrlNormalizer.normalize(relay) to listOf(filter)), timeoutMs)
) = fetchAll(newSubId(), mapOf(RelayUrlNormalizer.normalize(relay) to listOf(filter)), timeoutMs)
suspend fun INostrClient.query(
suspend fun INostrClient.fetchAll(
relay: String,
filters: List<Filter>,
timeoutMs: Long = 30_000L,
) = query(newSubId(), mapOf(RelayUrlNormalizer.normalize(relay) to filters), timeoutMs)
) = fetchAll(newSubId(), mapOf(RelayUrlNormalizer.normalize(relay) to filters), timeoutMs)
suspend fun INostrClient.query(
suspend fun INostrClient.fetchAll(
subscriptionId: String = newSubId(),
relay: String,
filters: List<Filter>,
timeoutMs: Long = 30_000L,
) = query(subscriptionId, mapOf(RelayUrlNormalizer.normalize(relay) to filters), timeoutMs)
) = fetchAll(subscriptionId, mapOf(RelayUrlNormalizer.normalize(relay) to filters), timeoutMs)
suspend fun INostrClient.query(
suspend fun INostrClient.fetchAll(
relay: NormalizedRelayUrl,
filter: Filter,
timeoutMs: Long = 30_000L,
) = query(newSubId(), mapOf(relay to listOf(filter)), timeoutMs)
) = fetchAll(newSubId(), mapOf(relay to listOf(filter)), timeoutMs)
suspend fun INostrClient.query(
suspend fun INostrClient.fetchAll(
relay: NormalizedRelayUrl,
filters: List<Filter>,
timeoutMs: Long = 30_000L,
) = query(newSubId(), mapOf(relay to filters), timeoutMs)
) = fetchAll(newSubId(), mapOf(relay to filters), timeoutMs)
suspend fun INostrClient.query(
suspend fun INostrClient.fetchAll(
subscriptionId: String = newSubId(),
relay: NormalizedRelayUrl,
filters: List<Filter>,
timeoutMs: Long = 30_000L,
) = query(subscriptionId, mapOf(relay to filters), timeoutMs)
) = fetchAll(subscriptionId, mapOf(relay to filters), timeoutMs)
suspend fun INostrClient.query(
suspend fun INostrClient.fetchAll(
subscriptionId: String = newSubId(),
filters: Map<NormalizedRelayUrl, List<Filter>>,
timeoutMs: Long = 30_000L,
@@ -82,7 +82,7 @@ suspend fun INostrClient.query(
val remaining = filters.keys.toMutableSet()
val listener =
object : IRequestListener {
object : SubscriptionListener {
override fun onEvent(
event: Event,
isLive: Boolean,
@@ -119,7 +119,7 @@ suspend fun INostrClient.query(
}
try {
openReqSubscription(subscriptionId, filters, listener)
subscribe(subscriptionId, filters, listener)
withTimeoutOrNull(timeoutMs) {
while (remaining.isNotEmpty()) {
@@ -128,7 +128,7 @@ suspend fun INostrClient.query(
}
}
} finally {
close(subscriptionId)
unsubscribe(subscriptionId)
doneChannel.close()
}
@@ -22,7 +22,7 @@ package com.vitorpamplona.quartz.nip01Core.relay.client.accessories
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.newSubId
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
@@ -51,7 +51,7 @@ import kotlin.math.min
* @param onEvent Called for every event received (in page order, after each EOSE).
* @return Total number of events received across all pages.
*/
suspend fun INostrClient.reqBypassingRelayLimits(
suspend fun INostrClient.fetchAllPages(
relay: NormalizedRelayUrl,
filters: List<Filter>,
timeoutMs: Long = 30_000L,
@@ -95,7 +95,7 @@ suspend fun INostrClient.reqBypassingRelayLimits(
try {
val listener =
object : IRequestListener {
object : SubscriptionListener {
override fun onEvent(
event: Event,
isLive: Boolean,
@@ -144,16 +144,16 @@ suspend fun INostrClient.reqBypassingRelayLimits(
}
}
openReqSubscription(subId, mapOf(relay to remainingFilters), listener)
subscribe(subId, mapOf(relay to remainingFilters), listener)
withTimeoutOrNull(timeoutMs) {
doneChannel.receive()
}
close(subId)
unsubscribe(subId)
doneChannel.close()
} finally {
close(subId)
unsubscribe(subId)
doneChannel.close()
}
@@ -168,14 +168,14 @@ suspend fun INostrClient.reqBypassingRelayLimits(
return totalEvents
}
suspend fun INostrClient.reqBypassingRelayLimits(
suspend fun INostrClient.fetchAllPages(
relay: String,
filters: List<Filter>,
timeoutMs: Long = 30_000L,
onNewPage: ((Long) -> Unit)? = null,
onEvent: (Event) -> Unit,
): Int =
reqBypassingRelayLimits(
fetchAllPages(
relay = RelayUrlNormalizer.normalize(relay),
filters = filters,
timeoutMs = timeoutMs,
@@ -22,7 +22,7 @@ package com.vitorpamplona.quartz.nip01Core.relay.client.accessories
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.newSubId
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
@@ -31,46 +31,46 @@ import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED
import kotlinx.coroutines.withTimeoutOrNull
suspend fun INostrClient.downloadFirstEvent(
suspend fun INostrClient.fetchFirst(
relay: String,
filter: Filter,
) = downloadFirstEvent(newSubId(), mapOf(RelayUrlNormalizer.normalize(relay) to listOf(filter)))
) = fetchFirst(newSubId(), mapOf(RelayUrlNormalizer.normalize(relay) to listOf(filter)))
suspend fun INostrClient.downloadFirstEvent(
suspend fun INostrClient.fetchFirst(
relay: String,
filters: List<Filter>,
) = downloadFirstEvent(newSubId(), mapOf(RelayUrlNormalizer.normalize(relay) to filters))
) = fetchFirst(newSubId(), mapOf(RelayUrlNormalizer.normalize(relay) to filters))
suspend fun INostrClient.downloadFirstEvent(
suspend fun INostrClient.fetchFirst(
subscriptionId: String = newSubId(),
relay: String,
filters: List<Filter>,
) = downloadFirstEvent(subscriptionId, mapOf(RelayUrlNormalizer.normalize(relay) to filters))
) = fetchFirst(subscriptionId, mapOf(RelayUrlNormalizer.normalize(relay) to filters))
suspend fun INostrClient.downloadFirstEvent(
suspend fun INostrClient.fetchFirst(
relay: NormalizedRelayUrl,
filter: Filter,
) = downloadFirstEvent(newSubId(), mapOf(relay to listOf(filter)))
) = fetchFirst(newSubId(), mapOf(relay to listOf(filter)))
suspend fun INostrClient.downloadFirstEvent(
suspend fun INostrClient.fetchFirst(
relay: NormalizedRelayUrl,
filters: List<Filter>,
) = downloadFirstEvent(newSubId(), mapOf(relay to filters))
) = fetchFirst(newSubId(), mapOf(relay to filters))
suspend fun INostrClient.downloadFirstEvent(
suspend fun INostrClient.fetchFirst(
subscriptionId: String = newSubId(),
relay: NormalizedRelayUrl,
filters: List<Filter>,
) = downloadFirstEvent(subscriptionId, mapOf(relay to filters))
) = fetchFirst(subscriptionId, mapOf(relay to filters))
suspend fun INostrClient.downloadFirstEvent(
suspend fun INostrClient.fetchFirst(
subscriptionId: String = newSubId(),
filters: Map<NormalizedRelayUrl, List<Filter>>,
): Event? {
val resultChannel = Channel<Event?>(UNLIMITED)
val listener =
object : IRequestListener {
object : SubscriptionListener {
override fun onEvent(
event: Event,
isLive: Boolean,
@@ -106,13 +106,13 @@ suspend fun INostrClient.downloadFirstEvent(
val result =
try {
openReqSubscription(subscriptionId, filters, listener)
subscribe(subscriptionId, filters, listener)
withTimeoutOrNull(30000) {
resultChannel.receive()
}
} finally {
close(subscriptionId)
unsubscribe(subscriptionId)
}
resultChannel.close()
@@ -22,7 +22,7 @@ package com.vitorpamplona.quartz.nip01Core.relay.client.accessories
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.IRelayClientListener
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.OkMessage
@@ -41,42 +41,42 @@ class Result(
)
@OptIn(DelicateCoroutinesApi::class)
suspend fun INostrClient.sendAndWaitForResponse(
suspend fun INostrClient.publishAndConfirm(
event: Event,
relayList: Set<NormalizedRelayUrl>,
timeoutInSeconds: Long = 15,
): Boolean = sendAndWaitForResponseDetailed(event, relayList, timeoutInSeconds).any { it.value }
): Boolean = publishAndConfirmDetailed(event, relayList, timeoutInSeconds).any { it.value }
/**
* Sends an event to the given relays and waits for OK responses.
* Returns per-relay results: relay URL -> accepted (true/false).
*/
@OptIn(DelicateCoroutinesApi::class)
suspend fun INostrClient.sendAndWaitForResponseDetailed(
suspend fun INostrClient.publishAndConfirmDetailed(
event: Event,
relayList: Set<NormalizedRelayUrl>,
timeoutInSeconds: Long = 15,
): Map<NormalizedRelayUrl, Boolean> {
val resultChannel = Channel<Result>(UNLIMITED)
Log.d("sendAndWaitForResponse", "Waiting for ${relayList.size} responses")
Log.d("publishAndConfirm", "Waiting for ${relayList.size} responses")
val subscription =
object : IRelayClientListener {
object : RelayConnectionListener {
override fun onCannotConnect(
relay: IRelayClient,
errorMessage: String,
) {
if (relay.url in relayList) {
resultChannel.trySend(Result(relay.url, false))
Log.d("sendAndWaitForResponse", "Error from relay ${relay.url}: $errorMessage")
Log.d("publishAndConfirm", "Error from relay ${relay.url}: $errorMessage")
}
}
override fun onDisconnected(relay: IRelayClient) {
if (relay.url in relayList) {
resultChannel.trySend(Result(relay.url, false))
Log.d("sendAndWaitForResponse", "Disconnected from relay ${relay.url}")
Log.d("publishAndConfirm", "Disconnected from relay ${relay.url}")
}
}
@@ -91,7 +91,7 @@ suspend fun INostrClient.sendAndWaitForResponseDetailed(
is OkMessage -> {
if (msg.eventId == event.id) {
resultChannel.trySend(Result(relay.url, msg.success))
Log.d("sendAndWaitForResponse", "onSendResponse Received response for ${msg.eventId} from relay ${relay.url} message ${msg.message} success ${msg.success}")
Log.d("publishAndConfirm", "onSendResponse Received response for ${msg.eventId} from relay ${relay.url} message ${msg.message} success ${msg.success}")
}
}
}
@@ -100,7 +100,7 @@ suspend fun INostrClient.sendAndWaitForResponseDetailed(
val receivedResults =
try {
subscribe(subscription)
addConnectionListener(subscription)
// subscribe before sending the result.
val resultSubscription =
@@ -123,20 +123,20 @@ suspend fun INostrClient.sendAndWaitForResponseDetailed(
receivedResults
}
send(event, relayList)
publish(event, relayList)
result
}
resultSubscription.await()
} finally {
unsubscribe(subscription)
removeConnectionListener(subscription)
}
// Clean up the channel
resultChannel.close()
Log.d("sendAndWaitForResponse", "Finished with ${receivedResults.size} results")
Log.d("publishAndConfirm", "Finished with ${receivedResults.size} results")
return receivedResults
}
@@ -22,21 +22,21 @@ package com.vitorpamplona.quartz.nip01Core.relay.client.accessories
import com.vitorpamplona.quartz.nip01Core.core.HexKey
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.IRelayClientListener
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.OkMessage
import com.vitorpamplona.quartz.utils.Log
/**
* Listens to NostrClient's onEvent messages for caching purposes.
* Listens to INostrClient's onEvent messages for caching purposes.
*/
class RelayInsertConfirmationCollector(
val client: INostrClient,
val onRelayReceived: (eventId: HexKey, relay: IRelayClient) -> Unit,
) {
private val clientListener =
object : IRelayClientListener {
object : RelayConnectionListener {
override fun onIncomingMessage(
relay: IRelayClient,
msgStr: String,
@@ -50,12 +50,12 @@ class RelayInsertConfirmationCollector(
init {
Log.d("RelayInsertConfirmationCollector", "Init, Subscribe")
client.subscribe(clientListener)
client.addConnectionListener(clientListener)
}
fun destroy() {
// makes sure to run
Log.d("RelayInsertConfirmationCollector", "Destroy, Unsubscribe")
client.unsubscribe(clientListener)
client.removeConnectionListener(clientListener)
}
}
@@ -21,7 +21,7 @@
package com.vitorpamplona.quartz.nip01Core.relay.client.accessories
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.IRelayClientListener
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.AuthMessage
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.ClosedMessage
@@ -38,7 +38,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.normalizer.displayUrl
import com.vitorpamplona.quartz.utils.Log
/**
* Listens to NostrClient's onNotify messages from the relay
* Listens to INostrClient's onNotify messages from the relay
*/
class RelayLogger(
val client: INostrClient,
@@ -48,7 +48,7 @@ class RelayLogger(
fun logTag(url: NormalizedRelayUrl) = "Relay ${url.displayUrl()}"
private val clientListener =
object : IRelayClientListener {
object : RelayConnectionListener {
override fun onIncomingMessage(
relay: IRelayClient,
msgStr: String,
@@ -110,12 +110,12 @@ class RelayLogger(
init {
Log.d("RelayLogger", "Init, Subscribe")
client.subscribe(clientListener)
client.addConnectionListener(clientListener)
}
fun destroy() {
// makes sure to run
Log.d("RelayLogger", "Destroy, Unsubscribe")
client.unsubscribe(clientListener)
client.removeConnectionListener(clientListener)
}
}
@@ -21,14 +21,14 @@
package com.vitorpamplona.quartz.nip01Core.relay.client.accessories
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.IRelayClientListener
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.NotifyMessage
import com.vitorpamplona.quartz.utils.Log
/**
* Listens to NostrClient's onNotify messages from the relay
* Listens to INostrClient's onNotify messages from the relay
*/
class RelayNotifier(
val client: INostrClient,
@@ -39,7 +39,7 @@ class RelayNotifier(
}
private val clientListener =
object : IRelayClientListener {
object : RelayConnectionListener {
override fun onIncomingMessage(
relay: IRelayClient,
msgStr: String,
@@ -54,12 +54,12 @@ class RelayNotifier(
init {
Log.d(TAG, "Init, Subscribe")
client.subscribe(clientListener)
client.addConnectionListener(clientListener)
}
fun destroy() {
// makes sure to run
Log.d(TAG, "Destroy, Unsubscribe")
client.unsubscribe(clientListener)
client.removeConnectionListener(clientListener)
}
}
@@ -21,13 +21,13 @@
package com.vitorpamplona.quartz.nip01Core.relay.client.accessories
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.IRelayClientListener
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
import com.vitorpamplona.quartz.utils.Log
/**
* Listens to NostrClient's onNotify messages from the relay
* Listens to INostrClient's onNotify messages from the relay
*/
class RelayOfflineTracker(
val client: INostrClient,
@@ -39,7 +39,7 @@ class RelayOfflineTracker(
var cannotConnectRelays = setOf<NormalizedRelayUrl>()
private val clientListener =
object : IRelayClientListener {
object : RelayConnectionListener {
override fun onConnected(
relay: IRelayClient,
pingMillis: Int,
@@ -58,12 +58,12 @@ class RelayOfflineTracker(
init {
Log.d(TAG, "Init, Subscribe")
client.subscribe(clientListener)
client.addConnectionListener(clientListener)
}
fun destroy() {
// makes sure to run
Log.d(TAG, "Destroy, Unsubscribe")
client.unsubscribe(clientListener)
client.removeConnectionListener(clientListener)
}
}
@@ -21,7 +21,7 @@
package com.vitorpamplona.quartz.nip01Core.relay.client.auth
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.IRelayClientListener
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.AuthMessage
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message
@@ -53,7 +53,7 @@ class RelayAuthenticator(
private val authStatus = mutableMapOf<NormalizedRelayUrl, RelayAuthStatus>()
private val clientListener =
object : IRelayClientListener {
object : RelayConnectionListener {
override fun onIncomingMessage(
relay: IRelayClient,
msgStr: String,
@@ -95,7 +95,7 @@ class RelayAuthenticator(
) {
// if this is the OK of an auth event, renew all subscriptions and resend all outgoing events.
if (authStatus[relay.url]?.checkAuthResults(msg.eventId, msg.success) == true) {
client.renewFilters(relay)
client.syncFilters(relay)
}
}
@@ -103,12 +103,12 @@ class RelayAuthenticator(
init {
Log.d("RelayAuthenticator", "Init, Subscribe")
client.subscribe(clientListener)
client.addConnectionListener(clientListener)
}
fun destroy() {
// makes sure to run
Log.d("RelayAuthenticator", "Destroy, Unsubscribe")
client.unsubscribe(clientListener)
client.removeConnectionListener(clientListener)
}
}
@@ -21,7 +21,7 @@
package com.vitorpamplona.quartz.nip01Core.relay.client.counts
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.IRelayClientListener
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.ClosedMessage
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.CountMessage
@@ -40,7 +40,7 @@ class RelayActiveCountStates(
fun subGetOrCreate(relay: NormalizedRelayUrl): CountQueryState<String> = queryStates[relay] ?: CountQueryState<String>().also { queryStates.put(relay, it) }
private val clientListener =
object : IRelayClientListener {
object : RelayConnectionListener {
override fun onConnecting(relay: IRelayClient) {
queryStates.put(relay.url, CountQueryState())
}
@@ -75,12 +75,12 @@ class RelayActiveCountStates(
init {
Log.d("RelaySubStateMachine", "Init, Subscribe")
client.subscribe(clientListener)
client.addConnectionListener(clientListener)
}
fun destroy() {
// makes sure to run
Log.d("RelaySubStateMachine", "Destroy, Unsubscribe")
client.unsubscribe(clientListener)
client.removeConnectionListener(clientListener)
}
}
@@ -24,9 +24,9 @@ import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message
import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.Command
open class RedirectRelayClientListener(
val listener: IRelayClientListener,
) : IRelayClientListener {
open class RedirectConnectionListener(
val listener: RelayConnectionListener,
) : RelayConnectionListener {
override fun onConnecting(relay: IRelayClient) {
listener.onConnecting(relay)
}
@@ -24,7 +24,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message
import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.Command
interface IRelayClientListener {
interface RelayConnectionListener {
fun onConnecting(relay: IRelayClient) {}
/**
@@ -72,4 +72,4 @@ interface IRelayClientListener {
) {}
}
object EmptyClientListener : IRelayClientListener
object EmptyConnectionListener : RelayConnectionListener
@@ -20,9 +20,9 @@
*/
package com.vitorpamplona.quartz.nip01Core.relay.client.pool
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.ReqSubStatus
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.RequestSubscriptionState
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.ClosedMessage
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.EoseMessage
@@ -51,7 +51,7 @@ class PoolRequests {
* to what the app whants to do.
*/
private val desiredSubs = LargeCache<String, Map<NormalizedRelayUrl, List<Filter>>>()
private val desiredSubListeners = LargeCache<String, IRequestListener>()
private val desiredSubListeners = LargeCache<String, SubscriptionListener>()
val desiredRelays = MutableStateFlow(setOf<NormalizedRelayUrl>())
/**
@@ -101,7 +101,7 @@ class PoolRequests {
fun addOrUpdate(
subId: String,
filters: Map<NormalizedRelayUrl, List<Filter>>,
listener: IRequestListener?,
listener: SubscriptionListener?,
): Set<NormalizedRelayUrl> {
// saves old relays
val oldRelays = desiredSubs.get(subId)?.keys ?: emptySet()
@@ -165,15 +165,15 @@ class PoolRequests {
when (cmd) {
is ReqCmd -> {
subState(cmd.subId).onOpenReq(relay, cmd.filters)
desiredSubListeners.get(cmd.subId)?.onStartReq(
desiredSubListeners.get(cmd.subId)?.onSubscriptionStarted(
relay = relay.url,
forFilters = cmd.filters,
)
}
is CloseCmd -> {
subState(cmd.subId).onCloseReq(relay)
desiredSubListeners.get(cmd.subId)?.onCloseReq(
subState(cmd.subId).onSubscriptionClosed(relay)
desiredSubListeners.get(cmd.subId)?.onSubscriptionClosed(
relay = relay.url,
)
}
@@ -20,8 +20,8 @@
*/
package com.vitorpamplona.quartz.nip01Core.relay.client.pool
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.EmptyClientListener
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.IRelayClientListener
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.EmptyConnectionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
import com.vitorpamplona.quartz.nip01Core.relay.client.single.basic.BasicRelayClient
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message
@@ -56,8 +56,8 @@ import kotlinx.coroutines.flow.update
*/
class RelayPool(
val websocketBuilder: WebsocketBuilder,
val listener: IRelayClientListener = EmptyClientListener,
) : IRelayClientListener {
val listener: RelayConnectionListener = EmptyConnectionListener,
) : RelayConnectionListener {
private val relays = LargeCache<NormalizedRelayUrl, IRelayClient>()
private val _connectedRelays = MutableStateFlow<Set<NormalizedRelayUrl>>(emptySet())
@@ -26,12 +26,12 @@ import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
import com.vitorpamplona.quartz.utils.RandomInstance
class NostrClientDynamicReq(
class DynamicSubscription(
val client: INostrClient,
val filter: () -> Map<NormalizedRelayUrl, List<Filter>>,
val onEvent: (event: Event) -> Unit = {},
) : IRequestListener,
IOpenNostrRequest {
) : SubscriptionListener,
SubscriptionHandle {
val subId = RandomInstance.randomChars(10)
override fun onEvent(
@@ -47,16 +47,16 @@ class NostrClientDynamicReq(
* Creates or Updates the filter with relays. This method should be called
* everytime the filter changes.
*/
override fun updateFilter() = client.openReqSubscription(subId, filter(), this)
override fun refresh() = client.subscribe(subId, filter(), this)
override fun close() = client.close(subId)
override fun close() = client.unsubscribe(subId)
init {
updateFilter()
refresh()
}
}
fun INostrClient.req(
fun INostrClient.subscribe(
filters: () -> Map<NormalizedRelayUrl, List<Filter>>,
onEvent: (event: Event) -> Unit = {},
): IOpenNostrRequest = NostrClientDynamicReq(this, filters, onEvent)
): SubscriptionHandle = DynamicSubscription(this, filters, onEvent)
@@ -37,22 +37,22 @@ import kotlinx.coroutines.flow.callbackFlow
* 3. Closes the flow (completes) when the relay sends EOSE,
* cancelling the subscription automatically via [awaitClose].
*/
fun INostrClient.reqUntilEoseAsFlow(
fun INostrClient.fetchAsFlow(
relay: String,
filters: List<Filter>,
) = reqUntilEoseAsFlow(RelayUrlNormalizer.normalize(relay), filters)
) = fetchAsFlow(RelayUrlNormalizer.normalize(relay), filters)
fun INostrClient.reqUntilEoseAsFlow(
fun INostrClient.fetchAsFlow(
relay: String,
filter: Filter,
) = reqUntilEoseAsFlow(RelayUrlNormalizer.normalize(relay), listOf(filter))
) = fetchAsFlow(RelayUrlNormalizer.normalize(relay), listOf(filter))
fun INostrClient.reqUntilEoseAsFlow(
fun INostrClient.fetchAsFlow(
relay: NormalizedRelayUrl,
filter: Filter,
) = reqUntilEoseAsFlow(relay, listOf(filter))
) = fetchAsFlow(relay, listOf(filter))
fun INostrClient.reqUntilEoseAsFlow(
fun INostrClient.fetchAsFlow(
relay: NormalizedRelayUrl,
filters: List<Filter>,
): Flow<List<Event>> =
@@ -62,7 +62,7 @@ fun INostrClient.reqUntilEoseAsFlow(
var currentEvents = listOf<Event>()
val listener =
object : IRequestListener {
object : SubscriptionListener {
override fun onEvent(
event: Event,
isLive: Boolean,
@@ -84,9 +84,9 @@ fun INostrClient.reqUntilEoseAsFlow(
}
}
openReqSubscription(subId, mapOf(relay to filters), listener)
subscribe(subId, mapOf(relay to filters), listener)
awaitClose {
close(subId)
unsubscribe(subId)
}
}
@@ -42,22 +42,22 @@ import kotlinx.coroutines.flow.callbackFlow
* - They will be ignored if they are already in the list.
* - They will be added to the beginning of the list if they are new.
*/
fun INostrClient.reqAsFlow(
fun INostrClient.subscribeAsFlow(
relay: String,
filters: List<Filter>,
) = reqAsFlow(RelayUrlNormalizer.normalize(relay), filters)
) = subscribeAsFlow(RelayUrlNormalizer.normalize(relay), filters)
fun INostrClient.reqAsFlow(
fun INostrClient.subscribeAsFlow(
relay: String,
filter: Filter,
) = reqAsFlow(RelayUrlNormalizer.normalize(relay), listOf(filter))
) = subscribeAsFlow(RelayUrlNormalizer.normalize(relay), listOf(filter))
fun INostrClient.reqAsFlow(
fun INostrClient.subscribeAsFlow(
relay: NormalizedRelayUrl,
filter: Filter,
) = reqAsFlow(relay, listOf(filter))
) = subscribeAsFlow(relay, listOf(filter))
fun INostrClient.reqAsFlow(
fun INostrClient.subscribeAsFlow(
relay: NormalizedRelayUrl,
filters: List<Filter>,
): Flow<List<Event>> =
@@ -68,7 +68,7 @@ fun INostrClient.reqAsFlow(
var currentEvents = listOf<Event>()
val listener =
object : IRequestListener {
object : SubscriptionListener {
override fun onEvent(
event: Event,
isLive: Boolean,
@@ -98,9 +98,9 @@ fun INostrClient.reqAsFlow(
}
}
openReqSubscription(subId, mapOf(relay to filters), listener)
subscribe(subId, mapOf(relay to filters), listener)
awaitClose {
close(subId)
unsubscribe(subId)
}
}
@@ -21,7 +21,7 @@
package com.vitorpamplona.quartz.nip01Core.relay.client.reqs
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.IRelayClientListener
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.ClosedMessage
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.EoseMessage
@@ -41,7 +41,7 @@ class RelayActiveRequestStates(
fun subGetOrCreate(relay: NormalizedRelayUrl): RequestSubscriptionState<String> = subStates[relay] ?: RequestSubscriptionState<String>().also { subStates.put(relay, it) }
private val clientListener =
object : IRelayClientListener {
object : RelayConnectionListener {
override fun onConnecting(relay: IRelayClient) {
subStates[relay.url] = RequestSubscriptionState()
}
@@ -66,7 +66,7 @@ class RelayActiveRequestStates(
) {
when (cmd) {
is ReqCmd -> subGetOrCreate(relay.url).onOpenReq(cmd.subId, cmd.filters)
is CloseCmd -> subGetOrCreate(relay.url).onCloseReq(cmd.subId)
is CloseCmd -> subGetOrCreate(relay.url).onSubscriptionClosed(cmd.subId)
}
}
@@ -77,12 +77,12 @@ class RelayActiveRequestStates(
init {
Log.d("RelaySubStateMachine", "Init, Subscribe")
client.subscribe(clientListener)
client.addConnectionListener(clientListener)
}
fun destroy() {
// makes sure to run
Log.d("RelaySubStateMachine", "Destroy, Unsubscribe")
client.unsubscribe(clientListener)
client.removeConnectionListener(clientListener)
}
}
@@ -80,7 +80,7 @@ class RequestSubscriptionState<T> {
lastKnownFilterStates[reference] = filters
}
fun onCloseReq(reference: T) {
fun onSubscriptionClosed(reference: T) {
subStates[reference] = ReqSubStatus.CLOSED
filterStates.remove(reference)
}
@@ -27,12 +27,12 @@ import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer
import com.vitorpamplona.quartz.utils.RandomInstance
class NostrClientStaticReq(
class StaticSubscription(
val client: INostrClient,
val filter: Map<NormalizedRelayUrl, List<Filter>>,
val onEvent: (event: Event) -> Unit = {},
) : IRequestListener,
IOpenNostrRequest {
) : SubscriptionListener,
SubscriptionHandle {
val subId = RandomInstance.randomChars(10)
override fun onEvent(
@@ -48,50 +48,50 @@ class NostrClientStaticReq(
* Creates or Updates the filter with relays. This method should be called
* everytime the filter changes.
*/
override fun updateFilter() = client.openReqSubscription(subId, filter, this)
override fun refresh() = client.subscribe(subId, filter, this)
override fun close() = client.close(subId)
override fun close() = client.unsubscribe(subId)
init {
updateFilter()
refresh()
}
}
fun INostrClient.req(
fun INostrClient.subscribe(
relay: NormalizedRelayUrl,
filters: List<Filter>,
onEvent: (event: Event) -> Unit = {},
): IOpenNostrRequest = NostrClientStaticReq(this, mapOf(relay to filters), onEvent)
): SubscriptionHandle = StaticSubscription(this, mapOf(relay to filters), onEvent)
fun INostrClient.req(
fun INostrClient.subscribe(
relay: NormalizedRelayUrl,
filter: Filter,
onEvent: (event: Event) -> Unit = {},
): IOpenNostrRequest = NostrClientStaticReq(this, mapOf(relay to listOf(filter)), onEvent)
): SubscriptionHandle = StaticSubscription(this, mapOf(relay to listOf(filter)), onEvent)
fun INostrClient.req(
fun INostrClient.subscribe(
relays: List<NormalizedRelayUrl>,
filters: List<Filter>,
onEvent: (event: Event) -> Unit = {},
): IOpenNostrRequest = NostrClientStaticReq(this, relays.associateWith { filters }, onEvent)
): SubscriptionHandle = StaticSubscription(this, relays.associateWith { filters }, onEvent)
fun INostrClient.req(
fun INostrClient.subscribe(
relays: List<NormalizedRelayUrl>,
filter: Filter,
onEvent: (event: Event) -> Unit = {},
): IOpenNostrRequest = NostrClientStaticReq(this, relays.associateWith { listOf(filter) }, onEvent)
): SubscriptionHandle = StaticSubscription(this, relays.associateWith { listOf(filter) }, onEvent)
// -----------------------------------
// Helper methods with relay as string
// -----------------------------------
fun INostrClient.req(
fun INostrClient.subscribe(
relay: String,
filters: List<Filter>,
onEvent: (event: Event) -> Unit = {},
): IOpenNostrRequest = NostrClientStaticReq(this, mapOf(RelayUrlNormalizer.normalize(relay) to filters), onEvent)
): SubscriptionHandle = StaticSubscription(this, mapOf(RelayUrlNormalizer.normalize(relay) to filters), onEvent)
fun INostrClient.req(
fun INostrClient.subscribe(
relay: String,
filter: Filter,
onEvent: (event: Event) -> Unit = {},
): IOpenNostrRequest = NostrClientStaticReq(this, mapOf(RelayUrlNormalizer.normalize(relay) to listOf(filter)), onEvent)
): SubscriptionHandle = StaticSubscription(this, mapOf(RelayUrlNormalizer.normalize(relay) to listOf(filter)), onEvent)
@@ -20,8 +20,8 @@
*/
package com.vitorpamplona.quartz.nip01Core.relay.client.reqs
interface IOpenNostrRequest {
fun updateFilter()
interface SubscriptionHandle {
fun refresh()
fun close()
}
@@ -24,7 +24,7 @@ import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
interface IRequestListener {
interface SubscriptionListener {
fun onEose(
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
@@ -49,10 +49,10 @@ interface IRequestListener {
forFilters: List<Filter>?,
) {}
fun onStartReq(
fun onSubscriptionStarted(
relay: String,
forFilters: List<Filter>,
) {}
fun onCloseReq(relay: String) {}
fun onSubscriptionClosed(relay: String) {}
}
@@ -21,14 +21,14 @@
package com.vitorpamplona.quartz.nip01Core.relay.client.reqs.stats
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.IRelayClientListener
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.EventMessage
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message
import com.vitorpamplona.quartz.utils.Log
/**
* Listens to NostrClient's onNotify messages from the relay
* Listens to INostrClient's onNotify messages from the relay
*/
class RelayReqStats(
val client: INostrClient,
@@ -36,7 +36,7 @@ class RelayReqStats(
private val stats = ReqStatsRepository()
private val clientListener =
object : IRelayClientListener {
object : RelayConnectionListener {
override fun onIncomingMessage(
relay: IRelayClient,
msgStr: String,
@@ -56,12 +56,12 @@ class RelayReqStats(
init {
Log.d("RelaySubStats", "Init, Subscribe")
client.subscribe(clientListener)
client.addConnectionListener(clientListener)
}
fun destroy() {
// makes sure to run
Log.d("RelaySubStats", "Destroy, Unsubscribe")
client.unsubscribe(clientListener)
client.removeConnectionListener(clientListener)
}
}
@@ -21,7 +21,7 @@
package com.vitorpamplona.quartz.nip01Core.relay.client.single.basic
import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.IRelayClientListener
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
import com.vitorpamplona.quartz.nip01Core.relay.client.single.basic.BasicRelayClient.Companion.DELAY_TO_RECONNECT_IN_SECS
import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.Command
@@ -55,7 +55,7 @@ import kotlin.coroutines.cancellation.CancellationException
open class BasicRelayClient(
override val url: NormalizedRelayUrl,
val socketBuilder: WebsocketBuilder,
val listener: IRelayClientListener,
val listener: RelayConnectionListener,
) : IRelayClient {
companion object {
// minimum wait time to reconnect: 1 second
@@ -22,9 +22,9 @@ package com.vitorpamplona.quartz.nip01Core.relay.client.single.standalone
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.core.HexKey
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.EmptyClientListener
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.IRelayClientListener
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RedirectRelayClientListener
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.EmptyConnectionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RedirectConnectionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
import com.vitorpamplona.quartz.nip01Core.relay.client.single.basic.BasicRelayClient
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message
@@ -46,7 +46,7 @@ import com.vitorpamplona.quartz.utils.cache.LargeCache
class StandaloneRelayClient(
url: NormalizedRelayUrl,
socketBuilder: WebsocketBuilder,
listener: IRelayClientListener = EmptyClientListener,
listener: RelayConnectionListener = EmptyConnectionListener,
) {
private val outbox = LargeCache<HexKey, Event>()
private val reqs = LargeCache<String, List<Filter>>()
@@ -56,14 +56,14 @@ class StandaloneRelayClient(
BasicRelayClient(
url,
socketBuilder,
object : RedirectRelayClientListener(listener) {
object : RedirectConnectionListener(listener) {
override fun onConnected(
relay: IRelayClient,
pingMillis: Int,
compressed: Boolean,
) {
super.onConnected(relay, pingMillis, compressed)
renewFilters()
syncFilters()
}
override fun onIncomingMessage(
@@ -83,7 +83,7 @@ class StandaloneRelayClient(
},
)
fun renewFilters() {
fun syncFilters() {
outbox.forEach { id, event ->
client.sendOrConnectAndSync(EventCmd(event))
}
@@ -22,7 +22,7 @@ package com.vitorpamplona.quartz.nip01Core.relay.client.stats
import androidx.collection.LruCache
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.IRelayClientListener
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.ClosedMessage
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message
@@ -47,7 +47,7 @@ class RelayStats(
fun get(url: NormalizedRelayUrl): RelayStat = innerCache[url] ?: throw IllegalArgumentException("Should never happen")
private val clientListener =
object : IRelayClientListener {
object : RelayConnectionListener {
override fun onConnecting(relay: IRelayClient) {
super.onConnecting(relay)
with(get(relay.url)) {
@@ -116,12 +116,12 @@ class RelayStats(
init {
Log.d("RelayStats", "Init, Subscribe")
client.subscribe(clientListener)
client.addConnectionListener(clientListener)
}
fun destroy() {
// makes sure to run
Log.d("RelayStats", "Destroy, Unsubscribe")
client.unsubscribe(clientListener)
client.removeConnectionListener(clientListener)
}
}
@@ -20,14 +20,14 @@
*/
package com.vitorpamplona.quartz.nip01Core.relay.client.subscriptions
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.newSubId
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
data class Subscription(
val id: String = newSubId(),
val listener: IRequestListener,
val listener: SubscriptionListener,
) {
private var currentVersion: Map<NormalizedRelayUrl, List<Filter>>? = null // Inactive when null
@@ -21,13 +21,13 @@
package com.vitorpamplona.quartz.nip01Core.relay.client.subscriptions
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener
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.utils.cache.LargeCache
/**
* Manages Nostr subscriptions using a [NostrClient], allowing subscriptions to be created, modified,
* Manages Nostr subscriptions using a [INostrClient], allowing subscriptions to be created, modified,
* and synchronized with relay filters. Subscriptions are stored in a cache and processed through
* [updateRelays] to update relay filters dynamically. Also tracks event statistics and EOSE (End of
* Stored Events) events, and provides utility methods to interact with subscriptions like dismissal.
@@ -53,7 +53,7 @@ class SubscriptionController(
fun requestNewSubscription(
subId: String,
listener: IRequestListener,
listener: SubscriptionListener,
): Subscription = Subscription(subId, listener).also { subscriptions.put(it.id, it) }
fun dismissSubscription(subId: String) = getSub(subId)?.let { dismissSubscription(it) }
@@ -61,7 +61,7 @@ class SubscriptionController(
fun dismissSubscription(subscription: Subscription) {
subscription.reset()
subscriptions.remove(subscription.id)
client.close(subscription.id)
client.unsubscribe(subscription.id)
}
fun updateRelays() {
@@ -77,23 +77,23 @@ class SubscriptionController(
fun updateRelaysIfNeeded(
subId: String,
listener: IRequestListener,
listener: SubscriptionListener,
newFilters: Map<NormalizedRelayUrl, List<Filter>>?,
oldFilters: Map<NormalizedRelayUrl, List<Filter>>?,
) {
if (oldFilters != null) {
if (newFilters == null) {
// was active and is not active anymore, just close.
client.close(subId)
client.unsubscribe(subId)
} else {
client.openReqSubscription(subId, newFilters, listener)
client.subscribe(subId, newFilters, listener)
}
} else {
if (newFilters == null) {
// was not active and is still not active, does nothing
} else {
// was not active and becomes active, sends the entire filter.
client.openReqSubscription(subId, newFilters, listener)
client.subscribe(subId, newFilters, listener)
}
}
}
@@ -23,7 +23,7 @@ package com.vitorpamplona.quartz.nip46RemoteSigner.signer
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.client.reqs.req
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.StaticSubscription
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
import com.vitorpamplona.quartz.nip01Core.signers.EventTemplate
@@ -67,13 +67,16 @@ class NostrSignerRemote(
)
val subscription =
client.req(
relays = relays.toList(),
filter =
Filter(
kinds = listOf(NostrConnectEvent.KIND),
tags = mapOf("p" to listOf(signer.pubKey)),
),
StaticSubscription(
client,
relays.associateWith {
listOf(
Filter(
kinds = listOf(NostrConnectEvent.KIND),
tags = mapOf("p" to listOf(signer.pubKey)),
),
)
},
) { event ->
if (event is NostrConnectEvent) {
scope.launch {
@@ -83,7 +86,7 @@ class NostrSignerRemote(
}
fun openSubscription() {
subscription.updateFilter()
subscription.refresh()
}
fun closeSubscription() {
@@ -77,7 +77,7 @@ class RemoteSignerManager(
awaitingRequests.put(request.id, continuation)
client.send(event, relayList = relayList)
client.publish(event, relayList = relayList)
}
return when (result) {
@@ -21,7 +21,7 @@
package com.vitorpamplona.quartz.nip77Negentropy
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.IRelayClientListener
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
@@ -58,18 +58,18 @@ interface INegentropyListener {
/**
* Manages NIP-77 negentropy sync sessions across multiple relays.
*
* This class acts as a [IRelayClientListener] that intercepts NEG-MSG and NEG-ERR
* This class acts as a [RelayConnectionListener] that intercepts NEG-MSG and NEG-ERR
* messages and drives the reconciliation protocol. It maintains active sessions
* per relay and subscription ID.
*
* Usage:
* 1. Register this as a listener on the NostrClient
* 1. Register this as a listener on the INostrClient
* 2. Call [startSync] with a filter and local events to begin reconciliation
* 3. Receive results via [INegentropyListener] callbacks
*/
class NegentropyManager(
private val listener: INegentropyListener,
) : IRelayClientListener {
) : RelayConnectionListener {
private val activeSessions = mutableMapOf<String, Pair<NormalizedRelayUrl, NegentropySession>>()
fun startSync(
@@ -23,8 +23,8 @@ package com.vitorpamplona.quartz.nip46RemoteSigner.signer
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.IRelayClientListener
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
@@ -37,7 +37,7 @@ import kotlin.test.assertNotNull
import kotlin.test.assertTrue
/**
* INostrClient that records openReqSubscription calls for verification.
* INostrClient that records subscribe calls for verification.
* Used instead of mockk since commonTest doesn't have mockk.
*/
private class TrackingNostrClient : INostrClient {
@@ -64,33 +64,33 @@ private class TrackingNostrClient : INostrClient {
override fun isActive() = false
override fun renewFilters(relay: IRelayClient) {}
override fun syncFilters(relay: IRelayClient) {}
override fun openReqSubscription(
override fun subscribe(
subId: String,
filters: Map<NormalizedRelayUrl, List<Filter>>,
listener: IRequestListener?,
listener: SubscriptionListener?,
) {
subscriptions.add(SubscriptionRecord(subId, filters))
}
override fun queryCount(
override fun count(
subId: String,
filters: Map<NormalizedRelayUrl, List<Filter>>,
) {}
override fun close(subId: String) {}
override fun unsubscribe(subId: String) {}
override fun send(
override fun publish(
event: Event,
relayList: Set<NormalizedRelayUrl>,
) {
sentEvents.add(event to relayList)
}
override fun subscribe(listener: IRelayClientListener) {}
override fun addConnectionListener(listener: RelayConnectionListener) {}
override fun unsubscribe(listener: IRelayClientListener) {}
override fun removeConnectionListener(listener: RelayConnectionListener) {}
override fun getReqFiltersOrNull(subId: String): Map<NormalizedRelayUrl, List<Filter>>? = null
@@ -120,7 +120,7 @@ class NostrSignerRemoteIsolationTest {
val trackingClient = TrackingNostrClient()
val ephemeralSigner = NostrSignerInternal(KeyPair())
// Construction triggers client.req() which calls openReqSubscription
// Construction triggers client.subscribe() which calls subscribe
NostrSignerRemote(
signer = ephemeralSigner,
remotePubkey = validHex,
@@ -19,10 +19,8 @@
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
package com.vitorpamplona.quartz.nip01Core.relay
import com.vitorpamplona.quartz.nip01Core.metadata.MetadataEvent
import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.downloadFirstEvent
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchFirst
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
@@ -40,7 +38,7 @@ class NostrClientFirstEventTest : BaseNostrClientTest() {
val client = NostrClient(socketBuilder, appScope)
val event =
client.downloadFirstEvent(
client.fetchFirst(
relay = "wss://nos.lol",
filter =
Filter(
@@ -19,11 +19,9 @@
* 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.IRequestListener
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
@@ -50,7 +48,7 @@ class NostrClientManualSubTest : BaseNostrClientTest() {
val mySubId = "test-sub-id-1"
val listener =
object : IRequestListener {
object : SubscriptionListener {
override fun onEvent(
event: Event,
isLive: Boolean,
@@ -79,7 +77,7 @@ class NostrClientManualSubTest : BaseNostrClientTest() {
),
)
client.openReqSubscription(mySubId, filters, listener)
client.subscribe(mySubId, filters, listener)
withTimeoutOrNull(10000) {
while (events.size < 101) {
@@ -90,7 +88,7 @@ class NostrClientManualSubTest : BaseNostrClientTest() {
resultChannel.close()
client.close(mySubId)
client.unsubscribe(mySubId)
client.disconnect()
appScope.cancel()
@@ -19,9 +19,7 @@
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
package com.vitorpamplona.quartz.nip01Core.relay
import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.queryCountSuspend
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.count
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.normalizeRelayUrl
import junit.framework.TestCase.assertTrue
@@ -45,7 +43,7 @@ class NostrClientQueryCountTest : BaseNostrClientTest() {
val appScope = CoroutineScope(Dispatchers.Default + SupervisorJob())
val client = NostrClient(socketBuilder, appScope)
val result = client.queryCountSuspend(relay = fiatjaf, filter = metadata)
val result = client.count(fiatjaf, metadata)
assertTrue((result?.count ?: 0) > 1)
@@ -59,7 +57,7 @@ class NostrClientQueryCountTest : BaseNostrClientTest() {
val appScope = CoroutineScope(Dispatchers.Default + SupervisorJob())
val client = NostrClient(socketBuilder, appScope)
val result = client.queryCountSuspend(relay = fiatjaf, filter = Filter())
val result = client.count(fiatjaf, Filter())
assertTrue((result?.count ?: 0) > 1)
@@ -73,18 +71,17 @@ class NostrClientQueryCountTest : BaseNostrClientTest() {
val appScope = CoroutineScope(Dispatchers.Default + SupervisorJob())
val client = NostrClient(socketBuilder, appScope)
val result =
client.queryCountSuspend(
filters =
mapOf(
fiatjaf to listOf(metadata, outboxRelays),
utxo to listOf(metadata, outboxRelays),
),
val results =
client.count(
mapOf(
fiatjaf to listOf(metadata, outboxRelays),
utxo to listOf(metadata, outboxRelays),
),
)
result.forEach { url, result ->
println("${url.url}: ${result.count}")
assertTrue(result.count > 1)
results.forEach { (url, countResult) ->
println("${url.url}: ${countResult.count}")
assertTrue(countResult.count > 1)
}
client.disconnect()
@@ -19,10 +19,8 @@
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
package com.vitorpamplona.quartz.nip01Core.relay
import com.vitorpamplona.quartz.nip01Core.metadata.MetadataEvent
import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.IRelayClientListener
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.EoseMessage
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.EventMessage
@@ -56,7 +54,7 @@ class NostrClientRepeatSubTest : BaseNostrClientTest() {
val mySubId = "test-sub-id-2"
val listener =
object : IRelayClientListener {
object : RelayConnectionListener {
override fun onIncomingMessage(
relay: IRelayClient,
msgStr: String,
@@ -79,7 +77,7 @@ class NostrClientRepeatSubTest : BaseNostrClientTest() {
}
}
client.subscribe(listener)
client.addConnectionListener(listener)
val filters =
mapOf(
@@ -121,10 +119,10 @@ class NostrClientRepeatSubTest : BaseNostrClientTest() {
Log.d("Test", "Processing message ${events.size}")
// simulates an update in the middle of the sub
if (events.size == 1) {
client.openReqSubscription(mySubId, filtersShouldIgnore)
client.subscribe(mySubId, filtersShouldIgnore)
}
if (events.size == 5) {
client.openReqSubscription(mySubId, filtersShouldSendAfterEOSE)
client.subscribe(mySubId, filtersShouldSendAfterEOSE)
}
events.add(resultChannel.receive())
}
@@ -132,12 +130,12 @@ class NostrClientRepeatSubTest : BaseNostrClientTest() {
}
launch {
client.openReqSubscription(mySubId, filters)
client.subscribe(mySubId, filters)
}
}
client.close(mySubId)
client.unsubscribe(listener)
client.unsubscribe(mySubId)
client.removeConnectionListener(listener)
client.disconnect()
appScope.cancel()
@@ -19,11 +19,9 @@
* 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.accessories.reqBypassingRelayLimits
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchAllPages
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip02FollowList.ContactListEvent
import kotlinx.coroutines.CoroutineScope
@@ -46,7 +44,7 @@ class NostrClientReqBypassingRelayLimitsTest : BaseNostrClientTest() {
// nos.lol returns only 500 events per req
val totalFound =
client.reqBypassingRelayLimits(
client.fetchAllPages(
relay = "wss://nos.lol",
filters =
listOf(
@@ -81,7 +79,7 @@ class NostrClientReqBypassingRelayLimitsTest : BaseNostrClientTest() {
// nos.lol returns only 500 events per req
val totalFound =
client.reqBypassingRelayLimits(
client.fetchAllPages(
relay = "wss://nos.lol",
filters =
listOf(
@@ -19,10 +19,8 @@
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
package com.vitorpamplona.quartz.nip01Core.relay
import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.sendAndWaitForResponse
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.publishAndConfirm
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.normalizeRelayUrl
import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerInternal
import com.vitorpamplona.quartz.nip10Notes.TextNoteEvent
@@ -46,13 +44,13 @@ class NostrClientSendAndWaitTest : BaseNostrClientTest() {
val event = randomSigner.sign(TextNoteEvent.build("Hello World"))
val resultDamus =
client.sendAndWaitForResponse(
client.publishAndConfirm(
event = event,
relayList = setOf("wss://nostr.bitcoiner.social".normalizeRelayUrl()),
)
val resultNos =
client.sendAndWaitForResponse(
client.publishAndConfirm(
event = event,
relayList = setOf("wss://nos.lol".normalizeRelayUrl()),
)
@@ -19,11 +19,9 @@
* 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.reqAsFlow
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.subscribeAsFlow
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.utils.Log
import kotlinx.coroutines.CoroutineScope
@@ -53,7 +51,7 @@ class NostrClientSubscriptionAsFlowTest : BaseNostrClientTest() {
val client = NostrClient(socketBuilder, appScope)
val flow =
client.reqAsFlow(
client.subscribeAsFlow(
relay = "wss://nos.lol",
filter =
Filter(
@@ -92,7 +90,7 @@ class NostrClientSubscriptionAsFlowTest : BaseNostrClientTest() {
val client = NostrClient(socketBuilder, appScope)
val flow =
client.reqAsFlow(
client.subscribeAsFlow(
relay = "wss://nos.lol",
filter =
Filter(
@@ -19,12 +19,11 @@
* 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.req
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.StaticSubscription
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
@@ -47,13 +46,17 @@ class NostrClientSubscriptionTest : BaseNostrClientTest() {
val events = mutableSetOf<Event>()
val sub =
client.req(
relay = "wss://nos.lol",
filter =
Filter(
kinds = listOf(MetadataEvent.KIND),
limit = 100,
),
StaticSubscription(
client,
mapOf(
RelayUrlNormalizer.normalize("wss://nos.lol") to
listOf(
Filter(
kinds = listOf(MetadataEvent.KIND),
limit = 100,
),
),
),
) { event ->
assertEquals(MetadataEvent.KIND, event.kind)
resultChannel.trySend(event)
@@ -19,11 +19,9 @@
* 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.client.reqs.fetchAsFlow
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.utils.Log
import kotlinx.coroutines.CoroutineScope
@@ -53,7 +51,7 @@ class NostrClientSubscriptionUntilEoseAsFlowTest : BaseNostrClientTest() {
val client = NostrClient(socketBuilder, appScope)
val flow =
client.reqUntilEoseAsFlow(
client.fetchAsFlow(
relay = "wss://nos.lol",
filter =
Filter(
@@ -92,7 +90,7 @@ class NostrClientSubscriptionUntilEoseAsFlowTest : BaseNostrClientTest() {
val client = NostrClient(socketBuilder, appScope)
val flow =
client.reqUntilEoseAsFlow(
client.fetchAsFlow(
relay = "wss://nos.lol",
filter =
Filter(