diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqSession.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqSession.kt index 745be13ce..acf8469fb 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqSession.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqSession.kt @@ -22,7 +22,17 @@ package com.vitorpamplona.nestsclient.moq import com.vitorpamplona.nestsclient.transport.WebTransportBidiStream import com.vitorpamplona.nestsclient.transport.WebTransportSession +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Job +import kotlinx.coroutines.channels.BufferOverflow +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.first +import kotlinx.coroutines.flow.receiveAsFlow +import kotlinx.coroutines.launch +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock import kotlinx.coroutines.withTimeout /** @@ -44,22 +54,25 @@ object MoqVersion { /** * Session wrapper over a [WebTransportSession] that speaks MoQ-transport. * - * Phase 3c-1 ships only [setup] — the CLIENT_SETUP / SERVER_SETUP handshake. - * Phase 3c-2 adds SUBSCRIBE + ANNOUNCE + object-stream multiplexing. - * - * Usage: - * ``` - * val session = MoqSession.client(webTransport).apply { - * setup(listOf(MoqVersion.DRAFT_17)) - * } - * // session.selectedVersion is now the version the server picked. - * ``` + * Lifecycle: + * 1. [client] / [server] attaches to a transport. No traffic yet. + * 2. [setup] runs the SETUP handshake synchronously. After it returns + * successfully, two background pumps start in [pumpScope]: + * * a control-stream pump that buffers chunks, decodes complete frames, + * and dispatches each (currently SUBSCRIBE_OK / SUBSCRIBE_ERROR); + * * a datagram pump that decodes OBJECT_DATAGRAMs and routes them to + * the matching per-track flow. + * 3. [subscribe] sends a SUBSCRIBE, awaits the OK, returns a + * [SubscribeHandle] whose [SubscribeHandle.objects] flow yields every + * OBJECT_DATAGRAM tagged with that subscription's track alias. + * 4. [close] cancels the pumps and closes the transport. */ class MoqSession private constructor( private val transport: WebTransportSession, /** The single bidi stream every MoQ exchange uses for control. */ private val controlStream: WebTransportBidiStream, private val role: Role, + private val pumpScope: CoroutineScope, ) { enum class Role { Client, Server } @@ -71,15 +84,39 @@ class MoqSession private constructor( var serverParameters: List = emptyList() private set + private val stateMutex = Mutex() + private val writeMutex = Mutex() + private var nextSubscribeId = 0L + + /** subscribe_id → CompletableDeferred awaiting SUBSCRIBE_OK / SUBSCRIBE_ERROR. */ + private val pendingSubscribes = HashMap>() + + /** track_alias → bounded channel that fans matching OBJECT_DATAGRAMs to the subscriber. + * + * We use a [Channel] with [BufferOverflow.DROP_OLDEST] (rather than a + * [MutableSharedFlow]) because objects must be buffered between + * [subscribe] returning and the caller attaching to [SubscribeHandle.objects], + * even when no collector exists yet. SharedFlow with `replay=0` drops + * pre-subscription emissions, which races real-time audio delivery. + */ + private val sinks = HashMap>() + + /** subscribe_id → track_alias, so [unsubscribe] can clean up the matching sink. */ + private val aliasBySubscribeId = HashMap() + + private var controlPumpJob: Job? = null + private var datagramPumpJob: Job? = null + private var closed = false + /** * Run the SETUP handshake. * * Client side: writes CLIENT_SETUP with [supportedVersions] + [clientParameters], - * then reads exactly one SERVER_SETUP, stores the result, and returns. + * then reads exactly one SERVER_SETUP, stores the result, and starts the + * background pumps. * Server side: reads exactly one CLIENT_SETUP, selects the first mutually - * supported version from [supportedVersions] (where the local list is the - * set of versions the server is willing to accept), writes SERVER_SETUP, - * and returns. + * supported version from [supportedVersions], writes SERVER_SETUP, and + * starts the background pumps. * * @throws MoqProtocolException if the peer sent an unexpected message, or * if no version overlap exists (server side). @@ -95,6 +132,7 @@ class MoqSession private constructor( Role.Server -> runServerSetup(supportedVersions, clientParameters) } } + startPumps() } private suspend fun runClientSetup( @@ -102,7 +140,7 @@ class MoqSession private constructor( clientParameters: List, ) { controlStream.write(MoqCodec.encode(ClientSetup(supportedVersions, clientParameters))) - val reply = readOneMessage() + val reply = readOneSetupMessage() val server = reply as? ServerSetup ?: throw MoqProtocolException("expected SERVER_SETUP, got ${reply.type.name}") @@ -119,7 +157,7 @@ class MoqSession private constructor( acceptedVersions: List, serverParameters: List, ) { - val incoming = readOneMessage() + val incoming = readOneSetupMessage() val client = incoming as? ClientSetup ?: throw MoqProtocolException("expected CLIENT_SETUP, got ${incoming.type.name}") @@ -134,14 +172,12 @@ class MoqSession private constructor( } /** - * Read exactly one full MoQ message from the control stream. - * - * Phase 3c-1 assumes a whole message fits in a single transport write - * (SETUP frames are only a handful of bytes and the peer always writes - * them atomically). Phase 3c-2 will replace this with a buffer-and-retry - * loop for messages that may fragment across chunks. + * Bootstrap helper: read exactly one SETUP-phase message from the control + * stream. Used only during the synchronous handshake (before the pump + * starts). SETUP frames always fit in a single transport write, so we can + * safely take the first chunk and decode. */ - private suspend fun readOneMessage(): MoqMessage { + private suspend fun readOneSetupMessage(): MoqMessage { val chunk = controlStream.incoming().first() val decoded = MoqCodec.decode(chunk) @@ -151,38 +187,260 @@ class MoqSession private constructor( return decoded.message } - /** Close the underlying transport. */ + // ---------------------------------------------------------------- Subscribe + + /** + * Send SUBSCRIBE and suspend until SUBSCRIBE_OK arrives. Returns a + * [SubscribeHandle] whose [SubscribeHandle.objects] flow emits every + * OBJECT_DATAGRAM tagged with this subscription's track alias. + * + * @throws MoqProtocolException if the publisher rejects the subscribe with + * SUBSCRIBE_ERROR, or if the session is closed mid-flight. + */ + suspend fun subscribe( + namespace: TrackNamespace, + trackName: ByteArray, + priority: Int = 0x80, + filter: SubscribeFilter = SubscribeFilter.LatestGroup, + objectBufferCapacity: Int = DEFAULT_OBJECT_BUFFER, + ): SubscribeHandle { + check(!closed) { "session is closed" } + + val subscribeId: Long + val trackAlias: Long + val deferred = CompletableDeferred() + val sink = + Channel( + capacity = objectBufferCapacity, + onBufferOverflow = BufferOverflow.DROP_OLDEST, + ) + + stateMutex.withLock { + check(!closed) { "session is closed" } + subscribeId = nextSubscribeId++ + trackAlias = subscribeId // 1:1 alias = id keeps things simple + pendingSubscribes[subscribeId] = deferred + sinks[trackAlias] = sink + aliasBySubscribeId[subscribeId] = trackAlias + } + + val frame = + MoqCodec.encode( + Subscribe( + subscribeId = subscribeId, + trackAlias = trackAlias, + namespace = namespace, + trackName = trackName, + subscriberPriority = priority, + filter = filter, + ), + ) + try { + writeMutex.withLock { controlStream.write(frame) } + val ok = deferred.await() + return SubscribeHandle( + subscribeId = subscribeId, + trackAlias = trackAlias, + ok = ok, + objects = sink.receiveAsFlow(), + unsubscribeAction = { unsubscribe(subscribeId) }, + ) + } catch (t: Throwable) { + // Roll back state on any failure (write error, deferred cancellation, + // SUBSCRIBE_ERROR, session close). + stateMutex.withLock { + pendingSubscribes.remove(subscribeId) + sinks.remove(trackAlias)?.close() + aliasBySubscribeId.remove(subscribeId) + } + throw t + } + } + + /** + * Send UNSUBSCRIBE and tear down all local state for [subscribeId]. Safe + * to call multiple times — second call is a no-op. + */ + suspend fun unsubscribe(subscribeId: Long) { + val alias = + stateMutex.withLock { + pendingSubscribes.remove(subscribeId)?.cancel() + aliasBySubscribeId.remove(subscribeId)?.also { sinks.remove(it)?.close() } + } ?: return + + if (closed) return + runCatching { + writeMutex.withLock { controlStream.write(MoqCodec.encode(Unsubscribe(subscribeId))) } + } + // Suppress write errors on unsubscribe — the session is being torn down + // and the alias is already gone from local state. `alias` is intentionally + // referenced so the compiler keeps the cleanup expression's side effects. + @Suppress("UNUSED_VARIABLE") + val ignored = alias + } + + // ---------------------------------------------------------------- Pumps + + private fun startPumps() { + controlPumpJob = + pumpScope.launch { + runControlPump() + } + datagramPumpJob = + pumpScope.launch { + runDatagramPump() + } + } + + private suspend fun runControlPump() { + var buffer = ByteArray(0) + controlStream.incoming().collect { chunk -> + buffer = + if (buffer.isEmpty()) { + chunk + } else { + val merged = ByteArray(buffer.size + chunk.size) + buffer.copyInto(merged, 0) + chunk.copyInto(merged, buffer.size) + merged + } + while (true) { + val decoded = + try { + MoqCodec.decode(buffer) ?: break + } catch (e: MoqCodecException) { + // Drop the corrupted buffer; keep the pump alive so the + // next valid frame can recover the session. + buffer = ByteArray(0) + break + } + buffer = buffer.copyOfRange(decoded.bytesConsumed, buffer.size) + dispatchControlMessage(decoded.message) + } + } + } + + private suspend fun dispatchControlMessage(msg: MoqMessage) { + when (msg) { + is SubscribeOk -> { + val deferred = + stateMutex.withLock { pendingSubscribes.remove(msg.subscribeId) } + deferred?.complete(msg) + } + + is SubscribeError -> { + val deferred = + stateMutex.withLock { + val d = pendingSubscribes.remove(msg.subscribeId) + aliasBySubscribeId.remove(msg.subscribeId)?.also { sinks.remove(it)?.close() } + d + } + deferred?.completeExceptionally( + MoqProtocolException( + "SUBSCRIBE rejected: code=${msg.errorCode} reason=${msg.reasonPhrase}", + ), + ) + } + + else -> { + // Other control messages (SETUP echoes, future ANNOUNCE/etc.) + // are silently dropped at this layer; Phase 3c-3 only needs the + // subscribe lifecycle. + } + } + } + + private suspend fun runDatagramPump() { + transport.incomingDatagrams().collect { datagram -> + val obj = + try { + MoqObjectDatagram.decode(datagram) + } catch (e: MoqCodecException) { + // Drop malformed datagrams — UDP-style transport, so partial + // / corrupted packets are normal. + return@collect + } + val sink = stateMutex.withLock { sinks[obj.trackAlias] } + // trySend on a DROP_OLDEST channel always succeeds (drops oldest on + // overflow). Returning value is intentionally ignored. + sink?.trySend(obj) + } + } + + /** Cancel the pumps and close the underlying transport. Idempotent. */ suspend fun close( code: Int = 0, reason: String = "", ) { - controlStream.finish() - transport.close(code, reason) + stateMutex.withLock { + if (closed) return + closed = true + // Fail any in-flight subscribe waiters cleanly. + for ((_, deferred) in pendingSubscribes) { + deferred.cancel() + } + pendingSubscribes.clear() + for ((_, ch) in sinks) ch.close() + sinks.clear() + aliasBySubscribeId.clear() + } + controlPumpJob?.cancel() + datagramPumpJob?.cancel() + runCatching { writeMutex.withLock { controlStream.finish() } } + runCatching { transport.close(code, reason) } } companion object { /** - * Attach to a [WebTransportSession] in the client role. Opens the - * control stream eagerly so [setup] doesn't need to manage stream - * acquisition. + * Default per-subscription buffer for unread objects. 64 frames at + * Opus 20 ms = ~1.3 s of audio backlog before DROP_OLDEST kicks in, + * which matches a typical real-time listener's tolerance. */ - suspend fun client(transport: WebTransportSession): MoqSession { + const val DEFAULT_OBJECT_BUFFER: Int = 64 + + /** + * Attach to a [WebTransportSession] in the client role. + * + * @param pumpScope where the post-handshake pumps live. Tests typically + * pass `backgroundScope` from `runTest`; production code passes the + * owning ViewModel scope so pumps are cancelled on screen exit. + */ + suspend fun client( + transport: WebTransportSession, + pumpScope: CoroutineScope, + ): MoqSession { val control = transport.openBidiStream() - return MoqSession(transport, control, Role.Client) + return MoqSession(transport, control, Role.Client, pumpScope) } /** * Attach to a [WebTransportSession] in the server role over an - * already-accepted control stream (usually the first bidi stream the - * peer opened). Used in tests — a real server accepts the first bidi. + * already-accepted control stream. Used in tests — a real server + * accepts the client's first bidi. */ fun server( transport: WebTransportSession, control: WebTransportBidiStream, - ): MoqSession = MoqSession(transport, control, Role.Server) + pumpScope: CoroutineScope, + ): MoqSession = MoqSession(transport, control, Role.Server, pumpScope) } } +/** + * Handle to an active subscription. [objects] emits every OBJECT_DATAGRAM the + * publisher delivers for this subscription's track. Call [unsubscribe] when + * done — equivalent to [MoqSession.unsubscribe] with this handle's id. + */ +class SubscribeHandle internal constructor( + val subscribeId: Long, + val trackAlias: Long, + val ok: SubscribeOk, + val objects: Flow, + private val unsubscribeAction: suspend () -> Unit, +) { + suspend fun unsubscribe() = unsubscribeAction() +} + /** Thrown when the peer violates the MoQ-transport state machine. */ class MoqProtocolException( message: String, 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 57eb412b5..7e36e9e3a 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/transport/FakeWebTransport.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/transport/FakeWebTransport.kt @@ -22,7 +22,7 @@ package com.vitorpamplona.nestsclient.transport import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.flow.Flow -import kotlinx.coroutines.flow.consumeAsFlow +import kotlinx.coroutines.flow.receiveAsFlow import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.withLock @@ -34,8 +34,10 @@ import kotlinx.coroutines.sync.withLock * only (no packet loss, no congestion); the real Kwik-backed transport will * exercise those codepaths separately. * - * [incomingDatagrams] and [FakeBidiStream.incoming] use [consumeAsFlow] - * semantics and therefore may only be collected once per channel. + * [incomingDatagrams] and [FakeBidiStream.incoming] use [receiveAsFlow] + * semantics: a `take(1)` / `first()` followed by a long-running `collect` + * works on the same underlying channel, and cancelling a consumer does not + * close the channel for future readers. */ class FakeWebTransport private constructor( private val outboundDatagrams: Channel, @@ -60,7 +62,7 @@ class FakeWebTransport private constructor( return local } - override fun incomingUniStreams(): Flow = inboundUniStreams.consumeAsFlow() + override fun incomingUniStreams(): Flow = inboundUniStreams.receiveAsFlow() override suspend fun sendDatagram(payload: ByteArray): Boolean { if (!open) return false @@ -68,7 +70,7 @@ class FakeWebTransport private constructor( return true } - override fun incomingDatagrams(): Flow = inboundDatagrams.consumeAsFlow() + override fun incomingDatagrams(): Flow = inboundDatagrams.receiveAsFlow() override suspend fun close( code: Int, @@ -89,7 +91,7 @@ class FakeWebTransport private constructor( * Flow of peer-opened bidi streams (i.e. the local endpoint of a stream the * other side created via [openBidiStream]). */ - fun peerOpenedBidiStreams(): Flow = inboundBidiStreams.consumeAsFlow() + fun peerOpenedBidiStreams(): Flow = inboundBidiStreams.receiveAsFlow() companion object { /** @@ -130,7 +132,7 @@ class FakeBidiStream internal constructor( private val write: Channel, private val read: Channel, ) : WebTransportBidiStream { - override fun incoming(): Flow = read.consumeAsFlow() + override fun incoming(): Flow = read.receiveAsFlow() override suspend fun write(chunk: ByteArray) { write.send(chunk) @@ -144,5 +146,5 @@ class FakeBidiStream internal constructor( class FakeReadStream internal constructor( private val read: Channel, ) : WebTransportReadStream { - override fun incoming(): Flow = read.consumeAsFlow() + override fun incoming(): Flow = read.receiveAsFlow() } diff --git a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/MoqSessionTest.kt b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/MoqSessionTest.kt index 353e7a60d..74836e72d 100644 --- a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/MoqSessionTest.kt +++ b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/MoqSessionTest.kt @@ -23,8 +23,11 @@ package com.vitorpamplona.nestsclient.moq import com.vitorpamplona.nestsclient.transport.FakeWebTransport import kotlinx.coroutines.async import kotlinx.coroutines.flow.first +import kotlinx.coroutines.flow.take +import kotlinx.coroutines.flow.toList import kotlinx.coroutines.test.runTest import kotlin.test.Test +import kotlin.test.assertContentEquals import kotlin.test.assertEquals import kotlin.test.assertFailsWith @@ -36,7 +39,7 @@ class MoqSessionTest { val clientJob = async { - val session = MoqSession.client(clientSide) + val session = MoqSession.client(clientSide, backgroundScope) session.setup(listOf(MoqVersion.DRAFT_17)) session.selectedVersion } @@ -44,16 +47,11 @@ class MoqSessionTest { val serverJob = async { val control = serverSide.peerOpenedBidiStreams().first() - val session = MoqSession.server(serverSide, control) + val session = MoqSession.server(serverSide, control, backgroundScope) session.setup( supportedVersions = listOf(MoqVersion.DRAFT_17, MoqVersion.DRAFT_11), clientParameters = - listOf( - SetupParameter( - SetupParameter.KEY_MAX_SUBSCRIBE_ID, - byteArrayOf(0x10), - ), - ), + listOf(SetupParameter(SetupParameter.KEY_MAX_SUBSCRIBE_ID, byteArrayOf(0x10))), ) session.selectedVersion } @@ -69,8 +67,7 @@ class MoqSessionTest { val clientJob = async { - val session = MoqSession.client(clientSide) - // Client prefers 17, falls back to 11. + val session = MoqSession.client(clientSide, backgroundScope) session.setup(listOf(MoqVersion.DRAFT_17, MoqVersion.DRAFT_11)) session.selectedVersion } @@ -78,8 +75,7 @@ class MoqSessionTest { val serverJob = async { val control = serverSide.peerOpenedBidiStreams().first() - val session = MoqSession.server(serverSide, control) - // Server only speaks 11 → overlap is 11. + val session = MoqSession.server(serverSide, control, backgroundScope) session.setup(supportedVersions = listOf(MoqVersion.DRAFT_11)) session.selectedVersion } @@ -95,23 +91,184 @@ class MoqSessionTest { val clientJob = async { - val session = MoqSession.client(clientSide) + val session = MoqSession.client(clientSide, backgroundScope) runCatching { session.setup(listOf(MoqVersion.DRAFT_17)) } } val serverJob = async { val control = serverSide.peerOpenedBidiStreams().first() - val session = MoqSession.server(serverSide, control) + val session = MoqSession.server(serverSide, control, backgroundScope) assertFailsWith { session.setup(supportedVersions = listOf(MoqVersion.DRAFT_11)) } } serverJob.await() - // Client's result: its control stream closes with no reply, which throws. - // We just verify that some throwable propagated. val clientOutcome = clientJob.await() assert(clientOutcome.isFailure) { "client setup should have failed, got: $clientOutcome" } } + + /** + * End-to-end: client subscribes; a *raw* server peer (no MoqSession on + * the server side, since that would compete with the test's manual reads + * for control-stream items) responds with SUBSCRIBE_OK and three + * OBJECT_DATAGRAMs. The client's flow should yield them in order. + */ + @Test + fun subscribe_completes_on_subscribe_ok_then_delivers_datagrams_in_order() = + runTest { + val (clientSide, serverSide) = FakeWebTransport.pair() + val clientSession = MoqSession.client(clientSide, backgroundScope) + + // Server-side raw peer: handshake + serve one subscription manually. + val serverWork = + async { + val serverControl = serverSide.peerOpenedBidiStreams().first() + // Handshake: read CLIENT_SETUP, write SERVER_SETUP. + val clientSetupFrame = serverControl.incoming().first() + val clientSetup = MoqCodec.decode(clientSetupFrame)!!.message as ClientSetup + val version = clientSetup.supportedVersions.first() + serverControl.write(MoqCodec.encode(ServerSetup(selectedVersion = version))) + + // Wait for SUBSCRIBE, reply with SUBSCRIBE_OK + three datagrams. + val sub = MoqCodec.decode(serverControl.incoming().first())!!.message as Subscribe + serverControl.write( + MoqCodec.encode( + SubscribeOk( + subscribeId = sub.subscribeId, + expiresMs = 60_000, + groupOrder = 0, + contentExists = false, + ), + ), + ) + repeat(3) { i -> + val obj = + MoqObject( + trackAlias = sub.trackAlias, + groupId = 0, + objectId = i.toLong(), + publisherPriority = 0x80, + payload = byteArrayOf(i.toByte()), + ) + serverSide.sendDatagram(MoqObjectDatagram.encode(obj)) + } + sub + } + + clientSession.setup(listOf(MoqVersion.DRAFT_17)) + + val handle = + clientSession.subscribe( + namespace = TrackNamespace.of("nests", "test-room"), + trackName = "speaker-1".encodeToByteArray(), + ) + val sub = serverWork.await() + + assertEquals(sub.subscribeId, handle.subscribeId) + assertEquals(sub.trackAlias, handle.trackAlias) + assertEquals(60_000L, handle.ok.expiresMs) + + val received = handle.objects.take(3).toList() + assertEquals(listOf(0L, 1L, 2L), received.map { it.objectId }) + assertContentEquals(byteArrayOf(0), received[0].payload) + assertContentEquals(byteArrayOf(1), received[1].payload) + assertContentEquals(byteArrayOf(2), received[2].payload) + + clientSession.close() + } + + @Test + fun subscribe_throws_MoqProtocolException_when_publisher_replies_with_subscribe_error() = + runTest { + val (clientSide, serverSide) = FakeWebTransport.pair() + val clientSession = MoqSession.client(clientSide, backgroundScope) + + val rejector = + async { + val serverControl = serverSide.peerOpenedBidiStreams().first() + val clientSetup = + MoqCodec.decode(serverControl.incoming().first())!!.message as ClientSetup + serverControl.write(MoqCodec.encode(ServerSetup(clientSetup.supportedVersions.first()))) + + val sub = + MoqCodec.decode(serverControl.incoming().first())!!.message as Subscribe + serverControl.write( + MoqCodec.encode( + SubscribeError( + subscribeId = sub.subscribeId, + errorCode = 404, + reasonPhrase = "track not found", + trackAlias = sub.trackAlias, + ), + ), + ) + } + + clientSession.setup(listOf(MoqVersion.DRAFT_17)) + + val ex = + assertFailsWith { + clientSession.subscribe( + namespace = TrackNamespace.of("nests", "test-room"), + trackName = "missing".encodeToByteArray(), + ) + } + rejector.await() + assert("404" in ex.message!!) { "error message should mention the code: ${ex.message}" } + + clientSession.close() + } + + @Test + fun datagrams_for_unknown_track_alias_are_dropped_silently() = + runTest { + val (clientSide, serverSide) = FakeWebTransport.pair() + val clientSession = MoqSession.client(clientSide, backgroundScope) + + val handshake = + async { + val serverControl = serverSide.peerOpenedBidiStreams().first() + val cs = MoqCodec.decode(serverControl.incoming().first())!!.message as ClientSetup + serverControl.write(MoqCodec.encode(ServerSetup(cs.supportedVersions.first()))) + } + clientSession.setup(listOf(MoqVersion.DRAFT_17)) + handshake.await() + + // Send a datagram for a track no one subscribed to. The pump should + // silently drop it; the session stays usable + closes cleanly. + serverSide.sendDatagram( + MoqObjectDatagram.encode( + MoqObject( + trackAlias = 9_999, + groupId = 0, + objectId = 0, + publisherPriority = 0x80, + payload = byteArrayOf(0xAA.toByte()), + ), + ), + ) + testScheduler.runCurrent() + clientSession.close() + } + + /** + * Existing FakeWebTransport bidi-stream test still passes after the + * receiveAsFlow refactor: write + finish + collect should still surface + * the chunk and complete cleanly. + */ + @Test + fun fake_bidi_stream_finish_completes_the_consumer_flow() = + runTest { + val (a, b) = FakeWebTransport.pair() + val clientStream = a.openBidiStream() + clientStream.write(byteArrayOf(0xAA.toByte())) + clientStream.finish() + + val serverStream = b.peerOpenedBidiStreams().first() + val received = serverStream.incoming().toList() + assertEquals(1, received.size) + assertContentEquals(byteArrayOf(0xAA.toByte()), received.single()) + } }