feat(nestsClient): MoQ varint + SETUP handshake codec
Phase 3c-1 of the Clubhouse/nests integration. Lands a MoQ-transport control-plane codec (draft-ietf-moq-transport) and a session wrapper that runs the SETUP handshake over any WebTransportSession. Purely commonMain + fully tested end-to-end against FakeWebTransport — no network or Kwik dependency required. commonMain (nestsclient.moq): - `Varint` — QUIC variable-length integer codec per RFC 9000 §16. Encode/decode/size, with typed truncation handling (decode returns null so callers can buffer more and retry). - `MoqWriter` / `MoqReader` — append-only byte writer + bounds-checked reader used by the message codec. - `MoqCodecException` — typed error for malformed frames. - `MoqMessage` sealed class + `MoqMessageType` enum + `SetupParameter`. Phase 3c-1 covers just `ClientSetup` / `ServerSetup` (message types 0x40 / 0x41). - `MoqCodec.encode/decode` — wraps payload with `type (varint) + length (varint)`. Rejects unknown types and trailing bytes inside a declared payload window. - `MoqSession.client/server` — attaches to a WebTransportSession and runs the CLIENT_SETUP / SERVER_SETUP handshake with version negotiation + configurable timeout. - `MoqVersion` constants for draft-11 and draft-17. Tests: - `VarintTest` — all four RFC 9000 §A.1 sample vectors (1/2/4/8 byte), boundary round-trips, negative/overflow rejection, short-buffer returns null, bytesConsumed accuracy. - `MoqCodecTest` — CLIENT_SETUP / SERVER_SETUP round-trip (empty and multi-param), multi-version negotiation, exact wire layout for ServerSetup(1L), truncated-frame returns null, concatenated frames decode with offset, unknown type rejected, trailing bytes rejected, zero-length parameter values allowed. - `MoqSessionTest` — full SETUP exchange over FakeWebTransport (both sides happy path, client falling back to older version, server rejecting when no version overlap + client timing out cleanly). SUBSCRIBE / ANNOUNCE / OBJECT_STREAM / OBJECT_DATAGRAM land in Phase 3c-2 on top of the same MoqSession. https://claude.ai/code/session_013nVLALALKaHVgHm9u5Cg8D
This commit is contained in:
Claude
@@ -0,0 +1,130 @@
|
||||
/*
|
||||
* 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
|
||||
|
||||
/**
|
||||
* Append-only byte buffer used by the MoQ encoders. Doubles in capacity when
|
||||
* full. Kept intentionally minimal so this module has no buffer-library
|
||||
* dependency — helps keep the KMP surface thin.
|
||||
*/
|
||||
class MoqWriter(
|
||||
initialCapacity: Int = 64,
|
||||
) {
|
||||
private var buf: ByteArray = ByteArray(initialCapacity)
|
||||
private var pos: Int = 0
|
||||
|
||||
val size: Int get() = pos
|
||||
|
||||
fun toByteArray(): ByteArray = buf.copyOf(pos)
|
||||
|
||||
fun writeByte(value: Int) {
|
||||
ensure(1)
|
||||
buf[pos++] = value.toByte()
|
||||
}
|
||||
|
||||
fun writeBytes(bytes: ByteArray) {
|
||||
ensure(bytes.size)
|
||||
bytes.copyInto(buf, pos)
|
||||
pos += bytes.size
|
||||
}
|
||||
|
||||
fun writeVarint(value: Long) {
|
||||
ensure(Varint.size(value))
|
||||
pos += Varint.writeTo(value, buf, pos)
|
||||
}
|
||||
|
||||
fun writeVarint(value: Int) = writeVarint(value.toLong())
|
||||
|
||||
/** Write a length-prefixed byte array (varint length + contents). */
|
||||
fun writeLengthPrefixedBytes(bytes: ByteArray) {
|
||||
writeVarint(bytes.size.toLong())
|
||||
writeBytes(bytes)
|
||||
}
|
||||
|
||||
/** Write a length-prefixed UTF-8 string. */
|
||||
fun writeLengthPrefixedString(s: String) = writeLengthPrefixedBytes(s.encodeToByteArray())
|
||||
|
||||
private fun ensure(more: Int) {
|
||||
if (pos + more > buf.size) {
|
||||
var newSize = buf.size * 2
|
||||
while (newSize < pos + more) newSize *= 2
|
||||
buf = buf.copyOf(newSize)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Read cursor over a fully-buffered frame payload. Throws [MoqCodecException]
|
||||
* on under-read so callers get a typed error surface instead of opaque
|
||||
* [IndexOutOfBoundsException]s.
|
||||
*/
|
||||
class MoqReader(
|
||||
private val src: ByteArray,
|
||||
private var pos: Int = 0,
|
||||
private val end: Int = src.size,
|
||||
) {
|
||||
val remaining: Int get() = end - pos
|
||||
|
||||
fun hasMore(): Boolean = pos < end
|
||||
|
||||
fun readByte(): Int {
|
||||
require(1)
|
||||
return src[pos++].toInt() and 0xFF
|
||||
}
|
||||
|
||||
fun readBytes(n: Int): ByteArray {
|
||||
require(n)
|
||||
val out = src.copyOfRange(pos, pos + n)
|
||||
pos += n
|
||||
return out
|
||||
}
|
||||
|
||||
fun readVarint(): Long {
|
||||
val dec =
|
||||
Varint.decode(src, pos)
|
||||
?: throw MoqCodecException("truncated varint at offset=$pos (remaining=$remaining)")
|
||||
require(dec.bytesConsumed)
|
||||
pos += dec.bytesConsumed
|
||||
return dec.value
|
||||
}
|
||||
|
||||
fun readLengthPrefixedBytes(): ByteArray {
|
||||
val len = readVarint()
|
||||
if (len < 0 || len > Int.MAX_VALUE) throw MoqCodecException("absurd length prefix: $len")
|
||||
return readBytes(len.toInt())
|
||||
}
|
||||
|
||||
fun readLengthPrefixedString(): String = readLengthPrefixedBytes().decodeToString()
|
||||
|
||||
private fun require(n: Int) {
|
||||
if (pos + n > end) {
|
||||
throw MoqCodecException(
|
||||
"short read at offset=$pos: wanted $n bytes, have $remaining",
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Thrown when an encoded MoQ frame is malformed or truncated. */
|
||||
class MoqCodecException(
|
||||
message: String,
|
||||
cause: Throwable? = null,
|
||||
) : RuntimeException(message, cause)
|
||||
@@ -0,0 +1,144 @@
|
||||
/*
|
||||
* 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
|
||||
|
||||
/**
|
||||
* Encode/decode MoQ control-stream messages per draft-ietf-moq-transport.
|
||||
*
|
||||
* Wire format for every control message:
|
||||
*
|
||||
* message_type (varint) | message_length (varint) | payload bytes
|
||||
*
|
||||
* The payload layout is per-message. Payload encoders here produce only the
|
||||
* payload; [encode] wraps it with type+length for on-the-wire use.
|
||||
*/
|
||||
object MoqCodec {
|
||||
/**
|
||||
* Encode a full control-stream frame (type + length + payload).
|
||||
*/
|
||||
fun encode(message: MoqMessage): ByteArray {
|
||||
val payload = encodePayload(message)
|
||||
val frame = MoqWriter(payload.size + 8)
|
||||
frame.writeVarint(message.type.code)
|
||||
frame.writeVarint(payload.size.toLong())
|
||||
frame.writeBytes(payload)
|
||||
return frame.toByteArray()
|
||||
}
|
||||
|
||||
/**
|
||||
* Decode one control-stream message from [src] starting at [offset].
|
||||
*
|
||||
* Returns the message plus the number of bytes consumed, or null if [src]
|
||||
* doesn't yet contain a whole frame (caller should buffer more data from
|
||||
* the control stream and retry).
|
||||
*/
|
||||
fun decode(
|
||||
src: ByteArray,
|
||||
offset: Int = 0,
|
||||
): DecodeResult? {
|
||||
val typeDec = Varint.decode(src, offset) ?: return null
|
||||
val lenDec = Varint.decode(src, offset + typeDec.bytesConsumed) ?: return null
|
||||
val payloadStart = offset + typeDec.bytesConsumed + lenDec.bytesConsumed
|
||||
val payloadEnd = payloadStart + lenDec.value.toInt()
|
||||
if (payloadEnd > src.size) return null
|
||||
|
||||
val type =
|
||||
MoqMessageType.fromCode(typeDec.value)
|
||||
?: throw MoqCodecException("unknown MoQ message type: 0x${typeDec.value.toString(16)}")
|
||||
|
||||
val reader = MoqReader(src, payloadStart, payloadEnd)
|
||||
val message =
|
||||
when (type) {
|
||||
MoqMessageType.ClientSetup -> decodeClientSetup(reader)
|
||||
MoqMessageType.ServerSetup -> decodeServerSetup(reader)
|
||||
}
|
||||
if (reader.hasMore()) {
|
||||
throw MoqCodecException(
|
||||
"trailing bytes in ${type.name} payload (consumed=${payloadEnd - payloadStart - reader.remaining}, total=${payloadEnd - payloadStart})",
|
||||
)
|
||||
}
|
||||
return DecodeResult(message, payloadEnd - offset)
|
||||
}
|
||||
|
||||
private fun encodePayload(message: MoqMessage): ByteArray =
|
||||
when (message) {
|
||||
is ClientSetup -> encodeClientSetup(message)
|
||||
is ServerSetup -> encodeServerSetup(message)
|
||||
}
|
||||
|
||||
private fun encodeClientSetup(message: ClientSetup): ByteArray {
|
||||
val w = MoqWriter()
|
||||
w.writeVarint(message.supportedVersions.size.toLong())
|
||||
for (v in message.supportedVersions) w.writeVarint(v)
|
||||
encodeParameters(w, message.parameters)
|
||||
return w.toByteArray()
|
||||
}
|
||||
|
||||
private fun decodeClientSetup(r: MoqReader): ClientSetup {
|
||||
val nVersions = r.readVarint().toInt()
|
||||
if (nVersions < 0) throw MoqCodecException("negative version count: $nVersions")
|
||||
val versions = ArrayList<Long>(nVersions)
|
||||
repeat(nVersions) { versions.add(r.readVarint()) }
|
||||
val params = decodeParameters(r)
|
||||
return ClientSetup(versions, params)
|
||||
}
|
||||
|
||||
private fun encodeServerSetup(message: ServerSetup): ByteArray {
|
||||
val w = MoqWriter()
|
||||
w.writeVarint(message.selectedVersion)
|
||||
encodeParameters(w, message.parameters)
|
||||
return w.toByteArray()
|
||||
}
|
||||
|
||||
private fun decodeServerSetup(r: MoqReader): ServerSetup {
|
||||
val version = r.readVarint()
|
||||
val params = decodeParameters(r)
|
||||
return ServerSetup(version, params)
|
||||
}
|
||||
|
||||
private fun encodeParameters(
|
||||
w: MoqWriter,
|
||||
params: List<SetupParameter>,
|
||||
) {
|
||||
w.writeVarint(params.size.toLong())
|
||||
for (p in params) {
|
||||
w.writeVarint(p.key)
|
||||
w.writeLengthPrefixedBytes(p.value)
|
||||
}
|
||||
}
|
||||
|
||||
private fun decodeParameters(r: MoqReader): List<SetupParameter> {
|
||||
val n = r.readVarint().toInt()
|
||||
if (n < 0) throw MoqCodecException("negative parameter count: $n")
|
||||
val out = ArrayList<SetupParameter>(n)
|
||||
repeat(n) {
|
||||
val key = r.readVarint()
|
||||
val value = r.readLengthPrefixedBytes()
|
||||
out.add(SetupParameter(key, value))
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
data class DecodeResult(
|
||||
val message: MoqMessage,
|
||||
val bytesConsumed: Int,
|
||||
)
|
||||
}
|
||||
@@ -0,0 +1,108 @@
|
||||
/*
|
||||
* 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
|
||||
|
||||
/**
|
||||
* Subset of the MoQ-transport control-plane messages needed for a listener
|
||||
* talking to a nests server. Per draft-ietf-moq-transport, control messages
|
||||
* on the control stream share the wire layout:
|
||||
*
|
||||
* message_type (varint) | message_length (varint) | payload...
|
||||
*
|
||||
* This phase (3c-1) covers only the setup handshake. SUBSCRIBE / ANNOUNCE /
|
||||
* OBJECT messages arrive in Phase 3c-2.
|
||||
*/
|
||||
sealed class MoqMessage {
|
||||
abstract val type: MoqMessageType
|
||||
}
|
||||
|
||||
/**
|
||||
* Message type codes per draft-ietf-moq-transport-17. Held as enum so future
|
||||
* draft revisions can extend without breaking call sites.
|
||||
*/
|
||||
enum class MoqMessageType(
|
||||
val code: Long,
|
||||
) {
|
||||
ClientSetup(0x40),
|
||||
ServerSetup(0x41),
|
||||
;
|
||||
|
||||
companion object {
|
||||
private val byCode = entries.associateBy { it.code }
|
||||
|
||||
fun fromCode(code: Long): MoqMessageType? = byCode[code]
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Wire-level setup parameter: key is a varint, value is an opaque byte string
|
||||
* (MoQ doesn't prescribe a single type for values — some draft revisions use
|
||||
* varints, others length-prefixed bytes; nests uses bytes in either case).
|
||||
*/
|
||||
data class SetupParameter(
|
||||
val key: Long,
|
||||
val value: ByteArray,
|
||||
) {
|
||||
override fun equals(other: Any?): Boolean {
|
||||
if (this === other) return true
|
||||
if (other !is SetupParameter) return false
|
||||
return key == other.key && value.contentEquals(other.value)
|
||||
}
|
||||
|
||||
override fun hashCode(): Int = 31 * key.hashCode() + value.contentHashCode()
|
||||
|
||||
companion object {
|
||||
/** ROLE parameter (key 0x00), draft-specific values. Kept as bytes. */
|
||||
const val KEY_ROLE: Long = 0x00L
|
||||
|
||||
/** PATH parameter (key 0x01). */
|
||||
const val KEY_PATH: Long = 0x01L
|
||||
|
||||
/** MAX_SUBSCRIBE_ID (key 0x02 in most drafts). */
|
||||
const val KEY_MAX_SUBSCRIBE_ID: Long = 0x02L
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* CLIENT_SETUP (0x40): sent by the client immediately after the WebTransport
|
||||
* session opens, on the first bidi stream (the control stream).
|
||||
*/
|
||||
data class ClientSetup(
|
||||
val supportedVersions: List<Long>,
|
||||
val parameters: List<SetupParameter> = emptyList(),
|
||||
) : MoqMessage() {
|
||||
override val type: MoqMessageType = MoqMessageType.ClientSetup
|
||||
|
||||
init {
|
||||
require(supportedVersions.isNotEmpty()) { "must offer at least one version" }
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* SERVER_SETUP (0x41): response to CLIENT_SETUP. Contains the one version the
|
||||
* server selected from [ClientSetup.supportedVersions].
|
||||
*/
|
||||
data class ServerSetup(
|
||||
val selectedVersion: Long,
|
||||
val parameters: List<SetupParameter> = emptyList(),
|
||||
) : MoqMessage() {
|
||||
override val type: MoqMessageType = MoqMessageType.ServerSetup
|
||||
}
|
||||
@@ -0,0 +1,190 @@
|
||||
/*
|
||||
* 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 com.vitorpamplona.nestsclient.transport.WebTransportBidiStream
|
||||
import com.vitorpamplona.nestsclient.transport.WebTransportSession
|
||||
import kotlinx.coroutines.flow.first
|
||||
import kotlinx.coroutines.withTimeout
|
||||
|
||||
/**
|
||||
* MoQ-transport draft version constants. Value shape: `0xff00000000 | draft`
|
||||
* is how some draft tests label versions; draft-ietf-moq-transport-17 is
|
||||
* commonly represented as 0xff000011 but implementations vary. The actual
|
||||
* version used on the wire is negotiated in [MoqSession.setup].
|
||||
*
|
||||
* Listed as `Long` to survive the MSB-set encoding on JVM.
|
||||
*/
|
||||
object MoqVersion {
|
||||
/** draft-ietf-moq-transport-11 (a popular stable draft). */
|
||||
const val DRAFT_11: Long = 0xff00000bL
|
||||
|
||||
/** draft-ietf-moq-transport-17 (newer draft). */
|
||||
const val DRAFT_17: Long = 0xff000011L
|
||||
}
|
||||
|
||||
/**
|
||||
* Session wrapper over a [WebTransportSession] that speaks MoQ-transport.
|
||||
*
|
||||
* Phase 3c-1 ships only [setup] — the CLIENT_SETUP / SERVER_SETUP handshake.
|
||||
* Phase 3c-2 adds SUBSCRIBE + ANNOUNCE + object-stream multiplexing.
|
||||
*
|
||||
* Usage:
|
||||
* ```
|
||||
* val session = MoqSession.client(webTransport).apply {
|
||||
* setup(listOf(MoqVersion.DRAFT_17))
|
||||
* }
|
||||
* // session.selectedVersion is now the version the server picked.
|
||||
* ```
|
||||
*/
|
||||
class MoqSession private constructor(
|
||||
private val transport: WebTransportSession,
|
||||
/** The single bidi stream every MoQ exchange uses for control. */
|
||||
private val controlStream: WebTransportBidiStream,
|
||||
private val role: Role,
|
||||
) {
|
||||
enum class Role { Client, Server }
|
||||
|
||||
/** Server's selected version; null until [setup] returns. */
|
||||
var selectedVersion: Long? = null
|
||||
private set
|
||||
|
||||
/** Parameters the server sent back in SERVER_SETUP. Empty until [setup] returns. */
|
||||
var serverParameters: List<SetupParameter> = emptyList()
|
||||
private set
|
||||
|
||||
/**
|
||||
* Run the SETUP handshake.
|
||||
*
|
||||
* Client side: writes CLIENT_SETUP with [supportedVersions] + [clientParameters],
|
||||
* then reads exactly one SERVER_SETUP, stores the result, and returns.
|
||||
* Server side: reads exactly one CLIENT_SETUP, selects the first mutually
|
||||
* supported version from [supportedVersions] (where the local list is the
|
||||
* set of versions the server is willing to accept), writes SERVER_SETUP,
|
||||
* and returns.
|
||||
*
|
||||
* @throws MoqProtocolException if the peer sent an unexpected message, or
|
||||
* if no version overlap exists (server side).
|
||||
*/
|
||||
suspend fun setup(
|
||||
supportedVersions: List<Long>,
|
||||
clientParameters: List<SetupParameter> = emptyList(),
|
||||
handshakeTimeoutMs: Long = 10_000,
|
||||
) {
|
||||
withTimeout(handshakeTimeoutMs) {
|
||||
when (role) {
|
||||
Role.Client -> runClientSetup(supportedVersions, clientParameters)
|
||||
Role.Server -> runServerSetup(supportedVersions, clientParameters)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun runClientSetup(
|
||||
supportedVersions: List<Long>,
|
||||
clientParameters: List<SetupParameter>,
|
||||
) {
|
||||
controlStream.write(MoqCodec.encode(ClientSetup(supportedVersions, clientParameters)))
|
||||
val reply = readOneMessage()
|
||||
val server =
|
||||
reply as? ServerSetup
|
||||
?: throw MoqProtocolException("expected SERVER_SETUP, got ${reply.type.name}")
|
||||
if (server.selectedVersion !in supportedVersions) {
|
||||
throw MoqProtocolException(
|
||||
"server picked version 0x${server.selectedVersion.toString(16)} which we did not offer",
|
||||
)
|
||||
}
|
||||
selectedVersion = server.selectedVersion
|
||||
serverParameters = server.parameters
|
||||
}
|
||||
|
||||
private suspend fun runServerSetup(
|
||||
acceptedVersions: List<Long>,
|
||||
serverParameters: List<SetupParameter>,
|
||||
) {
|
||||
val incoming = readOneMessage()
|
||||
val client =
|
||||
incoming as? ClientSetup
|
||||
?: throw MoqProtocolException("expected CLIENT_SETUP, got ${incoming.type.name}")
|
||||
val overlap =
|
||||
client.supportedVersions.firstOrNull { it in acceptedVersions }
|
||||
?: throw MoqProtocolException(
|
||||
"no mutually-supported MoQ version (client offered ${client.supportedVersions})",
|
||||
)
|
||||
controlStream.write(MoqCodec.encode(ServerSetup(overlap, serverParameters)))
|
||||
this.selectedVersion = overlap
|
||||
this.serverParameters = serverParameters
|
||||
}
|
||||
|
||||
/**
|
||||
* Read exactly one full MoQ message from the control stream.
|
||||
*
|
||||
* Phase 3c-1 assumes a whole message fits in a single transport write
|
||||
* (SETUP frames are only a handful of bytes and the peer always writes
|
||||
* them atomically). Phase 3c-2 will replace this with a buffer-and-retry
|
||||
* loop for messages that may fragment across chunks.
|
||||
*/
|
||||
private suspend fun readOneMessage(): MoqMessage {
|
||||
val chunk = controlStream.incoming().first()
|
||||
val decoded =
|
||||
MoqCodec.decode(chunk)
|
||||
?: throw MoqProtocolException(
|
||||
"control-stream chunk did not contain a complete MoQ frame (size=${chunk.size})",
|
||||
)
|
||||
return decoded.message
|
||||
}
|
||||
|
||||
/** Close the underlying transport. */
|
||||
suspend fun close(
|
||||
code: Int = 0,
|
||||
reason: String = "",
|
||||
) {
|
||||
controlStream.finish()
|
||||
transport.close(code, reason)
|
||||
}
|
||||
|
||||
companion object {
|
||||
/**
|
||||
* Attach to a [WebTransportSession] in the client role. Opens the
|
||||
* control stream eagerly so [setup] doesn't need to manage stream
|
||||
* acquisition.
|
||||
*/
|
||||
suspend fun client(transport: WebTransportSession): MoqSession {
|
||||
val control = transport.openBidiStream()
|
||||
return MoqSession(transport, control, Role.Client)
|
||||
}
|
||||
|
||||
/**
|
||||
* Attach to a [WebTransportSession] in the server role over an
|
||||
* already-accepted control stream (usually the first bidi stream the
|
||||
* peer opened). Used in tests — a real server accepts the first bidi.
|
||||
*/
|
||||
fun server(
|
||||
transport: WebTransportSession,
|
||||
control: WebTransportBidiStream,
|
||||
): MoqSession = MoqSession(transport, control, Role.Server)
|
||||
}
|
||||
}
|
||||
|
||||
/** Thrown when the peer violates the MoQ-transport state machine. */
|
||||
class MoqProtocolException(
|
||||
message: String,
|
||||
cause: Throwable? = null,
|
||||
) : RuntimeException(message, cause)
|
||||
@@ -0,0 +1,122 @@
|
||||
/*
|
||||
* 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
|
||||
|
||||
/**
|
||||
* QUIC variable-length integer codec, per RFC 9000 §16.
|
||||
*
|
||||
* The two most significant bits of the first byte encode the total length:
|
||||
* 00 → 1 byte, 6-bit value (0 .. 63)
|
||||
* 01 → 2 bytes, 14-bit value (0 .. 16_383)
|
||||
* 10 → 4 bytes, 30-bit value (0 .. 1_073_741_823)
|
||||
* 11 → 8 bytes, 62-bit value (0 .. 4_611_686_018_427_387_903)
|
||||
*
|
||||
* MoQ messages, parameters, and most length prefixes use this encoding.
|
||||
*/
|
||||
object Varint {
|
||||
const val MAX_VALUE: Long = (1L shl 62) - 1
|
||||
|
||||
/** Number of bytes required to encode [value]. */
|
||||
fun size(value: Long): Int {
|
||||
require(value >= 0) { "varint must be non-negative: $value" }
|
||||
require(value <= MAX_VALUE) { "varint overflow: $value" }
|
||||
return when {
|
||||
value < (1L shl 6) -> 1
|
||||
value < (1L shl 14) -> 2
|
||||
value < (1L shl 30) -> 4
|
||||
else -> 8
|
||||
}
|
||||
}
|
||||
|
||||
/** Encode [value] into a fresh ByteArray. */
|
||||
fun encode(value: Long): ByteArray {
|
||||
val buf = ByteArray(size(value))
|
||||
writeTo(value, buf, 0)
|
||||
return buf
|
||||
}
|
||||
|
||||
/**
|
||||
* Write [value] to [dst] starting at [offset]. Returns the number of bytes
|
||||
* written. [dst] must have room for [size] bytes starting at [offset].
|
||||
*/
|
||||
fun writeTo(
|
||||
value: Long,
|
||||
dst: ByteArray,
|
||||
offset: Int,
|
||||
): Int {
|
||||
val size = size(value)
|
||||
when (size) {
|
||||
1 -> {
|
||||
dst[offset] = value.toByte()
|
||||
}
|
||||
|
||||
2 -> {
|
||||
dst[offset] = (0x40 or ((value ushr 8).toInt() and 0x3F)).toByte()
|
||||
dst[offset + 1] = (value and 0xFF).toByte()
|
||||
}
|
||||
|
||||
4 -> {
|
||||
dst[offset] = (0x80 or ((value ushr 24).toInt() and 0x3F)).toByte()
|
||||
dst[offset + 1] = ((value ushr 16) and 0xFF).toByte()
|
||||
dst[offset + 2] = ((value ushr 8) and 0xFF).toByte()
|
||||
dst[offset + 3] = (value and 0xFF).toByte()
|
||||
}
|
||||
|
||||
8 -> {
|
||||
dst[offset] = (0xC0 or ((value ushr 56).toInt() and 0x3F)).toByte()
|
||||
dst[offset + 1] = ((value ushr 48) and 0xFF).toByte()
|
||||
dst[offset + 2] = ((value ushr 40) and 0xFF).toByte()
|
||||
dst[offset + 3] = ((value ushr 32) and 0xFF).toByte()
|
||||
dst[offset + 4] = ((value ushr 24) and 0xFF).toByte()
|
||||
dst[offset + 5] = ((value ushr 16) and 0xFF).toByte()
|
||||
dst[offset + 6] = ((value ushr 8) and 0xFF).toByte()
|
||||
dst[offset + 7] = (value and 0xFF).toByte()
|
||||
}
|
||||
}
|
||||
return size
|
||||
}
|
||||
|
||||
/**
|
||||
* Decode a varint starting at [offset] in [src]. Returns [DecodeResult] with
|
||||
* the value and the number of bytes consumed, or null if [src] doesn't
|
||||
* contain enough bytes (caller should buffer more and retry).
|
||||
*/
|
||||
fun decode(
|
||||
src: ByteArray,
|
||||
offset: Int = 0,
|
||||
): DecodeResult? {
|
||||
if (offset >= src.size) return null
|
||||
val first = src[offset].toInt() and 0xFF
|
||||
val length = 1 shl ((first ushr 6) and 0x03)
|
||||
if (offset + length > src.size) return null
|
||||
|
||||
var value = (first and 0x3F).toLong()
|
||||
for (i in 1 until length) {
|
||||
value = (value shl 8) or ((src[offset + i].toInt() and 0xFF).toLong())
|
||||
}
|
||||
return DecodeResult(value, length)
|
||||
}
|
||||
|
||||
data class DecodeResult(
|
||||
val value: Long,
|
||||
val bytesConsumed: Int,
|
||||
)
|
||||
}
|
||||
Reference in New Issue
Block a user