diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteHandles.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteHandles.kt index 9f69f1f00..86ecbbee8 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteHandles.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteHandles.kt @@ -54,6 +54,28 @@ class MoqLiteAnnouncesHandle internal constructor( suspend fun close() = close.invoke() } +/** + * Active probe handle returned by [MoqLiteSession.probe]. Mirrors + * kixelated's `Subscriber::run_probe_stream` + * (`rs/moq-lite/src/lite/subscriber.rs`): subscriber opens a bidi, + * writes `ControlType::Probe`, then reads size-prefixed + * [MoqLiteProbe] messages indefinitely until the publisher FINs or + * the consumer calls [close]. + * + * [updates] is a cold flow that emits every Probe message the + * publisher pushes — typically a single bitrate hint at session + * start, but a publisher running an ABR codec MAY emit multiple + * over time. For a fixed-rate Opus producer (which is what + * Amethyst's nests speaker is) the publisher emits one and FINs; + * the flow then completes naturally and consumers see end-of-flow. + */ +class MoqLiteProbeHandle internal constructor( + val updates: Flow, + private val close: suspend () -> Unit, +) { + suspend fun close() = close.invoke() +} + /** Thrown when subscribe is rejected (Drop) or the response stream dies. */ class MoqLiteSubscribeException( message: String, diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt index 4749f5f15..40282f6bf 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt @@ -691,6 +691,67 @@ class MoqLiteSession internal constructor( sub.frames.close() } + /** + * Open a Probe bidi against the relay / publisher (audit L3). + * Mirrors kixelated's `Subscriber::run_probe_stream` + * (`rs/moq-lite/src/lite/subscriber.rs`): subscriber writes + * `ControlType::Probe` (varint 4), then reads size-prefixed + * [MoqLiteProbe] messages until the publisher FINs. + * + * The publisher reciprocates by writing one or more [MoqLiteProbe] + * messages indicating its expected outbound bitrate. Amethyst's + * own publisher path is fixed-rate Opus and emits exactly one + * Probe before FIN, but third-party (e.g. browser-side + * `kixelated/moq` watcher → relay-served broadcast) publishers + * MAY emit multiple over time as their ABR estimates change. + * + * The returned handle's `updates` flow is cold — collecting it + * starts the read pump on this session's scope. On collector + * cancellation OR [MoqLiteProbeHandle.close], the bidi FINs and + * the pump exits. + * + * No application-level use of probe data exists in nests today + * (we don't run ABR), but the API completes the moq-lite Lite-03 + * protocol surface. Useful for diagnostic tools that want to + * read a publisher's bitrate without subscribing to its data + * track. + */ + suspend fun probe(): MoqLiteProbeHandle { + ensureOpen() + val bidi = transport.openBidiStream() + bidi.write(Varint.encode(MoqLiteControlType.Probe.code)) + val updates = + MutableSharedFlow( + replay = 8, + onBufferOverflow = BufferOverflow.DROP_OLDEST, + ) + val pump = + scope.launch { + val buffer = MoqLiteFrameBuffer() + try { + bidi.incoming().collect { chunk -> + buffer.push(chunk) + while (true) { + val payload = buffer.readSizePrefixed() ?: break + updates.emit(MoqLiteCodec.decodeProbe(payload)) + } + } + } catch (ce: CancellationException) { + throw ce + } catch (_: Throwable) { + // Probe bidi died — best-effort silent. The + // updates flow simply stops emitting. + } + } + return MoqLiteProbeHandle( + updates = updates, + close = { + runCatching { bidi.finish() } + pump.cancelAndJoin() + }, + ) + } + // ==================================================================== // Publisher side // ==================================================================== diff --git a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSessionTest.kt b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSessionTest.kt index cf09542b8..3e419fee4 100644 --- a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSessionTest.kt +++ b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSessionTest.kt @@ -940,6 +940,50 @@ class MoqLiteSessionTest { session.close() } + @Test + fun subscriber_probe_writes_control_type_and_decodes_publisher_bitrate() = + runBlocking { + // Lite-03 audit L3: the subscriber side opens a Probe bidi + // by writing ControlType=Probe and reads size-prefixed + // MoqLiteProbe messages from the publisher. Mirrors + // kixelated's `Subscriber::run_probe_stream` + // (`rs/moq-lite/src/lite/subscriber.rs`). + val (clientSide, serverSide) = FakeWebTransport.pair() + val session = MoqLiteSession.client(clientSide, pumpScope) + + // Server side: read the subscriber's control byte, then + // write two Probe messages. + val serverPump = + async { + val bidi = serverSide.peerOpenedBidiStreams().first() + val buf = MoqLiteFrameBuffer() + // Read the leading ControlType varint. The session + // writes it as a single chunk via bidi.write. + val ctChunk = withTimeout(2_000) { bidi.incoming().first() } + buf.push(ctChunk) + val ct = buf.readVarint() + assertEquals(MoqLiteControlType.Probe.code, ct, "subscriber writes ControlType=Probe") + + // Push two probes (different bitrates) then FIN. + bidi.write(MoqLiteCodec.encodeProbe(MoqLiteProbe(bitrate = 32_000L))) + bidi.write(MoqLiteCodec.encodeProbe(MoqLiteProbe(bitrate = 64_000L))) + bidi.finish() + } + + val handle = session.probe() + // Drain both probes from the cold flow. SharedFlow with + // replay=8 means we can collect after the publisher has + // already pushed without losing emissions. + val received = withTimeout(2_000) { handle.updates.take(2).toList() } + assertEquals(2, received.size) + assertEquals(32_000L, received[0].bitrate) + assertEquals(64_000L, received[1].bitrate) + + serverPump.await() + handle.close() + session.close() + } + @Test fun unsubscribe_FINs_the_subscribe_bidi() = runBlocking {