feat(nests): align moq-lite catalog with kixelated/hang shape
Switch RoomSpeakerCatalog to the canonical hang catalog format (camelCase audio.renditions[<track>] with container.kind), wrap audio frames in the legacy varint(timestamp_us) prefix on publish, and strip it on subscribe. Also reply to inbound Probe/Fetch/Session bidis with a bitrate hint or clean FIN instead of hanging.
This commit is contained in:
Claude
+44
-3
@@ -82,14 +82,36 @@ class MoqLiteNestsListener internal constructor(
|
||||
override suspend fun subscribeSpeaker(
|
||||
speakerPubkeyHex: String,
|
||||
maxLatencyMs: Long,
|
||||
): SubscribeHandle = wrapSubscription(broadcast = speakerPubkeyHex, track = AUDIO_TRACK, maxLatencyMs = maxLatencyMs)
|
||||
): SubscribeHandle =
|
||||
wrapSubscription(
|
||||
broadcast = speakerPubkeyHex,
|
||||
track = AUDIO_TRACK,
|
||||
maxLatencyMs = maxLatencyMs,
|
||||
// Audio frames arrive in kixelated/moq `hang` "legacy"
|
||||
// container layout: each moq-lite frame is
|
||||
// `varint(timestamp_us) + raw_opus_packet`. Strip the
|
||||
// leading varint so downstream decoders see a pristine
|
||||
// Opus packet exactly as if it came from
|
||||
// [com.vitorpamplona.nestsclient.audio.OpusEncoder].
|
||||
stripLegacyTimestamp = true,
|
||||
)
|
||||
|
||||
override suspend fun subscribeCatalog(speakerPubkeyHex: String): SubscribeHandle = wrapSubscription(broadcast = speakerPubkeyHex, track = CATALOG_TRACK, maxLatencyMs = 0L)
|
||||
override suspend fun subscribeCatalog(speakerPubkeyHex: String): SubscribeHandle =
|
||||
wrapSubscription(
|
||||
broadcast = speakerPubkeyHex,
|
||||
track = CATALOG_TRACK,
|
||||
maxLatencyMs = 0L,
|
||||
// Catalog frames carry raw JSON bytes — no container
|
||||
// wrapping (the catalog itself is what tells you which
|
||||
// container the audio track uses).
|
||||
stripLegacyTimestamp = false,
|
||||
)
|
||||
|
||||
private suspend fun wrapSubscription(
|
||||
broadcast: String,
|
||||
track: String,
|
||||
maxLatencyMs: Long,
|
||||
stripLegacyTimestamp: Boolean,
|
||||
): SubscribeHandle {
|
||||
check(state.value is NestsListenerState.Connected) {
|
||||
"NestsListener.subscribe requires Connected state, was ${state.value}"
|
||||
@@ -112,12 +134,13 @@ class MoqLiteNestsListener internal constructor(
|
||||
val objectIdSeq = AtomicLong(0L)
|
||||
val mapped =
|
||||
handle.frames.map { frame ->
|
||||
val payload = if (stripLegacyTimestamp) stripLegacyTimestampPrefix(frame.payload) else frame.payload
|
||||
MoqObject(
|
||||
trackAlias = handle.id,
|
||||
groupId = frame.groupSequence,
|
||||
objectId = objectIdSeq.getAndIncrement(),
|
||||
publisherPriority = MoqLiteSession.DEFAULT_PRIORITY,
|
||||
payload = frame.payload,
|
||||
payload = payload,
|
||||
)
|
||||
}
|
||||
|
||||
@@ -242,5 +265,23 @@ class MoqLiteNestsListener internal constructor(
|
||||
* [subscribeCatalog].
|
||||
*/
|
||||
const val CATALOG_TRACK: String = "catalog.json"
|
||||
|
||||
/**
|
||||
* Strip the leading `varint(timestamp_us)` prefix from a
|
||||
* kixelated/moq `hang` "legacy" container frame, returning
|
||||
* the bare codec payload (e.g. an Opus packet) for downstream
|
||||
* decoders. The varint length is encoded in the top 2 bits of
|
||||
* the first byte per RFC 9000 §16: `00`→1, `01`→2, `10`→4,
|
||||
* `11`→8 bytes. Returns the payload unchanged when the
|
||||
* varint header would overrun the payload (malformed frame —
|
||||
* surface upstream rather than silently mask).
|
||||
*/
|
||||
internal fun stripLegacyTimestampPrefix(payload: ByteArray): ByteArray {
|
||||
if (payload.isEmpty()) return payload
|
||||
val tag = (payload[0].toInt() ushr 6) and 0x3
|
||||
val varintLen = 1 shl tag
|
||||
if (payload.size < varintLen) return payload
|
||||
return payload.copyOfRange(varintLen, payload.size)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+18
-6
@@ -203,17 +203,29 @@ class MoqLiteNestsSpeaker internal constructor(
|
||||
companion object {
|
||||
/**
|
||||
* moq-lite catalog manifest broadcast on the
|
||||
* [MoqLiteNestsListener.CATALOG_TRACK] track. The shape mirrors
|
||||
* what the kixelated/moq browser watcher parses
|
||||
* ([com.vitorpamplona.amethyst.commons.viewmodels.RoomSpeakerCatalog]):
|
||||
* a versioned envelope with an `audio[]` of track descriptors.
|
||||
* [MoqLiteNestsListener.CATALOG_TRACK] track. Verbatim
|
||||
* canonical kixelated/moq `hang` shape (camelCase, nested
|
||||
* `audio.renditions[<trackName>]`, `container.kind = "legacy"`).
|
||||
* The rendition key MUST equal the moq-lite track we publish
|
||||
* audio frames on ([MoqLiteNestsListener.AUDIO_TRACK]) so the
|
||||
* watcher's "subscribe to this rendition" path resolves to our
|
||||
* actual audio stream.
|
||||
*
|
||||
* `container.kind = "legacy"` declares the wire layout the
|
||||
* watcher must expect inside each moq-lite frame: a single
|
||||
* `varint(timestamp_us)` prefix followed by the raw codec
|
||||
* payload (no MOOF/MDAT wrapping). The publisher side honours
|
||||
* this contract in
|
||||
* [com.vitorpamplona.nestsclient.audio.NestMoqLiteBroadcaster].
|
||||
*
|
||||
* Hard-coded because the speaker's encoder is fixed at
|
||||
* 48 kHz mono Opus today; if [OpusEncoder] becomes parameterised
|
||||
* this should be derived from the encoder config instead.
|
||||
*/
|
||||
const val SPEAKER_CATALOG_JSON: String =
|
||||
"{\"version\":1,\"audio\":[{\"track\":\"audio/data\"," +
|
||||
"\"codec\":\"opus\",\"sample_rate\":48000,\"channel_count\":1}]}"
|
||||
"{\"audio\":{\"renditions\":{\"" + MoqLiteNestsListener.AUDIO_TRACK + "\":{" +
|
||||
"\"codec\":\"opus\",\"container\":{\"kind\":\"legacy\"}," +
|
||||
"\"sampleRate\":48000,\"numberOfChannels\":1}}}}"
|
||||
|
||||
/**
|
||||
* How often the catalog group is re-emitted. The relay drops
|
||||
|
||||
+23
-1
@@ -21,11 +21,14 @@
|
||||
package com.vitorpamplona.nestsclient.audio
|
||||
|
||||
import com.vitorpamplona.nestsclient.moq.lite.MoqLitePublisherHandle
|
||||
import com.vitorpamplona.quic.Varint
|
||||
import kotlinx.coroutines.CancellationException
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.Job
|
||||
import kotlinx.coroutines.cancelAndJoin
|
||||
import kotlinx.coroutines.launch
|
||||
import kotlin.time.TimeMark
|
||||
import kotlin.time.TimeSource
|
||||
|
||||
/**
|
||||
* Mirror of [NestBroadcaster] but driving a moq-lite
|
||||
@@ -195,6 +198,21 @@ class NestMoqLiteBroadcaster(
|
||||
// and the relay would see two unrelated uni streams under
|
||||
// the same logical group.
|
||||
var lastPublisher: MoqLitePublisherHandle = publisher
|
||||
// kixelated/moq `hang` "legacy" container wire format:
|
||||
// every frame inside a moq-lite group is
|
||||
// varint(timestamp_us) + raw_codec_payload
|
||||
// (`rs/hang/src/container/frame.rs`,
|
||||
// Timescale<1_000_000>). Watchers that read our
|
||||
// catalog's `container.kind = "legacy"` declaration
|
||||
// will skip the leading varint as a microsecond
|
||||
// timestamp; they MUST receive a real timestamp or
|
||||
// their decoder picks up garbage bytes ahead of the
|
||||
// Opus packet. We use a monotonic mark captured at
|
||||
// capture-loop start so timestamps are robust to mute
|
||||
// gaps and survive publisher hot-swap (the wire is
|
||||
// single-broadcast, single-encoder; timestamps are
|
||||
// codec-payload metadata, not stream-position).
|
||||
val frameStartMark: TimeMark = TimeSource.Monotonic.markNow()
|
||||
try {
|
||||
while (true) {
|
||||
val pcm = capture.readFrame() ?: break
|
||||
@@ -233,7 +251,11 @@ class NestMoqLiteBroadcaster(
|
||||
// for the production cliff this works around.
|
||||
val sendOutcome =
|
||||
runCatching {
|
||||
val accepted = current.send(opus)
|
||||
val tsBytes = Varint.encode(frameStartMark.elapsedNow().inWholeMicroseconds)
|
||||
val payload = ByteArray(tsBytes.size + opus.size)
|
||||
tsBytes.copyInto(payload, 0)
|
||||
opus.copyInto(payload, tsBytes.size)
|
||||
val accepted = current.send(payload)
|
||||
framesInCurrentGroup += 1
|
||||
if (framesInCurrentGroup >= framesPerGroup) {
|
||||
current.endGroup()
|
||||
|
||||
+14
@@ -251,6 +251,20 @@ object MoqLiteCodec {
|
||||
return MoqLiteProbe(bitrate = bitrate)
|
||||
}
|
||||
|
||||
/**
|
||||
* Encode a single Lite-03 Probe message body
|
||||
* (`lite/probe.rs` — `bitrate: u62` only; `rtt` is Lite-04+).
|
||||
* The publisher writes these size-prefixed onto a Probe bidi the
|
||||
* subscriber opened, advertising the publisher's expected
|
||||
* bandwidth. Wrapping (size prefix) is the caller's responsibility,
|
||||
* matching [encodeAnnouncePlease] / [encodeAnnounce].
|
||||
*/
|
||||
fun encodeProbe(probe: MoqLiteProbe): ByteArray {
|
||||
val body = MoqWriter()
|
||||
body.writeVarint(probe.bitrate)
|
||||
return wrapSizePrefixed(body)
|
||||
}
|
||||
|
||||
// ---------------- internals ----------------
|
||||
|
||||
/**
|
||||
|
||||
+53
-3
@@ -883,9 +883,50 @@ class MoqLiteSession internal constructor(
|
||||
dispatched = true
|
||||
}
|
||||
|
||||
else -> {
|
||||
// Lite-03 treats Session/Fetch/Probe as
|
||||
// separate flows; we don't implement them.
|
||||
MoqLiteControlType.Probe -> {
|
||||
// Subscriber-opened bidi asking us (the
|
||||
// publisher) for a bitrate hint. Per
|
||||
// `lite/probe.rs:Lite03+`, the publisher
|
||||
// writes one or more size-prefixed
|
||||
// `Probe { bitrate: u62 }` messages on
|
||||
// this bidi. We're a fixed-rate Opus
|
||||
// publisher, so emit a single hint and
|
||||
// FIN our write side — the peer treats
|
||||
// FIN as "no further updates" rather than
|
||||
// an error. Better than the old behaviour
|
||||
// of FINing without writing anything,
|
||||
// which left the subscriber's ABR
|
||||
// estimator with no signal.
|
||||
runCatching {
|
||||
bidi.write(MoqLiteCodec.encodeProbe(MoqLiteProbe(bitrate = NESTS_AUDIO_BITRATE_HINT_BPS)))
|
||||
bidi.finish()
|
||||
}
|
||||
dispatched = true
|
||||
}
|
||||
|
||||
MoqLiteControlType.Fetch -> {
|
||||
// Subscriber-opened bidi requesting a
|
||||
// historical group (`lite/fetch.rs`).
|
||||
// Audio rooms are live-only — we have no
|
||||
// group history to serve. FINing the
|
||||
// write side without any reply is the
|
||||
// spec-clean way to signal "no groups
|
||||
// available"; the subscriber's wait on
|
||||
// its receive side resolves to
|
||||
// end-of-stream and it falls back to a
|
||||
// live Subscribe. Ignoring inbound bytes
|
||||
// (the request body) is fine: we don't
|
||||
// need to know which group was requested
|
||||
// because we couldn't serve any of them.
|
||||
runCatching { bidi.finish() }
|
||||
dispatched = true
|
||||
}
|
||||
|
||||
MoqLiteControlType.Session -> {
|
||||
// ControlType=0 was the Lite-01/02 setup
|
||||
// exchange. Lite-03 doesn't use it; if a
|
||||
// legacy peer sends one, FIN cleanly so
|
||||
// their bidi resolves rather than hanging.
|
||||
runCatching { bidi.finish() }
|
||||
dispatched = true
|
||||
}
|
||||
@@ -1194,6 +1235,15 @@ class MoqLiteSession internal constructor(
|
||||
/** moq-lite priority byte midpoint — neutral default. */
|
||||
const val DEFAULT_PRIORITY: Int = 0x80
|
||||
|
||||
/**
|
||||
* Bitrate hint (bits/sec) we report on inbound moq-lite Probe
|
||||
* bidis as a publisher. Mirrors the upper-bound of an Opus
|
||||
* voice profile at 48 kHz mono, ≈32 kbps. Subscriber-side ABR
|
||||
* estimators use this to size their forward queue; we emit the
|
||||
* single hint and FIN since our encoder runs at a fixed bitrate.
|
||||
*/
|
||||
const val NESTS_AUDIO_BITRATE_HINT_BPS: Long = 32_000L
|
||||
|
||||
// Diagnostic: log "send returned false" once every N invocations.
|
||||
// At 50 fps and N=50 → ≤ 1 log/sec for a sustained no-sub window.
|
||||
private const val SEND_LOG_THROTTLE: Long = 50L
|
||||
|
||||
Reference in New Issue
Block a user