mirror of
https://github.com/vitorpamplona/amethyst.git
synced 2026-07-22 15:52:22 +00:00
feat(audio-rooms): moq-lite codec primitives + message types (phase 5a/5b)
First two phases of the moq-lite (Lite-03) implementation per
nestsClient/plans/2026-04-26-moq-lite-gap.md. No production wiring
yet — the production helpers still call the IETF MoqSession.
What landed:
- MoqLitePath — mandatory wire-boundary normalisation (strip
leading/trailing/duplicate `/`), join, path-component-aware
stripPrefix. Mirrors rs/moq-lite/src/path.rs semantics.
- MoqLiteAlpn / MoqLiteControlType / MoqLiteDataType /
MoqLiteAnnounceStatus / MoqLiteSubscribeResponseType — wire
enums for ALPN ("moq-lite-03"), per-bidi ControlType varints,
Group uni-stream type byte, announce status, subscribe
response type.
- Message data classes — AnnouncePlease, Announce, Subscribe,
SubscribeOk, SubscribeDrop, GroupHeader, Probe.
- MoqLiteCodec — encode/decode for each message, with the
size-prefix envelope baked in. Handles the off-by-one
`0 = None, n = Some(n−1)` trick for startGroup/endGroup,
coerces booleans to 0/1, validates priority fits in u8,
rejects unknown status/response bytes. Path normalisation
applied at every encode + decode boundary.
- MoqLitePathTest, MoqLiteCodecTest — round-trip tests for
every codec entry point + negative paths (oversized priority,
invalid ordered byte, unknown status, trailing garbage).
All MoqWriter / MoqReader / MoqCodecException primitives reused
from the IETF MoQ codec — same varint and length-prefix shapes.
This commit is contained in:
Claude
@@ -0,0 +1,261 @@
|
||||
/*
|
||||
* 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.moq.MoqReader
|
||||
import com.vitorpamplona.nestsclient.moq.MoqWriter
|
||||
|
||||
/**
|
||||
* Encode/decode moq-lite (Lite-03) messages. Every message on the wire
|
||||
* is **size-prefixed**: a varint length, then the payload bytes.
|
||||
*
|
||||
* This codec produces the size-prefixed envelope on encode and consumes
|
||||
* it on decode. The control-stream framing in `MoqLiteSession` reads
|
||||
* one size-prefix at a time, slices the payload, and dispatches.
|
||||
*
|
||||
* `MoqLitePath.normalize` is applied to every path string at the wire
|
||||
* boundary so a non-normalised input from the application layer can't
|
||||
* silently produce wire bytes that the relay won't match against
|
||||
* `claims.root` or against another peer's broadcast lookup.
|
||||
*
|
||||
* The off-by-one encoding for `startGroup` / `endGroup` (Subscribe
|
||||
* Lite-03 only) is `0 = None, n = Some(n − 1)`. The data-class fields
|
||||
* carry the `Some(value)` form (or null for None); the codec applies
|
||||
* the `+1` / `−1` shift here.
|
||||
*
|
||||
* Source: `kixelated/moq-rs/rs/moq-lite/src/lite/{announce,subscribe,
|
||||
* group}.rs`, the equivalent `@moq/lite/lite/` JS files, and varint
|
||||
* encoding per RFC 9000 §16.
|
||||
*/
|
||||
object MoqLiteCodec {
|
||||
// ---------------- AnnouncePlease ----------------
|
||||
|
||||
fun encodeAnnouncePlease(msg: MoqLiteAnnouncePlease): ByteArray {
|
||||
val body = MoqWriter()
|
||||
body.writeLengthPrefixedString(MoqLitePath.normalize(msg.prefix))
|
||||
return wrapSizePrefixed(body)
|
||||
}
|
||||
|
||||
fun decodeAnnouncePlease(payload: ByteArray): MoqLiteAnnouncePlease {
|
||||
val r = MoqReader(payload)
|
||||
val prefix = MoqLitePath.normalize(r.readLengthPrefixedString())
|
||||
ensureFullyConsumed(r, "AnnouncePlease")
|
||||
return MoqLiteAnnouncePlease(prefix = prefix)
|
||||
}
|
||||
|
||||
// ---------------- Announce ----------------
|
||||
|
||||
fun encodeAnnounce(msg: MoqLiteAnnounce): ByteArray {
|
||||
val body = MoqWriter()
|
||||
body.writeByte(msg.status.code)
|
||||
body.writeLengthPrefixedString(MoqLitePath.normalize(msg.suffix))
|
||||
body.writeVarint(msg.hops)
|
||||
return wrapSizePrefixed(body)
|
||||
}
|
||||
|
||||
fun decodeAnnounce(payload: ByteArray): MoqLiteAnnounce {
|
||||
val r = MoqReader(payload)
|
||||
val statusByte = r.readByte()
|
||||
val status =
|
||||
MoqLiteAnnounceStatus.fromCode(statusByte)
|
||||
?: throw MoqCodecException("unknown moq-lite Announce status byte: $statusByte")
|
||||
val suffix = MoqLitePath.normalize(r.readLengthPrefixedString())
|
||||
val hops = r.readVarint()
|
||||
ensureFullyConsumed(r, "Announce")
|
||||
return MoqLiteAnnounce(status = status, suffix = suffix, hops = hops)
|
||||
}
|
||||
|
||||
// ---------------- Subscribe ----------------
|
||||
|
||||
fun encodeSubscribe(msg: MoqLiteSubscribe): ByteArray {
|
||||
val body = MoqWriter()
|
||||
body.writeVarint(msg.id)
|
||||
body.writeLengthPrefixedString(MoqLitePath.normalize(msg.broadcast))
|
||||
body.writeLengthPrefixedString(msg.track) // tracks are opaque, no normalize
|
||||
body.writeByte(msg.priority)
|
||||
body.writeByte(if (msg.ordered) 1 else 0)
|
||||
body.writeVarint(msg.maxLatencyMillis)
|
||||
body.writeVarint(encodeOptionalGroup(msg.startGroup))
|
||||
body.writeVarint(encodeOptionalGroup(msg.endGroup))
|
||||
return wrapSizePrefixed(body)
|
||||
}
|
||||
|
||||
fun decodeSubscribe(payload: ByteArray): MoqLiteSubscribe {
|
||||
val r = MoqReader(payload)
|
||||
val id = r.readVarint()
|
||||
val broadcast = MoqLitePath.normalize(r.readLengthPrefixedString())
|
||||
val track = r.readLengthPrefixedString()
|
||||
val priority = r.readByte()
|
||||
val ordered = decodeOrderedByte(r.readByte())
|
||||
val maxLatencyMillis = r.readVarint()
|
||||
val startGroup = decodeOptionalGroup(r.readVarint())
|
||||
val endGroup = decodeOptionalGroup(r.readVarint())
|
||||
ensureFullyConsumed(r, "Subscribe")
|
||||
return MoqLiteSubscribe(
|
||||
id = id,
|
||||
broadcast = broadcast,
|
||||
track = track,
|
||||
priority = priority,
|
||||
ordered = ordered,
|
||||
maxLatencyMillis = maxLatencyMillis,
|
||||
startGroup = startGroup,
|
||||
endGroup = endGroup,
|
||||
)
|
||||
}
|
||||
|
||||
// ---------------- Subscribe response ----------------
|
||||
|
||||
fun encodeSubscribeOk(msg: MoqLiteSubscribeOk): ByteArray {
|
||||
val body = MoqWriter()
|
||||
body.writeVarint(MoqLiteSubscribeResponseType.Ok.code)
|
||||
body.writeByte(msg.priority)
|
||||
body.writeByte(if (msg.ordered) 1 else 0)
|
||||
body.writeVarint(msg.maxLatencyMillis)
|
||||
body.writeVarint(encodeOptionalGroup(msg.startGroup))
|
||||
body.writeVarint(encodeOptionalGroup(msg.endGroup))
|
||||
return wrapSizePrefixed(body)
|
||||
}
|
||||
|
||||
fun encodeSubscribeDrop(msg: MoqLiteSubscribeDrop): ByteArray {
|
||||
val body = MoqWriter()
|
||||
body.writeVarint(MoqLiteSubscribeResponseType.Drop.code)
|
||||
body.writeVarint(msg.errorCode)
|
||||
body.writeLengthPrefixedString(msg.reasonPhrase)
|
||||
return wrapSizePrefixed(body)
|
||||
}
|
||||
|
||||
/**
|
||||
* Decode either an [MoqLiteSubscribeOk] or [MoqLiteSubscribeDrop]
|
||||
* — the response stream tells which by its leading varint type.
|
||||
*/
|
||||
sealed class SubscribeResponse {
|
||||
data class Ok(
|
||||
val ok: MoqLiteSubscribeOk,
|
||||
) : SubscribeResponse()
|
||||
|
||||
data class Dropped(
|
||||
val drop: MoqLiteSubscribeDrop,
|
||||
) : SubscribeResponse()
|
||||
}
|
||||
|
||||
fun decodeSubscribeResponse(payload: ByteArray): SubscribeResponse {
|
||||
val r = MoqReader(payload)
|
||||
val typeCode = r.readVarint()
|
||||
val type =
|
||||
MoqLiteSubscribeResponseType.fromCode(typeCode)
|
||||
?: throw MoqCodecException("unknown moq-lite SubscribeResponse type: $typeCode")
|
||||
return when (type) {
|
||||
MoqLiteSubscribeResponseType.Ok -> {
|
||||
val priority = r.readByte()
|
||||
val ordered = decodeOrderedByte(r.readByte())
|
||||
val maxLatencyMillis = r.readVarint()
|
||||
val startGroup = decodeOptionalGroup(r.readVarint())
|
||||
val endGroup = decodeOptionalGroup(r.readVarint())
|
||||
ensureFullyConsumed(r, "SubscribeOk")
|
||||
SubscribeResponse.Ok(
|
||||
MoqLiteSubscribeOk(
|
||||
priority = priority,
|
||||
ordered = ordered,
|
||||
maxLatencyMillis = maxLatencyMillis,
|
||||
startGroup = startGroup,
|
||||
endGroup = endGroup,
|
||||
),
|
||||
)
|
||||
}
|
||||
|
||||
MoqLiteSubscribeResponseType.Drop -> {
|
||||
val errorCode = r.readVarint()
|
||||
val reason = r.readLengthPrefixedString()
|
||||
ensureFullyConsumed(r, "SubscribeDrop")
|
||||
SubscribeResponse.Dropped(
|
||||
MoqLiteSubscribeDrop(errorCode = errorCode, reasonPhrase = reason),
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------- Group header ----------------
|
||||
|
||||
fun encodeGroupHeader(msg: MoqLiteGroupHeader): ByteArray {
|
||||
val body = MoqWriter()
|
||||
body.writeVarint(msg.subscribeId)
|
||||
body.writeVarint(msg.sequence)
|
||||
return wrapSizePrefixed(body)
|
||||
}
|
||||
|
||||
fun decodeGroupHeader(payload: ByteArray): MoqLiteGroupHeader {
|
||||
val r = MoqReader(payload)
|
||||
val subscribeId = r.readVarint()
|
||||
val sequence = r.readVarint()
|
||||
ensureFullyConsumed(r, "GroupHeader")
|
||||
return MoqLiteGroupHeader(subscribeId = subscribeId, sequence = sequence)
|
||||
}
|
||||
|
||||
// ---------------- Probe ----------------
|
||||
|
||||
fun decodeProbe(payload: ByteArray): MoqLiteProbe {
|
||||
val r = MoqReader(payload)
|
||||
val bitrate = r.readVarint()
|
||||
ensureFullyConsumed(r, "Probe")
|
||||
return MoqLiteProbe(bitrate = bitrate)
|
||||
}
|
||||
|
||||
// ---------------- internals ----------------
|
||||
|
||||
/**
|
||||
* Wrap a body buffer in a varint size prefix. Every moq-lite
|
||||
* control-stream message uses this envelope (see e.g.
|
||||
* `lite/announce.rs:64-81`, `lite/subscribe.rs:25-72`).
|
||||
*/
|
||||
private fun wrapSizePrefixed(body: MoqWriter): ByteArray {
|
||||
val payload = body.toByteArray()
|
||||
val out = MoqWriter(payload.size + 8)
|
||||
out.writeLengthPrefixedBytes(payload)
|
||||
return out.toByteArray()
|
||||
}
|
||||
|
||||
/**
|
||||
* `0 = None, n = Some(n − 1)`. The encoding lets the wire format
|
||||
* collapse "no bound" into a single zero byte.
|
||||
*/
|
||||
private fun encodeOptionalGroup(value: Long?): Long = if (value == null) 0L else value + 1
|
||||
|
||||
private fun decodeOptionalGroup(raw: Long): Long? = if (raw == 0L) null else raw - 1
|
||||
|
||||
private fun decodeOrderedByte(b: Int): Boolean =
|
||||
when (b) {
|
||||
0 -> false
|
||||
1 -> true
|
||||
else -> throw MoqCodecException("moq-lite ordered byte must be 0/1, got $b")
|
||||
}
|
||||
|
||||
private fun ensureFullyConsumed(
|
||||
r: MoqReader,
|
||||
msg: String,
|
||||
) {
|
||||
if (r.hasMore()) {
|
||||
throw MoqCodecException(
|
||||
"trailing $msg payload bytes (${r.remaining} left) — wire format mismatch",
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,239 @@
|
||||
/*
|
||||
* 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
|
||||
|
||||
/**
|
||||
* moq-lite ALPN strings. Lite-03 is preferred; `"moql"` is the legacy
|
||||
* combined ALPN that requires an in-band SETUP exchange.
|
||||
*
|
||||
* Source: `kixelated/moq-rs/rs/moq-lite/src/version.rs:21-26`,
|
||||
* `@moq/lite/connection/connect.js:277`.
|
||||
*/
|
||||
object MoqLiteAlpn {
|
||||
const val LITE_03: String = "moq-lite-03"
|
||||
const val LEGACY: String = "moql"
|
||||
}
|
||||
|
||||
/**
|
||||
* ControlType varint discriminator written as the first datum on every
|
||||
* client-initiated bidi stream. Selects which message body the peer
|
||||
* should expect to read next.
|
||||
*
|
||||
* Source: `rs/moq-lite/src/lite/stream.rs:7-15`.
|
||||
*/
|
||||
enum class MoqLiteControlType(
|
||||
val code: Long,
|
||||
) {
|
||||
/** Lite-01/02 only — unused on Lite-03. Reserved here for completeness. */
|
||||
Session(0L),
|
||||
Announce(1L),
|
||||
Subscribe(2L),
|
||||
Fetch(3L),
|
||||
Probe(4L),
|
||||
;
|
||||
|
||||
companion object {
|
||||
private val byCode = entries.associateBy { it.code }
|
||||
|
||||
fun fromCode(code: Long): MoqLiteControlType? = byCode[code]
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* DataType varint written as the first byte of every uni stream that
|
||||
* carries payload. Lite-03 currently uses `Group=0` only.
|
||||
*
|
||||
* Source: `rs/moq-lite/src/lite/stream.rs:32-36`.
|
||||
*/
|
||||
enum class MoqLiteDataType(
|
||||
val code: Long,
|
||||
) {
|
||||
Group(0L),
|
||||
;
|
||||
|
||||
companion object {
|
||||
private val byCode = entries.associateBy { it.code }
|
||||
|
||||
fun fromCode(code: Long): MoqLiteDataType? = byCode[code]
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Status byte at the head of an [MoqLiteAnnounce] payload. `Active=1`
|
||||
* is sent on first publish; `Ended=0` on explicit unannounce. Disconnect
|
||||
* is NOT signalled with `Ended` — the bidi just closes.
|
||||
*
|
||||
* Source: `rs/moq-lite/src/lite/announce.rs:84-90`.
|
||||
*/
|
||||
enum class MoqLiteAnnounceStatus(
|
||||
val code: Int,
|
||||
) {
|
||||
Ended(0),
|
||||
Active(1),
|
||||
;
|
||||
|
||||
companion object {
|
||||
fun fromCode(code: Int): MoqLiteAnnounceStatus? =
|
||||
when (code) {
|
||||
0 -> Ended
|
||||
1 -> Active
|
||||
else -> null
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Type byte at the head of a [MoqLiteSubscribeResponse] payload.
|
||||
*
|
||||
* Source: `rs/moq-lite/src/lite/subscribe.rs:271-328`,
|
||||
* `@moq/lite/lite/subscribe.js:264-282`.
|
||||
*/
|
||||
enum class MoqLiteSubscribeResponseType(
|
||||
val code: Long,
|
||||
) {
|
||||
Ok(0L),
|
||||
Drop(1L),
|
||||
;
|
||||
|
||||
companion object {
|
||||
private val byCode = entries.associateBy { it.code }
|
||||
|
||||
fun fromCode(code: Long): MoqLiteSubscribeResponseType? = byCode[code]
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* "I'm interested in broadcasts under this prefix" — the first message
|
||||
* the subscriber writes on an Announce bidi.
|
||||
*
|
||||
* Wire layout (size-prefixed):
|
||||
* prefix string (varint length + UTF-8)
|
||||
*
|
||||
* Empty prefix means "everything".
|
||||
*/
|
||||
data class MoqLiteAnnouncePlease(
|
||||
val prefix: String,
|
||||
)
|
||||
|
||||
/**
|
||||
* Per-broadcast announce update streamed by the publisher (or relay)
|
||||
* back to the subscriber on the same Announce bidi. One message per
|
||||
* known broadcast at attach time, then live updates as broadcasts come
|
||||
* and go.
|
||||
*
|
||||
* Wire layout (size-prefixed):
|
||||
* status u8 (0 = Ended, 1 = Active)
|
||||
* suffix string (broadcast path with the requested `prefix`
|
||||
* stripped — `MoqLitePath.join(prefix, suffix)`
|
||||
* reconstitutes the absolute path)
|
||||
* hops u62 (relay routing depth, Lite-03 only)
|
||||
*/
|
||||
data class MoqLiteAnnounce(
|
||||
val status: MoqLiteAnnounceStatus,
|
||||
val suffix: String,
|
||||
val hops: Long,
|
||||
)
|
||||
|
||||
/**
|
||||
* "Subscribe me to (broadcast, track)" — the first message the
|
||||
* subscriber writes on a Subscribe bidi.
|
||||
*
|
||||
* Wire layout (size-prefixed):
|
||||
* id u62 varint (subscriber-chosen, monotonically increasing)
|
||||
* broadcast string (absolute broadcast path, normalized)
|
||||
* track string (opaque app string — `"audio/data"` /
|
||||
* `"catalog.json"` for nests)
|
||||
* priority u8 (raw byte 0..255)
|
||||
* ordered u8 (Lite-03; 0/1)
|
||||
* maxLatency varint (Lite-03; *milliseconds*; 0 = unlimited)
|
||||
* startGroup varint (Lite-03; 0 = "from latest",
|
||||
* else group_seq + 1)
|
||||
* endGroup varint (Lite-03; 0 = "no end",
|
||||
* else group_seq + 1)
|
||||
*/
|
||||
data class MoqLiteSubscribe(
|
||||
val id: Long,
|
||||
val broadcast: String,
|
||||
val track: String,
|
||||
val priority: Int,
|
||||
val ordered: Boolean,
|
||||
val maxLatencyMillis: Long,
|
||||
val startGroup: Long?,
|
||||
val endGroup: Long?,
|
||||
) {
|
||||
init {
|
||||
require(priority in 0..255) { "moq-lite priority must fit in a byte: $priority" }
|
||||
require(maxLatencyMillis >= 0) { "maxLatencyMillis must be non-negative: $maxLatencyMillis" }
|
||||
if (startGroup != null) require(startGroup >= 0) { "startGroup must be non-negative: $startGroup" }
|
||||
if (endGroup != null) require(endGroup >= 0) { "endGroup must be non-negative: $endGroup" }
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Publisher's accept reply to a [MoqLiteSubscribe]. Echoes the
|
||||
* negotiated subscription parameters; the publisher may have narrowed
|
||||
* `startGroup` / `endGroup` from the subscriber's request.
|
||||
*/
|
||||
data class MoqLiteSubscribeOk(
|
||||
val priority: Int,
|
||||
val ordered: Boolean,
|
||||
val maxLatencyMillis: Long,
|
||||
val startGroup: Long?,
|
||||
val endGroup: Long?,
|
||||
)
|
||||
|
||||
/**
|
||||
* Publisher's reject / drop reply. moq-lite has no SUBSCRIBE_ERROR
|
||||
* code; failures during a live subscription are conveyed by
|
||||
* RESET_STREAM, but the publisher can pre-emptively decline a
|
||||
* subscription with [MoqLiteSubscribeDrop] before any group flows.
|
||||
*
|
||||
* Decode-only as far as the client cares — we never send Drop back
|
||||
* upstream.
|
||||
*/
|
||||
data class MoqLiteSubscribeDrop(
|
||||
val errorCode: Long,
|
||||
val reasonPhrase: String,
|
||||
)
|
||||
|
||||
/**
|
||||
* Header at the start of a Group uni stream. After the
|
||||
* [MoqLiteDataType.Group] type byte, the publisher writes one
|
||||
* size-prefixed [MoqLiteGroupHeader] payload, then a sequence of
|
||||
* `varint(size) + payload` frames until QUIC FIN.
|
||||
*/
|
||||
data class MoqLiteGroupHeader(
|
||||
val subscribeId: Long,
|
||||
val sequence: Long,
|
||||
)
|
||||
|
||||
/**
|
||||
* Probe message written by the *publisher* on a subscriber-initiated
|
||||
* Probe bidi (ControlType=4). `bitrate` is encoded as a u62 varint in
|
||||
* a size-prefixed body. Decode-only for now — the listener path
|
||||
* exposes the most recent reading, and we don't initiate probes from
|
||||
* the speaker side.
|
||||
*
|
||||
* Source: `rs/moq-lite/src/lite/probe.rs`.
|
||||
*/
|
||||
data class MoqLiteProbe(
|
||||
val bitrate: Long,
|
||||
)
|
||||
@@ -0,0 +1,101 @@
|
||||
/*
|
||||
* 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
|
||||
|
||||
/**
|
||||
* Mandatory path-normalization for moq-lite (Lite-03), mirroring
|
||||
* `kixelated/moq-rs/rs/moq-lite/src/path.rs`. moq-lite represents
|
||||
* broadcast and announce paths as plain UTF-8 strings; the wire format
|
||||
* is `varint(length) + UTF-8 bytes`. Both peers are required to
|
||||
* normalize before encoding and after decoding — without that, a wire
|
||||
* path `"/foo//bar/"` and `"foo/bar"` would be treated as distinct and
|
||||
* broadcast lookups silently fail.
|
||||
*
|
||||
* Normalization rules (mirroring `Path::new` semantics):
|
||||
* - strip every leading `/`
|
||||
* - strip every trailing `/`
|
||||
* - collapse runs of `/` into a single `/`
|
||||
*
|
||||
* Empty paths are valid (the relay's "everything under root" prefix).
|
||||
*/
|
||||
object MoqLitePath {
|
||||
/**
|
||||
* Apply the moq-lite normalization rules. Returns the canonical path
|
||||
* string for [s].
|
||||
*/
|
||||
fun normalize(s: String): String {
|
||||
if (s.isEmpty()) return s
|
||||
val out = StringBuilder(s.length)
|
||||
var lastWasSlash = true
|
||||
for (i in s.indices) {
|
||||
val c = s[i]
|
||||
if (c == '/') {
|
||||
if (!lastWasSlash) out.append('/')
|
||||
lastWasSlash = true
|
||||
} else {
|
||||
out.append(c)
|
||||
lastWasSlash = false
|
||||
}
|
||||
}
|
||||
// Strip the trailing `/` left behind by an input ending in `/`.
|
||||
if (out.isNotEmpty() && out[out.length - 1] == '/') {
|
||||
out.deleteCharAt(out.length - 1)
|
||||
}
|
||||
return out.toString()
|
||||
}
|
||||
|
||||
/**
|
||||
* Concatenate two path components per moq-lite semantics. Each side
|
||||
* is normalized first; if either is empty, returns the other (no
|
||||
* leading or trailing `/`). Otherwise emits `"<prefix>/<suffix>"`.
|
||||
*/
|
||||
fun join(
|
||||
prefix: String,
|
||||
suffix: String,
|
||||
): String {
|
||||
val a = normalize(prefix)
|
||||
val b = normalize(suffix)
|
||||
if (a.isEmpty()) return b
|
||||
if (b.isEmpty()) return a
|
||||
return "$a/$b"
|
||||
}
|
||||
|
||||
/**
|
||||
* If [path] starts with [prefix] (path-component-aware — `"foo"` does
|
||||
* NOT match `"foobar"`), return the suffix that remains. Returns
|
||||
* null if the prefix does not match.
|
||||
*
|
||||
* Both inputs are normalized before comparison.
|
||||
*/
|
||||
fun stripPrefix(
|
||||
prefix: String,
|
||||
path: String,
|
||||
): String? {
|
||||
val p = normalize(prefix)
|
||||
val full = normalize(path)
|
||||
if (p.isEmpty()) return full
|
||||
if (full == p) return ""
|
||||
if (full.length > p.length && full.startsWith(p) && full[p.length] == '/') {
|
||||
return full.substring(p.length + 1)
|
||||
}
|
||||
return null
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,264 @@
|
||||
/*
|
||||
* 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.moq.MoqReader
|
||||
import com.vitorpamplona.quic.Varint
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertContentEquals
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertFailsWith
|
||||
import kotlin.test.assertIs
|
||||
import kotlin.test.assertTrue
|
||||
|
||||
/**
|
||||
* Encode/decode round-trip tests for every moq-lite (Lite-03) message
|
||||
* type the client uses. Each test encodes a hand-built data class to
|
||||
* bytes, peels off the size-prefix to confirm the framing envelope,
|
||||
* then decodes the payload and asserts equality with the input.
|
||||
*
|
||||
* Wire-format expectations are anchored against the byte layout
|
||||
* documented in `nestsClient/plans/2026-04-26-moq-lite-gap.md` (which
|
||||
* cites `kixelated/moq-rs/rs/moq-lite/src/`).
|
||||
*/
|
||||
class MoqLiteCodecTest {
|
||||
@Test
|
||||
fun announcePlease_round_trips() {
|
||||
val msg = MoqLiteAnnouncePlease(prefix = "nests/30312:abc:room")
|
||||
val encoded = MoqLiteCodec.encodeAnnouncePlease(msg)
|
||||
val payload = peelSizePrefix(encoded)
|
||||
assertEquals(msg, MoqLiteCodec.decodeAnnouncePlease(payload))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun announcePlease_normalizes_prefix_on_encode_and_decode() {
|
||||
val raw = MoqLiteAnnouncePlease(prefix = "/nests//30312:abc:room/")
|
||||
val payload = peelSizePrefix(MoqLiteCodec.encodeAnnouncePlease(raw))
|
||||
// The wire bytes are normalised — no leading/trailing/duplicate `/`.
|
||||
val r = MoqReader(payload)
|
||||
assertEquals("nests/30312:abc:room", r.readLengthPrefixedString())
|
||||
assertTrue(!r.hasMore())
|
||||
}
|
||||
|
||||
@Test
|
||||
fun announce_active_round_trips() {
|
||||
val msg = MoqLiteAnnounce(status = MoqLiteAnnounceStatus.Active, suffix = "speakerPubkey", hops = 0L)
|
||||
val payload = peelSizePrefix(MoqLiteCodec.encodeAnnounce(msg))
|
||||
// Byte 0 is the status byte (literal 1), then a 13-byte string,
|
||||
// then a varint hops.
|
||||
assertEquals(1, payload[0].toInt())
|
||||
assertEquals(msg, MoqLiteCodec.decodeAnnounce(payload))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun announce_ended_with_relay_hops_round_trips() {
|
||||
val msg = MoqLiteAnnounce(status = MoqLiteAnnounceStatus.Ended, suffix = "fff", hops = 7L)
|
||||
val payload = peelSizePrefix(MoqLiteCodec.encodeAnnounce(msg))
|
||||
assertEquals(0, payload[0].toInt())
|
||||
assertEquals(msg, MoqLiteCodec.decodeAnnounce(payload))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun announce_rejects_unknown_status_byte() {
|
||||
// Hand-craft a payload with an invalid status byte. Wrap in the
|
||||
// size-prefix envelope just like a real message would arrive.
|
||||
val handCraft = byteArrayOf(0x05.toByte()) + lengthPrefixed("x") + Varint.encode(0L)
|
||||
assertFailsWith<MoqCodecException> { MoqLiteCodec.decodeAnnounce(handCraft) }
|
||||
}
|
||||
|
||||
@Test
|
||||
fun subscribe_round_trips_with_all_groups_set() {
|
||||
val msg =
|
||||
MoqLiteSubscribe(
|
||||
id = 42L,
|
||||
broadcast = "nests/30312:abc:room",
|
||||
track = "audio/data",
|
||||
priority = 0x80,
|
||||
ordered = true,
|
||||
maxLatencyMillis = 250L,
|
||||
startGroup = 5L,
|
||||
endGroup = 9L,
|
||||
)
|
||||
val payload = peelSizePrefix(MoqLiteCodec.encodeSubscribe(msg))
|
||||
assertEquals(msg, MoqLiteCodec.decodeSubscribe(payload))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun subscribe_round_trips_with_null_group_bounds() {
|
||||
val msg =
|
||||
MoqLiteSubscribe(
|
||||
id = 1L,
|
||||
broadcast = "speakerPubkey",
|
||||
track = "catalog.json",
|
||||
priority = 0,
|
||||
ordered = false,
|
||||
maxLatencyMillis = 0L,
|
||||
startGroup = null,
|
||||
endGroup = null,
|
||||
)
|
||||
val payload = peelSizePrefix(MoqLiteCodec.encodeSubscribe(msg))
|
||||
assertEquals(msg, MoqLiteCodec.decodeSubscribe(payload))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun subscribe_off_by_one_group_encoding_uses_zero_for_none() {
|
||||
val msg = subscribeWith(startGroup = null, endGroup = null)
|
||||
val payload = peelSizePrefix(MoqLiteCodec.encodeSubscribe(msg))
|
||||
// Skip past id (varint), broadcast/track strings, priority byte,
|
||||
// ordered byte, maxLatency varint — pull the last two varints.
|
||||
val r = MoqReader(payload)
|
||||
r.readVarint() // id
|
||||
r.readLengthPrefixedString() // broadcast
|
||||
r.readLengthPrefixedString() // track
|
||||
r.readByte() // priority
|
||||
r.readByte() // ordered
|
||||
r.readVarint() // maxLatency
|
||||
assertEquals(0L, r.readVarint(), "startGroup=None encodes as 0 on the wire")
|
||||
assertEquals(0L, r.readVarint(), "endGroup=None encodes as 0 on the wire")
|
||||
}
|
||||
|
||||
@Test
|
||||
fun subscribe_off_by_one_some_zero_encodes_as_one() {
|
||||
val msg = subscribeWith(startGroup = 0L, endGroup = 0L)
|
||||
val payload = peelSizePrefix(MoqLiteCodec.encodeSubscribe(msg))
|
||||
val r = MoqReader(payload)
|
||||
r.readVarint()
|
||||
r.readLengthPrefixedString()
|
||||
r.readLengthPrefixedString()
|
||||
r.readByte()
|
||||
r.readByte()
|
||||
r.readVarint()
|
||||
assertEquals(1L, r.readVarint(), "Some(0) encodes as 1 (off-by-one trick)")
|
||||
assertEquals(1L, r.readVarint(), "Some(0) encodes as 1 (off-by-one trick)")
|
||||
}
|
||||
|
||||
@Test
|
||||
fun subscribe_rejects_oversized_priority_at_construction() {
|
||||
assertFailsWith<IllegalArgumentException> {
|
||||
MoqLiteSubscribe(
|
||||
id = 0L,
|
||||
broadcast = "x",
|
||||
track = "y",
|
||||
priority = 256,
|
||||
ordered = false,
|
||||
maxLatencyMillis = 0L,
|
||||
startGroup = null,
|
||||
endGroup = null,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun subscribeOk_round_trips() {
|
||||
val msg =
|
||||
MoqLiteSubscribeOk(
|
||||
priority = 0x80,
|
||||
ordered = true,
|
||||
maxLatencyMillis = 100L,
|
||||
startGroup = null,
|
||||
endGroup = null,
|
||||
)
|
||||
val payload = peelSizePrefix(MoqLiteCodec.encodeSubscribeOk(msg))
|
||||
val resp = MoqLiteCodec.decodeSubscribeResponse(payload)
|
||||
val ok = assertIs<MoqLiteCodec.SubscribeResponse.Ok>(resp)
|
||||
assertEquals(msg, ok.ok)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun subscribeDrop_round_trips() {
|
||||
val msg = MoqLiteSubscribeDrop(errorCode = 0x12L, reasonPhrase = "publisher gone")
|
||||
val payload = peelSizePrefix(MoqLiteCodec.encodeSubscribeDrop(msg))
|
||||
val resp = MoqLiteCodec.decodeSubscribeResponse(payload)
|
||||
val drop = assertIs<MoqLiteCodec.SubscribeResponse.Dropped>(resp)
|
||||
assertEquals(msg, drop.drop)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun groupHeader_round_trips() {
|
||||
val msg = MoqLiteGroupHeader(subscribeId = 42L, sequence = 7L)
|
||||
val payload = peelSizePrefix(MoqLiteCodec.encodeGroupHeader(msg))
|
||||
assertEquals(msg, MoqLiteCodec.decodeGroupHeader(payload))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun decoders_reject_trailing_garbage() {
|
||||
// Build a valid AnnouncePlease payload then append a stray byte.
|
||||
val payload = peelSizePrefix(MoqLiteCodec.encodeAnnouncePlease(MoqLiteAnnouncePlease("foo")))
|
||||
val tampered = payload + byteArrayOf(0)
|
||||
assertFailsWith<MoqCodecException> { MoqLiteCodec.decodeAnnouncePlease(tampered) }
|
||||
}
|
||||
|
||||
@Test
|
||||
fun decoders_reject_invalid_ordered_byte() {
|
||||
val msg = subscribeWith(startGroup = null, endGroup = null)
|
||||
val payload = peelSizePrefix(MoqLiteCodec.encodeSubscribe(msg)).copyOf()
|
||||
// Patch the ordered byte. Layout: varint id (1B for id=0),
|
||||
// string broadcast (1B len + bytes), string track (1B len + bytes),
|
||||
// priority byte, ordered byte. So index = 1 + 1 + b.length + 1 + t.length + 1.
|
||||
val idx = 1 + 1 + msg.broadcast.length + 1 + msg.track.length + 1
|
||||
payload[idx] = 7
|
||||
assertFailsWith<MoqCodecException> { MoqLiteCodec.decodeSubscribe(payload) }
|
||||
}
|
||||
|
||||
private fun subscribeWith(
|
||||
startGroup: Long?,
|
||||
endGroup: Long?,
|
||||
) = MoqLiteSubscribe(
|
||||
id = 0L,
|
||||
broadcast = "b",
|
||||
track = "t",
|
||||
priority = 0,
|
||||
ordered = false,
|
||||
maxLatencyMillis = 0L,
|
||||
startGroup = startGroup,
|
||||
endGroup = endGroup,
|
||||
)
|
||||
|
||||
private fun lengthPrefixed(s: String): ByteArray {
|
||||
val bytes = s.encodeToByteArray()
|
||||
return Varint.encode(bytes.size.toLong()) + bytes
|
||||
}
|
||||
|
||||
/**
|
||||
* Peel the varint size-prefix off an encoded message. Tests work
|
||||
* directly on the *payload* (no envelope) since the session-layer
|
||||
* is the one that frames the envelope onto/off the wire.
|
||||
*/
|
||||
private fun peelSizePrefix(encoded: ByteArray): ByteArray {
|
||||
val sizeDecode = Varint.decode(encoded) ?: error("envelope missing size prefix")
|
||||
val payload = encoded.copyOfRange(sizeDecode.bytesConsumed, encoded.size)
|
||||
assertEquals(
|
||||
sizeDecode.value.toInt(),
|
||||
payload.size,
|
||||
"size prefix must equal payload length",
|
||||
)
|
||||
return payload
|
||||
}
|
||||
|
||||
@Suppress("unused")
|
||||
private fun byteHex(bytes: ByteArray): String = bytes.joinToString(" ") { (it.toInt() and 0xFF).toString(16).padStart(2, '0') }
|
||||
|
||||
@Test
|
||||
fun checks_byteHex_helper_compiles() {
|
||||
assertContentEquals(byteArrayOf(0x10, 0x20), byteArrayOf(0x10, 0x20))
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,83 @@
|
||||
/*
|
||||
* 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 kotlin.test.Test
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertNull
|
||||
|
||||
class MoqLitePathTest {
|
||||
@Test
|
||||
fun normalize_strips_leading_trailing_and_collapses_duplicates() {
|
||||
assertEquals("foo/bar", MoqLitePath.normalize("/foo/bar/"))
|
||||
assertEquals("foo/bar", MoqLitePath.normalize("foo//bar"))
|
||||
assertEquals("foo/bar", MoqLitePath.normalize("///foo///bar///"))
|
||||
assertEquals("foo", MoqLitePath.normalize("/foo/"))
|
||||
assertEquals("", MoqLitePath.normalize(""))
|
||||
assertEquals("", MoqLitePath.normalize("/"))
|
||||
assertEquals("", MoqLitePath.normalize("/////"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun normalize_preserves_already_canonical_paths() {
|
||||
assertEquals("foo/bar", MoqLitePath.normalize("foo/bar"))
|
||||
assertEquals(
|
||||
"nests/30312:abc:room/${"f".repeat(64)}",
|
||||
MoqLitePath.normalize("nests/30312:abc:room/${"f".repeat(64)}"),
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun join_handles_empty_sides() {
|
||||
assertEquals("foo", MoqLitePath.join("", "foo"))
|
||||
assertEquals("foo", MoqLitePath.join("foo", ""))
|
||||
assertEquals("", MoqLitePath.join("", ""))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun join_inserts_exactly_one_slash() {
|
||||
assertEquals("a/b", MoqLitePath.join("a", "b"))
|
||||
// Sides are normalized first — the join can't produce `//`.
|
||||
assertEquals("a/b", MoqLitePath.join("a/", "/b"))
|
||||
assertEquals("nests/30312:abc:room/pubkey", MoqLitePath.join("nests/30312:abc:room", "pubkey"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun stripPrefix_is_path_component_aware() {
|
||||
assertEquals("baz", MoqLitePath.stripPrefix("foo/bar", "foo/bar/baz"))
|
||||
assertEquals("", MoqLitePath.stripPrefix("foo/bar", "foo/bar"))
|
||||
// Prefix must end on a path-component boundary.
|
||||
assertNull(MoqLitePath.stripPrefix("foo", "foobar"))
|
||||
assertNull(MoqLitePath.stripPrefix("foo/bar", "foo/barbaz"))
|
||||
// Disjoint paths return null.
|
||||
assertNull(MoqLitePath.stripPrefix("foo/bar", "baz"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun stripPrefix_normalizes_both_sides() {
|
||||
assertEquals("baz", MoqLitePath.stripPrefix("/foo/bar/", "/foo//bar/baz/"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun stripPrefix_with_empty_prefix_returns_normalized_path() {
|
||||
assertEquals("foo/bar", MoqLitePath.stripPrefix("", "/foo//bar/"))
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user