nick/amethyst
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

refactor(nests): extract MoqLiteFrame + handles + publisher interface (Audit-8)

Browse Source 
`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
This commit is contained in:
Claude
2026-05-06 19:50:24 +00:00
parent 6f649e76d8
commit e9d19e5de4
4 changed files with 223 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteFrame.kt
+40
View File
@@ -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()
}
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteHandles.kt
+61
View File
@@ -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<MoqLiteFrame>,
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<MoqLiteAnnounce>,
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)
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLitePublisherHandle.kt
+122
View File
@@ -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()
}
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt
-154
View File
@@ -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<MoqLiteFrame>,
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<MoqLiteAnnounce>,
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()
}