diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsListener.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsListener.kt index 48c0ea1dd..2a372a10a 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsListener.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsListener.kt @@ -59,16 +59,20 @@ class MoqLiteNestsListener internal constructor( ) : NestsListener { override val state: StateFlow = mutableState.asStateFlow() - override suspend fun subscribeSpeaker(speakerPubkeyHex: String): SubscribeHandle { + override suspend fun subscribeSpeaker(speakerPubkeyHex: String): SubscribeHandle = wrapSubscription(broadcast = speakerPubkeyHex, track = AUDIO_TRACK) + + override suspend fun subscribeCatalog(speakerPubkeyHex: String): SubscribeHandle = wrapSubscription(broadcast = speakerPubkeyHex, track = CATALOG_TRACK) + + private suspend fun wrapSubscription( + broadcast: String, + track: String, + ): SubscribeHandle { check(state.value is NestsListenerState.Connected) { - "NestsListener.subscribeSpeaker requires Connected state, was ${state.value}" + "NestsListener.subscribe requires Connected state, was ${state.value}" } val handle = try { - session.subscribe( - broadcast = speakerPubkeyHex, - track = AUDIO_TRACK, - ) + session.subscribe(broadcast = broadcast, track = track) } catch (e: MoqLiteSubscribeException) { // The IETF SubscribeHandle path conventionally surfaces // protocol-level rejections through the same exception diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsListener.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsListener.kt index 5754cec78..17c4684af 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsListener.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsListener.kt @@ -52,6 +52,26 @@ interface NestsListener { */ suspend fun subscribeSpeaker(speakerPubkeyHex: String): SubscribeHandle + /** + * Subscribe to a speaker's `catalog.json` track — moq-lite's + * canonical channel for "what is this broadcast publishing" JSON + * metadata (codec, sample rate, optional speaker-side hints). + * The publisher emits one JSON object per group; consumers + * typically read the latest frame and parse it. + * + * Throws on the IETF reference path because moq-transport-17 + * doesn't define a catalog convention. Callers that want + * cross-protocol behavior should `runCatching` this. + * + * @throws UnsupportedOperationException on the IETF listener. + * @throws MoqProtocolException if the publisher rejects the subscription. + * @throws IllegalStateException if the listener is not in [NestsListenerState.Connected]. + */ + suspend fun subscribeCatalog(speakerPubkeyHex: String): SubscribeHandle = + throw UnsupportedOperationException( + "subscribeCatalog is moq-lite-only; IETF listener has no catalog channel.", + ) + /** Tear down the MoQ session + underlying transport. Idempotent. */ suspend fun close() } diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt index 2faa674a0..27674720a 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt @@ -156,7 +156,20 @@ private class ReconnectingHandle( ) : NestsListener { override val state: StateFlow = mutableState.asStateFlow() - override suspend fun subscribeSpeaker(speakerPubkeyHex: String): SubscribeHandle { + override suspend fun subscribeSpeaker(speakerPubkeyHex: String): SubscribeHandle = reissuingSubscribe { listener -> listener.subscribeSpeaker(speakerPubkeyHex) } + + override suspend fun subscribeCatalog(speakerPubkeyHex: String): SubscribeHandle = reissuingSubscribe { listener -> listener.subscribeCatalog(speakerPubkeyHex) } + + /** + * Open a SubscribeHandle whose `objects` flow survives session + * swaps. Each time the wrapper opens a fresh inner listener, the + * pump cancels the prior subscription and calls [opener] against + * the new session, forwarding its frames into the consumer-facing + * [MutableSharedFlow]. Same shape for both `subscribeSpeaker` + * and `subscribeCatalog`; the only difference is which inner + * subscribe call to make. + */ + private fun reissuingSubscribe(opener: suspend (NestsListener) -> SubscribeHandle): SubscribeHandle { // Require a live (or just-connected) session at call time — // matches the existing IETF / moq-lite listener contract so // a caller can't accidentally subscribe in Idle and stall @@ -170,10 +183,6 @@ private class ReconnectingHandle( // Re-subscribe pump: every time activeListener changes, drop // the prior subscription (collectLatest cancels the inner // body) and open a new one against the fresh session. - // Wait for the inner listener to reach Connected before - // calling subscribeSpeaker — IETF + moq-lite listeners both - // throw IllegalStateException if subscribed before Connected, - // and a fresh session may go through Connecting first. val pumpJob = scope.launch { activeListener.collectLatest { listener -> @@ -186,36 +195,23 @@ private class ReconnectingHandle( } if (terminalOrConnected !is NestsListenerState.Connected) return@collectLatest val handle = - runCatching { listener.subscribeSpeaker(speakerPubkeyHex) } + runCatching { opener(listener) } .getOrNull() ?: return@collectLatest liveHandleRef.set(handle) try { handle.objects.collect { frames.emit(it) } } finally { - // Either the inner session ended, or we're - // being replaced because activeListener - // changed. Drop the dead reference; the - // inner unsubscribe is handled by the - // session's own teardown. if (liveHandleRef.get() === handle) liveHandleRef.set(null) } } } return SubscribeHandle( - // Synthetic ids — the consumer-facing handle is logical, - // not bound to any one session's wire ids. -1 is a - // reserved sentinel callers shouldn't compare against. subscribeId = -1L, trackAlias = -1L, ok = SYNTH_OK, objects = frames.asSharedFlow(), unsubscribeAction = { - // Capture the live underlying handle BEFORE cancelling - // the pump — the pump's finally clears `liveHandleRef` - // on cancellation, and we need a stable reference to - // forward the user-initiated unsubscribe to the live - // session's wire SUBSCRIBE_DONE. val live = liveHandleRef.getAndSet(null) pumpJob.cancel() live?.let { runCatching { it.unsubscribe() } } diff --git a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/NestsListenerCatalogTest.kt b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/NestsListenerCatalogTest.kt new file mode 100644 index 000000000..9e4aa44d2 --- /dev/null +++ b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/NestsListenerCatalogTest.kt @@ -0,0 +1,56 @@ +/* + * 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.nestsclient + +import com.vitorpamplona.nestsclient.moq.SubscribeHandle +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.StateFlow +import kotlinx.coroutines.flow.asStateFlow +import kotlinx.coroutines.test.runTest +import kotlin.test.Test +import kotlin.test.assertFailsWith + +class NestsListenerCatalogTest { + private class IetfStyleListener : NestsListener { + override val state: StateFlow = + MutableStateFlow(NestsListenerState.Idle).asStateFlow() + + override suspend fun subscribeSpeaker(speakerPubkeyHex: String): SubscribeHandle = throw UnsupportedOperationException("not under test") + + // Intentionally does NOT override subscribeCatalog so the + // interface's default body fires. + + override suspend fun close() = Unit + } + + @Test + fun ietfDefaultListenerRejectsCatalogSubscribe() = + runTest { + // The catalog channel is moq-lite only; the IETF listener + // path doesn't define one, so the interface default throws + // UnsupportedOperationException rather than silently + // returning a subscription that will never deliver data. + val listener = IetfStyleListener() + assertFailsWith { + listener.subscribeCatalog("speakerPubkey") + } + } +}