diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/AudioRoomViewModel.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/AudioRoomViewModel.kt index e264555b7..53465d50d 100644 --- a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/AudioRoomViewModel.kt +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/AudioRoomViewModel.kt @@ -171,6 +171,24 @@ class AudioRoomViewModel( private val _speakerCatalogs = MutableStateFlow>(emptyMap()) val speakerCatalogs: StateFlow> = _speakerCatalogs.asStateFlow() + /** + * Pubkeys the moq-lite relay has announced as actively + * broadcasting in this room. Populated by [observeAnnounces] + * once the listener is Connected; an Active announce adds the + * pubkey, an Ended announce removes it. Independent of the + * kind-10312 `publishing` flag — announces come straight from + * the relay's view of who has an open broadcast track, while + * presence is the speaker's self-reported state. The two + * usually agree but the announce flow is the more authoritative + * "is this broadcast really live" signal. + * + * Empty when the listener doesn't expose announces (IETF + * reference path); the UI falls back to `publishingNow` from + * presence in that case. + */ + private val _announcedSpeakers = MutableStateFlow>(emptySet()) + val announcedSpeakers: StateFlow> = _announcedSpeakers.asStateFlow() + /** * `true` once the local user has been kicked (#5) — the platform * layer flips this on a valid kind-4312 from a host/moderator and @@ -198,6 +216,7 @@ class AudioRoomViewModel( private var listener: NestsListener? = null private var connectJob: Job? = null private var stateObserverJob: Job? = null + private var announcesJob: Job? = null // Last in-flight `listener.close()` launched by teardown(). A subsequent // connect() awaits this before opening a fresh transport so two QUIC @@ -505,6 +524,31 @@ class AudioRoomViewModel( _uiState.update { it.copy(broadcast = targetState) } } + /** + * Drain the moq-lite ANNOUNCE flow into [announcedSpeakers]. + * Each Active emission adds the speaker pubkey; an inactive + * emission removes it. Best-effort — listeners that don't + * support announces (IETF reference path) throw on the first + * `collect` and we silently leave the set empty (the UI falls + * back to presence's `publishingNow` flag). + */ + private fun observeAnnounces(l: NestsListener) { + announcesJob?.cancel() + announcesJob = + viewModelScope.launch { + runCatching { + l.announces().collect { ann -> + if (closed) return@collect + if (ann.active) { + _announcedSpeakers.update { it + ann.pubkey } + } else { + _announcedSpeakers.update { it - ann.pubkey } + } + } + } + } + } + private fun observeListenerState(l: NestsListener) { stateObserverJob?.cancel() stateObserverJob = @@ -583,6 +627,7 @@ class AudioRoomViewModel( } listener = l observeListenerState(l) + observeAnnounces(l) } catch (ce: CancellationException) { throw ce } catch (t: Throwable) { @@ -746,6 +791,11 @@ class AudioRoomViewModel( connectJob = null stateObserverJob?.cancel() stateObserverJob = null + announcesJob?.cancel() + announcesJob = null + if (_announcedSpeakers.value.isNotEmpty()) { + _announcedSpeakers.value = emptySet() + } // Detach + suspend-stop each player on the cleanup scope. The // listener.close() below tears down the MoQ session and drops // every active subscription, so we don't need to call diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsListener.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsListener.kt index bdb94e8cd..51b61c843 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsListener.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsListener.kt @@ -23,11 +23,14 @@ package com.vitorpamplona.nestsclient import com.vitorpamplona.nestsclient.moq.MoqObject import com.vitorpamplona.nestsclient.moq.SubscribeHandle import com.vitorpamplona.nestsclient.moq.SubscribeOk +import com.vitorpamplona.nestsclient.moq.lite.MoqLiteAnnounceStatus import com.vitorpamplona.nestsclient.moq.lite.MoqLiteSession import com.vitorpamplona.nestsclient.moq.lite.MoqLiteSubscribeException +import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.asStateFlow +import kotlinx.coroutines.flow.flow import kotlinx.coroutines.flow.map import java.util.concurrent.atomic.AtomicLong @@ -108,6 +111,30 @@ class MoqLiteNestsListener internal constructor( ) } + override fun announces(): Flow = + flow { + check(state.value is NestsListenerState.Connected) { + "NestsListener.announces requires Connected state, was ${state.value}" + } + // Empty prefix → the relay sends every active broadcast + // under our session's namespace (one per speaker). The + // suffix is the broadcast's path component within the + // room — for nests this is the speaker pubkey hex. + val handle = session.announce(prefix = "") + try { + handle.updates.collect { announce -> + emit( + RoomAnnouncement( + pubkey = announce.suffix, + active = announce.status == MoqLiteAnnounceStatus.Active, + ), + ) + } + } finally { + runCatching { handle.close() } + } + } + override suspend fun close() { if (state.value is NestsListenerState.Closed) return runCatching { session.close() } diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsListener.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsListener.kt index 17c4684af..703674089 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsListener.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsListener.kt @@ -26,9 +26,11 @@ import com.vitorpamplona.nestsclient.moq.SubscribeFilter import com.vitorpamplona.nestsclient.moq.SubscribeHandle import com.vitorpamplona.nestsclient.moq.TrackNamespace import com.vitorpamplona.nestsclient.transport.WebTransportSession +import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.asStateFlow +import kotlinx.coroutines.flow.flow /** * High-level listener handle for an audio-room. Hides the layered HTTP + @@ -72,10 +74,49 @@ interface NestsListener { "subscribeCatalog is moq-lite-only; IETF listener has no catalog channel.", ) + /** + * Cold flow of moq-lite ANNOUNCE updates for the room's + * namespace. Each emission represents a publisher transitioning + * Active (broadcast started) or inactive (broadcast ended). + * Lets clients render an "actively broadcasting" indicator + * independent of the kind-10312 `publishing` presence tag — + * the JS reference's `useRoomAnnouncements` consumes this. + * + * Default body throws [UnsupportedOperationException] on the + * IETF reference path (`MoqSession` doesn't define an + * announce-prefix subscription that fits this shape). Callers + * mixing the two should `runCatching` the collect. + * + * @throws UnsupportedOperationException on the IETF listener. + * @throws IllegalStateException if the listener is not in [NestsListenerState.Connected]. + */ + fun announces(): Flow = + flow { + throw UnsupportedOperationException( + "announces() is moq-lite-only; IETF listener has no announce-prefix flow.", + ) + } + /** Tear down the MoQ session + underlying transport. Idempotent. */ suspend fun close() } +/** + * One moq-lite ANNOUNCE update. The JS reference uses these to + * drive the "live now" badge on each speaker's avatar. + * + * @param pubkey the suffix of the announce path — for nests + * audio rooms this is the speaker's pubkey hex (the broadcast + * path `/` minus the `` prefix the + * listener already established). + * @param active true on the `Active` status (broadcast came up); + * false on `Ended` (broadcast went down). + */ +data class RoomAnnouncement( + val pubkey: String, + val active: Boolean, +) + /** * Lifecycle states of a [NestsListener]. * diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt index 96eb98893..3403c4d69 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt @@ -28,6 +28,7 @@ import com.vitorpamplona.quartz.nip01Core.signers.NostrSigner import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Job import kotlinx.coroutines.delay +import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow @@ -35,6 +36,7 @@ import kotlinx.coroutines.flow.asSharedFlow import kotlinx.coroutines.flow.asStateFlow import kotlinx.coroutines.flow.collectLatest import kotlinx.coroutines.flow.first +import kotlinx.coroutines.flow.flow import kotlinx.coroutines.flow.onEach import kotlinx.coroutines.launch import kotlinx.coroutines.withTimeoutOrNull @@ -205,6 +207,32 @@ private class ReconnectingHandle( override suspend fun subscribeCatalog(speakerPubkeyHex: String): SubscribeHandle = reissuingSubscribe { listener -> listener.subscribeCatalog(speakerPubkeyHex) } + /** + * The announce flow is a cold per-collect stream rather than a + * SubscribeHandle, so it doesn't go through [reissuingSubscribe]. + * Instead, restart the inner collect on every activeListener + * change — `collectLatest` cancels the prior collection and + * re-runs against the fresh session. Best-effort: an inner + * listener that throws (IETF reference path) is logged-and- + * swallowed so a future moq-lite session keeps emitting. + */ + override fun announces(): Flow = + flow { + activeListener.collectLatest { listener -> + if (listener == null) return@collectLatest + val terminalOrConnected = + listener.state.first { state -> + state is NestsListenerState.Connected || + state is NestsListenerState.Closed || + state is NestsListenerState.Failed + } + if (terminalOrConnected !is NestsListenerState.Connected) return@collectLatest + runCatching { + listener.announces().collect { emit(it) } + } + } + } + /** * Open a SubscribeHandle whose `objects` flow survives session * swaps. Each time the wrapper opens a fresh inner listener, the diff --git a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/NestsListenerCatalogTest.kt b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/NestsListenerCatalogTest.kt index 9e4aa44d2..b9b1f1592 100644 --- a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/NestsListenerCatalogTest.kt +++ b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/NestsListenerCatalogTest.kt @@ -53,4 +53,19 @@ class NestsListenerCatalogTest { listener.subscribeCatalog("speakerPubkey") } } + + @Test + fun ietfDefaultListenerAnnouncesFlowThrowsOnCollect() = + runTest { + // The announce-prefix channel is moq-lite only as well — + // the IETF reference path's announces() default body + // throws UnsupportedOperationException at collect time. + // Callers mixing the two protocols can `runCatching` the + // collect to no-op cleanly. + val listener = IetfStyleListener() + val flow = listener.announces() + assertFailsWith { + flow.collect { /* unreachable */ } + } + } }