From e9d19e5de45738799477e7bd411c74aca7c77187 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 6 May 2026 19:50:24 +0000 Subject: [PATCH] refactor(nests): extract MoqLiteFrame + handles + publisher interface (Audit-8) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `MoqLiteSession.kt` was 1526 lines mixing the session state machine (announce pump, subscribe pump, group-stream demux, publisher state) with the public types its API hands back to consumers. The session- internal `PublisherStateImpl` inner class is tightly coupled to the session's outer scope (state lock, transport, scope.launch, openGroupStream) and is left in place — extracting it is a separate refactor with its own design choices. Extract the standalone caller-facing types: - `MoqLiteFrame.kt` — the `MoqLiteFrame` data class with its custom `equals` / `hashCode` for ByteArray content equality. - `MoqLiteHandles.kt` — `MoqLiteSubscribeHandle`, `MoqLiteAnnouncesHandle`, and `MoqLiteSubscribeException`. All three are "what the caller gets back from a subscribe / announce" + "how protocol-level rejections surface." - `MoqLitePublisherHandle.kt` — the public publisher interface that the session's internal `PublisherStateImpl` implements. `internal constructor` on the handles keeps them un-instantiable outside the package. Same package, same visibility, no call-site changes needed. `MoqLiteSession.kt` shrinks from 1526 to 1372 lines, focused on session lifecycle + the inner `PublisherStateImpl`. Tests + spotless green. https://claude.ai/code/session_014JfZJHSTvyYYWJbC9VbB47 --- .../nestsclient/moq/lite/MoqLiteFrame.kt | 40 +++++ .../nestsclient/moq/lite/MoqLiteHandles.kt | 61 +++++++ .../moq/lite/MoqLitePublisherHandle.kt | 122 ++++++++++++++ .../nestsclient/moq/lite/MoqLiteSession.kt | 154 ------------------ 4 files changed, 223 insertions(+), 154 deletions(-) create mode 100644 nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteFrame.kt create mode 100644 nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteHandles.kt create mode 100644 nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLitePublisherHandle.kt diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteFrame.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteFrame.kt new file mode 100644 index 000000000..73c827014 --- /dev/null +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteFrame.kt @@ -0,0 +1,40 @@ +/* + * 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 + +/** + * One frame received from a subscription. moq-lite's wire format + * carries no per-frame envelope beyond the size; [groupSequence] is + * pulled from the group header so consumers can detect group rollover + * (e.g. for keyframe boundaries). + */ +data class MoqLiteFrame( + val groupSequence: Long, + val payload: ByteArray, +) { + override fun equals(other: Any?): Boolean { + if (this === other) return true + if (other !is MoqLiteFrame) return false + return groupSequence == other.groupSequence && payload.contentEquals(other.payload) + } + + override fun hashCode(): Int = 31 * groupSequence.hashCode() + payload.contentHashCode() +} diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteHandles.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteHandles.kt new file mode 100644 index 000000000..9f69f1f00 --- /dev/null +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteHandles.kt @@ -0,0 +1,61 @@ +/* + * 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 kotlinx.coroutines.flow.Flow + +// Caller-facing handles returned by MoqLiteSession.subscribe / .announce, +// plus the typed protocol-level rejection exception. Kept together +// because they share a single concern: "what the consumer gets back +// from a subscribe / announce request and how it surfaces failure." + +/** + * Active subscription handle returned by [MoqLiteSession.subscribe]. + * [frames] emits every frame the publisher pushes; [unsubscribe] + * FINs the bidi to signal "no longer interested" (moq-lite has no + * UNSUBSCRIBE message — FIN is the protocol). + */ +class MoqLiteSubscribeHandle internal constructor( + val id: Long, + val ok: MoqLiteSubscribeOk, + val frames: Flow, + private val unsubscribeAction: suspend () -> Unit, +) { + suspend fun unsubscribe() = unsubscribeAction() +} + +/** + * Active announce-discovery handle returned by [MoqLiteSession.announce]. + * [updates] emits every [MoqLiteAnnounce] update the relay streams + * back; [close] FINs the bidi to stop receiving updates. + */ +class MoqLiteAnnouncesHandle internal constructor( + val updates: Flow, + private val close: suspend () -> Unit, +) { + suspend fun close() = close.invoke() +} + +/** Thrown when subscribe is rejected (Drop) or the response stream dies. */ +class MoqLiteSubscribeException( + message: String, + cause: Throwable? = null, +) : RuntimeException(message, cause) diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLitePublisherHandle.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLitePublisherHandle.kt new file mode 100644 index 000000000..143917b78 --- /dev/null +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLitePublisherHandle.kt @@ -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.lite + +/** + * Active publisher handle returned by [MoqLiteSession.publish]. + * + * Lifecycle: + * 1. Call [startGroup] (or [send] which auto-starts a fresh group on + * first call) to begin pushing frames for one Opus group. + * 2. Call [send] for each frame (one Opus packet = one frame). + * 3. Call [endGroup] to FIN the current group's uni stream and start + * a fresh group on the next [send]. Group rollover is the + * publisher's call — typically every N seconds or every keyframe. + * 4. Call [close] when the broadcast ends — sends `Announce(Ended)` + * on every active announce bidi and FINs every group stream. + */ +interface MoqLitePublisherHandle { + /** + * The broadcast suffix this publisher claimed at [MoqLiteSession.publish]. + * Always normalised per [MoqLitePath]. + */ + val suffix: String + + /** + * The next group sequence number that will be assigned by [send] / + * [startGroup]. Snapshot-only — read AFTER the broadcaster has + * stopped sending into this publisher (typically just before the + * caller closes the publisher in a hot-swap), so the value is the + * highest-already-used sequence + 1. + * + * Used by [com.vitorpamplona.nestsclient.MoqLiteNestsSpeaker]'s + * hot-swap path to seed the new session's publisher with a + * monotonically-continuing sequence — without this, every JWT + * refresh restarts at sequence 0 and kixelated/hang's + * `Container.Consumer.#run` drops every group whose sequence is + * less than its current `#active` high-water mark, killing audio + * for the watcher until either `#active` rolls over or the + * watcher re-subscribes. + * + * `@Volatile` on the implementation; safe to read from any + * coroutine. The accept-tiny-race window between read and a + * concurrent `send` is closed in practice because the broadcaster + * is responsible for swapping its publisher reference BEFORE the + * caller reads this value (so no further sends land on this + * publisher). + */ + val nextSequence: Long + + /** + * Start a new group. Allocates a fresh sequence id and opens a new + * uni stream pre-loaded with `DataType=Group + GroupHeader`. Idempotent + * — calling [startGroup] when the previous group hasn't been ended + * is treated as an implicit [endGroup] then a new start. + */ + suspend fun startGroup() + + /** + * Push one [payload] (one Opus packet) as a `varint(size) + payload` + * frame on the current group's uni stream. Auto-starts a group if + * none is active. + * + * Returns false if no inbound subscriber is currently attached. + * Subscriber-less sends silently drop on the wire — the relay keeps + * the publisher's announce active either way, so unmute is + * sample-accurate. + */ + suspend fun send(payload: ByteArray): Boolean + + /** FIN the current group's uni stream. The next [send] starts a fresh group. */ + suspend fun endGroup() + + /** + * Register a callback that fires once each time a new inbound + * subscriber is registered against this publisher's track (i.e. + * each track-matching SUBSCRIBE bidi the relay opens to us). Used + * to push a "track-latest" payload — the canonical example is the + * broadcast catalog manifest, which a watcher needs to receive on + * subscribe but doesn't change between subscribers — without + * forcing the publisher to maintain a periodic re-emit loop. + * + * Called once per accepted SUBSCRIBE (track filter passed). Fires + * OUTSIDE the publisher's serialisation lock, so the hook can + * safely call [send] / [endGroup] without deadlocking. + * + * Caller MUST set the hook before any subscriber attaches (typically + * immediately after [com.vitorpamplona.nestsclient.moq.lite.MoqLiteSession.publish] + * returns) — there's no "fire-on-set for existing subscribers" + * replay. For the typical catalog use case the publisher is fresh + * when the hook is set, and the relay's SUBSCRIBE bidi takes a + * round-trip to arrive, so this is safe in practice. + * + * Pass `null` to clear the hook. Calling twice with non-null + * replaces the previous hook (no de-duplication). + */ + fun setOnNewSubscriber(hook: (suspend () -> Unit)?) + + /** + * Stop publishing. Sends `Announce(Ended)` on every active announce + * bidi, FINs the current group, and releases all per-publisher + * resources. Idempotent. + */ + suspend fun close() +} diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt index b5e49dc2c..3dc3825ed 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt @@ -34,7 +34,6 @@ import kotlinx.coroutines.Job import kotlinx.coroutines.cancelAndJoin import kotlinx.coroutines.channels.BufferOverflow import kotlinx.coroutines.channels.Channel -import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.consumeAsFlow import kotlinx.coroutines.launch @@ -1371,156 +1370,3 @@ class MoqLiteSession internal constructor( ): MoqLiteSession = MoqLiteSession(transport, pumpScope) } } - -/** - * One frame received from a subscription. moq-lite's wire format - * carries no per-frame envelope beyond the size; [groupSequence] is - * pulled from the group header so consumers can detect group rollover - * (e.g. for keyframe boundaries). - */ -data class MoqLiteFrame( - val groupSequence: Long, - val payload: ByteArray, -) { - override fun equals(other: Any?): Boolean { - if (this === other) return true - if (other !is MoqLiteFrame) return false - return groupSequence == other.groupSequence && payload.contentEquals(other.payload) - } - - override fun hashCode(): Int = 31 * groupSequence.hashCode() + payload.contentHashCode() -} - -/** - * Active subscription handle returned by [MoqLiteSession.subscribe]. - * [frames] emits every frame the publisher pushes; [unsubscribe] - * FINs the bidi to signal "no longer interested" (moq-lite has no - * UNSUBSCRIBE message — FIN is the protocol). - */ -class MoqLiteSubscribeHandle internal constructor( - val id: Long, - val ok: MoqLiteSubscribeOk, - val frames: Flow, - private val unsubscribeAction: suspend () -> Unit, -) { - suspend fun unsubscribe() = unsubscribeAction() -} - -/** - * Active announce-discovery handle returned by [MoqLiteSession.announce]. - * [updates] emits every [MoqLiteAnnounce] update the relay streams - * back; [close] FINs the bidi to stop receiving updates. - */ -class MoqLiteAnnouncesHandle internal constructor( - val updates: Flow, - private val close: suspend () -> Unit, -) { - suspend fun close() = close.invoke() -} - -/** Thrown when subscribe is rejected (Drop) or the response stream dies. */ -class MoqLiteSubscribeException( - message: String, - cause: Throwable? = null, -) : RuntimeException(message, cause) - -/** - * Active publisher handle returned by [MoqLiteSession.publish]. - * - * Lifecycle: - * 1. Call [startGroup] (or [send] which auto-starts a fresh group on - * first call) to begin pushing frames for one Opus group. - * 2. Call [send] for each frame (one Opus packet = one frame). - * 3. Call [endGroup] to FIN the current group's uni stream and start - * a fresh group on the next [send]. Group rollover is the - * publisher's call — typically every N seconds or every keyframe. - * 4. Call [close] when the broadcast ends — sends `Announce(Ended)` - * on every active announce bidi and FINs every group stream. - */ -interface MoqLitePublisherHandle { - /** - * The broadcast suffix this publisher claimed at [MoqLiteSession.publish]. - * Always normalised per [MoqLitePath]. - */ - val suffix: String - - /** - * The next group sequence number that will be assigned by [send] / - * [startGroup]. Snapshot-only — read AFTER the broadcaster has - * stopped sending into this publisher (typically just before the - * caller closes the publisher in a hot-swap), so the value is the - * highest-already-used sequence + 1. - * - * Used by [com.vitorpamplona.nestsclient.MoqLiteNestsSpeaker]'s - * hot-swap path to seed the new session's publisher with a - * monotonically-continuing sequence — without this, every JWT - * refresh restarts at sequence 0 and kixelated/hang's - * `Container.Consumer.#run` drops every group whose sequence is - * less than its current `#active` high-water mark, killing audio - * for the watcher until either `#active` rolls over or the - * watcher re-subscribes. - * - * `@Volatile` on the implementation; safe to read from any - * coroutine. The accept-tiny-race window between read and a - * concurrent `send` is closed in practice because the broadcaster - * is responsible for swapping its publisher reference BEFORE the - * caller reads this value (so no further sends land on this - * publisher). - */ - val nextSequence: Long - - /** - * Start a new group. Allocates a fresh sequence id and opens a new - * uni stream pre-loaded with `DataType=Group + GroupHeader`. Idempotent - * — calling [startGroup] when the previous group hasn't been ended - * is treated as an implicit [endGroup] then a new start. - */ - suspend fun startGroup() - - /** - * Push one [payload] (one Opus packet) as a `varint(size) + payload` - * frame on the current group's uni stream. Auto-starts a group if - * none is active. - * - * Returns false if no inbound subscriber is currently attached. - * Subscriber-less sends silently drop on the wire — the relay keeps - * the publisher's announce active either way, so unmute is - * sample-accurate. - */ - suspend fun send(payload: ByteArray): Boolean - - /** FIN the current group's uni stream. The next [send] starts a fresh group. */ - suspend fun endGroup() - - /** - * Register a callback that fires once each time a new inbound - * subscriber is registered against this publisher's track (i.e. - * each track-matching SUBSCRIBE bidi the relay opens to us). Used - * to push a "track-latest" payload — the canonical example is the - * broadcast catalog manifest, which a watcher needs to receive on - * subscribe but doesn't change between subscribers — without - * forcing the publisher to maintain a periodic re-emit loop. - * - * Called once per accepted SUBSCRIBE (track filter passed). Fires - * OUTSIDE the publisher's serialisation lock, so the hook can - * safely call [send] / [endGroup] without deadlocking. - * - * Caller MUST set the hook before any subscriber attaches (typically - * immediately after [com.vitorpamplona.nestsclient.moq.lite.MoqLiteSession.publish] - * returns) — there's no "fire-on-set for existing subscribers" - * replay. For the typical catalog use case the publisher is fresh - * when the hook is set, and the relay's SUBSCRIBE bidi takes a - * round-trip to arrive, so this is safe in practice. - * - * Pass `null` to clear the hook. Calling twice with non-null - * replaces the previous hook (no de-duplication). - */ - fun setOnNewSubscriber(hook: (suspend () -> Unit)?) - - /** - * Stop publishing. Sends `Announce(Ended)` on every active announce - * bidi, FINs the current group, and releases all per-publisher - * resources. Idempotent. - */ - suspend fun close() -}