From 4e136cacf4a6563eaf3d308c0d84b195ce19782d Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 26 Apr 2026 17:24:17 +0000 Subject: [PATCH] feat(audio-rooms): MoqLiteSession listener-side (phase 5c-listener) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Listener-side moq-lite (Lite-03) session that wraps a WebTransportSession and exposes: - announce(prefix) — opens an Announce bidi with ControlType=Announce + AnnouncePlease(prefix), surfaces a flow of relay Announce updates (Active/Ended) - subscribe(broadcast, track, ...) — opens a Subscribe bidi with ControlType=Subscribe + body, reads the size-prefixed response, returns a MoqLiteSubscribeHandle whose frames flow yields each frame the publisher pushes - close() — cancels pumps, FINs every active subscribe bidi (moq-lite UNSUBSCRIBE is "FIN the bidi"), closes the transport Group demux: a single inbound-uni-stream pump per session reads the DataType byte (Group=0) + size-prefixed group header, then routes each subsequent (size, payload) frame to the matching subscriptionsBySubscribeId entry's frame channel. Speaker side stops at the announce/subscribe interface — moq-lite publisher flows depend on accept_bi() / server-initiated bidi streams (see clarifying agent summary in plans/2026-04-26-moq-lite-gap.md). That's a follow-up; the WebTransportSession interface needs an acceptBidiStream() method first. FakeWebTransport.openPeerUniStream() — new helper so tests can inject group data from the peer side without going through the session's outbound uni stream API. Tests in MoqLiteSessionTest — runBlocking-based (not runTest) so real-time channel coordination works without virtual-time artifacts: - subscribe_writes_request_and_returns_handle_on_ok - subscribe_throws_on_drop_response - announce_streams_relay_updates - groups_are_demuxed_by_subscribeId (no cross-talk) - unsubscribe_FINs_the_subscribe_bidi --- .../nestsclient/moq/lite/MoqLiteFraming.kt | 137 ++++++ .../nestsclient/moq/lite/MoqLiteSession.kt | 410 ++++++++++++++++++ .../nestsclient/transport/FakeWebTransport.kt | 34 ++ .../moq/lite/MoqLiteSessionTest.kt | 309 +++++++++++++ 4 files changed, 890 insertions(+) create mode 100644 nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteFraming.kt create mode 100644 nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt create mode 100644 nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSessionTest.kt diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteFraming.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteFraming.kt new file mode 100644 index 000000000..3050aecc2 --- /dev/null +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteFraming.kt @@ -0,0 +1,137 @@ +/* + * 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.moq.lite + +import com.vitorpamplona.nestsclient.moq.MoqCodecException +import com.vitorpamplona.quic.Varint +import kotlinx.coroutines.CancellationException + +/** + * Streaming frame buffer for moq-lite. Every wire datum on a control + * stream is a varint length followed by `length` bytes; every uni + * stream starts with a one-varint type byte then a size-prefixed + * group header then a sequence of size-prefixed frames until QUIC FIN. + * + * Network reads arrive as opaque chunks, so this class holds a rolling + * buffer and exposes: + * - [readVarint] — pull one varint when enough bytes have arrived + * - [readSizePrefixed] — pull one size-prefixed payload (varint + * length + payload bytes) when complete + * + * Both return null when more bytes are needed; the caller buffers + * additional chunks via [push] and retries. + * + * Not thread-safe — used from a single coroutine per stream. + */ +class MoqLiteFrameBuffer { + private var buf: ByteArray = ByteArray(0) + private var pos: Int = 0 + + /** Append [chunk] to the buffer, growing/compacting as needed. */ + fun push(chunk: ByteArray) { + if (chunk.isEmpty()) return + compact() + val needed = buf.size + chunk.size + // Power-of-two doubling matches MoqWriter's growth policy and + // avoids quadratic copies under bursty arrivals. + var newCap = if (buf.size == 0) 64 else buf.size + while (newCap < needed) newCap *= 2 + val grown = ByteArray(newCap) + buf.copyInto(grown, 0, 0, buf.size) + chunk.copyInto(grown, buf.size, 0, chunk.size) + buf = grown.copyOf(needed) + } + + /** Try to read one varint. Returns null if not enough bytes. */ + fun readVarint(): Long? { + val dec = Varint.decode(buf, pos) ?: return null + pos += dec.bytesConsumed + return dec.value + } + + /** + * Try to read one size-prefixed payload. Returns the payload bytes + * (no length prefix) or null if either the length varint or the + * payload itself isn't fully buffered yet. On invalid lengths + * (negative, > Int.MAX_VALUE) throws. + */ + fun readSizePrefixed(): ByteArray? { + val savedPos = pos + val len = readVarint() ?: return null + if (len < 0 || len > Int.MAX_VALUE) { + throw MoqCodecException("absurd moq-lite size prefix: $len") + } + if (pos + len.toInt() > buf.size) { + // Roll the cursor back so the next call sees the same + // varint and only commits when the whole payload arrives. + pos = savedPos + return null + } + val payload = buf.copyOfRange(pos, pos + len.toInt()) + pos += len.toInt() + return payload + } + + /** Bytes still buffered after [pos] — exposed for diagnostic / EOF detection. */ + val remaining: Int get() = buf.size - pos + + /** Drop the consumed prefix when half the buffer is dead weight. */ + private fun compact() { + if (pos == 0) return + if (pos < buf.size / 2) return + val live = buf.size - pos + val newBuf = ByteArray(live) + buf.copyInto(newBuf, 0, pos, buf.size) + buf = newBuf + pos = 0 + } +} + +/** + * Pull *one* size-prefixed message off a chunk channel. Drains chunks + * one at a time into [buffer] until [MoqLiteFrameBuffer.readSizePrefixed] + * returns a full payload. Returns null if [chunks] closes before a + * complete message arrives. + * + * Channel-based rather than Flow-based — `collect { throw EarlyExit }` + * is fragile under virtual-time test dispatchers and doesn't compose + * well with multiple sequential reads on the same channel. The caller + * owns the channel and is responsible for keeping it pumped (typically + * a single launched collector that forwards `incoming` chunks into the + * channel). + */ +internal suspend fun readOneSizePrefixedFrom( + chunks: kotlinx.coroutines.channels.ReceiveChannel, + buffer: MoqLiteFrameBuffer, +): ByteArray? { + while (true) { + buffer.readSizePrefixed()?.let { return it } + val next = + try { + chunks.receive() + } catch (ce: CancellationException) { + throw ce + } catch (_: kotlinx.coroutines.channels.ClosedReceiveChannelException) { + return null + } + buffer.push(next) + } +} 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 new file mode 100644 index 000000000..00b0fdffc --- /dev/null +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt @@ -0,0 +1,410 @@ +/* + * 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.moq.lite + +import com.vitorpamplona.nestsclient.moq.MoqCodecException +import com.vitorpamplona.nestsclient.transport.WebTransportSession +import com.vitorpamplona.quic.Varint +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Job +import kotlinx.coroutines.cancelAndJoin +import kotlinx.coroutines.channels.BufferOverflow +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.MutableSharedFlow +import kotlinx.coroutines.flow.consumeAsFlow +import kotlinx.coroutines.flow.firstOrNull +import kotlinx.coroutines.launch +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock + +/** + * Listener-side moq-lite (Lite-03) session. Wraps a connected + * [WebTransportSession] and exposes: + * + * - [announce] — open an Announce bidi with a prefix, observe live + * `Active` / `Ended` updates from the relay. + * - [subscribe] — open a Subscribe bidi for a `(broadcast, track)` + * pair, await SubscribeOk, return a [MoqLiteSubscribeHandle] whose + * [MoqLiteSubscribeHandle.frames] yields each frame the publisher + * pushes. + * + * Speaker-side is **not yet implemented** — the publisher direction + * needs server-initiated bidi acceptance (relay → us) which the + * current `WebTransportSession` interface does not surface. Tracked in + * `nestsClient/plans/2026-04-26-moq-lite-gap.md` phase-5c-speaker. + * + * Wire-protocol scope: Lite-03 — no SETUP, no datagrams, one fresh + * client-initiated bidi per request, group data on uni streams. See + * the gap doc for the byte-level layout. + */ +class MoqLiteSession internal constructor( + private val transport: WebTransportSession, + private val scope: CoroutineScope, +) { + private val state = Mutex() + private val subscriptionsBySubscribeId: MutableMap = HashMap() + private var nextSubscribeId: Long = 0L + private var groupPump: Job? = null + + @Volatile private var closed: Boolean = false + + val isClosed: Boolean get() = closed + + /** + * Open an Announce bidi against the relay. Sends + * `ControlType=Announce` + `AnnouncePlease(prefix)`, then surfaces + * a flow of [MoqLiteAnnounce] updates the relay streams back. + * + * The returned [MoqLiteAnnouncesHandle] keeps the bidi open until + * [MoqLiteAnnouncesHandle.close] is called. FINing the bidi is + * how moq-lite signals "no longer interested" — there is no + * ANNOUNCE_CANCEL message. + */ + suspend fun announce(prefix: String): MoqLiteAnnouncesHandle { + ensureOpen() + val bidi = transport.openBidiStream() + bidi.write(Varint.encode(MoqLiteControlType.Announce.code)) + bidi.write(MoqLiteCodec.encodeAnnouncePlease(MoqLiteAnnouncePlease(prefix))) + + val updates = MutableSharedFlow(replay = 0, extraBufferCapacity = 64) + 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.decodeAnnounce(payload)) + } + } + } catch (ce: CancellationException) { + throw ce + } catch (_: Throwable) { + // Flow terminated (peer FIN or transport close). + // The Announce stream's emit-side just stops; consumers + // see an end-of-flow. + } + } + return MoqLiteAnnouncesHandle( + updates = updates, + close = { + runCatching { bidi.finish() } + pump.cancelAndJoin() + }, + ) + } + + /** + * Open a Subscribe bidi for `(broadcast, track)`. Sends + * `ControlType=Subscribe` + the [MoqLiteSubscribe] body, awaits + * the [MoqLiteCodec.SubscribeResponse] reply, and on Ok returns a + * [MoqLiteSubscribeHandle] whose `frames` flow yields each frame + * the publisher pushes through the relay (one [MoqLiteFrame] per + * payload, bundled into [MoqLiteFrame] structs that carry the + * group sequence). + * + * Throws [MoqLiteSubscribeException] if the publisher / relay + * replies with Drop, or if the response stream tears down before + * any reply arrives. + */ + suspend fun subscribe( + broadcast: String, + track: String, + priority: Int = DEFAULT_PRIORITY, + ordered: Boolean = true, + maxLatencyMillis: Long = 0L, + startGroup: Long? = null, + endGroup: Long? = null, + ): MoqLiteSubscribeHandle { + ensureOpen() + val id = + state.withLock { + check(!closed) { "session is closed" } + val next = nextSubscribeId++ + next + } + val request = + MoqLiteSubscribe( + id = id, + broadcast = broadcast, + track = track, + priority = priority, + ordered = ordered, + maxLatencyMillis = maxLatencyMillis, + startGroup = startGroup, + endGroup = endGroup, + ) + val bidi = transport.openBidiStream() + bidi.write(Varint.encode(MoqLiteControlType.Subscribe.code)) + bidi.write(MoqLiteCodec.encodeSubscribe(request)) + + // moq-lite's subscribe-response is a single size-prefixed + // message on the response side of the bidi. Read incoming + // chunks into a buffer until the buffer holds a full payload, + // then stop. We don't need a separate collector pump because + // post-Ok the bidi is idle (group data flows on its own uni + // streams). + val responseBuffer = MoqLiteFrameBuffer() + val responsePayload = readSubscribeResponseFromBidi(bidi.incoming(), responseBuffer, id) + when (val resp = MoqLiteCodec.decodeSubscribeResponse(responsePayload)) { + is MoqLiteCodec.SubscribeResponse.Dropped -> { + throw MoqLiteSubscribeException( + "publisher rejected subscribe id=$id: errorCode=${resp.drop.errorCode} " + + "reason='${resp.drop.reasonPhrase}'", + ) + } + + is MoqLiteCodec.SubscribeResponse.Ok -> { + val frames = Channel(capacity = DEFAULT_FRAME_BUFFER, onBufferOverflow = BufferOverflow.DROP_OLDEST) + val sub = + ListenerSubscription( + id = id, + request = request, + ok = resp.ok, + bidi = bidi, + frames = frames, + ) + state.withLock { + subscriptionsBySubscribeId[id] = sub + if (groupPump == null) groupPump = scope.launch { pumpUniStreams() } + } + return MoqLiteSubscribeHandle( + id = id, + ok = resp.ok, + frames = frames.consumeAsFlow(), + unsubscribeAction = { unsubscribe(id) }, + ) + } + } + } + + /** + * Drain inbound uni streams and route each one's group frames to + * the matching subscription. The relay opens a fresh uni stream + * per (subscribe id, group sequence); the first byte is + * [MoqLiteDataType] (0 = Group), followed by a size-prefixed + * [MoqLiteGroupHeader], followed by `varint(size) + payload` + * frames until QUIC FIN. + * + * One pump per session — started lazily on the first subscribe. + */ + private suspend fun pumpUniStreams() { + try { + transport.incomingUniStreams().collect { stream -> + scope.launch { drainOneGroup(stream) } + } + } catch (ce: CancellationException) { + throw ce + } catch (_: Throwable) { + // Transport closed — subscriptions will surface end-of-flow + // via their own bidi pumps as well. + } + } + + private suspend fun drainOneGroup(stream: com.vitorpamplona.nestsclient.transport.WebTransportReadStream) { + val buffer = MoqLiteFrameBuffer() + var typeRead = false + var subscribeId: Long = -1L + var groupSequence: Long = -1L + var headerRead = false + try { + stream.incoming().collect { chunk -> + buffer.push(chunk) + if (!typeRead) { + val type = buffer.readVarint() ?: return@collect + if (type != MoqLiteDataType.Group.code) { + throw MoqCodecException("unknown moq-lite uni-stream type code: $type") + } + typeRead = true + } + if (!headerRead) { + val payload = buffer.readSizePrefixed() ?: return@collect + val hdr = MoqLiteCodec.decodeGroupHeader(payload) + subscribeId = hdr.subscribeId + groupSequence = hdr.sequence + headerRead = true + } + while (true) { + val frame = buffer.readSizePrefixed() ?: break + val sub = state.withLock { subscriptionsBySubscribeId[subscribeId] } + if (sub != null) { + sub.frames.trySend( + MoqLiteFrame( + groupSequence = groupSequence, + payload = frame, + ), + ) + } + // If the subscription has been closed already we + // silently drop the frame — the publisher hasn't + // observed the unsubscribe yet (its uni streams + // are independent of our bidi FIN). + } + } + } catch (ce: CancellationException) { + throw ce + } catch (_: Throwable) { + // Stream errored / FIN'd. Nothing to do — the next group + // arrives on a fresh uni stream. + } + } + + private suspend fun unsubscribe(id: Long) { + val sub = + state.withLock { subscriptionsBySubscribeId.remove(id) } ?: return + runCatching { sub.bidi.finish() } + sub.frames.close() + } + + suspend fun close() { + if (closed) return + closed = true + val toClose: List + state.withLock { + toClose = subscriptionsBySubscribeId.values.toList() + subscriptionsBySubscribeId.clear() + } + for (sub in toClose) { + runCatching { sub.bidi.finish() } + sub.frames.close() + } + groupPump?.cancelAndJoin() + runCatching { transport.close() } + } + + private fun ensureOpen() { + check(!closed) { "MoqLiteSession is closed" } + } + + /** + * Read the single size-prefixed subscribe response off a fresh + * subscribe bidi. moq-lite's subscribe-response is one wire write, + * so the first chunk we receive contains the entire payload (matches + * what the IETF [com.vitorpamplona.nestsclient.moq.MoqSession] does + * during its SETUP read). + * + * Throws [MoqLiteSubscribeException] if the bidi closes before any + * chunk arrives. + */ + private suspend fun readSubscribeResponseFromBidi( + incoming: kotlinx.coroutines.flow.Flow, + buffer: MoqLiteFrameBuffer, + id: Long, + ): ByteArray { + val chunk = + incoming.firstOrNull() + ?: throw MoqLiteSubscribeException("subscribe stream FIN before reply for id=$id") + buffer.push(chunk) + return buffer.readSizePrefixed() + ?: throw MoqLiteSubscribeException( + "first chunk on subscribe response did not carry a complete size-prefixed payload " + + "(id=$id, chunkSize=${chunk.size})", + ) + } + + private class ListenerSubscription( + val id: Long, + val request: MoqLiteSubscribe, + val ok: MoqLiteSubscribeOk, + val bidi: com.vitorpamplona.nestsclient.transport.WebTransportBidiStream, + val frames: Channel, + ) + + companion object { + /** moq-lite priority byte midpoint — neutral default. */ + const val DEFAULT_PRIORITY: Int = 0x80 + + /** + * Per-subscription channel buffer for inbound frames. 128 audio + * frames at Opus 20 ms ≈ 2.5 s of backlog before DROP_OLDEST, + * matching a real-time listener's tolerance. + */ + const val DEFAULT_FRAME_BUFFER: Int = 128 + + /** + * Attach to a [WebTransportSession] in the client role. Lite-03 + * has no SETUP — the WT handshake itself is the handshake — so + * this returns a ready-to-use session immediately. + * + * @param pumpScope where the inbound-uni-stream + per-bidi pumps + * live. Production code passes the owning ViewModel scope. + */ + fun client( + transport: WebTransportSession, + pumpScope: CoroutineScope, + ): MoqLiteSession = MoqLiteSession(transport, pumpScope) + } +} + +/** + * One frame received from a subscription. moq-lite's wire format + * carries no per-frame envelope beyond the size; [groupSequence] is + * pulled from the group header so consumers can detect group rollover + * (e.g. for keyframe boundaries). + */ +data class MoqLiteFrame( + val groupSequence: Long, + val payload: ByteArray, +) { + override fun equals(other: Any?): Boolean { + if (this === other) return true + if (other !is MoqLiteFrame) return false + return groupSequence == other.groupSequence && payload.contentEquals(other.payload) + } + + override fun hashCode(): Int = 31 * groupSequence.hashCode() + payload.contentHashCode() +} + +/** + * Active subscription handle returned by [MoqLiteSession.subscribe]. + * [frames] emits every frame the publisher pushes; [unsubscribe] + * FINs the bidi to signal "no longer interested" (moq-lite has no + * UNSUBSCRIBE message — FIN is the protocol). + */ +class MoqLiteSubscribeHandle internal constructor( + val id: Long, + val ok: MoqLiteSubscribeOk, + val frames: Flow, + private val unsubscribeAction: suspend () -> Unit, +) { + suspend fun unsubscribe() = unsubscribeAction() +} + +/** + * Active announce-discovery handle returned by [MoqLiteSession.announce]. + * [updates] emits every [MoqLiteAnnounce] update the relay streams + * back; [close] FINs the bidi to stop receiving updates. + */ +class MoqLiteAnnouncesHandle 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, + cause: Throwable? = null, +) : RuntimeException(message, cause) diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/transport/FakeWebTransport.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/transport/FakeWebTransport.kt index d24aefddf..6cc32c967 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/transport/FakeWebTransport.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/transport/FakeWebTransport.kt @@ -46,6 +46,7 @@ class FakeWebTransport private constructor( private val outboundBidiStreams: Channel, private val inboundBidiStreams: Channel, private val inboundUniStreams: Channel, + private val outboundUniStreams: Channel, ) : WebTransportSession { private val stateLock = Mutex() private var open = true @@ -86,6 +87,7 @@ class FakeWebTransport private constructor( outboundBidiStreams.close() inboundBidiStreams.close() inboundUniStreams.close() + outboundUniStreams.close() } /** @@ -94,6 +96,19 @@ class FakeWebTransport private constructor( */ fun peerOpenedBidiStreams(): Flow = inboundBidiStreams.receiveAsFlow() + /** + * Open a uni stream from this side toward the paired peer. Tests use this + * to simulate a publisher pushing a moq-lite group uni stream — write the + * full sequence of chunks via [WebTransportWriteStream.write] then call + * [WebTransportWriteStream.finish] to FIN. + */ + suspend fun openPeerUniStream(): WebTransportWriteStream { + stateLock.withLock { check(open) { "session closed" } } + val pipe = Channel(Channel.BUFFERED) + outboundUniStreams.send(FakeReadStream(pipe)) + return ChannelWriteStream(pipe) + } + companion object { /** * Create two linked fakes that act as "client" and "server" endpoints of @@ -115,6 +130,7 @@ class FakeWebTransport private constructor( outboundBidiStreams = aToBBidi, inboundBidiStreams = bToABidi, inboundUniStreams = bToAUni, + outboundUniStreams = aToBUni, ) val b = FakeWebTransport( @@ -123,6 +139,7 @@ class FakeWebTransport private constructor( outboundBidiStreams = bToABidi, inboundBidiStreams = aToBBidi, inboundUniStreams = aToBUni, + outboundUniStreams = bToAUni, ) return a to b } @@ -149,3 +166,20 @@ class FakeReadStream internal constructor( ) : WebTransportReadStream { override fun incoming(): Flow = read.receiveAsFlow() } + +/** + * Write-only adapter over a [Channel]. Used by [FakeWebTransport.openPeerUniStream] + * so a test can drive a peer-initiated uni stream by writing chunks then + * FIN'ing via [finish]. + */ +private class ChannelWriteStream( + private val channel: Channel, +) : WebTransportWriteStream { + override suspend fun write(chunk: ByteArray) { + channel.send(chunk) + } + + override suspend fun finish() { + channel.close() + } +} 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 new file mode 100644 index 000000000..3542aee09 --- /dev/null +++ b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSessionTest.kt @@ -0,0 +1,309 @@ +/* + * 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.moq.lite + +import com.vitorpamplona.nestsclient.transport.FakeBidiStream +import com.vitorpamplona.nestsclient.transport.FakeWebTransport +import com.vitorpamplona.quic.Varint +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.async +import kotlinx.coroutines.cancelAndJoin +import kotlinx.coroutines.flow.first +import kotlinx.coroutines.flow.take +import kotlinx.coroutines.flow.toList +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withTimeout +import kotlin.test.AfterTest +import kotlin.test.Test +import kotlin.test.assertContentEquals +import kotlin.test.assertEquals +import kotlin.test.assertFailsWith + +/** + * Drives a [MoqLiteSession] from the listener side against a fake peer + * that plays the relay role. Validates: + * + * - subscribe() writes the right ControlType + body, blocks until the + * peer's SubscribeOk arrives, then yields a SubscribeHandle whose + * frames flow surfaces every incoming group. + * - subscribe() throws MoqLiteSubscribeException on a Drop response. + * - announce() writes the AnnouncePlease and surfaces every Announce + * update the peer streams back. + * - Group uni streams are demuxed by subscribeId — frames for sub A + * never leak into sub B. + * + * No transport mock magic — these tests use the production codec to + * encode/decode bytes, so a wire-shape regression on either side fails + * here. + */ +class MoqLiteSessionTest { + /** Real-time scope for session pumps. Cancelled in [tearDown] so each + * test ends with a clean coroutine ledger. + */ + private val supervisor = SupervisorJob() + private val pumpScope = CoroutineScope(supervisor) + + @AfterTest + fun tearDown() = + runBlocking { + supervisor.cancelAndJoin() + } + + @Test + fun subscribe_writes_request_and_returns_handle_on_ok() = + runBlocking { + val (clientSide, serverSide) = FakeWebTransport.pair() + val session = MoqLiteSession.client(clientSide, pumpScope) + + val peerHandlesSubscribe = + async { + val bidi = serverSide.peerOpenedBidiStreams().first() + val req = readSubscribeRequest(bidi) + assertEquals("speakerPubkey", req.broadcast) + assertEquals("audio/data", req.track) + assertEquals(MoqLiteSession.DEFAULT_PRIORITY, req.priority) + val ok = + MoqLiteSubscribeOk( + priority = req.priority, + ordered = req.ordered, + maxLatencyMillis = req.maxLatencyMillis, + startGroup = null, + endGroup = null, + ) + bidi.write(MoqLiteCodec.encodeSubscribeOk(ok)) + req + } + + val handle = session.subscribe("speakerPubkey", "audio/data") + val request = peerHandlesSubscribe.await() + assertEquals(request.id, handle.id) + assertEquals(0L, handle.id, "first subscribe id is 0") + + // Now push one group with two frames from the server side. + val uni = serverSide.openPeerUniStream() + uni.write(Varint.encode(MoqLiteDataType.Group.code)) + uni.write(MoqLiteCodec.encodeGroupHeader(MoqLiteGroupHeader(subscribeId = handle.id, sequence = 7L))) + uni.write(framePayload(byteArrayOf(0x10, 0x11))) + uni.write(framePayload(byteArrayOf(0x20, 0x21))) + uni.finish() + + val frames = + withTimeout(2_000) { + handle.frames.take(2).toList() + } + assertEquals(2, frames.size) + assertEquals(7L, frames[0].groupSequence) + assertEquals(7L, frames[1].groupSequence) + assertContentEquals(byteArrayOf(0x10, 0x11), frames[0].payload) + assertContentEquals(byteArrayOf(0x20, 0x21), frames[1].payload) + + session.close() + } + + @Test + fun subscribe_throws_on_drop_response() = + runBlocking { + val (clientSide, serverSide) = FakeWebTransport.pair() + val session = MoqLiteSession.client(clientSide, pumpScope) + + val peer = + async { + val bidi = serverSide.peerOpenedBidiStreams().first() + readSubscribeRequest(bidi) + bidi.write( + MoqLiteCodec.encodeSubscribeDrop( + MoqLiteSubscribeDrop(errorCode = 4L, reasonPhrase = "no such broadcast"), + ), + ) + } + + assertFailsWith { + session.subscribe("nope", "audio/data") + } + peer.await() + session.close() + } + + @Test + fun announce_streams_relay_updates() = + runBlocking { + val (clientSide, serverSide) = FakeWebTransport.pair() + val session = MoqLiteSession.client(clientSide, pumpScope) + + val peer = + async { + val bidi = serverSide.peerOpenedBidiStreams().first() + val chunks = bidi.incoming().take(2).toList() + val controlByte = MoqLiteFrameBuffer().apply { push(chunks[0]) }.readVarint() + assertEquals(MoqLiteControlType.Announce.code, controlByte) + val plea = + MoqLiteFrameBuffer().apply { push(chunks[1]) }.readSizePrefixed() + ?: error("AnnouncePlease chunk did not contain a complete size-prefixed payload") + assertEquals("nests/30312:abc:room", MoqLiteCodec.decodeAnnouncePlease(plea).prefix) + + // Send two Announce updates back. + bidi.write( + MoqLiteCodec.encodeAnnounce( + MoqLiteAnnounce( + MoqLiteAnnounceStatus.Active, + "speakerOne", + hops = 1L, + ), + ), + ) + bidi.write( + MoqLiteCodec.encodeAnnounce( + MoqLiteAnnounce( + MoqLiteAnnounceStatus.Active, + "speakerTwo", + hops = 1L, + ), + ), + ) + } + + val announces = session.announce("nests/30312:abc:room") + val updates = withTimeout(2_000) { announces.updates.take(2).toList() } + peer.await() + + assertEquals(2, updates.size) + assertEquals("speakerOne", updates[0].suffix) + assertEquals(MoqLiteAnnounceStatus.Active, updates[0].status) + assertEquals("speakerTwo", updates[1].suffix) + + announces.close() + session.close() + } + + @Test + fun groups_are_demuxed_by_subscribeId() = + runBlocking { + val (clientSide, serverSide) = FakeWebTransport.pair() + val session = MoqLiteSession.client(clientSide, pumpScope) + + // Set up two parallel subscriptions — peer accepts both, replies + // with Ok, then pushes one group per subscription out of order. + val subAck = + async { + val bidiA = serverSide.peerOpenedBidiStreams().first() + readSubscribeRequest(bidiA) + bidiA.write(MoqLiteCodec.encodeSubscribeOk(okFor(0L))) + bidiA + } + val handleA = session.subscribe("speakerA", "audio/data") + subAck.await() + + val subAck2 = + async { + val bidiB = serverSide.peerOpenedBidiStreams().first() + readSubscribeRequest(bidiB) + bidiB.write(MoqLiteCodec.encodeSubscribeOk(okFor(1L))) + bidiB + } + val handleB = session.subscribe("speakerB", "audio/data") + subAck2.await() + + assertEquals(0L, handleA.id) + assertEquals(1L, handleB.id) + + // Push one group for A with payload "a", one for B with payload "b". + val uniB = serverSide.openPeerUniStream() + uniB.write(Varint.encode(MoqLiteDataType.Group.code)) + uniB.write(MoqLiteCodec.encodeGroupHeader(MoqLiteGroupHeader(subscribeId = handleB.id, sequence = 0L))) + uniB.write(framePayload("b".encodeToByteArray())) + uniB.finish() + + val uniA = serverSide.openPeerUniStream() + uniA.write(Varint.encode(MoqLiteDataType.Group.code)) + uniA.write(MoqLiteCodec.encodeGroupHeader(MoqLiteGroupHeader(subscribeId = handleA.id, sequence = 0L))) + uniA.write(framePayload("a".encodeToByteArray())) + uniA.finish() + + val fromA = withTimeout(2_000) { handleA.frames.first() } + val fromB = withTimeout(2_000) { handleB.frames.first() } + assertContentEquals("a".encodeToByteArray(), fromA.payload) + assertContentEquals("b".encodeToByteArray(), fromB.payload) + + handleA.unsubscribe() + handleB.unsubscribe() + session.close() + } + + @Test + fun unsubscribe_FINs_the_subscribe_bidi() = + runBlocking { + val (clientSide, serverSide) = FakeWebTransport.pair() + val session = MoqLiteSession.client(clientSide, pumpScope) + + var peerBidi: FakeBidiStream? = null + val peer = + async { + val bidi = serverSide.peerOpenedBidiStreams().first() + readSubscribeRequest(bidi) + bidi.write(MoqLiteCodec.encodeSubscribeOk(okFor(0L))) + peerBidi = bidi + // Drain whatever the listener writes after Ok — moq-lite + // unsubscribe is a FIN, which surfaces here as the + // incoming flow completing. + bidi.incoming().toList() + } + + val handle = session.subscribe("speakerX", "audio/data") + handle.unsubscribe() + withTimeout(2_000) { peer.await() } + assertEquals(true, peerBidi != null, "peer received the bidi") + + session.close() + } + + // ---------- helpers ---------- + + /** + * Read a Subscribe request from the peer side of the bidi. Each + * `bidi.write` on the session side becomes one channel send (so + * one chunk on this end), and the session writes ControlType then + * the encoded body — exactly two chunks. Pull both with a single + * [take]+[toList] and parse them, no streaming buffer needed. + */ + private suspend fun readSubscribeRequest(bidi: FakeBidiStream): MoqLiteSubscribe { + val chunks = bidi.incoming().take(2).toList() + check(chunks.size == 2) { "expected 2 chunks (control byte + body), got ${chunks.size}" } + val controlByte = MoqLiteFrameBuffer().apply { push(chunks[0]) }.readVarint() + assertEquals(MoqLiteControlType.Subscribe.code, controlByte) + val payload = + MoqLiteFrameBuffer().apply { push(chunks[1]) }.readSizePrefixed() + ?: error("subscribe body chunk did not contain a complete size-prefixed payload") + return MoqLiteCodec.decodeSubscribe(payload) + } + + private fun framePayload(bytes: ByteArray): ByteArray = Varint.encode(bytes.size.toLong()) + bytes + + private fun okFor( + @Suppress("UNUSED_PARAMETER") id: Long, + ) = MoqLiteSubscribeOk( + priority = MoqLiteSession.DEFAULT_PRIORITY, + ordered = true, + maxLatencyMillis = 0L, + startGroup = null, + endGroup = null, + ) +}