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
+261
@@ -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",
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
+239
@@ -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,
|
||||
)
|
||||
+101
@@ -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
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user