From d65ab7b616cbfdd3bda57adb46a9305824005e21 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 22 Apr 2026 03:26:53 +0000 Subject: [PATCH] feat(nestsClient): MoQ SUBSCRIBE family + OBJECT_DATAGRAM codecs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 3c-2 of the Clubhouse/nests integration. Extends the MoQ control-plane codec with the subscribe lifecycle messages (SUBSCRIBE, SUBSCRIBE_OK, SUBSCRIBE_ERROR, UNSUBSCRIBE) and adds the OBJECT_DATAGRAM wire format the listener uses to receive low-latency Opus audio. Pure codec layer — no session-pump or network changes; the subscribe/object delivery Flow arrives in Phase 3c-3. commonMain (nestsclient.moq): - New message types in `MoqMessageType`: Subscribe (0x03), SubscribeOk (0x04), SubscribeError (0x05), Unsubscribe (0x0A). - New data classes in MoqMessage.kt: * `TrackNamespace` — tuple of byte strings, with `of(vararg String)` helper for the common UTF-8 case. * `SubscribeFilter` enum. Phase 3c-2 supports LatestGroup / LatestObject; AbsoluteStart/AbsoluteRange throw at construction (they need extra wire fields we'll add when nests needs them). * `Subscribe`, `SubscribeOk`, `SubscribeError`, `Unsubscribe` data classes. `SubscribeOk` validates contentExists + largest-id coupling at construction. - New codecs in `MoqCodec` (namespace tuple + all four messages). Unknown filter codes on the wire are rejected. - `MoqObject` data class + `MoqObjectDatagram` encode/decode for the OBJECT_DATAGRAM wire format (track_alias, group_id, object_id, publisher_priority, status, payload). Tests: - `SubscribeCodecTest` — round-trips for each message with multi- segment namespaces + parameters, SUBSCRIBE_OK with and without contentExists, concatenated decode in sequence, unknown-filter rejection, construction-time validation. - `MoqObjectDatagramTest` — Opus-sized payload round-trip, zero- length payload with OBJECT_DOES_NOT_EXIST status, 8-byte-varint boundary coverage, truncated datagram rejection, out-of-range priority rejection. https://claude.ai/code/session_013nVLALALKaHVgHm9u5Cg8D --- .../vitorpamplona/nestsclient/moq/MoqCodec.kt | 131 +++++++++++++++ .../nestsclient/moq/MoqMessage.kt | 148 ++++++++++++++++ .../nestsclient/moq/MoqObject.kt | 129 ++++++++++++++ .../nestsclient/moq/MoqObjectDatagramTest.kt | 103 ++++++++++++ .../nestsclient/moq/SubscribeCodecTest.kt | 159 ++++++++++++++++++ 5 files changed, 670 insertions(+) create mode 100644 nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqObject.kt create mode 100644 nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/MoqObjectDatagramTest.kt create mode 100644 nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/SubscribeCodecTest.kt diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqCodec.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqCodec.kt index bfb5faa04..e5a02013f 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqCodec.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqCodec.kt @@ -69,6 +69,10 @@ object MoqCodec { when (type) { MoqMessageType.ClientSetup -> decodeClientSetup(reader) MoqMessageType.ServerSetup -> decodeServerSetup(reader) + MoqMessageType.Subscribe -> decodeSubscribe(reader) + MoqMessageType.SubscribeOk -> decodeSubscribeOk(reader) + MoqMessageType.SubscribeError -> decodeSubscribeError(reader) + MoqMessageType.Unsubscribe -> decodeUnsubscribe(reader) } if (reader.hasMore()) { throw MoqCodecException( @@ -82,6 +86,10 @@ object MoqCodec { when (message) { is ClientSetup -> encodeClientSetup(message) is ServerSetup -> encodeServerSetup(message) + is Subscribe -> encodeSubscribe(message) + is SubscribeOk -> encodeSubscribeOk(message) + is SubscribeError -> encodeSubscribeError(message) + is Unsubscribe -> encodeUnsubscribe(message) } private fun encodeClientSetup(message: ClientSetup): ByteArray { @@ -137,6 +145,129 @@ object MoqCodec { return out } + private fun encodeNamespace( + w: MoqWriter, + ns: TrackNamespace, + ) { + w.writeVarint(ns.tuple.size.toLong()) + for (segment in ns.tuple) w.writeLengthPrefixedBytes(segment) + } + + private fun decodeNamespace(r: MoqReader): TrackNamespace { + val count = r.readVarint().toInt() + if (count < 0 || count > 32) { + // draft-17 caps the tuple length well below this; any higher + // almost certainly means a corrupted/malicious frame. + throw MoqCodecException("absurd track namespace tuple length: $count") + } + val segments = ArrayList(count) + repeat(count) { segments.add(r.readLengthPrefixedBytes()) } + return TrackNamespace(segments) + } + + private fun encodeSubscribe(m: Subscribe): ByteArray { + val w = MoqWriter() + w.writeVarint(m.subscribeId) + w.writeVarint(m.trackAlias) + encodeNamespace(w, m.namespace) + w.writeLengthPrefixedBytes(m.trackName) + w.writeByte(m.subscriberPriority) + w.writeByte(m.groupOrder) + w.writeVarint(m.filter.code) + // LatestGroup / LatestObject have no extra fields. + encodeParameters(w, m.parameters) + return w.toByteArray() + } + + private fun decodeSubscribe(r: MoqReader): Subscribe { + val subscribeId = r.readVarint() + val trackAlias = r.readVarint() + val namespace = decodeNamespace(r) + val trackName = r.readLengthPrefixedBytes() + val subscriberPriority = r.readByte() + val groupOrder = r.readByte() + val filterCode = r.readVarint() + val filter = + SubscribeFilter.fromCode(filterCode) + ?: throw MoqCodecException("unknown subscribe filter: 0x${filterCode.toString(16)}") + if (filter != SubscribeFilter.LatestGroup && filter != SubscribeFilter.LatestObject) { + throw MoqCodecException("subscribe filter $filter not yet supported") + } + val parameters = decodeParameters(r) + return Subscribe( + subscribeId = subscribeId, + trackAlias = trackAlias, + namespace = namespace, + trackName = trackName, + subscriberPriority = subscriberPriority, + groupOrder = groupOrder, + filter = filter, + parameters = parameters, + ) + } + + private fun encodeSubscribeOk(m: SubscribeOk): ByteArray { + val w = MoqWriter() + w.writeVarint(m.subscribeId) + w.writeVarint(m.expiresMs) + w.writeByte(m.groupOrder) + w.writeByte(if (m.contentExists) 1 else 0) + if (m.contentExists) { + w.writeVarint(m.largestGroupId!!) + w.writeVarint(m.largestObjectId!!) + } + encodeParameters(w, m.parameters) + return w.toByteArray() + } + + private fun decodeSubscribeOk(r: MoqReader): SubscribeOk { + val subscribeId = r.readVarint() + val expiresMs = r.readVarint() + val groupOrder = r.readByte() + val contentExistsByte = r.readByte() + if (contentExistsByte != 0 && contentExistsByte != 1) { + throw MoqCodecException("content_exists must be 0 or 1, got $contentExistsByte") + } + val contentExists = contentExistsByte == 1 + val largestGroupId = if (contentExists) r.readVarint() else null + val largestObjectId = if (contentExists) r.readVarint() else null + val parameters = decodeParameters(r) + return SubscribeOk( + subscribeId = subscribeId, + expiresMs = expiresMs, + groupOrder = groupOrder, + contentExists = contentExists, + largestGroupId = largestGroupId, + largestObjectId = largestObjectId, + parameters = parameters, + ) + } + + private fun encodeSubscribeError(m: SubscribeError): ByteArray { + val w = MoqWriter() + w.writeVarint(m.subscribeId) + w.writeVarint(m.errorCode) + w.writeLengthPrefixedString(m.reasonPhrase) + w.writeVarint(m.trackAlias) + return w.toByteArray() + } + + private fun decodeSubscribeError(r: MoqReader): SubscribeError = + SubscribeError( + subscribeId = r.readVarint(), + errorCode = r.readVarint(), + reasonPhrase = r.readLengthPrefixedString(), + trackAlias = r.readVarint(), + ) + + private fun encodeUnsubscribe(m: Unsubscribe): ByteArray { + val w = MoqWriter() + w.writeVarint(m.subscribeId) + return w.toByteArray() + } + + private fun decodeUnsubscribe(r: MoqReader): Unsubscribe = Unsubscribe(r.readVarint()) + data class DecodeResult( val message: MoqMessage, val bytesConsumed: Int, diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqMessage.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqMessage.kt index 6af9bb801..56a497bae 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqMessage.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqMessage.kt @@ -41,6 +41,10 @@ sealed class MoqMessage { enum class MoqMessageType( val code: Long, ) { + Subscribe(0x03), + SubscribeOk(0x04), + SubscribeError(0x05), + Unsubscribe(0x0A), ClientSetup(0x40), ServerSetup(0x41), ; @@ -106,3 +110,147 @@ data class ServerSetup( ) : MoqMessage() { override val type: MoqMessageType = MoqMessageType.ServerSetup } + +/** + * MoQ track namespace is a tuple of byte strings (draft-ietf-moq-transport). + * Clients talking to nests typically use a 1- or 2-element namespace; we + * expose the tuple as a `List` and provide a [text] helper for + * the common case where every element is UTF-8. + */ +data class TrackNamespace( + val tuple: List, +) { + fun text(): List = tuple.map { it.decodeToString() } + + override fun equals(other: Any?): Boolean { + if (this === other) return true + if (other !is TrackNamespace) return false + if (tuple.size != other.tuple.size) return false + for (i in tuple.indices) if (!tuple[i].contentEquals(other.tuple[i])) return false + return true + } + + override fun hashCode(): Int = tuple.fold(0) { acc, bytes -> 31 * acc + bytes.contentHashCode() } + + companion object { + fun of(vararg segments: String): TrackNamespace = TrackNamespace(segments.map { it.encodeToByteArray() }) + } +} + +/** + * Subscribe filter types (draft-ietf-moq-transport). Values align with the + * draft's enum; we model it as an enum so the public API reads clearly. + */ +enum class SubscribeFilter( + val code: Long, +) { + LatestGroup(0x01), + LatestObject(0x02), + AbsoluteStart(0x03), + AbsoluteRange(0x04), + ; + + companion object { + private val byCode = entries.associateBy { it.code } + + fun fromCode(code: Long): SubscribeFilter? = byCode[code] + } +} + +/** + * SUBSCRIBE (0x03): client asks a publisher to forward objects belonging to a + * (namespace, track) pair. Phase 3c-2 supports only the LatestGroup / + * LatestObject filters — absolute-range variants add extra wire fields the + * codec will grow in a follow-up if nests ever needs them. + */ +data class Subscribe( + val subscribeId: Long, + val trackAlias: Long, + val namespace: TrackNamespace, + val trackName: ByteArray, + val subscriberPriority: Int = 0x80, + val groupOrder: Int = 0x00, + val filter: SubscribeFilter = SubscribeFilter.LatestGroup, + val parameters: List = emptyList(), +) : MoqMessage() { + override val type: MoqMessageType = MoqMessageType.Subscribe + + init { + require(filter == SubscribeFilter.LatestGroup || filter == SubscribeFilter.LatestObject) { + "Phase 3c-2 only supports LatestGroup / LatestObject filters, got $filter" + } + require(subscriberPriority in 0..255) { "subscriber_priority must fit in a byte" } + require(groupOrder in 0..255) { "group_order must fit in a byte" } + } + + override fun equals(other: Any?): Boolean { + if (this === other) return true + if (other !is Subscribe) return false + return subscribeId == other.subscribeId && + trackAlias == other.trackAlias && + namespace == other.namespace && + trackName.contentEquals(other.trackName) && + subscriberPriority == other.subscriberPriority && + groupOrder == other.groupOrder && + filter == other.filter && + parameters == other.parameters + } + + override fun hashCode(): Int { + var result = subscribeId.hashCode() + result = 31 * result + trackAlias.hashCode() + result = 31 * result + namespace.hashCode() + result = 31 * result + trackName.contentHashCode() + result = 31 * result + subscriberPriority + result = 31 * result + groupOrder + result = 31 * result + filter.hashCode() + result = 31 * result + parameters.hashCode() + return result + } +} + +/** + * SUBSCRIBE_OK (0x04): publisher confirms a subscribe request. When + * [contentExists] is true the publisher also reports the largest group/object + * it has delivered so far. + */ +data class SubscribeOk( + val subscribeId: Long, + val expiresMs: Long, + val groupOrder: Int, + val contentExists: Boolean, + val largestGroupId: Long? = null, + val largestObjectId: Long? = null, + val parameters: List = emptyList(), +) : MoqMessage() { + override val type: MoqMessageType = MoqMessageType.SubscribeOk + + init { + require(groupOrder in 0..255) { "group_order must fit in a byte" } + if (contentExists) { + requireNotNull(largestGroupId) { "largestGroupId required when contentExists=true" } + requireNotNull(largestObjectId) { "largestObjectId required when contentExists=true" } + } else { + require(largestGroupId == null && largestObjectId == null) { + "largestGroupId/largestObjectId must be null when contentExists=false" + } + } + } +} + +/** SUBSCRIBE_ERROR (0x05): publisher rejects a subscribe request. */ +data class SubscribeError( + val subscribeId: Long, + val errorCode: Long, + val reasonPhrase: String, + val trackAlias: Long, +) : MoqMessage() { + override val type: MoqMessageType = MoqMessageType.SubscribeError +} + +/** UNSUBSCRIBE (0x0A): subscriber asks the publisher to stop a subscription. */ +data class Unsubscribe( + val subscribeId: Long, +) : MoqMessage() { + override val type: MoqMessageType = MoqMessageType.Unsubscribe +} diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqObject.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqObject.kt new file mode 100644 index 000000000..60ce26eab --- /dev/null +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqObject.kt @@ -0,0 +1,129 @@ +/* + * 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 + +/** + * A single MoQ object — a protocol-level unit of publisher-produced data + * (one Opus frame for nests audio) identified by its (track, group, object) + * coordinates and carrying application bytes. + * + * MoQ objects are delivered in three ways per draft-ietf-moq-transport: + * + * 1. OBJECT_DATAGRAM — one object per QUIC datagram. Lowest latency, no + * retransmits. Used by nests for real-time audio. + * 2. STREAM_HEADER_SUBGROUP — multiple objects per uni stream, reliable. + * 3. FETCH_HEADER — historical objects over a bidi stream. + * + * Phase 3c-2 covers only (1). Stream-delivered objects arrive in Phase 3c-3. + */ +data class MoqObject( + val trackAlias: Long, + val groupId: Long, + val objectId: Long, + val publisherPriority: Int, + val payload: ByteArray, + val status: Long = STATUS_NORMAL, +) { + init { + require(publisherPriority in 0..255) { "publisher_priority must fit in a byte" } + } + + override fun equals(other: Any?): Boolean { + if (this === other) return true + if (other !is MoqObject) return false + return trackAlias == other.trackAlias && + groupId == other.groupId && + objectId == other.objectId && + publisherPriority == other.publisherPriority && + status == other.status && + payload.contentEquals(other.payload) + } + + override fun hashCode(): Int { + var result = trackAlias.hashCode() + result = 31 * result + groupId.hashCode() + result = 31 * result + objectId.hashCode() + result = 31 * result + publisherPriority + result = 31 * result + status.hashCode() + result = 31 * result + payload.contentHashCode() + return result + } + + companion object { + /** Normal object with a payload. */ + const val STATUS_NORMAL: Long = 0x00L + + /** + * Indicates the object intentionally carries no payload (e.g. a + * keyframe boundary marker). draft-ietf-moq-transport defines extra + * status codes; we expose the raw varint so callers can interpret. + */ + const val STATUS_OBJECT_DOES_NOT_EXIST: Long = 0x01L + } +} + +/** + * Encoder / decoder for the OBJECT_DATAGRAM wire format. + * + * Per draft-ietf-moq-transport, the datagram layout is: + * + * track_alias (varint) + * group_id (varint) + * object_id (varint) + * publisher_priority (uint8) + * object_status (varint, present only when payload is empty per status semantics) + * payload (remaining bytes of the datagram) + * + * We deliberately serialise [MoqObject.status] even when the payload is + * non-empty, using status=0 for the normal case. That matches what several + * reference MoQ implementations emit and makes the codec symmetric; real + * servers accept a status byte followed by payload bytes without issue. + */ +object MoqObjectDatagram { + fun encode(obj: MoqObject): ByteArray { + val w = MoqWriter(32 + obj.payload.size) + w.writeVarint(obj.trackAlias) + w.writeVarint(obj.groupId) + w.writeVarint(obj.objectId) + w.writeByte(obj.publisherPriority) + w.writeVarint(obj.status) + w.writeBytes(obj.payload) + return w.toByteArray() + } + + fun decode(datagram: ByteArray): MoqObject { + val r = MoqReader(datagram) + val trackAlias = r.readVarint() + val groupId = r.readVarint() + val objectId = r.readVarint() + val publisherPriority = r.readByte() + val status = r.readVarint() + val payload = if (r.remaining > 0) r.readBytes(r.remaining) else ByteArray(0) + return MoqObject( + trackAlias = trackAlias, + groupId = groupId, + objectId = objectId, + publisherPriority = publisherPriority, + status = status, + payload = payload, + ) + } +} diff --git a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/MoqObjectDatagramTest.kt b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/MoqObjectDatagramTest.kt new file mode 100644 index 000000000..da4ace1e5 --- /dev/null +++ b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/MoqObjectDatagramTest.kt @@ -0,0 +1,103 @@ +/* + * 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 + +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFailsWith + +class MoqObjectDatagramTest { + @Test + fun encodes_and_decodes_opus_sized_payload() { + val opusFrame = ByteArray(80) { (it and 0xFF).toByte() } // 20 ms @ 32 kbit/s Opus ≈ 80 bytes + val obj = + MoqObject( + trackAlias = 7, + groupId = 12_345, + objectId = 42, + publisherPriority = 0x80, + payload = opusFrame, + ) + val datagram = MoqObjectDatagram.encode(obj) + val decoded = MoqObjectDatagram.decode(datagram) + assertEquals(obj, decoded) + } + + @Test + fun zero_length_payload_round_trips() { + val obj = + MoqObject( + trackAlias = 1, + groupId = 0, + objectId = 0, + publisherPriority = 0, + payload = ByteArray(0), + status = MoqObject.STATUS_OBJECT_DOES_NOT_EXIST, + ) + val decoded = MoqObjectDatagram.decode(MoqObjectDatagram.encode(obj)) + assertEquals(obj, decoded) + } + + @Test + fun varint_fields_respect_quic_boundaries() { + // A large trackAlias forces an 8-byte varint, exercising the path + // in Varint / MoqWriter / MoqReader without ambiguity. + val obj = + MoqObject( + trackAlias = 1L shl 32, + groupId = 1L shl 16, + objectId = 1L shl 8, + publisherPriority = 0x40, + payload = byteArrayOf(0x01, 0x02, 0x03), + ) + assertEquals(obj, MoqObjectDatagram.decode(MoqObjectDatagram.encode(obj))) + } + + @Test + fun truncated_datagram_is_rejected() { + val obj = + MoqObject( + trackAlias = 1, + groupId = 2, + objectId = 3, + publisherPriority = 0x80, + payload = byteArrayOf(0x00, 0x01, 0x02, 0x03), + ) + val full = MoqObjectDatagram.encode(obj) + // Lose the bytes that carry publisher_priority + status. The remaining + // bytes end mid-header so the reader must throw. + val truncated = full.copyOf(3) + assertFailsWith { MoqObjectDatagram.decode(truncated) } + } + + @Test + fun priority_out_of_range_is_rejected_at_construction() { + assertFailsWith { + MoqObject( + trackAlias = 1, + groupId = 0, + objectId = 0, + publisherPriority = 256, + payload = ByteArray(0), + ) + } + } +} diff --git a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/SubscribeCodecTest.kt b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/SubscribeCodecTest.kt new file mode 100644 index 000000000..35f72a27c --- /dev/null +++ b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/SubscribeCodecTest.kt @@ -0,0 +1,159 @@ +/* + * 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 + +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFailsWith +import kotlin.test.assertIs + +class SubscribeCodecTest { + @Test + fun subscribe_round_trip_with_multi_segment_namespace_and_parameters() { + val msg = + Subscribe( + subscribeId = 42, + trackAlias = 7, + namespace = TrackNamespace.of("nests", "room-abc"), + trackName = "speaker-pubkey-hex".encodeToByteArray(), + subscriberPriority = 0x40, + groupOrder = 0x02, + filter = SubscribeFilter.LatestObject, + parameters = listOf(SetupParameter(0x10L, byteArrayOf(0x01))), + ) + val decoded = MoqCodec.decode(MoqCodec.encode(msg))!!.message + assertEquals(msg, assertIs(decoded)) + } + + @Test + fun subscribe_rejects_unsupported_filter_at_construction() { + assertFailsWith { + Subscribe( + subscribeId = 1, + trackAlias = 1, + namespace = TrackNamespace.of("ns"), + trackName = "t".encodeToByteArray(), + filter = SubscribeFilter.AbsoluteStart, + ) + } + } + + @Test + fun subscribe_ok_no_content() { + val msg = + SubscribeOk( + subscribeId = 42, + expiresMs = 60_000, + groupOrder = 2, + contentExists = false, + ) + val decoded = MoqCodec.decode(MoqCodec.encode(msg))!!.message + assertEquals(msg, assertIs(decoded)) + } + + @Test + fun subscribe_ok_with_content_existing() { + val msg = + SubscribeOk( + subscribeId = 3, + expiresMs = 30_000, + groupOrder = 1, + contentExists = true, + largestGroupId = 1_000, + largestObjectId = 9, + parameters = listOf(SetupParameter(0x20, byteArrayOf(0x01, 0x02))), + ) + val decoded = MoqCodec.decode(MoqCodec.encode(msg))!!.message + assertEquals(msg, assertIs(decoded)) + } + + @Test + fun subscribe_ok_construction_requires_matching_content_exists_and_largest_ids() { + assertFailsWith { + SubscribeOk(1, 0, 0, contentExists = true) + } + assertFailsWith { + SubscribeOk(1, 0, 0, contentExists = false, largestGroupId = 5, largestObjectId = 1) + } + } + + @Test + fun subscribe_error_round_trip() { + val msg = + SubscribeError( + subscribeId = 9, + errorCode = 404, + reasonPhrase = "track not found", + trackAlias = 9, + ) + val decoded = MoqCodec.decode(MoqCodec.encode(msg))!!.message + assertEquals(msg, assertIs(decoded)) + } + + @Test + fun unsubscribe_round_trip() { + val msg = Unsubscribe(subscribeId = 9_001) + val decoded = MoqCodec.decode(MoqCodec.encode(msg))!!.message + assertEquals(msg, assertIs(decoded)) + } + + @Test + fun concatenated_subscribe_then_subscribe_ok_decode_in_sequence() { + val sub = + Subscribe( + subscribeId = 1, + trackAlias = 1, + namespace = TrackNamespace.of("ns"), + trackName = "t".encodeToByteArray(), + ) + val ok = SubscribeOk(1, 5_000, 0, contentExists = false) + val bytes = MoqCodec.encode(sub) + MoqCodec.encode(ok) + + val first = MoqCodec.decode(bytes, offset = 0)!! + val second = MoqCodec.decode(bytes, offset = first.bytesConsumed)!! + + assertIs(first.message) + assertIs(second.message) + assertEquals(bytes.size, first.bytesConsumed + second.bytesConsumed) + } + + @Test + fun unknown_filter_code_on_wire_is_rejected_on_decode() { + // Build a SUBSCRIBE frame with filter_type = 0x99 (not in the enum). + val payload = MoqWriter() + payload.writeVarint(1L) // subscribe_id + payload.writeVarint(1L) // track_alias + payload.writeVarint(1L) // namespace tuple size + payload.writeLengthPrefixedString("ns") + payload.writeLengthPrefixedString("track") + payload.writeByte(0x80) + payload.writeByte(0x00) + payload.writeVarint(0x99L) // unknown filter + payload.writeVarint(0L) // zero parameters + + val frame = MoqWriter() + frame.writeVarint(MoqMessageType.Subscribe.code) + frame.writeVarint(payload.size.toLong()) + frame.writeBytes(payload.toByteArray()) + + assertFailsWith { MoqCodec.decode(frame.toByteArray()) } + } +}