Merge pull request #2696 from vitorpamplona/claude/moq-speaker-animation-okK2f

feat: animate local speaker ring from MoQ frame transmission
This commit is contained in:
Vitor Pamplona
2026-05-01 22:03:18 -04:00
committed by GitHub
11 changed files with 108 additions and 26 deletions
@@ -479,7 +479,21 @@ class NestViewModel(
} }
speaker = s speaker = s
observeSpeakerState(s) observeSpeakerState(s)
val handle = s.startBroadcasting() // The speaker-side `onLevel` callback is the ground
// truth for the local-speaking ring: it fires only
// after a frame actually reaches the wire (see
// NestMoqLiteBroadcaster.start). Reusing the same
// `onSpeakerActivity` / `onAudioLevel` plumbing the
// remote speakers use means the local avatar gets
// the same ring + glow + 250 ms timeout fade as
// any other speaker — no separate UI branch needed.
val handle =
s.startBroadcasting(
onLevel = { level ->
onSpeakerActivity(speakerPubkeyHex)
onAudioLevel(speakerPubkeyHex, level)
},
)
broadcastHandle = handle broadcastHandle = handle
_uiState.update { it.copy(broadcast = BroadcastUiState.Broadcasting(isMuted = false)) } _uiState.update { it.copy(broadcast = BroadcastUiState.Broadcasting(isMuted = false)) }
} catch (ce: CancellationException) { } catch (ce: CancellationException) {
@@ -65,7 +65,7 @@ class MoqLiteNestsSpeaker internal constructor(
private val gate = Mutex() private val gate = Mutex()
private var activeHandle: MoqLiteBroadcastHandle? = null private var activeHandle: MoqLiteBroadcastHandle? = null
override suspend fun startBroadcasting(): BroadcastHandle { override suspend fun startBroadcasting(onLevel: (Float) -> Unit): BroadcastHandle {
gate.withLock { gate.withLock {
val current = state.value val current = state.value
check(current is NestsSpeakerState.Connected) { check(current is NestsSpeakerState.Connected) {
@@ -104,6 +104,7 @@ class MoqLiteNestsSpeaker internal constructor(
// and the room is silently mute. // and the room is silently mute.
onBroadcastTerminalFailure() onBroadcastTerminalFailure()
}, },
onLevel = onLevel,
) )
} }
} catch (t: Throwable) { } catch (t: Throwable) {
@@ -260,7 +260,7 @@ private fun failedSpeaker(state: MutableStateFlow<NestsSpeakerState>): NestsSpea
object : NestsSpeaker { object : NestsSpeaker {
override val state = state override val state = state
override suspend fun startBroadcasting(): BroadcastHandle = error("speaker never connected: ${state.value}") override suspend fun startBroadcasting(onLevel: (Float) -> Unit): BroadcastHandle = error("speaker never connected: ${state.value}")
override suspend fun close() { override suspend fun close() {
if (state.value !is NestsSpeakerState.Closed) { if (state.value !is NestsSpeakerState.Closed) {
@@ -51,11 +51,20 @@ interface NestsSpeaker {
* Begin announcing our speaker track and pumping mic frames out as * Begin announcing our speaker track and pumping mic frames out as
* OBJECT_DATAGRAMs. * OBJECT_DATAGRAMs.
* *
* [onLevel] fires once per Opus frame that successfully reaches
* the wire, with the peak amplitude of the underlying PCM frame
* normalized to `[0, 1]`. Default no-op so callers that don't
* render a local-speaking ring pay zero cost. The callback runs
* on the speaker's coroutine scope (typically the caller's
* `viewModelScope`) so the consumer must keep its handler
* lightweight. See [com.vitorpamplona.nestsclient.audio.NestMoqLiteBroadcaster.start]
* for the ground-truth-vs-UI-state rationale.
*
* @throws IllegalStateException if a broadcast is already running on * @throws IllegalStateException if a broadcast is already running on
* this speaker or the session is not [NestsSpeakerState.Connected]. * this speaker or the session is not [NestsSpeakerState.Connected].
* @throws MoqProtocolException if the peer rejects the ANNOUNCE. * @throws MoqProtocolException if the peer rejects the ANNOUNCE.
*/ */
suspend fun startBroadcasting(): BroadcastHandle suspend fun startBroadcasting(onLevel: (Float) -> Unit = { /* no-op */ }): BroadcastHandle
/** Tear down the MoQ session + transport. Idempotent. */ /** Tear down the MoQ session + transport. Idempotent. */
suspend fun close() suspend fun close()
@@ -151,7 +160,7 @@ class DefaultNestsSpeaker internal constructor(
private val gate = Mutex() private val gate = Mutex()
private var activeHandle: DefaultBroadcastHandle? = null private var activeHandle: DefaultBroadcastHandle? = null
override suspend fun startBroadcasting(): BroadcastHandle { override suspend fun startBroadcasting(onLevel: (Float) -> Unit): BroadcastHandle {
gate.withLock { gate.withLock {
val current = state.value val current = state.value
check(current is NestsSpeakerState.Connected) { check(current is NestsSpeakerState.Connected) {
@@ -174,7 +183,7 @@ class DefaultNestsSpeaker internal constructor(
publisher = publisher, publisher = publisher,
scope = scope, scope = scope,
) )
broadcaster.start() broadcaster.start(onLevel = onLevel)
mutableState.value = mutableState.value =
NestsSpeakerState.Broadcasting( NestsSpeakerState.Broadcasting(
room = current.room, room = current.room,
@@ -273,7 +273,7 @@ private class ReconnectingSpeakerHandle(
@Volatile private var activeBroadcast: ReissuingBroadcastHandle? = null @Volatile private var activeBroadcast: ReissuingBroadcastHandle? = null
override suspend fun startBroadcasting(): BroadcastHandle = override suspend fun startBroadcasting(onLevel: (Float) -> Unit): BroadcastHandle =
gate.withLock { gate.withLock {
check(state.value !is NestsSpeakerState.Closed) { check(state.value !is NestsSpeakerState.Closed) {
"startBroadcasting on a closed speaker" "startBroadcasting on a closed speaker"
@@ -291,7 +291,7 @@ private class ReconnectingSpeakerHandle(
?: error("no live session — wait for state == Connected before startBroadcasting") ?: error("no live session — wait for state == Connected before startBroadcasting")
val handle = val handle =
ReissuingBroadcastHandle(activeSpeaker, scope) { closed -> ReissuingBroadcastHandle(activeSpeaker, scope, onLevel) { closed ->
if (activeBroadcast === closed) activeBroadcast = null if (activeBroadcast === closed) activeBroadcast = null
} }
handle.start() handle.start()
@@ -325,6 +325,13 @@ private class ReconnectingSpeakerHandle(
private class ReissuingBroadcastHandle( private class ReissuingBroadcastHandle(
private val activeSpeaker: StateFlow<NestsSpeaker?>, private val activeSpeaker: StateFlow<NestsSpeaker?>,
private val scope: CoroutineScope, private val scope: CoroutineScope,
/**
* Forwarded to every underlying [NestsSpeaker.startBroadcasting]
* the pump opens, so the local-speaking ring keeps animating
* across session recycles. Mirrors how [desiredMuted] is
* replayed the user-observed signal is monotonic.
*/
private val onLevel: (Float) -> Unit,
private val onClose: (ReissuingBroadcastHandle) -> Unit, private val onClose: (ReissuingBroadcastHandle) -> Unit,
) : BroadcastHandle { ) : BroadcastHandle {
@Volatile private var desiredMuted: Boolean = false @Volatile private var desiredMuted: Boolean = false
@@ -362,7 +369,7 @@ private class ReissuingBroadcastHandle(
if (closed) return@collectLatest if (closed) return@collectLatest
val handle = val handle =
try { try {
sp.startBroadcasting() sp.startBroadcasting(onLevel)
} catch (ce: kotlinx.coroutines.CancellationException) { } catch (ce: kotlinx.coroutines.CancellationException) {
// Don't `return@collectLatest` on cancel — // Don't `return@collectLatest` on cancel —
// propagate so the launched pumpJob actually // propagate so the launched pumpJob actually
@@ -0,0 +1,39 @@
/*
* 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.audio
/**
* Peak amplitude of a 16-bit PCM frame, normalized to `[0, 1]`. Peak
* (vs RMS) is jittery on its own but responds instantly to onsets,
* which suits a visual ring that the UI smooths via animateDpAsState.
*
* Shared between [NestPlayer] (decoded remote audio) and the speaker
* broadcasters (raw mic capture) so the on-screen ring around the
* local avatar uses the same scale as remote speakers' rings.
*/
internal fun peakAmplitude(pcm: ShortArray): Float {
var maxAbs = 0
for (s in pcm) {
val abs = if (s.toInt() < 0) -s.toInt() else s.toInt()
if (abs > maxAbs) maxAbs = abs
}
return (maxAbs / 32768f).coerceIn(0f, 1f)
}
@@ -74,6 +74,8 @@ class NestBroadcaster(
*/ */
onTerminalFailure: () -> Unit = { /* swallow */ }, onTerminalFailure: () -> Unit = { /* swallow */ },
onError: (AudioException) -> Unit = { /* swallow */ }, onError: (AudioException) -> Unit = { /* swallow */ },
/** See [NestMoqLiteBroadcaster.start]'s `onLevel` kdoc. */
onLevel: (Float) -> Unit = { /* no-op */ },
) { ) {
check(!stopped) { "NestBroadcaster already stopped" } check(!stopped) { "NestBroadcaster already stopped" }
check(job == null) { "NestBroadcaster.start already called" } check(job == null) { "NestBroadcaster.start already called" }
@@ -120,6 +122,7 @@ class NestBroadcaster(
sendOutcome sendOutcome
.onSuccess { .onSuccess {
consecutiveSendErrors = 0 consecutiveSendErrors = 0
onLevel(peakAmplitude(pcm))
}.onFailure { t -> }.onFailure { t ->
if (t is CancellationException) throw t if (t is CancellationException) throw t
consecutiveSendErrors += 1 consecutiveSendErrors += 1
@@ -116,6 +116,23 @@ class NestMoqLiteBroadcaster(
fun start( fun start(
onTerminalFailure: () -> Unit = { /* swallow */ }, onTerminalFailure: () -> Unit = { /* swallow */ },
onError: (AudioException) -> Unit = { /* swallow */ }, onError: (AudioException) -> Unit = { /* swallow */ },
/**
* Fires once per Opus frame that successfully reaches
* [MoqLitePublisherHandle.send] i.e. the frame is on the
* wire from this client's perspective. The argument is the
* peak amplitude of the underlying PCM frame, normalized to
* `[0, 1]` (same scale [NestPlayer] reports for remote
* speakers).
*
* This deliberately taps **after** the mute gate and **after**
* a successful send, so the local "I'm speaking" ring on the
* UI tracks ground truth (frames actually leaving) rather
* than any UI-side mute / role state that could lie.
*
* Default no-op so callers that don't render a local ring
* (tests, headless interop) pay zero cost.
*/
onLevel: (Float) -> Unit = { /* no-op */ },
) { ) {
check(!stopped) { "NestMoqLiteBroadcaster already stopped" } check(!stopped) { "NestMoqLiteBroadcaster already stopped" }
check(job == null) { "NestMoqLiteBroadcaster.start already called" } check(job == null) { "NestMoqLiteBroadcaster.start already called" }
@@ -180,6 +197,12 @@ class NestMoqLiteBroadcaster(
sendOutcome sendOutcome
.onSuccess { .onSuccess {
consecutiveSendErrors = 0 consecutiveSendErrors = 0
// Fire the speaking-ring tap only on
// a successful send. If the publisher
// throws (transport gone, peer dead),
// the frame didn't actually go out and
// the UI shouldn't claim we're talking.
onLevel(peakAmplitude(pcm))
}.onFailure { t -> }.onFailure { t ->
if (t is CancellationException) throw t if (t is CancellationException) throw t
consecutiveSendErrors += 1 consecutiveSendErrors += 1
@@ -112,20 +112,6 @@ class NestPlayer(
} }
} }
/**
* Peak amplitude of a 16-bit PCM frame, normalized to `[0, 1]`. Peak
* (vs RMS) is jittery on its own but responds instantly to onsets,
* which suits a visual ring that the UI smooths via animateDpAsState.
*/
private fun peakAmplitude(pcm: ShortArray): Float {
var maxAbs = 0
for (s in pcm) {
val abs = if (s.toInt() < 0) -s.toInt() else s.toInt()
if (abs > maxAbs) maxAbs = abs
}
return (maxAbs / 32768f).coerceIn(0f, 1f)
}
/** /**
* Stop playback, cancel the decode loop, release the decoder. Idempotent. * Stop playback, cancel the decode loop, release the decoder. Idempotent.
* *
@@ -138,7 +138,7 @@ class ReconnectingNestsSpeakerTest {
val handles: MutableList<ScriptedBroadcastHandle> = val handles: MutableList<ScriptedBroadcastHandle> =
java.util.concurrent.CopyOnWriteArrayList() java.util.concurrent.CopyOnWriteArrayList()
override suspend fun startBroadcasting(): BroadcastHandle { override suspend fun startBroadcasting(onLevel: (Float) -> Unit): BroadcastHandle {
val handle = ScriptedBroadcastHandle() val handle = ScriptedBroadcastHandle()
handles += handle handles += handle
// Increment AFTER appending to `handles` so the volatile // Increment AFTER appending to `handles` so the volatile
@@ -176,7 +176,7 @@ class ReconnectingNestsSpeakerTest {
override val state: StateFlow<NestsSpeakerState> = override val state: StateFlow<NestsSpeakerState> =
MutableStateFlow(NestsSpeakerState.Failed("scripted-fail")).asStateFlow() MutableStateFlow(NestsSpeakerState.Failed("scripted-fail")).asStateFlow()
override suspend fun startBroadcasting(): BroadcastHandle = error("never connected") override suspend fun startBroadcasting(onLevel: (Float) -> Unit): BroadcastHandle = error("never connected")
override suspend fun close() {} override suspend fun close() {}
} }
@@ -103,7 +103,7 @@ class NestBroadcasterTest {
val broadcaster = NestBroadcaster(capture, encoder, publisher, backgroundScope) val broadcaster = NestBroadcaster(capture, encoder, publisher, backgroundScope)
val errors = mutableListOf<AudioException>() val errors = mutableListOf<AudioException>()
broadcaster.start { errors.add(it) } broadcaster.start(onError = { errors.add(it) })
capture.awaitDrained() capture.awaitDrained()
assertEquals(1, errors.size, "exactly one encode error should have been reported") assertEquals(1, errors.size, "exactly one encode error should have been reported")