diff --git a/amethyst/src/main/AndroidManifest.xml b/amethyst/src/main/AndroidManifest.xml index a006b8382..f49058bee 100644 --- a/amethyst/src/main/AndroidManifest.xml +++ b/amethyst/src/main/AndroidManifest.xml @@ -37,6 +37,11 @@ + + + diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/nests/NestForegroundService.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/nests/NestForegroundService.kt index c79670aff..aefe23703 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/nests/NestForegroundService.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/nests/NestForegroundService.kt @@ -28,13 +28,23 @@ import android.app.Service import android.content.Context import android.content.Intent import android.content.pm.ServiceInfo +import android.media.AudioDeviceCallback +import android.media.AudioDeviceInfo +import android.media.AudioFocusRequest +import android.media.AudioManager +import android.net.ConnectivityManager +import android.net.Network import android.os.Build import android.os.IBinder import android.os.PowerManager import androidx.core.app.NotificationCompat import androidx.core.content.ContextCompat import com.vitorpamplona.amethyst.R +import com.vitorpamplona.amethyst.commons.viewmodels.NestAudioFocusBus +import com.vitorpamplona.amethyst.commons.viewmodels.NestAudioFocusState +import com.vitorpamplona.amethyst.commons.viewmodels.NestNetworkChangeBus import com.vitorpamplona.amethyst.ui.MainActivity +import com.vitorpamplona.quartz.utils.Log /** * Process-anchor for an active audio-room session. Holds a partial wake-lock @@ -57,7 +67,42 @@ import com.vitorpamplona.amethyst.ui.MainActivity class NestForegroundService : Service() { private var wakeLock: PowerManager.WakeLock? = null private var promoted = false - private var audioFocusRequest: android.media.AudioFocusRequest? = null + private var audioFocusRequest: AudioFocusRequest? = null + + /** + * Logs route changes (Bluetooth headset attach/detach, wired + * headset, USB audio, speakerphone) so field reports of + * "audio cut out when I plugged in headphones" can be correlated + * with a concrete device-add / device-remove event. + * + * Doesn't drive playback decisions — Android's AudioTrack + + * AudioRecord auto-route to whichever device the OS treats as + * active, so the brief silence on a route swap is unavoidable + * without going through the (intrusive) `MODE_IN_COMMUNICATION` + + * `setCommunicationDevice` flow that this service deliberately + * avoids. v1 ships observability only; future work could pause + * playback briefly across a route change to mask the swap. + */ + private val deviceCallback = + object : AudioDeviceCallback() { + override fun onAudioDevicesAdded(addedDevices: Array?) { + addedDevices?.forEach { dev -> + Log.i("NestAudio") { + "audio device added: type=${dev.type} name='${dev.productName}' " + + "isSink=${dev.isSink} isSource=${dev.isSource}" + } + } + } + + override fun onAudioDevicesRemoved(removedDevices: Array?) { + removedDevices?.forEach { dev -> + Log.i("NestAudio") { + "audio device removed: type=${dev.type} name='${dev.productName}' " + + "isSink=${dev.isSink} isSource=${dev.isSource}" + } + } + } + } override fun onCreate() { super.onCreate() @@ -67,13 +112,122 @@ class NestForegroundService : Service() { .newWakeLock(PowerManager.PARTIAL_WAKE_LOCK, "amethyst:audio-room") .apply { setReferenceCounted(false) } requestAudioFocus() + registerAudioDeviceCallback() + registerNetworkCallback() + } + + private fun registerAudioDeviceCallback() { + runCatching { + val mgr = getSystemService(Context.AUDIO_SERVICE) as AudioManager + // null Handler → callback runs on the main looper, which is + // the cheapest path for "log a line" handlers like this. + mgr.registerAudioDeviceCallback(deviceCallback, null) + } + } + + private fun unregisterAudioDeviceCallback() { + runCatching { + val mgr = getSystemService(Context.AUDIO_SERVICE) as AudioManager + mgr.unregisterAudioDeviceCallback(deviceCallback) + } } /** - * Request transient-may-duck audio focus so an inbound phone - * call lowers the room volume cleanly instead of mixing two - * voices on top of each other. Acquired once per service - * lifetime; released in [onDestroy]. + * Track the current default network so the NetworkCallback can + * distinguish "first onAvailable after register" (no-op) from "the + * default network just changed under us" (publish). Volatile because + * the callback fires on a binder thread; the publish path is + * lock-free via [NestNetworkChangeBus]. + * + * `seenInitialNetwork` is the registration-time suppression flag — + * set true on the first onAvailable and never cleared. The earlier + * shape used `previous != null` as the suppression check, which + * also suppressed the WiFi-loss-then-cellular-available case + * (`onLost` clears [currentDefaultNetwork] back to null, then the + * follow-up `onAvailable` looks identical to the registration + * callback). That broke the most important scenario this whole + * code path exists for — a clean Wi-Fi ↔ cellular handover. + */ + @Volatile private var currentDefaultNetwork: Network? = null + + @Volatile private var seenInitialNetwork: Boolean = false + + /** + * Listens for the device's default-network changing (Wi-Fi ↔ + * cellular handover, plane mode toggle, hotspot swap) and signals + * every active [com.vitorpamplona.amethyst.commons.viewmodels.NestViewModel] + * to recycle its QUIC session. Without this nudge the QUIC + * connection sitting on the now-dead socket would have to wait + * for its PTO (~30 s) before the wrapper notices — a long + * audible silence on every handover. + * + * The callback also fires once right after registration with the + * current default network — we suppress that emission so the VM + * doesn't recycle on every service start. + */ + private val networkCallback = + object : ConnectivityManager.NetworkCallback() { + override fun onAvailable(network: Network) { + val previous = currentDefaultNetwork + currentDefaultNetwork = network + if (!seenInitialNetwork) { + // Registration-time callback — fires once with the + // currently-active default network. Suppress so the + // VM doesn't recycle on every service start. + seenInitialNetwork = true + return + } + if (previous != network) { + Log.i("NestNet") { + "default network changed ($previous → $network), recycling QUIC sessions" + } + NestNetworkChangeBus.publish() + } + } + + override fun onLost(network: Network) { + if (currentDefaultNetwork == network) { + // We've lost the current default. Don't publish here — + // the next onAvailable (with the replacement network) + // will, and recycling NOW means the wrapper would try + // to handshake on no network at all and fail-then- + // backoff. Just clear so the next onAvailable is + // recognised as a change. + currentDefaultNetwork = null + } + } + } + + private fun registerNetworkCallback() { + runCatching { + val mgr = getSystemService(Context.CONNECTIVITY_SERVICE) as ConnectivityManager + mgr.registerDefaultNetworkCallback(networkCallback) + } + } + + private fun unregisterNetworkCallback() { + runCatching { + val mgr = getSystemService(Context.CONNECTIVITY_SERVICE) as ConnectivityManager + mgr.unregisterNetworkCallback(networkCallback) + } + currentDefaultNetwork = null + seenInitialNetwork = false + } + + /** + * Request audio focus for the duration of the audio-room session + * and route the system's focus-change events into [NestAudioFocusBus] + * so every active [com.vitorpamplona.amethyst.commons.viewmodels.NestViewModel] + * can react. + * + * Why we actually handle the focus change (vs the previous no-op + * listener): the platform only auto-ducks streams that opted into + * auto-ducking (CONTENT_TYPE_MUSIC, etc.) — a `CONTENT_TYPE_SPEECH` + * stream is left alone, so without a real listener an inbound phone + * call would mix on top of the room audio and a Maps voice prompt + * would be inaudible against an active speaker. The bus carries + * the translated state to the VM, which silences the listener + * playback and the broadcast mic for the duration of the loss. * * Matches the playback `AudioAttributes` we set on `AudioTrack` * in `AudioTrackPlayer` (USAGE_MEDIA + CONTENT_TYPE_SPEECH) so @@ -82,7 +236,7 @@ class NestForegroundService : Service() { */ private fun requestAudioFocus() { if (audioFocusRequest != null) return - val mgr = getSystemService(Context.AUDIO_SERVICE) as android.media.AudioManager + val mgr = getSystemService(Context.AUDIO_SERVICE) as AudioManager val attrs = android.media.AudioAttributes .Builder() @@ -90,26 +244,62 @@ class NestForegroundService : Service() { .setContentType(android.media.AudioAttributes.CONTENT_TYPE_SPEECH) .build() val request = - android.media.AudioFocusRequest - .Builder(android.media.AudioManager.AUDIOFOCUS_GAIN) + AudioFocusRequest + .Builder(AudioManager.AUDIOFOCUS_GAIN) .setAudioAttributes(attrs) .setAcceptsDelayedFocusGain(false) - .setOnAudioFocusChangeListener({ /* no-op — duck-on-loss handled by the OS */ }) - .build() - // Best-effort: a refused request just means the OS will - // duck/pause us based on its own policy. Don't fail the - // service start over a focus denial. - runCatching { mgr.requestAudioFocus(request) } + .setOnAudioFocusChangeListener { focusChange -> + NestAudioFocusBus.publish(translateFocusChange(focusChange)) + }.build() + // Strict GRANTED check: anything else — FAILED (active call + // blocking us), DELAYED (we set [setAcceptsDelayedFocusGain] + // false so this shouldn't happen but treat it as not-granted + // defensively), or a swallowed exception from runCatching — + // means we don't actually own playback yet, so the VM must + // start muted. The earlier shape used `!= FAILED` which fell + // through to Granted on `null` (exception path) and on + // DELAYED (= 2), playing audio over a call that the OS hadn't + // released to us. + val result = runCatching { mgr.requestAudioFocus(request) }.getOrNull() + if (result == AudioManager.AUDIOFOCUS_REQUEST_GRANTED) { + NestAudioFocusBus.publish(NestAudioFocusState.Granted) + } else { + NestAudioFocusBus.publish(NestAudioFocusState.TransientLoss) + } audioFocusRequest = request } private fun abandonAudioFocus() { val request = audioFocusRequest ?: return - val mgr = getSystemService(Context.AUDIO_SERVICE) as android.media.AudioManager + val mgr = getSystemService(Context.AUDIO_SERVICE) as AudioManager runCatching { mgr.abandonAudioFocusRequest(request) } audioFocusRequest = null + // Reset to Granted so a future foreground-service start that + // happens before the next focus request lands doesn't inherit + // a stale "muted because we lost focus" state. + NestAudioFocusBus.publish(NestAudioFocusState.Granted) } + private fun translateFocusChange(focusChange: Int): NestAudioFocusState = + when (focusChange) { + AudioManager.AUDIOFOCUS_GAIN, + AudioManager.AUDIOFOCUS_GAIN_TRANSIENT, + AudioManager.AUDIOFOCUS_GAIN_TRANSIENT_MAY_DUCK, + AudioManager.AUDIOFOCUS_GAIN_TRANSIENT_EXCLUSIVE, + -> NestAudioFocusState.Granted + + AudioManager.AUDIOFOCUS_LOSS_TRANSIENT, + AudioManager.AUDIOFOCUS_LOSS_TRANSIENT_CAN_DUCK, + -> NestAudioFocusState.TransientLoss + + AudioManager.AUDIOFOCUS_LOSS -> NestAudioFocusState.Loss + + // Unknown future codes: be defensive and treat as Granted + // so a vendor-specific extension can't silently mute the + // room forever. + else -> NestAudioFocusState.Granted + } + override fun onStartCommand( intent: Intent?, flags: Int, @@ -217,6 +407,8 @@ class NestForegroundService : Service() { wakeLock?.takeIf { it.isHeld }?.release() wakeLock = null abandonAudioFocus() + unregisterAudioDeviceCallback() + unregisterNetworkCallback() super.onDestroy() } diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestAudioFocusBus.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestAudioFocusBus.kt new file mode 100644 index 000000000..4e61fe35e --- /dev/null +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestAudioFocusBus.kt @@ -0,0 +1,92 @@ +/* + * 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.amethyst.commons.viewmodels + +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.StateFlow +import kotlinx.coroutines.flow.asStateFlow + +/** + * Process-wide audio-focus signal published by the platform-side audio + * focus listener (Android: `NestForegroundService`'s + * `OnAudioFocusChangeListener`) and consumed by every active + * [NestViewModel]. + * + * Decoupled from the platform audio APIs so commons can stay free of + * `android.media.AudioManager` references — the platform layer + * translates focus-change codes into the small enum below before + * publishing. + * + * Why a singleton: there's at most one foreground audio-room session + * per process (the foreground service is unique), and every ViewModel + * wants to observe the same focus signal, so keying by some other + * dimension would just complicate the wiring. The bus survives + * activity / VM rotations naturally. + */ +object NestAudioFocusBus { + private val _state = MutableStateFlow(NestAudioFocusState.Granted) + + /** + * The latest audio-focus state. Defaults to [NestAudioFocusState.Granted] + * — i.e. "we own playback" — so consumers don't have to special-case + * the boot-up window before the foreground service has a chance to + * register its listener. + */ + val state: StateFlow = _state.asStateFlow() + + /** + * Publish a new focus state. Only the platform-side listener calls + * this; consumers observe via [state]. + */ + fun publish(newState: NestAudioFocusState) { + _state.value = newState + } +} + +/** + * Audio-focus state translated from the platform's `AUDIOFOCUS_*` codes. + * Maps them onto the three actions an audio-room app actually cares + * about: keep playing, pause-because-something-else-is-playing, and + * stop-because-something-else-is-now-the-primary-audio-source. + */ +enum class NestAudioFocusState { + /** We own playback. Normal operation. */ + Granted, + + /** + * We've lost focus temporarily — typical triggers are an inbound + * phone call, a maps voice prompt, a system alarm. The + * [NestViewModel] reacts by silencing playback + the broadcast + * mic (the user-visible "muted" state stays unchanged so it + * restores on regain). Audio pipeline keeps running so resume + * is sample-accurate. + */ + TransientLoss, + + /** + * Permanent focus loss — another app has taken over as the + * primary audio source for the foreseeable future. Same effective + * action as [TransientLoss] in v1 (silence both directions), but + * the distinction is preserved so future enhancements can tear + * down the audio device entirely on long-form loss. + */ + Loss, +} diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestNetworkChangeBus.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestNetworkChangeBus.kt new file mode 100644 index 000000000..3a0a5f458 --- /dev/null +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestNetworkChangeBus.kt @@ -0,0 +1,80 @@ +/* + * 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.amethyst.commons.viewmodels + +import kotlinx.coroutines.channels.BufferOverflow +import kotlinx.coroutines.flow.MutableSharedFlow +import kotlinx.coroutines.flow.SharedFlow +import kotlinx.coroutines.flow.asSharedFlow + +/** + * Process-wide signal published by the platform-side connectivity + * listener (Android: `NestForegroundService`'s + * `ConnectivityManager.NetworkCallback`) and consumed by every active + * [NestViewModel] to recycle its [com.vitorpamplona.nestsclient.NestsListener] + * / [com.vitorpamplona.nestsclient.NestsSpeaker] sessions. + * + * Why this exists: when a phone hands over from Wi-Fi to cellular (or + * the other way), the local socket binding's source IP changes. The + * QUIC connection sitting on the previous socket isn't notified of + * the change — it'll keep retransmitting into the void until its PTO + * fires (`~30 s` for an idle connection). Without an external nudge + * the user hears 30 seconds of silence before the wrapper's reconnect + * loop notices the failure. With this bus, the wrapper recycles the + * QUIC session the moment the platform sees the network change, + * shrinking the audible gap to a single re-handshake (≈ 1 s on + * typical mobile networks). + * + * Decoupled from `android.net.ConnectivityManager` so commons stays + * platform-free — the platform layer translates `onAvailable` / + * `onLost` callbacks into a single `publish()` event before the + * VM observes it. + */ +object NestNetworkChangeBus { + /** + * `extraBufferCapacity = 1, DROP_OLDEST` — the consumer only cares + * that *a* network change happened, not how many. A burst of + * onLost / onAvailable calls during a Wi-Fi flap collapses to a + * single recycle event, which is exactly what we want (one + * handshake instead of N). + */ + private val _events = + MutableSharedFlow( + replay = 0, + extraBufferCapacity = 1, + onBufferOverflow = BufferOverflow.DROP_OLDEST, + ) + + /** + * Hot flow of network-change events. Consumers ([NestViewModel]) + * collect this and call `recycleSession()` on their listener + + * speaker. + */ + val events: SharedFlow = _events.asSharedFlow() + + /** + * Publish a network-change event. Only the platform-side listener + * calls this; consumers observe via [events]. + */ + fun publish() { + _events.tryEmit(Unit) + } +} diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestViewModel.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestViewModel.kt index 896c7398b..b06bd9f36 100644 --- a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestViewModel.kt +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestViewModel.kt @@ -34,6 +34,7 @@ import com.vitorpamplona.nestsclient.NestsRoomConfig import com.vitorpamplona.nestsclient.NestsSpeaker import com.vitorpamplona.nestsclient.NestsSpeakerState import com.vitorpamplona.nestsclient.audio.AudioCapture +import com.vitorpamplona.nestsclient.audio.AudioException import com.vitorpamplona.nestsclient.audio.AudioPlayer import com.vitorpamplona.nestsclient.audio.NestPlayer import com.vitorpamplona.nestsclient.audio.OpusDecoder @@ -282,6 +283,34 @@ class NestViewModel( private var speakerStateJob: Job? = null private var speakerConnectJob: Job? = null + /** + * Mirror of [NestAudioFocusBus.state] flipped to a boolean: `true` + * while another app holds audio focus (inbound phone call, Maps + * voice prompt, system alarm). Composed with the user-visible + * mute states (`NestUiState.isMuted` for listeners, + * `BroadcastUiState.Broadcasting.isMuted` for the speaker mic) via + * [effectiveListenMuted] / [effectiveMicMuted] so the user-chosen + * state is preserved across a focus loss + regain cycle. + * + * Maintained by [audioFocusObserverJob], started in [launchConnect] + * (so a never-connected VM doesn't subscribe needlessly). + */ + @Volatile private var focusMuted: Boolean = false + private var audioFocusObserverJob: Job? = null + + /** + * Subscribed in [ensureNetworkChangeObserverStarted]; calls + * [com.vitorpamplona.nestsclient.NestsListener.recycleSession] + * (and the speaker counterpart) when the platform reports a + * default-network change. Without this, a Wi-Fi → cellular + * handover leaves the QUIC connection sitting on the now-dead + * socket until its PTO fires (~30 s) — the wrapper's reconnect + * loop only learns about the failure at PTO. With the bus + * nudge, recycle happens immediately and the audible gap + * shrinks to a single re-handshake. + */ + private var networkChangeObserverJob: Job? = null + /** * Push the latest known speaker set from the room event. The user's * own pubkey (when broadcasting) is filtered out so we don't subscribe @@ -317,13 +346,97 @@ class NestViewModel( teardown(targetState = ConnectionUiState.Idle, finalCleanup = false) } + ensureAudioFocusObserverStarted() + ensureNetworkChangeObserverStarted() launchConnect() } + /** + * Subscribe to [NestAudioFocusBus] for the lifetime of this VM (or + * until [teardown] cancels). On loss, mute every active listener + * + the broadcast mic without touching the user-visible mute + * states; on regain, restore the user's choice. Idempotent — + * a second call while the job is alive is a no-op. + */ + private fun ensureAudioFocusObserverStarted() { + if (audioFocusObserverJob?.isActive == true) return + audioFocusObserverJob = + viewModelScope.launch { + NestAudioFocusBus.state.collect { focusState -> + if (closed) return@collect + val newFocusMuted = focusState != NestAudioFocusState.Granted + if (newFocusMuted == focusMuted) return@collect + focusMuted = newFocusMuted + // Re-apply both directions. effectiveListenMuted() + // and the per-broadcast effective rebuild below + // both consume the just-updated [focusMuted] field. + applyEffectiveListenMute() + applyEffectiveMicMute() + } + } + } + + /** + * Subscribe to [NestNetworkChangeBus] for the lifetime of this VM + * (or until [teardown]). On a default-network change, ask both + * the listener and the speaker wrappers to recycle their + * underlying sessions — much faster than waiting for the QUIC + * PTO on a dead socket. Idempotent. + */ + private fun ensureNetworkChangeObserverStarted() { + if (networkChangeObserverJob?.isActive == true) return + networkChangeObserverJob = + viewModelScope.launch { + NestNetworkChangeBus.events.collect { + if (closed) return@collect + // Best-effort. The wrapper-level recycleSession is a + // no-op on raw / non-reconnecting implementations + // (test fakes, IETF reference) so a network change + // there is silently ignored — the production paths + // wrap with [connectReconnectingNestsListener] / + // [connectReconnectingNestsSpeaker] which both + // override the hook to close the inner session. + listener?.runCatching { recycleSession() } + speaker?.runCatching { recycleSession() } + } + } + } + + /** + * Push the effective mic mute (`user choice OR focus-loss`) onto + * the live broadcast handle, if any. The user-visible + * [BroadcastUiState.Broadcasting.isMuted] stays the user's choice + * — UI only reflects that, never the focus-driven temporary + * silencing — so a focus regain is invisible to the user. + */ + private fun applyEffectiveMicMute() { + val handle = broadcastHandle ?: return + val ui = _uiState.value.broadcast + val userMuted = (ui as? BroadcastUiState.Broadcasting)?.isMuted ?: false + val effective = userMuted || focusMuted + viewModelScope.launch { + runCatching { handle.setMuted(effective) } + } + } + fun setMuted(muted: Boolean) { if (closed) return _uiState.update { it.copy(isMuted = muted) } - activeSubscriptions.values.forEach { it.player?.setMutedSafe(muted) } + applyEffectiveListenMute() + } + + /** + * Effective listener mute = user-chosen mute (from the talk-bar) + * OR audio-focus loss (phone call / nav prompt). Applied to every + * active player. The user-visible state (`NestUiState.isMuted`) + * stays the user's choice, so a focus regain restores it + * automatically. + */ + private fun effectiveListenMuted(): Boolean = _uiState.value.isMuted || focusMuted + + private fun applyEffectiveListenMute() { + val effective = effectiveListenMuted() + activeSubscriptions.values.forEach { it.player?.setMutedSafe(effective) } } /** @@ -523,7 +636,13 @@ class NestViewModel( if (closed) return val handle = broadcastHandle ?: return viewModelScope.launch { - val result = handle.runCatching { setMuted(muted) } + // Apply (user-mute OR audio-focus-loss) to the wire so the + // mic stays silent through a phone call without us losing + // the user's intent. UI then reflects the user choice (not + // the focus-effective state) so a focus-driven mute doesn't + // visually flip the talk button. + val effective = muted || focusMuted + val result = handle.runCatching { setMuted(effective) } if (closed) return@launch _uiState.update { val current = it.broadcast @@ -866,9 +985,26 @@ class NestViewModel( throw t } try { - val isMuted = _uiState.value.isMuted + // Apply (user-mute OR audio-focus-loss) so a speaker + // that comes on stage during a phone call attaches + // already silenced. The focus observer re-runs + // applyEffectiveListenMute() on regain, restoring the + // user's intent. + val isMuted = effectiveListenMuted() val isHushed = pubkey in _uiState.value.locallyHushed - val roomPlayer = NestPlayer(decoder, player, viewModelScope) + val roomPlayer = + NestPlayer( + decoder = decoder, + player = player, + scope = viewModelScope, + // ~100 ms of audio buffered before the AudioTrack + // starts consuming. Long enough to hide a typical + // Main-thread hiccup (Compose recomposition / GC) + // without making the join feel laggy. Empirically + // tuned in the audio-rooms audit; see NestPlayer + // kdoc for details. + prerollFrames = ROOM_PLAYER_PREROLL_FRAMES, + ) // Apply current mute + per-speaker hush state before play() // opens the device so the first frame respects them. player.setMutedSafe(isMuted) @@ -880,7 +1016,32 @@ class NestViewModel( val instrumented = handle.objects.onEach { onSpeakerActivity(pubkey) } roomPlayer.play( instrumented, - onError = { /* swallow per-packet decoder errors */ }, + onError = { err -> + // Per-packet decoder errors are noise — Opus PLC + // papers over a single bad frame. Swallow those. + // PlaybackFailed (or DeviceUnavailable from a + // deferred [AudioPlayer.beginPlayback], or the + // audio-priority dispatcher dying mid-stream) is + // terminal: the device is wedged for THIS + // subscription. Roll the slot back so a future + // reconcile can retry rather than leave a + // perma-spinning speaker tile. + when (err.kind) { + AudioException.Kind.DecoderError, + AudioException.Kind.EncoderError, + -> { + Unit + } + + AudioException.Kind.PlaybackFailed, + AudioException.Kind.DeviceUnavailable, + -> { + if (activeSubscriptions[pubkey] === slot) { + activeSubscriptions.remove(pubkey)?.let { closeSubscription(it) } + } + } + } + }, onLevel = { onAudioLevel(pubkey, it) }, ) slot.attach(handle, roomPlayer, player) @@ -928,6 +1089,20 @@ class NestViewModel( announcesJob = null levelEmitterJob?.cancel() levelEmitterJob = null + // Audio-focus observation only ends on the final teardown — + // a transient disconnect+reconnect (user retry, room swap) + // keeps the bus subscription alive so a focus loss that + // happens during the gap is still picked up. The observer + // is idempotent under [ensureAudioFocusObserverStarted], so + // the next [connect] call is safe whether or not we + // cancelled here. + if (finalCleanup) { + audioFocusObserverJob?.cancel() + audioFocusObserverJob = null + focusMuted = false + networkChangeObserverJob?.cancel() + networkChangeObserverJob = null + } rawAudioLevels.clear() if (_audioLevels.value.isNotEmpty()) { _audioLevels.value = emptyMap() @@ -1246,6 +1421,17 @@ sealed class BroadcastUiState { */ const val SPEAKING_TIMEOUT_MS: Long = 250L +/** + * Per-speaker pre-roll: number of decoded PCM frames buffered before the + * underlying [com.vitorpamplona.nestsclient.audio.AudioPlayer] starts + * consuming. 5 × 20 ms ≈ 100 ms of audio — long enough to mask a typical + * Main-thread stall (Compose recomposition / GC) without adding perceptible + * join latency. Combines with [com.vitorpamplona.nestsclient.audio.AudioTrackPlayer]'s + * ~250 ms AudioTrack buffer for ~350 ms of total slack between the + * arrival-of-frame and the underrun horizon. + */ +const val ROOM_PLAYER_PREROLL_FRAMES: Int = 5 + /** * Coalescing interval for [NestViewModel.audioLevels]. The decode loop * pushes a fresh peak every ~20 ms (one per Opus frame); we publish to diff --git a/nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/AudioRecordCapture.kt b/nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/AudioRecordCapture.kt index 1f5cfa1a1..b5645c85e 100644 --- a/nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/AudioRecordCapture.kt +++ b/nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/AudioRecordCapture.kt @@ -22,6 +22,10 @@ package com.vitorpamplona.nestsclient.audio import android.media.AudioRecord import android.media.MediaRecorder +import android.media.audiofx.AcousticEchoCanceler +import android.media.audiofx.AutomaticGainControl +import android.media.audiofx.NoiseSuppressor +import com.vitorpamplona.quartz.utils.Log import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.withContext import android.media.AudioFormat as AndroidAudioFormat @@ -32,6 +36,19 @@ import android.media.AudioFormat as AndroidAudioFormat * voice-chat libraries use, so it gets the platform's echo-cancellation and * noise-suppression filters when available. * + * **Audio effects (AEC / NS / AGC).** On most modern Android devices the + * `VOICE_COMMUNICATION` source is enough to engage the platform's echo + * canceller automatically. On a small set of older / OEM-customised + * devices it isn't — the AEC engine only attaches when the device is in + * `MODE_IN_COMMUNICATION`, which an audio-room app deliberately avoids + * driving (it reroutes everything through the call audio path and shows a + * "phone call" notification icon). To cover those devices without + * touching `AudioManager.mode` we explicitly attach the standalone + * [AcousticEchoCanceler] / [NoiseSuppressor] / [AutomaticGainControl] + * effects to the AudioRecord's session id. On devices where the source + * already engages them, attaching a second effect is a no-op (the + * platform deduplicates by session id) — so this is purely additive. + * * **Permission:** the caller is responsible for holding `RECORD_AUDIO` before * calling [start]; this class will throw [AudioException.Kind.DeviceUnavailable] * if the OS denies the resource. @@ -40,6 +57,9 @@ class AudioRecordCapture( private val source: Int = MediaRecorder.AudioSource.VOICE_COMMUNICATION, ) : AudioCapture { private var record: AudioRecord? = null + private var aec: AcousticEchoCanceler? = null + private var ns: NoiseSuppressor? = null + private var agc: AutomaticGainControl? = null private var stopped = false override fun start() { @@ -97,6 +117,34 @@ class AudioRecordCapture( ) } record = rec + // Attach standalone audio effects to the session. This covers + // devices where the VOICE_COMMUNICATION source alone doesn't + // engage AEC (typically because they only attach AEC under + // MODE_IN_COMMUNICATION). All three are best-effort — a device + // that doesn't support an effect leaves it null without + // affecting capture. + attachAudioEffects(rec.audioSessionId) + } + + private fun attachAudioEffects(sessionId: Int) { + if (AcousticEchoCanceler.isAvailable()) { + aec = + runCatching { AcousticEchoCanceler.create(sessionId)?.apply { enabled = true } } + .onFailure { Log.w("NestTx") { "AcousticEchoCanceler.create failed: ${it.message}" } } + .getOrNull() + } + if (NoiseSuppressor.isAvailable()) { + ns = + runCatching { NoiseSuppressor.create(sessionId)?.apply { enabled = true } } + .onFailure { Log.w("NestTx") { "NoiseSuppressor.create failed: ${it.message}" } } + .getOrNull() + } + if (AutomaticGainControl.isAvailable()) { + agc = + runCatching { AutomaticGainControl.create(sessionId)?.apply { enabled = true } } + .onFailure { Log.w("NestTx") { "AutomaticGainControl.create failed: ${it.message}" } } + .getOrNull() + } } override suspend fun readFrame(): ShortArray? { @@ -128,6 +176,16 @@ class AudioRecordCapture( override fun stop() { if (stopped) return stopped = true + // Release the audio effects BEFORE the AudioRecord — they hold + // a session-id reference and the platform expects effect + // teardown to precede the AudioRecord.release() that frees + // the session. + runCatching { aec?.release() } + runCatching { ns?.release() } + runCatching { agc?.release() } + aec = null + ns = null + agc = null val rec = record ?: return record = null runCatching { rec.stop() } diff --git a/nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/AudioTrackPlayer.kt b/nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/AudioTrackPlayer.kt index 01dfddd07..8434263ea 100644 --- a/nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/AudioTrackPlayer.kt +++ b/nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/AudioTrackPlayer.kt @@ -22,8 +22,12 @@ package com.vitorpamplona.nestsclient.audio import android.media.AudioAttributes import android.media.AudioTrack -import kotlinx.coroutines.Dispatchers +import android.os.Process +import kotlinx.coroutines.ExecutorCoroutineDispatcher +import kotlinx.coroutines.asCoroutineDispatcher import kotlinx.coroutines.withContext +import java.util.concurrent.ExecutorService +import java.util.concurrent.Executors import android.media.AudioFormat as AndroidAudioFormat /** @@ -45,9 +49,30 @@ import android.media.AudioFormat as AndroidAudioFormat * `MediaRecorder.AudioSource.VOICE_COMMUNICATION` regardless of the * playback usage. * - * Buffer sizing: 4× minimum so the producer can fall behind by ~80 ms before - * dropouts, which roughly matches the jitter the WebTransport datagram path - * introduces over typical mobile networks. + * Buffer sizing: target ~250 ms of slack, computed as + * `max(minBuffer * 16, 250 ms-equivalent)`. The previous 4× minimum (~80 ms + * by the device-reported floor) underran on devices with very small + * `getMinBufferSize` returns and on any handset whose decode loop got + * stalled by Compose recomposition / GC on Main. 250 ms matches the + * jitter-buffer depth Spaces / Clubhouse use for hands-free audio rooms, + * and combined with the per-subscription pre-roll in [NestPlayer] keeps + * the AudioTrack from underruning across typical mobile network jitter. + * + * Threading: writes go through a per-instance audio-priority single-thread + * dispatcher (`HandlerThread` + [Process.THREAD_PRIORITY_AUDIO]) instead of + * `Dispatchers.IO`. The previous shape did one IO dispatcher hop per Opus + * frame (~50 hops/sec/speaker) and contended with whatever else `Dispatchers.IO` + * was running; an audio-priority dedicated thread gets reliable scheduling + * and removes the contention. + * + * Two-phase startup: [start] allocates the AudioTrack (in stopped state) + * + spins up the writer thread. [beginPlayback] calls `AudioTrack.play()` + * to flip the device into the playing state. Splitting the two lets + * [NestPlayer] pre-roll several decoded frames into the AudioTrack's + * internal buffer BEFORE the hardware starts pulling samples, so playback + * begins with ~100 ms of buffered audio instead of underrunning on the + * first frame. AudioTrack in `MODE_STREAM` explicitly supports `write()` + * before `play()` per the platform docs — this is the intended pattern. */ class AudioTrackPlayer( private val usage: Int = AudioAttributes.USAGE_MEDIA, @@ -57,6 +82,18 @@ class AudioTrackPlayer( private var muted: Boolean = false private var volume: Float = 1f + /** + * Dedicated audio-priority single-thread executor for AudioTrack writes. + * Lazily created on [start] and shut down on [stop] so a never-started + * player doesn't leak a thread. `THREAD_PRIORITY_AUDIO` is the standard + * Linux nice level for VoIP / WebRTC playback paths on Android — it sits + * above background but below true audio-callback priority, so the OS + * scheduler keeps it running through GC / Compose recomposition without + * starving other threads. + */ + private var audioExecutor: ExecutorService? = null + private var audioDispatcher: ExecutorCoroutineDispatcher? = null + override fun start() { if (track != null) return @@ -79,7 +116,14 @@ class AudioTrackPlayer( "AudioTrack.getMinBufferSize returned $minBuffer for ${AudioFormat.SAMPLE_RATE_HZ} Hz", ) } - val bufferBytes = minBuffer * 4 + // Target ~250 ms of audio: enough headroom so the decode loop can + // miss its 20 ms cadence by an order of magnitude before the device + // underruns. Take the larger of `minBuffer * 16` and an explicit + // 250 ms-equivalent so devices that report a small minBuffer still + // get the same wall-clock slack. + val targetBytes250Ms = + (AudioFormat.SAMPLE_RATE_HZ / 4) * AudioFormat.BYTES_PER_SAMPLE * AudioFormat.CHANNELS + val bufferBytes = maxOf(minBuffer * 16, targetBytes250Ms) val newTrack = try { @@ -109,25 +153,66 @@ class AudioTrackPlayer( ) } + // NOTE: deliberately NOT calling `newTrack.play()` here — that's + // [beginPlayback]'s job. AudioTrack.write is legal in the not-yet- + // playing state for `MODE_STREAM`, which is the contract that lets + // [NestPlayer] pre-roll decoded frames into the buffer before the + // hardware starts consuming. + applyMuteVolume(newTrack) + // Spin up the audio-priority writer thread. The executor is private + // to this player instance so per-speaker NestPlayer pumps don't + // contend on a shared queue. Priority is set inside the thread's + // Runnable because Linux thread priority is per-OS-thread, not + // per-Java Thread; `Thread.setPriority` does NOT translate to a + // Linux nice level on Android. `Process.setThreadPriority` does. + val executor = + Executors.newSingleThreadExecutor { r -> + Thread( + { + runCatching { Process.setThreadPriority(Process.THREAD_PRIORITY_AUDIO) } + r.run() + }, + "nest-audio-writer", + ) + } + audioExecutor = executor + audioDispatcher = executor.asCoroutineDispatcher() + track = newTrack + } + + override fun beginPlayback() { + // Idempotent: AudioTrack.play() on an already-playing track is a + // no-op, and we gate on the track itself being non-null so a + // beginPlayback before start (or after stop) silently no-ops + // rather than blowing up — matches the rest of the player's + // tolerant-of-misordered-calls posture. + val t = track ?: return try { - newTrack.play() - } catch (t: Throwable) { - runCatching { newTrack.release() } + t.play() + } catch (e: Throwable) { + // PLAY-on-uninitialized AudioTrack is the only realistic + // failure here, and we already guard against an unallocated + // track above. Throw so [NestPlayer]'s outer catch surfaces + // it via `onError(AudioException.PlaybackFailed)` — same path + // that handles every other mid-stream device failure. throw AudioException( AudioException.Kind.DeviceUnavailable, "AudioTrack.play() rejected start", - t, + e, ) } - applyMuteVolume(newTrack) - track = newTrack } override suspend fun enqueue(pcm: ShortArray) { val t = track ?: throw AudioException(AudioException.Kind.PlaybackFailed, "player not started") - // AudioTrack.write blocks if the internal buffer is full. Run on IO so - // we don't stall a coroutine dispatcher backed by a small thread pool. - withContext(Dispatchers.IO) { + val dispatcher = + audioDispatcher + ?: throw AudioException(AudioException.Kind.PlaybackFailed, "audio dispatcher not initialized") + // AudioTrack.write blocks if the internal buffer is full. Run on the + // per-instance audio-priority writer thread so the WRITE_BLOCKING + // suspension is on a thread the OS schedules tightly, and so we + // don't compete with whatever else `Dispatchers.IO` is running. + withContext(dispatcher) { val written = t.write(pcm, 0, pcm.size, AudioTrack.WRITE_BLOCKING) if (written < 0) { throw AudioException( @@ -155,6 +240,16 @@ class AudioTrackPlayer( runCatching { t.flush() } runCatching { t.stop() } runCatching { t.release() } + // Tear down the audio-priority writer. close() on the + // ExecutorCoroutineDispatcher shuts down the executor (and the + // underlying single thread) — pending writes are abandoned rather + // than blocked on, since the AudioTrack itself has already been + // stopped + released two lines up so any further `write` would + // throw IllegalStateException anyway. + audioDispatcher?.close() + audioDispatcher = null + audioExecutor?.let { runCatching { it.shutdownNow() } } + audioExecutor = null } private fun applyMuteVolume(track: AudioTrack) { diff --git a/nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/MediaCodecOpusDecoder.kt b/nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/MediaCodecOpusDecoder.kt index 24e08e2cc..5f044d508 100644 --- a/nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/MediaCodecOpusDecoder.kt +++ b/nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/MediaCodecOpusDecoder.kt @@ -57,8 +57,33 @@ class MediaCodecOpusDecoder : OpusDecoder { override fun decode(opusPacket: ByteArray): ShortArray { check(!released) { "decoder released" } - val inputIndex = codec.dequeueInputBuffer(DEQUEUE_TIMEOUT_US) - if (inputIndex < 0) return ShortArray(0) + // Pre-sized output. Write decoded samples directly into the array + // via ShortBuffer.get(dst, off, len) — the previous shape went + // through ArrayList, which boxed every PCM sample + // (~48 000 alloc/sec/speaker on the audio hot path). + val out = ShortArray(AudioFormat.FRAME_SIZE_SAMPLES) + var outPos = 0 + + // 1. Acquire an input slot. On rare back-pressure (output buffers + // haven't been drained fast enough → all input slots are tied + // up), drain whatever output IS ready first to free slots, + // then retry input dequeue with a longer timeout. The previous + // shape returned `ShortArray(0)` immediately when the first + // 10 ms dequeue missed, turning every transient stall (thermal + // throttle, GC pause) into a 20 ms audio gap. + var inputIndex = codec.dequeueInputBuffer(DEQUEUE_TIMEOUT_US) + if (inputIndex < 0) { + outPos = drainAvailableOutput(out, outPos) + inputIndex = codec.dequeueInputBuffer(DEQUEUE_RETRY_TIMEOUT_US) + if (inputIndex < 0) { + // Genuine input starvation past the retry window. Drop + // this packet (presentationTimeUs is NOT advanced — Opus + // PLC will paper over the gap on the listener) and + // return whatever we drained from output so the player + // doesn't underrun on the back of one tight cycle. + return if (outPos > 0) out.copyOf(outPos) else ShortArray(0) + } + } val inputBuffer = codec.getInputBuffer(inputIndex) ?: throw AudioException( @@ -71,17 +96,25 @@ class MediaCodecOpusDecoder : OpusDecoder { // Advance presentation time by one 20 ms frame. presentationTimeUs += FRAME_DURATION_US - // Drain whatever output is ready right now. A single Opus packet - // typically yields exactly one output buffer, but on some devices the - // first call returns INFO_OUTPUT_FORMAT_CHANGED before the PCM frame. - // - // Write the decoded samples directly into a pre-sized ShortArray - // via ShortBuffer.get(dst, off, len) — the previous shape went via - // ArrayList, which boxed every PCM sample (one heap object - // per Short × 960 samples × 50 fps × N speakers ≈ 48 000 - // allocations/sec/speaker on the audio hot path). - val out = ShortArray(AudioFormat.FRAME_SIZE_SAMPLES) - var outPos = 0 + // 2. Drain whatever output is ready right now. A single Opus + // packet typically yields exactly one output buffer, but + // [drainAvailableOutput] tolerates an INFO_OUTPUT_FORMAT_CHANGED + // that some devices emit before the first PCM frame. + outPos = drainAvailableOutput(out, outPos) + return if (outPos == out.size) out else out.copyOf(outPos) + } + + /** + * Drain any output buffers MediaCodec has ready, copying samples into + * [out] starting at [startPos]. Returns the new write position. + * Stops on the first empty / EOS / TRY_AGAIN_LATER signal — never + * blocks past the [DEQUEUE_TIMEOUT_US] dequeue probe. + */ + private fun drainAvailableOutput( + out: ShortArray, + startPos: Int, + ): Int { + var outPos = startPos var formatChangeAbsorbed = false drain@ while (true) { val outputIndex = codec.dequeueOutputBuffer(bufferInfo, DEQUEUE_TIMEOUT_US) @@ -126,7 +159,7 @@ class MediaCodecOpusDecoder : OpusDecoder { } } } - return if (outPos == out.size) out else out.copyOf(outPos) + return outPos } override fun release() { @@ -138,6 +171,20 @@ class MediaCodecOpusDecoder : OpusDecoder { companion object { private const val DEQUEUE_TIMEOUT_US = 10_000L // 10 ms + + /** + * Fallback timeout after the first input dequeue misses + we + * drain pending output to free slots. Deliberately well under + * the 20 ms frame cadence so a tight back-pressure window is + * absorbed without falling off the per-frame budget. The + * previous code took a 0 ms retry (i.e. dropped the frame + * outright on the first miss), which turned every transient + * stall into ~20 ms of silence. 5 ms gives MediaCodec time to + * free a slot after the output drain without blowing the + * frame budget. + */ + private const val DEQUEUE_RETRY_TIMEOUT_US = 5_000L // 5 ms + private const val FRAME_DURATION_US = 20_000L // 20 ms private fun buildFormat(): MediaFormat { diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt index b1e9b3dac..c719a8cac 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt @@ -59,7 +59,8 @@ class MoqLiteNestsSpeaker internal constructor( * Defaults to [NestMoqLiteBroadcaster.DEFAULT_FRAMES_PER_GROUP]. */ private val framesPerGroup: Int = NestMoqLiteBroadcaster.DEFAULT_FRAMES_PER_GROUP, -) : NestsSpeaker { +) : NestsSpeaker, + HotSwappablePublisherSource { override val state: StateFlow = mutableState.asStateFlow() private val gate = Mutex() @@ -100,7 +101,7 @@ class MoqLiteNestsSpeaker internal constructor( NestMoqLiteBroadcaster( capture = captureFactory(), encoder = encoderFactory(), - publisher = publisher, + initialPublisher = publisher, scope = scope, framesPerGroup = framesPerGroup, ).also { @@ -113,7 +114,7 @@ class MoqLiteNestsSpeaker internal constructor( // the session — without this signal the // outward state stays on Broadcasting // and the room is silently mute. - onBroadcastTerminalFailure() + reportBroadcastTerminalFailure() }, onLevel = onLevel, ) @@ -139,6 +140,15 @@ class MoqLiteNestsSpeaker internal constructor( } } + /** + * [HotSwappablePublisherSource] implementation. See the interface + * kdoc — this method mints a fresh publisher on the session + * WITHOUT spinning up a broadcaster on top of it. Used by the + * reconnect wrapper's hot-swap path; not called from the + * non-reconnecting path which goes through [startBroadcasting]. + */ + override suspend fun openPublisherForHotSwap(track: String): MoqLitePublisherHandle = session.publish(broadcastSuffix = speakerPubkeyHex, track = track) + /** * Compare-and-clear that runs from inside [close] (already holds * [gate]) and from [MoqLiteBroadcastHandle.close] (doesn't). @@ -160,8 +170,13 @@ class MoqLiteNestsSpeaker internal constructor( * the reconnect orchestrator (`ReconnectingNestsSpeaker`) observes * a terminal state and recycles the session. No-op if the speaker * is already in a terminal state. + * + * Also exposed via [HotSwappablePublisherSource.reportBroadcastTerminalFailure] + * so the hot-swap pump (which owns its own long-lived broadcaster) + * can drive the same orchestrator-reconnect path the legacy + * `startBroadcasting` flow does. */ - private fun onBroadcastTerminalFailure() { + override fun reportBroadcastTerminalFailure() { val current = mutableState.value if (current is NestsSpeakerState.Failed || current is NestsSpeakerState.Closed) return mutableState.value = @@ -257,3 +272,40 @@ internal class MoqLiteBroadcastHandle( parent.broadcastClosed(this) } } + +/** + * Internal hot-swap seam: speakers that expose this interface let the + * reconnect wrapper retarget a long-lived + * [com.vitorpamplona.nestsclient.audio.NestMoqLiteBroadcaster] onto a + * freshly-opened moq-lite session's publisher without restarting the + * AudioRecord / Opus encoder pipeline. Implemented by + * [MoqLiteNestsSpeaker]; not implemented by the IETF reference + * [DefaultNestsSpeaker], which falls back to the close-then-restart path + * inside [com.vitorpamplona.nestsclient.connectReconnectingNestsSpeaker]. + * + * The wrapper uses an `as?` cast to detect support so this interface + * can stay package-internal — protocol consumers never see it. + */ +internal interface HotSwappablePublisherSource { + /** + * Open a fresh [MoqLitePublisherHandle] on the underlying moq-lite + * session. Caller owns the returned handle's lifetime (typically + * via [com.vitorpamplona.nestsclient.audio.NestMoqLiteBroadcaster.swapPublisher]'s + * close-the-old contract). + */ + suspend fun openPublisherForHotSwap(track: String): MoqLitePublisherHandle + + /** + * Surface a broadcast-pipeline terminal failure (e.g. sustained + * `publisher.send` errors past + * [com.vitorpamplona.nestsclient.audio.NestMoqLiteBroadcaster.MAX_CONSECUTIVE_SEND_ERRORS]) + * by flipping the speaker's state to [NestsSpeakerState.Failed]. + * Called by the hot-swap pump when the long-lived broadcaster's + * `onTerminalFailure` fires; lets the reconnect orchestrator + * observe the terminal state and recycle the session, matching + * the legacy + * [MoqLiteNestsSpeaker.startBroadcasting] path's failure + * propagation. + */ + fun reportBroadcastTerminalFailure() +} diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsListener.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsListener.kt index 0b9e7fd95..2cc2be72c 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsListener.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsListener.kt @@ -116,6 +116,27 @@ interface NestsListener { "announces() is moq-lite-only; IETF listener has no announce-prefix flow.", ) + /** + * Force the underlying transport / MoQ session to be torn down and + * a fresh one opened in its place — without permanently closing + * this listener. Used by the platform layer on a network change + * (Wi-Fi ↔ cellular handover) to skip the ~30 s QUIC PTO that + * would otherwise have to fire before the wrapper notices the + * old socket is dead. + * + * Default no-op for non-reconnecting implementations (raw + * [DefaultNestsListener] / [MoqLiteNestsListener]) — there's no + * orchestrator to drive a reconnect, so a force recycle would + * just close the listener. The reconnecting wrapper + * (`ReconnectingHandle` from [connectReconnectingNestsListener]) + * overrides to close the active inner listener so its + * orchestrator opens a fresh session. + */ + suspend fun recycleSession() { + // no-op — only the reconnecting wrapper has anywhere to + // recycle to. + } + /** Tear down the MoQ session + underlying transport. Idempotent. */ suspend fun close() } diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsSpeaker.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsSpeaker.kt index dd5720610..88f584cac 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsSpeaker.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsSpeaker.kt @@ -66,6 +66,27 @@ interface NestsSpeaker { */ suspend fun startBroadcasting(onLevel: (Float) -> Unit = { /* no-op */ }): BroadcastHandle + /** + * Force the underlying transport / MoQ session to be torn down and + * a fresh one opened in its place — without permanently closing + * this speaker. Mirror of [NestsListener.recycleSession]; same + * use case (network handover) and same default no-op for non- + * reconnecting implementations. + * + * The reconnecting wrapper from + * [connectReconnectingNestsSpeaker] overrides to close the + * inner speaker so its orchestrator opens a fresh session. + * For moq-lite speakers, the long-lived + * [com.vitorpamplona.nestsclient.audio.NestMoqLiteBroadcaster] + * keeps capturing through the recycle and the + * [BroadcastHandle.swapPublisher]-based pump retargets onto + * the new session's publisher with no audible gap. + */ + suspend fun recycleSession() { + // no-op — only the reconnecting wrapper has anywhere to + // recycle to. + } + /** Tear down the MoQ session + transport. Idempotent. */ suspend fun close() } diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt index 308b62f8d..6c4e726d9 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt @@ -340,6 +340,14 @@ private class ReconnectingHandle( } if (terminalOrConnected !is NestsListenerState.Connected) return@collectLatest + // Backoff for the "opener threw" retry path — exponential + // 250 → 500 → 1 000 ms with reset on success. The previous + // shape used a flat 1 000 ms which, combined with the + // 64-frame (~1.3 s) wrapper buffer, just barely hid the + // first-retry gap; a quick retry usually succeeds because + // moq-rs propagates announces in < 200 ms. + var subscribeRetryDelayMs = SUBSCRIBE_RETRY_BACKOFF_INITIAL_MS + while (currentCoroutineContext().isActive) { val handle = try { @@ -368,14 +376,21 @@ private class ReconnectingHandle( // backoff used between publisher cycles. Log.w("NestRx") { "ReconnectingHandle.opener threw ${t::class.simpleName}: ${t.message} " + - "— retrying after ${SUBSCRIBE_RETRY_BACKOFF_MS}ms" + "— retrying after ${subscribeRetryDelayMs}ms" } null } if (handle == null) { - delay(SUBSCRIBE_RETRY_BACKOFF_MS) + delay(subscribeRetryDelayMs) + subscribeRetryDelayMs = + (subscribeRetryDelayMs * 2) + .coerceAtMost(SUBSCRIBE_RETRY_BACKOFF_MAX_MS) continue } + // Success: reset the backoff so the next publisher- + // cycle gap starts at the floor again (rather than + // inheriting the last failure window's saturation). + subscribeRetryDelayMs = SUBSCRIBE_RETRY_BACKOFF_INITIAL_MS liveHandleRef.set(handle) try { handle.objects.collect { frames.emit(it) } @@ -405,6 +420,29 @@ private class ReconnectingHandle( ) } + /** + * Force-close the active inner listener so the orchestrator's + * `terminalAwait` returns Closed and the next loop iteration opens + * a fresh session. Used by the platform layer on a network change + * (Wi-Fi ↔ cellular) — without this, the wrapper would have to + * wait for the QUIC PTO to fire on the now-dead old socket + * (~30 s of silence) before noticing. + * + * The SubscribeHandle re-issuance pump cuts existing subs over to + * the new session as soon as it's Connected — same path the + * JWT-refresh recycle uses — so the consumer-facing + * [SubscribeHandle.objects] flow keeps emitting once the new + * session lands. + */ + override suspend fun recycleSession() { + val current = activeListener.value ?: return + // Best-effort. If the close races a concurrent reconnect path + // (extremely rare — we only call this on deliberate user / + // platform signals), the orchestrator absorbs both via its + // existing terminal-state handling. + runCatching { current.close() } + } + override suspend fun close() { orchestrator.cancel() runCatching { activeListener.value?.close() } @@ -427,14 +465,24 @@ private class ReconnectingHandle( // gone publisher doesn't spin the relay with re-subscribes. private const val RESUBSCRIBE_BACKOFF_MS = 100L - // Backoff after an opener throws (typically: subscribe-before- - // announce arrives at the relay before the publisher exists, - // and the relay FINs the bidi without a SubscribeOk/Drop). One - // second is well over moq-rs's typical announce-propagation - // latency (< 200 ms in production traces), so the next retry - // usually succeeds; long enough that a never-arriving publisher - // doesn't hammer the relay either. - private const val SUBSCRIBE_RETRY_BACKOFF_MS = 1_000L + // Exponential backoff for the opener-throws retry path. + // Typical case: subscribe-before-announce arrives at the relay + // before the publisher exists, and the relay FINs the bidi + // without a SubscribeOk/Drop reply. + // + // - INITIAL = 250 ms: under moq-rs's typical announce-propagation + // latency (< 200 ms in production traces), so the very first + // retry usually succeeds and the listener-side buffer hides + // the gap. + // - Doubles each miss → 500 ms → 1 000 ms. + // - MAX = 1 000 ms: matches the previous flat constant; long + // enough that a never-arriving publisher doesn't hammer the + // relay, short enough to stay under the wrapper SharedFlow's + // ~1.3 s buffer once playback is rolling on the next attempt. + // - Reset on first successful subscribe so a subsequent + // publisher-cycle gap doesn't inherit the previous saturation. + private const val SUBSCRIBE_RETRY_BACKOFF_INITIAL_MS = 250L + private const val SUBSCRIBE_RETRY_BACKOFF_MAX_MS = 1_000L private val SYNTH_OK = SubscribeOk( diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsSpeaker.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsSpeaker.kt index a7f627ef0..a6ed6e68d 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsSpeaker.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsSpeaker.kt @@ -21,6 +21,7 @@ package com.vitorpamplona.nestsclient import com.vitorpamplona.nestsclient.audio.AudioCapture +import com.vitorpamplona.nestsclient.audio.NestMoqLiteBroadcaster import com.vitorpamplona.nestsclient.audio.OpusEncoder import com.vitorpamplona.nestsclient.transport.WebTransportFactory import com.vitorpamplona.quartz.nip01Core.signers.NostrSigner @@ -68,15 +69,23 @@ import java.util.concurrent.atomic.AtomicReference * `close` cancels the re-issue pump and best-effort closes the * latest live handle. * - * **Audio gap during refresh** — the wrapper closes the current - * underlying speaker (which stops the mic capture + Opus encoder + - * publisher) before opening the next, so the listener side will - * hear ~50–150 ms of silence at each recycle boundary. That's the - * trade-off we pay for a clean session swap; the alternative - * (carrying the mic capture across sessions) would require deeper - * plumbing into the audio pipeline. Acceptable for v1 — - * 9-min-spaced 150 ms gaps are well below the noise floor of a - * voice call. + * **Audio gap during refresh — eliminated for moq-lite** — when the + * underlying speaker implements [HotSwappablePublisherSource] + * (which [MoqLiteNestsSpeaker] does), the wrapper keeps a single + * long-lived [com.vitorpamplona.nestsclient.audio.NestMoqLiteBroadcaster] + * alive across session recycles and only swaps the + * [com.vitorpamplona.nestsclient.moq.lite.MoqLitePublisherHandle] + * underneath it. The mic + encoder run continuously through the + * boundary and the listener hears no silence. The old session + * close runs in parallel with the new openOnce so the WebTransport + * teardown doesn't block the swap. + * + * **Audio gap during refresh — legacy path** — for speakers that + * don't implement [HotSwappablePublisherSource] (the IETF reference + * `DefaultNestsSpeaker` and test fakes), the wrapper falls back to + * close-then-restart: the listener hears ~50–150 ms of silence per + * recycle. Acceptable because the IETF path is reference-only — + * production runs the moq-lite hot-swap path. * * Cancellation: cancelling [scope] (typically the room screen's VM * scope) cancels the reconnect loop and closes both the active @@ -186,16 +195,41 @@ suspend fun connectReconnectingNestsSpeaker( } if (terminal == null) { // Refresh deadline hit before any terminal state — - // planned recycle, not a failure. Close the old - // speaker; don't bump `attempt` (it's not a - // backoff event) so the next openOnce() runs - // immediately. - try { - speaker.close() - } catch (ce: kotlinx.coroutines.CancellationException) { - throw ce - } catch (_: Throwable) { - // Best-effort. + // planned recycle, not a failure. Don't bump + // `attempt` (it's not a backoff event) so the next + // openOnce() runs immediately. + // + // Hand the old speaker's close to a separate + // launch so it runs IN PARALLEL with the next + // iteration's openOnce(). Sequence on the wire: + // + // 1. (concurrent) old speaker close — closes + // old session's publisher + WebTransport. + // 2. (concurrent) new openOnce() → sets + // activeSpeaker = newSpeaker. + // 3. Wrapper's hot-swap pump observes the + // activeSpeaker change, opens a publisher + // on the new session, swaps it into the + // long-lived broadcaster, closes the old + // publisher. + // + // The previous shape closed the old speaker + // synchronously BEFORE openOnce, which forced the + // mic + encoder to stop and re-open across the + // boundary (50–150 ms audible silence at the + // listener). With the close hoisted onto a + // sibling job, the broadcaster keeps capturing + // through the recycle and the listener hears + // continuous audio. + val toClose = speaker + scope.launch { + try { + toClose.close() + } catch (ce: kotlinx.coroutines.CancellationException) { + throw ce + } catch (_: Throwable) { + // Best-effort. + } } attempt = 0 refreshTriggered = true @@ -249,7 +283,14 @@ suspend fun connectReconnectingNestsSpeaker( throw NestsException(firstReady.reason, firstReady.cause) } - return ReconnectingSpeakerHandle(state, activeSpeaker, orchestrator, scope) + return ReconnectingSpeakerHandle( + mutableState = state, + activeSpeaker = activeSpeaker, + orchestrator = orchestrator, + scope = scope, + captureFactory = captureFactory, + encoderFactory = encoderFactory, + ) } private fun isUserCancelledSpeaker(state: NestsSpeakerState.Failed): Boolean { @@ -266,6 +307,17 @@ private class ReconnectingSpeakerHandle( private val activeSpeaker: MutableStateFlow, private val orchestrator: Job, private val scope: CoroutineScope, + /** + * Audio-pipeline factories propagated from + * [connectReconnectingNestsSpeaker]. Used by the hot-swap + * [ReissuingBroadcastHandle] to construct ONE long-lived + * [NestMoqLiteBroadcaster] that survives session recycles — + * without these the broadcaster would have to be re-built + * (which restarts the mic + encoder + adds a 50–150 ms + * audible gap) on every JWT refresh. + */ + private val captureFactory: () -> AudioCapture, + private val encoderFactory: () -> OpusEncoder, ) : NestsSpeaker { override val state: StateFlow = mutableState.asStateFlow() @@ -291,14 +343,32 @@ private class ReconnectingSpeakerHandle( ?: error("no live session — wait for state == Connected before startBroadcasting") val handle = - ReissuingBroadcastHandle(activeSpeaker, scope, onLevel) { closed -> - if (activeBroadcast === closed) activeBroadcast = null - } + ReissuingBroadcastHandle( + activeSpeaker = activeSpeaker, + scope = scope, + captureFactory = captureFactory, + encoderFactory = encoderFactory, + onLevel = onLevel, + onClose = { closed -> + if (activeBroadcast === closed) activeBroadcast = null + }, + ) handle.start() activeBroadcast = handle handle } + /** + * Force-close the active inner speaker so the orchestrator opens + * a fresh session against the (presumably new) network. Used by + * the platform layer on a network change. Mirror of the listener + * wrapper's [com.vitorpamplona.nestsclient.NestsListener.recycleSession]. + */ + override suspend fun recycleSession() { + val current = activeSpeaker.value ?: return + runCatching { current.close() } + } + override suspend fun close() { orchestrator.cancel() runCatching { activeBroadcast?.close() } @@ -310,26 +380,42 @@ private class ReconnectingSpeakerHandle( } /** - * Stable [BroadcastHandle] backed by a re-issuing pump. Each time - * the wrapper opens a fresh session the pump cancels its prior - * iteration, calls [NestsSpeaker.startBroadcasting] on the new - * session, replays the cached mute intent on the resulting - * underlying handle, and parks until the next session swap. + * Stable [BroadcastHandle] backed by a re-issuing pump. Two paths, + * picked per-iteration based on whether the active speaker exposes + * the [HotSwappablePublisherSource] hook: * - * `setMuted` updates the cached intent unconditionally and forwards - * to whichever live underlying handle exists at the time. If no - * underlying handle is up (e.g. a brief gap during recycle), the - * intent is replayed on the next handle the pump opens, so the - * user-observed mute state is monotonic across recycles. + * - **Hot-swap path** (moq-lite): the pump constructs ONE + * long-lived [NestMoqLiteBroadcaster] on the first session + * and, on every subsequent session swap, opens a fresh + * [com.vitorpamplona.nestsclient.moq.lite.MoqLitePublisherHandle] + * on the new session and atomically swaps it into the + * broadcaster via + * [NestMoqLiteBroadcaster.swapPublisher]. The mic + + * encoder + capture loop keep running through the swap, so + * listeners hear ZERO gap at the JWT-refresh boundary + * (vs the legacy ~50–150 ms close-then-restart silence). + * + * - **Legacy path** (IETF reference / test fakes): each session + * swap closes the prior `BroadcastHandle` from + * [NestsSpeaker.startBroadcasting] and opens a fresh one. Same + * behaviour as before this hot-swap was introduced. + * + * `setMuted` updates the cached intent unconditionally; the pump + * applies the latest intent to whichever underlying broadcaster / + * handle is currently live, so user-observed mute is monotonic + * across recycles. */ private class ReissuingBroadcastHandle( private val activeSpeaker: StateFlow, private val scope: CoroutineScope, + private val captureFactory: () -> AudioCapture, + private val encoderFactory: () -> OpusEncoder, /** - * 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. + * Forwarded to the underlying broadcaster (hot-swap path) or + * `sp.startBroadcasting` (legacy path) 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, @@ -337,17 +423,22 @@ private class ReissuingBroadcastHandle( @Volatile private var desiredMuted: Boolean = false @Volatile private var closed: Boolean = false + + /** Hot-swap path's long-lived broadcaster. Null until the first session arrives. */ + @Volatile private var hotSwapBroadcaster: NestMoqLiteBroadcaster? = null + + /** Legacy path's per-session handle. Cleared when the session swaps. */ private val liveHandle = AtomicReference(null) private var pumpJob: Job? = null override val isMuted: Boolean get() = desiredMuted fun start() { - // Re-broadcast pump: every time activeSpeaker changes, drop - // the prior broadcast (collectLatest cancels the inner - // body via awaitCancellation) and open a new one against - // the fresh session. The pattern mirrors the listener's - // SubscribeHandle re-issuance pump. + // Per-iteration pump: every time activeSpeaker changes, decide + // whether the new speaker supports hot swap. If yes, retarget + // the long-lived broadcaster onto its publisher; if no, fall + // back to the close-then-restart path the previous version + // used uniformly. pumpJob = scope.launch { activeSpeaker.collectLatest { sp -> @@ -367,60 +458,151 @@ private class ReissuingBroadcastHandle( return@collectLatest } if (closed) return@collectLatest - val handle = - try { - sp.startBroadcasting(onLevel) - } catch (ce: kotlinx.coroutines.CancellationException) { - // Don't `return@collectLatest` on cancel — - // propagate so the launched pumpJob actually - // dies on close/scope cancellation. The old - // `runCatching` shape ate the cancel. - throw ce - } catch (_: Throwable) { - null - } ?: return@collectLatest - if (closed) { - runCatching { handle.close() } - return@collectLatest - } - // Apply current mute intent BEFORE storing the - // handle so a setMuted that races us applies - // exactly once: either (a) we set intent → - // apply intent → store, and the racing setMuted - // sees the live handle and applies again (no-op - // on the broadcaster); or (b) the racing - // setMuted updates intent → we read intent → - // apply. Order doesn't matter; idempotent. - if (desiredMuted) { - runCatching { handle.setMuted(true) } - } - liveHandle.set(handle) - try { - // Park until activeSpeaker emits a new value - // (collectLatest cancels us) or close() runs - // (pumpJob.cancel). - awaitCancellation() - } finally { - // Clear our slot only if we still own it — - // close() may have already swapped in null. - if (liveHandle.get() === handle) liveHandle.set(null) - // Best-effort close on the way out: the user - // may have called wrapper.close (closed=true, - // pump cancelling), or activeSpeaker swapped - // (the prior speaker is about to be closed - // by the orchestrator anyway, but defensively - // closing here releases the broadcaster + - // publisher promptly rather than waiting for - // the speaker.close()). - runCatching { handle.close() } + + val hotSwap = sp as? HotSwappablePublisherSource + if (hotSwap != null) { + runHotSwapIteration(hotSwap) + } else { + runLegacyIteration(sp) } } } } + /** + * Hot-swap iteration body. On the first session: open a publisher, + * construct the long-lived broadcaster, start it. On subsequent + * sessions: open a publisher on the new session, swap into the + * existing broadcaster, close the OLD publisher (FINs its + * announce + group streams on the about-to-die session). + * + * Parks via [awaitCancellation] so [collectLatest] can cancel us + * cleanly when the next session arrives. The broadcaster is NOT + * stopped here — it lives across iterations and is only stopped + * when the wrapper itself closes. + */ + private suspend fun runHotSwapIteration(hotSwap: HotSwappablePublisherSource) { + val newPublisher = + try { + hotSwap.openPublisherForHotSwap(MoqLiteNestsListener.AUDIO_TRACK) + } catch (ce: kotlinx.coroutines.CancellationException) { + throw ce + } catch (_: Throwable) { + // Couldn't mint a publisher on this session (rare — + // the session is already past Connected). Bail; the + // next session swap will retry. + return + } + if (closed) { + runCatching { newPublisher.close() } + return + } + + val existing = hotSwapBroadcaster + if (existing == null) { + // First-ever iteration: build the broadcaster and start it. + val broadcaster = + NestMoqLiteBroadcaster( + capture = captureFactory(), + encoder = encoderFactory(), + initialPublisher = newPublisher, + scope = scope, + framesPerGroup = NestMoqLiteBroadcaster.DEFAULT_FRAMES_PER_GROUP, + ) + try { + broadcaster.start( + onTerminalFailure = { + // Bubble up via the speaker's state machine so + // the orchestrator's normal Failed-handling + // path recycles the session. Same shape the + // legacy `startBroadcasting`-internal path uses. + runCatching { hotSwap.reportBroadcastTerminalFailure() } + }, + onLevel = onLevel, + ) + } catch (t: Throwable) { + runCatching { newPublisher.close() } + throw t + } + // Apply current mute intent (a setMuted that arrived before + // the broadcaster existed has already updated desiredMuted). + if (desiredMuted) broadcaster.setMuted(true) + hotSwapBroadcaster = broadcaster + } else { + // Subsequent iteration: hot swap. The capture / encoder + // pipeline is already running and feeding [existing.publisher] + // (the soon-to-be-old one). Install the new publisher, + // grab the old, close it. The capture loop's next snapshot + // picks up the new publisher and resets its group counter. + val old = existing.swapPublisher(newPublisher) + // Re-apply mute on the broadcaster — it survives swap, but + // a setMuted that arrived during the gap between the last + // session's terminal state and this swap may have flipped + // [desiredMuted] without finding a live broadcaster (no, it + // would have seen [hotSwapBroadcaster] since that's + // long-lived; but the no-op on equality keeps this safe). + existing.setMuted(desiredMuted) + // Close the old publisher AFTER the broadcaster has the + // new one. This is the order that matters: if we closed + // first, the capture loop's next send would race the swap + // and might see the closed publisher. + if (old != null) runCatching { old.close() } + } + + try { + // Park until [collectLatest] cancels us on the next session + // swap, OR [close] cancels [pumpJob]. The broadcaster keeps + // running through the cancellation; only close() stops it. + awaitCancellation() + } finally { + // Intentionally do NOT close the broadcaster here. + // collectLatest cancels this iteration on every session + // swap; closing the broadcaster would create the exact + // 50–150 ms gap this whole path exists to avoid. + } + } + + /** + * Legacy iteration body — used for IETF reference speakers and any + * future [NestsSpeaker] that doesn't implement + * [HotSwappablePublisherSource]. Identical to the pre-hot-swap + * behaviour: open a fresh handle on each session, close it on + * cancellation. + */ + private suspend fun runLegacyIteration(sp: NestsSpeaker) { + val handle = + try { + sp.startBroadcasting(onLevel) + } catch (ce: kotlinx.coroutines.CancellationException) { + throw ce + } catch (_: Throwable) { + null + } ?: return + if (closed) { + runCatching { handle.close() } + return + } + if (desiredMuted) { + runCatching { handle.setMuted(true) } + } + liveHandle.set(handle) + try { + awaitCancellation() + } finally { + if (liveHandle.get() === handle) liveHandle.set(null) + runCatching { handle.close() } + } + } + override suspend fun setMuted(muted: Boolean) { if (closed) return desiredMuted = muted + // Apply to whichever live underlying exists. Hot-swap and + // legacy paths are mutually exclusive in steady state — a + // moq-lite session's wrapper never has a live legacy handle, + // and vice versa — but both reads are cheap and harmless if + // the unused one is null. + hotSwapBroadcaster?.setMuted(muted) liveHandle.get()?.let { runCatching { it.setMuted(muted) } } } @@ -428,6 +610,13 @@ private class ReissuingBroadcastHandle( if (closed) return closed = true pumpJob?.cancel() + // Tear down whichever path was active. For hot-swap, stopping + // the broadcaster releases the mic + encoder AND closes the + // current publisher (broadcaster.stop calls publisher.close + // internally). For legacy, the per-session handle's close + // releases its own broadcaster. + hotSwapBroadcaster?.let { runCatching { it.stop() } } + hotSwapBroadcaster = null liveHandle.getAndSet(null)?.let { runCatching { it.close() } } onClose(this) } diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/Audio.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/Audio.kt index 14b523c6f..e015a49c1 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/Audio.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/Audio.kt @@ -99,11 +99,37 @@ interface AudioCapture { /** * Sink for PCM audio playback. Implementations buffer internally — [enqueue] * may suspend if the device's playback buffer is full. + * + * **Two-phase startup.** [start] allocates the underlying device + per-instance + * resources but does NOT begin consuming samples; [beginPlayback] flips the + * device into the playing state. Splitting the two lets [com.vitorpamplona.nestsclient.audio.NestPlayer] + * pre-roll a few decoded frames into the device's buffer before playback + * starts, so the AudioTrack has slack the moment the hardware begins pulling + * samples. Calling [enqueue] between [start] and [beginPlayback] is allowed + * and is the intended pattern. Implementations that don't need the + * distinction (test fakes, software-only sinks) can leave the default + * [beginPlayback] no-op alone. */ interface AudioPlayer { - /** Allocate underlying audio resources and begin playback. */ + /** + * Allocate underlying audio resources. The returned device is in a + * "ready, not playing" state — [enqueue] is allowed but the hardware + * doesn't consume samples until [beginPlayback] is called. Throws on + * device-unavailable so callers (typically + * [com.vitorpamplona.nestsclient.audio.NestPlayer.play]) get a + * synchronous failure they can roll back the subscription on. + */ fun start() + /** + * Transition the allocated device into the playing state. The + * hardware begins pulling samples from whatever's already been + * [enqueue]d. Default no-op so test fakes / software-only sinks + * don't have to grow a method they'll never use; production + * Android implementations override to call `AudioTrack.play()`. + */ + fun beginPlayback() {} + /** * Feed one PCM frame (any length, but typically [AudioFormat.FRAME_SIZE_SAMPLES] * samples) into the playback queue. diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt index e849d1bd3..385a68188 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt @@ -39,7 +39,7 @@ import kotlinx.coroutines.launch class NestMoqLiteBroadcaster( private val capture: AudioCapture, private val encoder: OpusEncoder, - private val publisher: MoqLitePublisherHandle, + initialPublisher: MoqLitePublisherHandle, private val scope: CoroutineScope, /** * Number of consecutive Opus frames to pack into a single moq-lite @@ -98,6 +98,26 @@ class NestMoqLiteBroadcaster( @Volatile private var muted: Boolean = false + /** + * Active publisher the capture loop pushes frames into. Mutable + + * `@Volatile` so [swapPublisher] can atomically retarget the loop + * onto a fresh moq-lite session's publisher when the + * [com.vitorpamplona.nestsclient.connectReconnectingNestsSpeaker] + * orchestrator recycles the WebTransport session for a JWT + * refresh — without restarting capture or the encoder. The + * capture loop snapshots this reference once per frame, so a swap + * takes effect on the very next frame and any in-flight send on + * the previous publisher is allowed to complete (or fail + * gracefully, caught by `runCatching`). + * + * Per-instance group sequence: each publisher restarts at sequence + * 0, since the publisher state lives inside [MoqLitePublisherHandle] + * (one per moq-lite session). That's fine — listeners use group + * sequence for ordering within a single subscription's uni-stream- + * per-group flow, not across publisher cycles. + */ + @Volatile private var publisher: MoqLitePublisherHandle = initialPublisher + /** * Start capturing + encoding + publishing in the background. * Returns immediately. Calling twice is an error. If @@ -161,6 +181,16 @@ class NestMoqLiteBroadcaster( // we bail. publisher.send returning `false` (no inbound // subscriber) is NOT counted — empty rooms are normal. var consecutiveSendErrors = 0 + // Track which publisher we last sent to. On swapPublisher + // (JWT-refresh hot swap), the snapshot below picks up the + // new reference; we reset framesInCurrentGroup so the + // first frame on the new publisher starts a fresh group + // (the new publisher's group sequence is 0 anyway). Without + // this, a mid-group swap would feed the new publisher + // frames as if they continued the old publisher's group, + // and the relay would see two unrelated uni streams under + // the same logical group. + var lastPublisher: MoqLitePublisherHandle = publisher try { while (true) { val pcm = capture.readFrame() ?: break @@ -181,16 +211,28 @@ class NestMoqLiteBroadcaster( } if (opus.isEmpty()) continue if (muted) continue + // Snapshot the publisher reference once per frame. + // If [swapPublisher] mid-loop installed a new + // reference, pick it up here and reset the group + // counter so the new publisher's first frame + // starts a clean group. Frames already in flight + // on the previous publisher are allowed to + // complete (or fail gracefully on `runCatching`). + val current = publisher + if (current !== lastPublisher) { + framesInCurrentGroup = 0 + lastPublisher = current + } // Pack [framesPerGroup] consecutive Opus frames // into the same moq-lite group / QUIC uni stream // before FINning. See the [framesPerGroup] kdoc // for the production cliff this works around. val sendOutcome = runCatching { - publisher.send(opus) + current.send(opus) framesInCurrentGroup += 1 if (framesInCurrentGroup >= framesPerGroup) { - publisher.endGroup() + current.endGroup() framesInCurrentGroup = 0 } } @@ -235,11 +277,14 @@ class NestMoqLiteBroadcaster( } } // EOF on the capture side. Flush whatever's in the - // open group so the relay sees its FIN and the - // listener doesn't sit on a half-delivered group - // buffer. Mirrors the per-group endGroup above. + // open group on the most-recently-used publisher so + // the relay sees its FIN and the listener doesn't + // sit on a half-delivered group buffer. Use + // [lastPublisher] (not the live [publisher] field) + // because a swap-then-EOF would otherwise FIN a + // group on a publisher that didn't open it. if (framesInCurrentGroup > 0) { - runCatching { publisher.endGroup() } + runCatching { lastPublisher.endGroup() } } } catch (ce: CancellationException) { throw ce @@ -264,6 +309,38 @@ class NestMoqLiteBroadcaster( this.muted = muted } + /** + * Hot-swap the underlying publisher. Used by + * [com.vitorpamplona.nestsclient.connectReconnectingNestsSpeaker]'s + * orchestrator to retarget the broadcaster onto a fresh moq-lite + * session's publisher when the JWT-refresh window fires — without + * tearing down the AudioRecord / Opus encoder. Returns the previous + * publisher reference so the caller can close it after the new one + * is wired up. + * + * **Closing the returned old publisher** is the caller's + * responsibility — this method only swaps the reference so the + * capture loop's next snapshot picks up the new publisher. The + * caller typically does: + * + * 1. Open a new moq-lite session, mint a new publisher on it. + * 2. Call [swapPublisher] with the new publisher; cache the old. + * 3. Close the old publisher (FINs the announce bidi + current + * group's uni stream on the old session). + * 4. Close the old moq-lite session (drops the WebTransport). + * + * No-op if the broadcaster is already [stop]ped (returns null) — the + * capture loop is already gone, so swapping in a new publisher would + * just leak it. + */ + fun swapPublisher(newPublisher: MoqLitePublisherHandle): MoqLitePublisherHandle? { + if (stopped) return null + val old = publisher + if (old === newPublisher) return null + publisher = newPublisher + return old + } + /** * Stop the loop, release the mic, release the encoder, close the * moq-lite publisher (which sends `Announce(Ended)` on every active diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestPlayer.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestPlayer.kt index deec824af..fe641b994 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestPlayer.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestPlayer.kt @@ -45,7 +45,27 @@ class NestPlayer( private val decoder: OpusDecoder, private val player: AudioPlayer, private val scope: CoroutineScope, + /** + * Number of decoded PCM frames to buffer before starting the underlying + * [AudioPlayer]. Without pre-roll, the device begins consuming audio the + * instant the first frame arrives and underruns at the first decode that + * misses its 20 ms cadence — the most common cause of perceptible + * dropouts on devices where Compose recomposition or GC briefly stalls + * the decode loop. + * + * Production callers typically pass `5` (≈ 100 ms) — long enough to mask + * a typical Main-thread hiccup, short enough that listeners don't notice + * the join latency. Tests pass `0` to preserve the synchronous test + * scheduler behaviour the existing assertions rely on. + * + * Default is `0` so existing tests stand without modification. + */ + private val prerollFrames: Int = 0, ) { + init { + require(prerollFrames >= 0) { "prerollFrames must be >= 0, got $prerollFrames" } + } + private var job: Job? = null private var stopped = false @@ -73,9 +93,56 @@ class NestPlayer( check(!stopped) { "NestPlayer already stopped" } check(job == null) { "NestPlayer.play already called" } + // Allocate the device synchronously so a [AudioException.DeviceUnavailable] + // (audio policy denial, AudioTrack rejected, etc.) propagates to + // the caller — typically `NestViewModel.openSubscription`, which + // catches and rolls back the freshly-reserved subscription slot. + // Routing this failure through `onError` instead would attach the + // slot first and leave a permanent "Connecting…" spinner on the + // speaker tile when the device fails to allocate. + // + // Two-phase startup: [AudioPlayer.start] allocates without + // beginning playback; [AudioPlayer.beginPlayback] flips the + // device into the playing state. We delay [beginPlayback] until + // the pre-roll buffer is full so the first frames already + // populate the device's internal buffer when the hardware + // starts pulling samples. player.start() + + // Pre-roll buffer holds decoded PCM until either [prerollFrames] + // frames have arrived or the upstream flow ends. Once the + // threshold is met (or the flow ends with anything queued), we + // call [AudioPlayer.beginPlayback] and flush the buffer in a + // tight loop so the device starts playback with a populated + // buffer. A flow that never produces PCM (decoder always empty, + // no audio in the room) never calls [beginPlayback] — the + // allocated device sits in the "ready, not playing" state until + // [stop] tears it down. job = scope.launch { + val preroll = ArrayDeque(prerollFrames.coerceAtLeast(1)) + var playbackBegun = false + + suspend fun beginAndFlushIfNeeded() { + if (playbackBegun) return + if (preroll.isEmpty()) return + // Flush BEFORE [beginPlayback] so the device starts + // playing against an already-populated buffer rather + // than emitting silence for the microseconds it + // takes the flush loop to fill. AudioTrack + // MODE_STREAM explicitly supports write() before + // play() per the Android docs — that's the + // textbook pre-roll pattern. Order matters + // because [beginPlayback] is the moment the + // hardware starts pulling samples; getting + // [enqueue] in first means the very first sample + // pulled is from our pre-rolled audio, not silence. + while (preroll.isNotEmpty()) { + player.enqueue(preroll.removeFirst()) + } + player.beginPlayback() + playbackBegun = true + } try { objects.collect { obj -> val pcm = @@ -95,9 +162,21 @@ class NestPlayer( } if (pcm.isNotEmpty()) { onLevel(peakAmplitude(pcm)) - player.enqueue(pcm) + if (playbackBegun) { + player.enqueue(pcm) + } else { + preroll.addLast(pcm) + if (preroll.size >= prerollFrames) { + beginAndFlushIfNeeded() + } + } } } + // Flow ended without enough frames to fill the pre-roll + // (e.g. the publisher cycled before pre-roll was full, + // or the room ended). Flush whatever's queued so any + // already-decoded audio still reaches the device. + beginAndFlushIfNeeded() } catch (ce: CancellationException) { throw ce } catch (t: Throwable) { 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 13e4c9696..b6169fd03 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 @@ -202,8 +202,6 @@ class MoqLiteSession internal constructor( endGroup = endGroup, ) val bidi = transport.openBidiStream() - bidi.write(Varint.encode(MoqLiteControlType.Subscribe.code)) - bidi.write(MoqLiteCodec.encodeSubscribe(request)) // Single long-running collector for the bidi's whole lifetime. // The collector parses the SubscribeResponse inline, signals it @@ -228,16 +226,31 @@ class MoqLiteSession internal constructor( val frames = Channel(capacity = DEFAULT_FRAME_BUFFER, onBufferOverflow = BufferOverflow.DROP_OLDEST) val responseDeferred = CompletableDeferred() - // Pre-register the subscription BEFORE launching the collector. - // Otherwise: if the publisher FINs the bidi immediately after - // sending Ok, the collector's exit-cleanup races subscribe()'s - // post-await registration — collector exits, runs `remove(id)` - // against an empty map, no-ops; subscribe() then inserts; the - // frames channel is now in the map with no live collector to - // close it on transport tear-down. Consumer hangs forever. - // Pre-registering means the collector's idempotent - // remove+close cleanup always finds and closes the frames - // channel, regardless of timing. + // Pre-register the subscription BEFORE launching the collector + // AND before the SUBSCRIBE goes on the wire. The order matters + // for two reasons: + // + // 1. Collector race (original reason): if the publisher FINs + // the bidi immediately after sending Ok, the collector's + // exit-cleanup races subscribe()'s post-await registration + // — collector exits, runs `remove(id)` against an empty + // map, no-ops; subscribe() then inserts; the frames + // channel is now in the map with no live collector to + // close it on transport tear-down. Consumer hangs + // forever. Pre-registering means the collector's + // idempotent remove+close cleanup always finds and + // closes the frames channel, regardless of timing. + // + // 2. First-group race (audit fix #3): the relay can open + // the first group's uni stream BEFORE our subscribe() + // continuation re-enters [state] to register `id`. If + // the SUBSCRIBE bytes hit the wire before the map entry + // lands, [drainOneGroup] looks the id up against an + // empty map and silently drops the frame. Registering + // the id before writing the SUBSCRIBE closes the + // window — by the time the relay can have parsed the + // message and opened a uni stream in response, the map + // already has our entry. val sub = ListenerSubscription( id = id, @@ -249,18 +262,44 @@ class MoqLiteSession internal constructor( subscriptionsBySubscribeId[id] = sub if (groupPump == null) groupPump = scope.launch { pumpUniStreams() } } + // Now that the subscription is registered, push the SUBSCRIBE + // bytes. If `bidi.write` throws (transport torn down, peer + // reset) we'd otherwise leave an orphaned map entry whose + // frames channel never closes — the response collector hasn't + // been launched yet so its idempotent cleanup wouldn't run. + // Roll back explicitly on throw. + try { + bidi.write(Varint.encode(MoqLiteControlType.Subscribe.code)) + bidi.write(MoqLiteCodec.encodeSubscribe(request)) + } catch (t: Throwable) { + state.withLock { subscriptionsBySubscribeId.remove(id) } + frames.close() + runCatching { bidi.finish() } + throw t + } scope.launch { val responseBuffer = MoqLiteFrameBuffer() var responseParsed = false + // typeCode hoisted outside the collect lambda — `readVarint` + // advances `pos` permanently while `readSizePrefixed` rolls + // its own length-varint back on incomplete-body but does NOT + // roll back the type-code that's already been consumed. If + // the response chunks split between type and body (possible + // under MTU pressure / fragmentation), a per-tick `val + // typeCode = readVarint()` would read the body's size + // prefix as the type code on the next chunk and misframe + // the response. Mirrors the same fix in [handleInboundBidi]. + var typeCode: Long? = null try { bidi.incoming().collect { chunk -> if (!responseParsed) { responseBuffer.push(chunk) - val typeCode = responseBuffer.readVarint() ?: return@collect + if (typeCode == null) typeCode = responseBuffer.readVarint() + val tc = typeCode ?: return@collect val body = responseBuffer.readSizePrefixed() ?: return@collect val out = MoqWriter() - out.writeVarint(typeCode) + out.writeVarint(tc) out.writeVarint(body.size.toLong()) out.writeBytes(body) val resp = MoqLiteCodec.decodeSubscribeResponse(out.toByteArray()) diff --git a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/audio/NestPlayerTest.kt b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/audio/NestPlayerTest.kt index b84a05471..9d63e22ca 100644 --- a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/audio/NestPlayerTest.kt +++ b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/audio/NestPlayerTest.kt @@ -216,6 +216,130 @@ class NestPlayerTest { sut.stop() } + /** + * Pre-roll: with `prerollFrames=3`, [AudioPlayer.beginPlayback] + * must NOT fire until the third decoded frame has arrived, and + * once it fires the AudioPlayer's queue must already contain + * the buffered pre-roll (i.e. flushed atomically with playback + * start, not lazily on the next enqueue). + */ + @Test + fun preroll_defers_beginPlayback_until_threshold_is_met() = + runTest { + val channel = Channel(capacity = 8) + val decoder = FakeOpusDecoder { byteToShorts(it) } + val player = FakeAudioPlayer() + + val sut = + NestPlayer( + decoder = decoder, + player = player, + scope = this, + prerollFrames = 3, + ) + sut.play(channel.receiveAsFlow()) + testScheduler.runCurrent() + + // Frames 1 and 2: pre-roll buffer fills, beginPlayback NOT + // called yet, AudioPlayer hasn't seen a single enqueue. + channel.send(moqObject(byteArrayOf(0x01))) + channel.send(moqObject(byteArrayOf(0x02))) + testScheduler.advanceUntilIdle() + assertEquals(0, player.beginPlaybackCount, "beginPlayback before threshold") + assertEquals(0, player.queued.size, "no enqueue before threshold") + + // Frame 3 trips the threshold: beginPlayback fires exactly + // once, and the pre-rolled frames must ALREADY be sitting + // in the AudioPlayer queue at that moment (otherwise the + // device starts playback against an empty buffer and + // pre-roll's whole point is defeated). [NestPlayer.play] + // implements this by flushing-then-beginPlayback — + // AudioTrack MODE_STREAM explicitly supports write() + // before play(). + channel.send(moqObject(byteArrayOf(0x03))) + testScheduler.advanceUntilIdle() + assertEquals(1, player.beginPlaybackCount) + // The fake records the queue size at the moment + // beginPlayback fires. With flush-then-begin ordering, + // all 3 pre-rolled frames are already in the queue. + assertEquals(3, player.queuedAtBeginPlayback, "pre-roll flushed before beginPlayback") + assertEquals(3, player.queued.size, "buffer populated when device starts") + + // Subsequent frames bypass the buffer and go directly + // through enqueue. + channel.send(moqObject(byteArrayOf(0x04))) + testScheduler.advanceUntilIdle() + assertEquals(1, player.beginPlaybackCount, "beginPlayback only fires once") + assertEquals(4, player.queued.size) + + sut.stop() + } + + /** + * Pre-roll: a flow that ends BEFORE the pre-roll threshold fires + * must still flush its partial buffer to the AudioPlayer. + * Otherwise a fast-cycling publisher could leave already-decoded + * frames stranded forever. + */ + @Test + fun preroll_flushes_partial_buffer_when_flow_ends_early() = + runTest { + val decoder = FakeOpusDecoder { byteToShorts(it) } + val player = FakeAudioPlayer() + + val sut = + NestPlayer( + decoder = decoder, + player = player, + scope = this, + prerollFrames = 5, + ) + // Only 2 frames — pre-roll never reaches its 5-frame floor, + // but the upstream Flow ends so the loop's flush hook must + // begin playback and drain whatever's queued. + sut.play( + flowOf( + moqObject(byteArrayOf(0x01)), + moqObject(byteArrayOf(0x02)), + ), + ) + testScheduler.advanceUntilIdle() + + assertEquals(1, player.beginPlaybackCount) + assertEquals(2, player.queued.size) + assertContentEquals(byteToShorts(byteArrayOf(0x01)), player.queued[0]) + assertContentEquals(byteToShorts(byteArrayOf(0x02)), player.queued[1]) + + sut.stop() + } + + /** + * Pre-roll edge case: an empty flow shouldn't start playback at + * all. The AudioTrack stays in its allocated-but-not-playing + * state until [stop] tears it down. + */ + @Test + fun preroll_does_not_begin_playback_when_flow_emits_no_pcm() = + runTest { + val decoder = FakeOpusDecoder { ShortArray(0) } + val player = FakeAudioPlayer() + + val sut = + NestPlayer( + decoder = decoder, + player = player, + scope = this, + prerollFrames = 3, + ) + sut.play(flowOf(moqObject(byteArrayOf(0x01)), moqObject(byteArrayOf(0x02)))) + testScheduler.advanceUntilIdle() + + assertEquals(0, player.beginPlaybackCount, "no PCM → no playback") + assertEquals(0, player.queued.size) + + sut.stop() + } + // -- helpers ----------------------------------------------------------- private fun moqObject(payload: ByteArray): MoqObject = @@ -246,6 +370,26 @@ class NestPlayerTest { private class FakeAudioPlayer : AudioPlayer { var started = false private set + + /** + * Counts the [AudioPlayer.beginPlayback] invocations. Used by + * the pre-roll regression tests to assert that playback only + * begins AFTER `prerollFrames` decoded frames have arrived (or + * after the upstream flow ends with a partial buffer). The + * default no-op `beginPlayback` in the interface lets fakes + * skip overriding when they don't care; we override here so + * the tests can verify the pre-roll wiring. + * + * Also tracks the size of `queued` at the moment beginPlayback + * fired — pre-roll's contract is that the buffer is flushed + * IN A TIGHT LOOP after beginPlayback, so we can read the + * snapshot to verify the flush ordering. + */ + var beginPlaybackCount = 0 + private set + var queuedAtBeginPlayback: Int = -1 + private set + var stopCount = 0 private set val stopped: Boolean get() = stopCount > 0 @@ -257,6 +401,11 @@ class NestPlayerTest { started = true } + override fun beginPlayback() { + beginPlaybackCount++ + queuedAtBeginPlayback = queued.size + } + override suspend fun enqueue(pcm: ShortArray) { queued.add(pcm) }