From 5c32996f938fe3f6fb85231c935f299bd9fb72e4 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 26 Apr 2026 03:11:33 +0000 Subject: [PATCH] =?UTF-8?q?feat(audio-rooms):=20M5=20MoQ=20publisher=20pat?= =?UTF-8?q?h=20=E2=80=94=20ANNOUNCE=20+=20inbound=20SUBSCRIBE=20+=20OBJECT?= =?UTF-8?q?=20emit?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Lifts MoqSession from listener-only to bidirectional. A session can now ANNOUNCE a track namespace, register one TrackPublisher per track name under it, accept inbound SUBSCRIBEs from the peer, and emit OBJECT_DATAGRAMs that fan out to every attached subscriber. This is what the speaker / host path needs in nests — phases M6/M7 layer audio capture + a NestsSpeaker API on top. Public API additions: - `MoqSession.announce(namespace, parameters)` — sends ANNOUNCE, awaits ANNOUNCE_OK, returns an `AnnounceHandle`. - `AnnounceHandle.openTrack(name)` — registers a `TrackPublisher` for a track under the announced namespace. Idempotent insert is rejected. - `TrackPublisher.send(payload)` — encodes one MoQ OBJECT_DATAGRAM and emits it to every currently-attached subscriber. Group id is fixed at zero per the audio-rooms NIP draft; object ids are monotonic. - `TrackPublisher.close()` — sends SUBSCRIBE_DONE to attached subscribers, removes the track from its parent announce. - `AnnounceHandle.unannounce()` — sends UNANNOUNCE on the wire, closes every registered publisher. - `ErrorCode.TRACK_DOES_NOT_EXIST` and `SubscribeDoneStatus.{TRACK_ENDED, UNSUBSCRIBED}` constants for the SUBSCRIBE_ERROR / SUBSCRIBE_DONE values we emit. Internal additions: - `pendingAnnounces` keyed by namespace, mirroring the existing `pendingSubscribes` pattern. - `announces` map tracking live publishers per namespace. - `inboundSubscribers` + `publisherSubscribers` so the control-pump can fan an inbound SUBSCRIBE to the right TrackPublisher and so TrackPublisher.send can snapshot subscribers without per-OBJECT locking. - Control pump now routes ANNOUNCE_OK / ANNOUNCE_ERROR (publisher-side ack) and inbound SUBSCRIBE / UNSUBSCRIBE (we're being asked to send OBJECTs, or stop). Unknown-track SUBSCRIBE replies with SUBSCRIBE_ERROR(TRACK_DOES_NOT_EXIST) so the peer doesn't hang. - `close()` propagates session death into every AnnounceHandleImpl / TrackPublisherImpl so any in-flight `send` short-circuits cleanly. All shared-state mutation is funneled through the existing `stateMutex` (no `synchronized` — that's JVM-only and would break commonMain). The session-wide `writeMutex` continues to serialize control-stream writes. Tests (new in MoqSessionTest): - announce → ANNOUNCE_OK happy path; UNANNOUNCE wire frame on close. - ANNOUNCE_ERROR surfaces as MoqProtocolException. - End-to-end: announce + openTrack + peer SUBSCRIBE → SUBSCRIBE_OK + three OBJECT_DATAGRAMs round-trip with intact group/object ids. - send() returns false when no subscribers are attached (no buffering). - Inbound SUBSCRIBE for unknown track under an announced namespace replies SUBSCRIBE_ERROR. Verified: `:nestsClient:jvmTest` (12 tests, 0 failures) + `:quic:jvmTest` (green) + `./gradlew spotlessApply` clean. --- .../nestsclient/moq/MoqSession.kt | 415 +++++++++++++++++- .../nestsclient/moq/MoqSessionTest.kt | 204 +++++++++ 2 files changed, 614 insertions(+), 5 deletions(-) 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 823094fef..fc43f9121 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqSession.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqSession.kt @@ -106,6 +106,29 @@ class MoqSession private constructor( /** subscribe_id → track_alias, so [unsubscribe] can clean up the matching sink. */ private val aliasBySubscribeId = HashMap() + // ------- Publisher-side state (Phase M5) ---------------------------------- + + /** namespace → CompletableDeferred awaiting ANNOUNCE_OK / ANNOUNCE_ERROR. */ + private val pendingAnnounces = HashMap>() + + /** namespace → AnnounceHandleImpl (alive while we still publish that namespace). */ + private val announces = HashMap() + + /** + * Subscribers attached to one of our published tracks. Keyed by the + * subscribeId chosen by the remote subscriber. Each carries the + * (namespace, trackName, trackAlias, publisherPriority) tuple we need to + * format outbound OBJECT_DATAGRAMs. + */ + private val inboundSubscribers = HashMap() + + /** + * subscribers attached per published track. The list is rewritten + * (copy-on-write) under [stateMutex] so [TrackPublisherImpl.send] can + * read a snapshot without acquiring a lock per OBJECT. + */ + private val publisherSubscribers = HashMap>() + private var controlPumpJob: Job? = null private var datagramPumpJob: Job? = null private var closed = false @@ -259,6 +282,53 @@ class MoqSession private constructor( } } + // ---------------------------------------------------------------- Announce (publisher) + + /** + * Send ANNOUNCE for [namespace] and suspend until ANNOUNCE_OK arrives. + * Returns an [AnnounceHandle] you can [AnnounceHandle.openTrack] on for + * each speaker / sub-track you intend to publish under that namespace. + * + * The session keeps the announce alive until you call + * [AnnounceHandle.unannounce] (or [close]); during that time, inbound + * SUBSCRIBE messages whose namespace matches will be routed to the + * publisher you registered via [AnnounceHandle.openTrack]. + * + * @throws MoqProtocolException if the peer rejects with ANNOUNCE_ERROR + * or the session closes mid-flight. + * @throws IllegalStateException if [namespace] is already announced on + * this session. + */ + suspend fun announce( + namespace: TrackNamespace, + parameters: List = emptyList(), + ): AnnounceHandle { + check(!closed) { "session is closed" } + val deferred = CompletableDeferred() + val handle = AnnounceHandleImpl(namespace) + + stateMutex.withLock { + check(!closed) { "session is closed" } + check(!announces.containsKey(namespace)) { "namespace $namespace is already announced" } + check(!pendingAnnounces.containsKey(namespace)) { "namespace $namespace announce already in flight" } + announces[namespace] = handle + pendingAnnounces[namespace] = deferred + } + + val frame = MoqCodec.encode(Announce(namespace = namespace, parameters = parameters)) + try { + writeMutex.withLock { controlStream.write(frame) } + deferred.await() + return handle + } catch (t: Throwable) { + stateMutex.withLock { + pendingAnnounces.remove(namespace) + announces.remove(namespace) + } + throw t + } + } + /** * Send UNSUBSCRIBE and tear down all local state for [subscribeId]. Safe * to call multiple times — second call is a no-op. @@ -344,16 +414,117 @@ class MoqSession private constructor( ) } + // ----- Publisher-side ------------------------------------------- + + is AnnounceOk -> { + val deferred = stateMutex.withLock { pendingAnnounces.remove(msg.namespace) } + deferred?.complete(Unit) + } + + is AnnounceError -> { + val deferred = + stateMutex.withLock { + announces.remove(msg.namespace) + pendingAnnounces.remove(msg.namespace) + } + deferred?.completeExceptionally( + MoqProtocolException( + "ANNOUNCE rejected: code=${msg.errorCode} reason=${msg.reasonPhrase}", + ), + ) + } + + is Subscribe -> { + handleInboundSubscribe(msg) + } + + is Unsubscribe -> { + handleInboundUnsubscribe(msg) + } + else -> { - // Other control messages (echoed SETUP, future ANNOUNCE + - // SUBSCRIBE-receiving for the publisher path, etc.) are - // silently dropped — the listener path only needs the - // subscribe lifecycle. Publisher-side routing is Phase M5 - // in nestsClient/plans/2026-04-26-audio-rooms-completion.md. + // Echoed SETUP and other unhandled control messages are + // intentionally dropped — they don't change session state. } } } + /** + * Server (or peer) sent us a SUBSCRIBE for a track we ANNOUNCEd. Look up + * the matching publisher; on hit reply with SUBSCRIBE_OK and start + * routing this subscriber's OBJECTs. On miss reply with SUBSCRIBE_ERROR + * so the peer doesn't hang. + */ + private suspend fun handleInboundSubscribe(msg: Subscribe) { + val (publisher, sub) = + stateMutex.withLock { + val handle = announces[msg.namespace] + val publisher = handle?.publisherForLocked(msg.trackName) + if (publisher == null) { + null to null + } else { + val sub = + InboundSubscription( + subscribeId = msg.subscribeId, + trackAlias = msg.trackAlias, + namespace = msg.namespace, + trackName = msg.trackName, + publisher = publisher, + ) + inboundSubscribers[msg.subscribeId] = sub + val current = publisherSubscribers[publisher].orEmpty() + publisherSubscribers[publisher] = current + sub + publisher to sub + } + } + + if (publisher == null || sub == null) { + val err = + SubscribeError( + subscribeId = msg.subscribeId, + errorCode = ErrorCode.TRACK_DOES_NOT_EXIST, + reasonPhrase = "no publisher for that namespace+name", + trackAlias = msg.trackAlias, + ) + runCatching { writeMutex.withLock { controlStream.write(MoqCodec.encode(err)) } } + return + } + + val ok = + SubscribeOk( + subscribeId = msg.subscribeId, + expiresMs = 0L, + groupOrder = 0, + contentExists = false, + ) + runCatching { writeMutex.withLock { controlStream.write(MoqCodec.encode(ok)) } } + } + + private suspend fun handleInboundUnsubscribe(msg: Unsubscribe) { + val removed = + stateMutex.withLock { + val sub = inboundSubscribers.remove(msg.subscribeId) ?: return@withLock null + val current = publisherSubscribers[sub.publisher].orEmpty() + val updated = current.filter { it.subscribeId != msg.subscribeId } + if (updated.isEmpty()) { + publisherSubscribers.remove(sub.publisher) + } else { + publisherSubscribers[sub.publisher] = updated + } + sub + } ?: return + + // Best-effort SUBSCRIBE_DONE so the peer's local state matches. + val done = + SubscribeDone( + subscribeId = removed.subscribeId, + statusCode = SubscribeDoneStatus.UNSUBSCRIBED, + streamCount = 0L, + reasonPhrase = "unsubscribed", + ) + runCatching { writeMutex.withLock { controlStream.write(MoqCodec.encode(done)) } } + } + private suspend fun runDatagramPump() { transport.incomingDatagrams().collect { datagram -> val obj = @@ -387,6 +558,16 @@ class MoqSession private constructor( for ((_, ch) in sinks) ch.close() sinks.clear() aliasBySubscribeId.clear() + // Fail in-flight announce waiters and surface an exception, since + // the caller is awaiting on .await(). + for ((_, deferred) in pendingAnnounces) { + deferred.cancel() + } + pendingAnnounces.clear() + announces.values.forEach { it.markSessionClosedLocked() } + announces.clear() + inboundSubscribers.clear() + publisherSubscribers.clear() } controlPumpJob?.cancel() datagramPumpJob?.cancel() @@ -394,6 +575,201 @@ class MoqSession private constructor( runCatching { transport.close(code, reason) } } + // ---------------------------------------------------------------- Publisher implementations + + /** + * Public, addressable announce on this session. Returned by [announce] and + * lives until [unannounce] (or session [close]). + */ + interface AnnounceHandle { + val namespace: TrackNamespace + + /** + * Register a publisher for one track name under this namespace. nests + * uses the speaker's pubkey hex (UTF-8 bytes) as the track name. + * + * The returned [TrackPublisher] is the only way to push OBJECTs out; + * each call to [TrackPublisher.send] becomes one OBJECT_DATAGRAM + * fanned out to every active inbound subscriber for this track. + */ + suspend fun openTrack(name: ByteArray): TrackPublisher + + /** + * Withdraw the namespace: send UNANNOUNCE, send SUBSCRIBE_DONE to + * every inbound subscriber that's still attached, and close every + * registered [TrackPublisher]. Idempotent. + */ + suspend fun unannounce() + } + + /** A single outbound track. Push OBJECTs with [send]; tear down with [close]. */ + interface TrackPublisher { + val name: ByteArray + + /** + * Send one OBJECT (one Opus frame for nests audio) as an + * OBJECT_DATAGRAM to every inbound subscriber currently attached. + * Group/object ids are managed internally; this matches the + * audio-rooms NIP draft (one group, monotonic object ids). + * + * @return true if at least one datagram was queued for delivery. + * Returns false (and is a silent no-op) if no inbound subscriber + * exists yet — callers can keep producing audio without buffering. + */ + suspend fun send(payload: ByteArray): Boolean + + /** + * Stop publishing this track. Sends SUBSCRIBE_DONE to every attached + * subscriber, removes the track from the parent announce. Idempotent. + */ + suspend fun close() + } + + /** + * Internal accounting for a peer's inbound SUBSCRIBE on one of our + * published tracks. We send each `publisher.send(payload)` as a separate + * OBJECT_DATAGRAM keyed by this subscription's negotiated trackAlias. + */ + internal class InboundSubscription( + val subscribeId: Long, + val trackAlias: Long, + val namespace: TrackNamespace, + val trackName: ByteArray, + val publisher: TrackPublisherImpl, + ) + + internal inner class AnnounceHandleImpl( + override val namespace: TrackNamespace, + ) : AnnounceHandle { + // Non-suspending reads/writes on `tracks` are all guarded by the + // session's `stateMutex` (held by the suspend caller). + internal val tracks = HashMap() + internal var sessionClosed = false + + override suspend fun openTrack(name: ByteArray): TrackPublisher { + val key = TrackKey(name) + val publisher = TrackPublisherImpl(name) + stateMutex.withLock { + check(!closed) { "session is closed" } + check(!sessionClosed) { "namespace is unannounced" } + check(tracks[key] == null) { "track ${name.decodeToString()} already published" } + tracks[key] = publisher + publisher.parent = this + } + return publisher + } + + // Caller must hold stateMutex. + fun publisherForLocked(trackName: ByteArray): TrackPublisherImpl? = tracks[TrackKey(trackName)] + + override suspend fun unannounce() { + val toClose: List + stateMutex.withLock { + if (sessionClosed) return + sessionClosed = true + toClose = tracks.values.toList() + tracks.clear() + announces.remove(namespace) + } + toClose.forEach { runCatching { it.close() } } + if (closed) return + runCatching { + writeMutex.withLock { controlStream.write(MoqCodec.encode(Unannounce(namespace))) } + } + } + + /** + * Called from session.close while the caller already holds stateMutex. + * Drops every track without trying to send a wire UNANNOUNCE. + */ + fun markSessionClosedLocked() { + sessionClosed = true + tracks.values.forEach { it.markSessionDeadLocked() } + tracks.clear() + } + } + + internal inner class TrackPublisherImpl( + override val name: ByteArray, + ) : TrackPublisher { + var parent: AnnounceHandleImpl? = null + internal var nextObjectId = 0L + internal val groupId = 0L // single-group track per audio-rooms NIP draft + internal var trackClosed = false + internal var sessionDead = false + + override suspend fun send(payload: ByteArray): Boolean { + val snapshotAndId = + stateMutex.withLock { + if (trackClosed || sessionDead) return false + val snapshot = publisherSubscribers[this] + if (snapshot.isNullOrEmpty()) return false + val id = nextObjectId + nextObjectId += 1 + snapshot to id + } + val (snapshot, objectId) = snapshotAndId + var anySent = false + for (sub in snapshot) { + val datagram = + MoqObjectDatagram.encode( + MoqObject( + trackAlias = sub.trackAlias, + groupId = groupId, + objectId = objectId, + publisherPriority = DEFAULT_PUBLISHER_PRIORITY, + payload = payload, + ), + ) + runCatching { transport.sendDatagram(datagram) } + .onSuccess { anySent = true } + } + return anySent + } + + override suspend fun close() { + val snapshot: List + stateMutex.withLock { + if (trackClosed) return + trackClosed = true + snapshot = publisherSubscribers.remove(this).orEmpty() + snapshot.forEach { inboundSubscribers.remove(it.subscribeId) } + parent?.tracks?.remove(TrackKey(name)) + } + if (closed || sessionDead) return + for (sub in snapshot) { + val done = + SubscribeDone( + subscribeId = sub.subscribeId, + statusCode = SubscribeDoneStatus.TRACK_ENDED, + streamCount = 0L, + reasonPhrase = "track closed", + ) + runCatching { writeMutex.withLock { controlStream.write(MoqCodec.encode(done)) } } + } + } + + /** Caller already holds stateMutex (called from session.close). */ + fun markSessionDeadLocked() { + sessionDead = true + trackClosed = true + publisherSubscribers.remove(this) + } + } + + /** Equality wrapper for byte-array keys in [AnnounceHandleImpl.tracks]. */ + internal data class TrackKey( + val name: ByteArray, + ) { + override fun equals(other: Any?): Boolean { + if (this === other) return true + if (other !is TrackKey) return false + return name.contentEquals(other.name) + } + + override fun hashCode(): Int = name.contentHashCode() + } + companion object { /** * Default per-subscription buffer for unread objects. 64 frames at @@ -402,6 +778,14 @@ class MoqSession private constructor( */ const val DEFAULT_OBJECT_BUFFER: Int = 64 + /** + * Publisher priority byte used for outbound OBJECT_DATAGRAMs from + * [TrackPublisher.send]. 0x80 is the middle of the byte range — a + * sensible neutral default until nests cares about prioritising + * speakers over each other. + */ + const val DEFAULT_PUBLISHER_PRIORITY: Int = 0x80 + /** * Attach to a [WebTransportSession] in the client role. * @@ -450,3 +834,24 @@ class MoqProtocolException( message: String, cause: Throwable? = null, ) : RuntimeException(message, cause) + +/** + * MoQ-transport SUBSCRIBE_ERROR / ANNOUNCE_ERROR error codes. Draft revisions + * shuffle these around; the values here track the most stable subset. + */ +object ErrorCode { + /** Track the publisher knows nothing about. */ + const val TRACK_DOES_NOT_EXIST: Long = 0x04L +} + +/** + * MoQ-transport SUBSCRIBE_DONE status_code values. As above, draft-stable + * subset; not exhaustive. + */ +object SubscribeDoneStatus { + /** The publisher has no more objects coming for this track. */ + const val TRACK_ENDED: Long = 0x00L + + /** Subscriber explicitly UNSUBSCRIBEd; publisher is acknowledging. */ + const val UNSUBSCRIBED: Long = 0x01L +} 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 74836e72d..f91fd2556 100644 --- a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/MoqSessionTest.kt +++ b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/MoqSessionTest.kt @@ -21,6 +21,7 @@ package com.vitorpamplona.nestsclient.moq import com.vitorpamplona.nestsclient.transport.FakeWebTransport +import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.async import kotlinx.coroutines.flow.first import kotlinx.coroutines.flow.take @@ -271,4 +272,207 @@ class MoqSessionTest { assertEquals(1, received.size) assertContentEquals(byteArrayOf(0xAA.toByte()), received.single()) } + + // ----- Publisher path (M5) ------------------------------------------- + + @Test + fun announce_completes_on_announce_ok_then_unannounce_sends_unannounce_frame() = + runTest { + val (publisherSide, peerSide) = FakeWebTransport.pair() + val publisherSession = MoqSession.client(publisherSide, backgroundScope) + + val ns = TrackNamespace.of("nests", "test-room") + + val peer = + async { + val ctrl = peerSide.peerOpenedBidiStreams().first() + // Setup + val cs = MoqCodec.decode(ctrl.incoming().first())!!.message as ClientSetup + ctrl.write(MoqCodec.encode(ServerSetup(cs.supportedVersions.first()))) + // ANNOUNCE + val announce = MoqCodec.decode(ctrl.incoming().first())!!.message as Announce + assertEquals(ns, announce.namespace) + ctrl.write(MoqCodec.encode(AnnounceOk(announce.namespace))) + // Read UNANNOUNCE after publisher tears down + val unannounce = MoqCodec.decode(ctrl.incoming().first())!!.message as Unannounce + assertEquals(ns, unannounce.namespace) + } + + publisherSession.setup(listOf(MoqVersion.DRAFT_17)) + val handle = publisherSession.announce(ns) + assertEquals(ns, handle.namespace) + handle.unannounce() + peer.await() + + publisherSession.close() + } + + @Test + fun announce_throws_MoqProtocolException_when_peer_replies_announce_error() = + runTest { + val (publisherSide, peerSide) = FakeWebTransport.pair() + val publisherSession = MoqSession.client(publisherSide, backgroundScope) + val ns = TrackNamespace.of("nests", "no-perms") + + val peer = + async { + val ctrl = peerSide.peerOpenedBidiStreams().first() + val cs = MoqCodec.decode(ctrl.incoming().first())!!.message as ClientSetup + ctrl.write(MoqCodec.encode(ServerSetup(cs.supportedVersions.first()))) + val ann = MoqCodec.decode(ctrl.incoming().first())!!.message as Announce + ctrl.write( + MoqCodec.encode( + AnnounceError( + namespace = ann.namespace, + errorCode = 0x10, + reasonPhrase = "no permission", + ), + ), + ) + } + + publisherSession.setup(listOf(MoqVersion.DRAFT_17)) + val ex = assertFailsWith { publisherSession.announce(ns) } + assert("0x10" in ex.message!! || "no permission" in ex.message!!) + peer.await() + publisherSession.close() + } + + @Test + fun publisher_routes_inbound_subscribe_and_emits_object_datagrams() = + runTest { + val (publisherSide, peerSide) = FakeWebTransport.pair() + val publisherSession = MoqSession.client(publisherSide, backgroundScope) + val ns = TrackNamespace.of("nests", "test-room") + val trackName = "speaker-pub-1".encodeToByteArray() + + // Real nests servers wait for a viewer's UI to subscribe; they don't + // race ANNOUNCE_OK. The test models that with an explicit gate so + // openTrack runs before we forge an inbound SUBSCRIBE. + val publisherReady = CompletableDeferred() + + val peerJob = + async { + val ctrl = peerSide.peerOpenedBidiStreams().first() + // Setup + val cs = MoqCodec.decode(ctrl.incoming().first())!!.message as ClientSetup + ctrl.write(MoqCodec.encode(ServerSetup(cs.supportedVersions.first()))) + // ANNOUNCE + val announce = MoqCodec.decode(ctrl.incoming().first())!!.message as Announce + ctrl.write(MoqCodec.encode(AnnounceOk(announce.namespace))) + publisherReady.await() + // Send a SUBSCRIBE for the publisher's track + val subscribeId = 42L + val trackAlias = 7L + ctrl.write( + MoqCodec.encode( + Subscribe( + subscribeId = subscribeId, + trackAlias = trackAlias, + namespace = announce.namespace, + trackName = trackName, + ), + ), + ) + // Expect a SUBSCRIBE_OK reply + val ok = MoqCodec.decode(ctrl.incoming().first())!!.message as SubscribeOk + assertEquals(subscribeId, ok.subscribeId) + Triple(subscribeId, trackAlias, ok) + } + + publisherSession.setup(listOf(MoqVersion.DRAFT_17)) + val handle = publisherSession.announce(ns) + val publisher = handle.openTrack(trackName) + publisherReady.complete(Unit) + + val (subscribeId, trackAlias, _) = peerJob.await() + + // Now push 3 objects through the publisher; they should arrive on + // the peer's incoming-datagram channel. + repeat(3) { i -> + publisher.send(byteArrayOf(i.toByte())) + } + + val received = peerSide.incomingDatagrams().take(3).toList() + val decoded = received.map { MoqObjectDatagram.decode(it) } + assertEquals(listOf(0L, 1L, 2L), decoded.map { it.objectId }) + assertEquals(List(3) { trackAlias }, decoded.map { it.trackAlias }) + assertContentEquals(byteArrayOf(0), decoded[0].payload) + assertContentEquals(byteArrayOf(1), decoded[1].payload) + assertContentEquals(byteArrayOf(2), decoded[2].payload) + + // subscribeId is intentionally read so the test fails if the order + // ever flips silently. + assertEquals(42L, subscribeId) + + publisher.close() + handle.unannounce() + publisherSession.close() + } + + @Test + fun publisher_send_returns_false_when_no_subscribers_attached() = + runTest { + val (publisherSide, peerSide) = FakeWebTransport.pair() + val publisherSession = MoqSession.client(publisherSide, backgroundScope) + val ns = TrackNamespace.of("nests", "lonely") + + val peer = + async { + val ctrl = peerSide.peerOpenedBidiStreams().first() + val cs = MoqCodec.decode(ctrl.incoming().first())!!.message as ClientSetup + ctrl.write(MoqCodec.encode(ServerSetup(cs.supportedVersions.first()))) + val a = MoqCodec.decode(ctrl.incoming().first())!!.message as Announce + ctrl.write(MoqCodec.encode(AnnounceOk(a.namespace))) + } + + publisherSession.setup(listOf(MoqVersion.DRAFT_17)) + val handle = publisherSession.announce(ns) + peer.await() + val publisher = handle.openTrack("nobody-listens".encodeToByteArray()) + + assertEquals(false, publisher.send(byteArrayOf(1, 2, 3))) + + publisher.close() + handle.unannounce() + publisherSession.close() + } + + @Test + fun publisher_replies_subscribe_error_for_unknown_track_under_announced_namespace() = + runTest { + val (publisherSide, peerSide) = FakeWebTransport.pair() + val publisherSession = MoqSession.client(publisherSide, backgroundScope) + val ns = TrackNamespace.of("nests", "test-room") + + val peer = + async { + val ctrl = peerSide.peerOpenedBidiStreams().first() + val cs = MoqCodec.decode(ctrl.incoming().first())!!.message as ClientSetup + ctrl.write(MoqCodec.encode(ServerSetup(cs.supportedVersions.first()))) + val a = MoqCodec.decode(ctrl.incoming().first())!!.message as Announce + ctrl.write(MoqCodec.encode(AnnounceOk(a.namespace))) + // Send SUBSCRIBE for a track we never opened + ctrl.write( + MoqCodec.encode( + Subscribe( + subscribeId = 1L, + trackAlias = 1L, + namespace = a.namespace, + trackName = "ghost".encodeToByteArray(), + ), + ), + ) + val err = MoqCodec.decode(ctrl.incoming().first())!!.message as SubscribeError + assertEquals(1L, err.subscribeId) + assertEquals(ErrorCode.TRACK_DOES_NOT_EXIST, err.errorCode) + } + + publisherSession.setup(listOf(MoqVersion.DRAFT_17)) + val handle = publisherSession.announce(ns) + peer.await() + + handle.unannounce() + publisherSession.close() + } }