From 0afde79afd2e85e9e390946204a00667ac8d73e9 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 26 Apr 2026 03:22:02 +0000 Subject: [PATCH] =?UTF-8?q?feat(audio-rooms):=20M6=20+=20M7=20=E2=80=94=20?= =?UTF-8?q?AudioRoomBroadcaster=20+=20NestsSpeaker=20API?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit M6 (AudioRoomBroadcaster): Inverse of AudioRoomPlayer. Pulls PCM frames from an AudioCapture, runs them through an OpusEncoder, and pushes the resulting Opus packets into a MoqSession.TrackPublisher as OBJECT_DATAGRAMs. setMuted keeps the capture + encoder running so unmute is sample-accurate; encode failures are reported via onError but don't tear the loop down. M7 (NestsSpeaker): Mirror of NestsListener for the host / speaker path. - `NestsSpeaker.startBroadcasting()` — sends ANNOUNCE for the room's namespace (`["nests", roomId]`), opens a TrackPublisher named after this user's pubkey hex, wires an AudioRoomBroadcaster onto it. - `BroadcastHandle.setMuted` / `close()` — speaker-side equivalents of the listener's mute / disconnect controls. - `NestsSpeakerState` sealed hierarchy (Idle / Connecting{step} / Connected / Broadcasting{isMuted} / Failed / Closed). - `connectNestsSpeaker` orchestration mirrors `connectNestsListener` end-to-end (HTTP → WebTransport → MoQ setup), then returns a `DefaultNestsSpeaker` ready for `startBroadcasting`. Tests: - AudioRoomBroadcasterTest (5 tests): pcm-flow ordering, mute behavior, encoder warmup skip, encode-error tolerance, idempotent stop. - NestsSpeakerTest (2 tests): start/mute/close state transitions, double-start rejection. Bug caught + fixed during M7 testing: `DefaultNestsSpeaker.close()` held its `gate` mutex while calling `handle.close()` which then called back through `parent.broadcastClosed()` → self-deadlock. Fixed by making `broadcastClosed` lockless (StateFlow updates are atomic; the activeHandle compare-and-set is benign under the gate's exclusion of concurrent startBroadcasting calls). Verified: `:nestsClient:jvmTest` (78 tests, all green) + `./gradlew spotlessApply` clean. --- .../vitorpamplona/nestsclient/NestsConnect.kt | 110 ++++++++ .../vitorpamplona/nestsclient/NestsSpeaker.kt | 237 ++++++++++++++++++ .../nestsclient/audio/AudioRoomBroadcaster.kt | 141 +++++++++++ .../nestsclient/NestsSpeakerTest.kt | 170 +++++++++++++ .../audio/AudioRoomBroadcasterTest.kt | 210 ++++++++++++++++ 5 files changed, 868 insertions(+) create mode 100644 nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsSpeaker.kt create mode 100644 nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/AudioRoomBroadcaster.kt create mode 100644 nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/NestsSpeakerTest.kt create mode 100644 nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/audio/AudioRoomBroadcasterTest.kt diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsConnect.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsConnect.kt index 9a732e13b..b14e55fad 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsConnect.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsConnect.kt @@ -20,6 +20,8 @@ */ package com.vitorpamplona.nestsclient +import com.vitorpamplona.nestsclient.audio.AudioCapture +import com.vitorpamplona.nestsclient.audio.OpusEncoder import com.vitorpamplona.nestsclient.moq.MoqSession import com.vitorpamplona.nestsclient.moq.MoqVersion import com.vitorpamplona.nestsclient.moq.SubscribeHandle @@ -141,6 +143,114 @@ private fun failedListener(state: MutableStateFlow): NestsLi } } +/** + * Speaker / host counterpart of [connectNestsListener]. Walks the same + * HTTP → WebTransport → MoQ handshake; the difference is the post-setup + * step is `announce(...)` (driven by [NestsSpeaker.startBroadcasting]) + * instead of `subscribe(...)`. + * + * @param speakerPubkeyHex this user's pubkey hex, used as the MoQ track + * name when we ANNOUNCE — listeners look us up by exactly that name. + * @param captureFactory builds an [AudioCapture] (one per broadcast). + * Android passes `{ AudioRecordCapture() }`. + * @param encoderFactory builds an [OpusEncoder] (one per broadcast). + * Android passes `{ MediaCodecOpusEncoder() }`. + */ +suspend fun connectNestsSpeaker( + httpClient: NestsClient, + transport: WebTransportFactory, + scope: CoroutineScope, + serviceBase: String, + roomId: String, + signer: NostrSigner, + speakerPubkeyHex: String, + captureFactory: () -> AudioCapture, + encoderFactory: () -> OpusEncoder, + supportedMoqVersions: List = listOf(MoqVersion.DRAFT_17), +): NestsSpeaker { + val state = + MutableStateFlow( + NestsSpeakerState.Connecting(NestsSpeakerState.Connecting.ConnectStep.ResolvingRoom), + ) + + val roomInfo = + try { + httpClient.resolveRoom(serviceBase = serviceBase, roomId = roomId, signer = signer) + } catch (t: NestsException) { + state.value = NestsSpeakerState.Failed("Room resolution failed: ${t.message}", t) + return failedSpeaker(state) + } + + state.value = NestsSpeakerState.Connecting(NestsSpeakerState.Connecting.ConnectStep.OpeningTransport) + + val (authority, path) = + try { + parseEndpoint(roomInfo.endpoint) + } catch (t: Throwable) { + state.value = + NestsSpeakerState.Failed( + "Malformed MoQ endpoint URL '${roomInfo.endpoint}': ${t.message}", + t, + ) + return failedSpeaker(state) + } + + val webTransport = + try { + transport.connect(authority = authority, path = path, bearerToken = roomInfo.token) + } catch (t: WebTransportException) { + state.value = + NestsSpeakerState.Failed( + "WebTransport ${t.kind.name}: ${t.message}", + t, + ) + return failedSpeaker(state) + } + + state.value = NestsSpeakerState.Connecting(NestsSpeakerState.Connecting.ConnectStep.MoqHandshake) + + val moq = + try { + MoqSession.client(webTransport, scope).also { it.setup(supportedMoqVersions) } + } catch (t: Throwable) { + runCatching { webTransport.close(0, "moq setup failed") } + state.value = NestsSpeakerState.Failed("MoQ handshake failed: ${t.message}", t) + return failedSpeaker(state) + } + + val negotiatedVersion = + moq.selectedVersion ?: run { + runCatching { moq.close() } + state.value = NestsSpeakerState.Failed("MoQ session reported no negotiated version") + return failedSpeaker(state) + } + + state.value = NestsSpeakerState.Connected(roomInfo, negotiatedVersion) + return DefaultNestsSpeaker( + session = moq, + roomNamespace = TrackNamespace.of("nests", roomId), + speakerTrackName = speakerPubkeyHex.encodeToByteArray(), + captureFactory = captureFactory, + encoderFactory = encoderFactory, + scope = scope, + mutableState = state, + ) +} + +/** Mirror of [failedListener] for the speaker path. */ +private fun failedSpeaker(state: MutableStateFlow): NestsSpeaker = + object : NestsSpeaker { + override val state = state + + override suspend fun startBroadcasting(): BroadcastHandle = error("speaker never connected: ${state.value}") + + override suspend fun close() { + if (state.value !is NestsSpeakerState.Closed) { + state.value = NestsSpeakerState.Closed + } + } + } + /** * Split a typical nests endpoint URL such as `https://relay.example.com/moq` * or `https://relay.example.com:4443/api/v1/moq?room=abc` into the diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsSpeaker.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsSpeaker.kt new file mode 100644 index 000000000..a0e920e10 --- /dev/null +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsSpeaker.kt @@ -0,0 +1,237 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.nestsclient + +import com.vitorpamplona.nestsclient.audio.AudioCapture +import com.vitorpamplona.nestsclient.audio.AudioRoomBroadcaster +import com.vitorpamplona.nestsclient.audio.OpusEncoder +import com.vitorpamplona.nestsclient.moq.MoqSession +import com.vitorpamplona.nestsclient.moq.TrackNamespace +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.StateFlow +import kotlinx.coroutines.flow.asStateFlow +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock + +/** + * High-level handle for the speaker / host audio path. Mirror of + * [NestsListener]: hides HTTP + WebTransport + MoQ wiring under one + * observable state machine, with the added ability to broadcast our own + * speaker track. + * + * Open one [NestsSpeaker] per audio-room screen where we have host or + * speaker permission. Call [startBroadcasting] when the user taps "Talk"; + * call [BroadcastHandle.close] when they stop talking or leave the room. + */ +interface NestsSpeaker { + /** Connection / broadcast state — drives the speaker UI's status chip. */ + val state: StateFlow + + /** + * Begin announcing our speaker track and pumping mic frames out as + * OBJECT_DATAGRAMs. + * + * @throws IllegalStateException if a broadcast is already running on + * this speaker or the session is not [NestsSpeakerState.Connected]. + * @throws com.vitorpamplona.nestsclient.moq.MoqProtocolException if the + * peer rejects the ANNOUNCE. + */ + suspend fun startBroadcasting(): BroadcastHandle + + /** Tear down the MoQ session + transport. Idempotent. */ + suspend fun close() +} + +/** Active broadcast on one [NestsSpeaker]. Returned from [NestsSpeaker.startBroadcasting]. */ +interface BroadcastHandle { + /** + * Toggle whether mic frames reach the wire. The capture pipeline keeps + * running, so unmute is sample-accurate. Default unmuted. + */ + suspend fun setMuted(muted: Boolean) + + /** Whether the next OBJECT_DATAGRAM the peer would receive is silenced. */ + val isMuted: Boolean + + /** + * Stop publishing. Releases the mic + encoder, sends UNANNOUNCE + + * SUBSCRIBE_DONE on the wire. Idempotent. + */ + suspend fun close() +} + +/** Lifecycle states of a [NestsSpeaker]. */ +sealed class NestsSpeakerState { + data object Idle : NestsSpeakerState() + + data class Connecting( + val step: ConnectStep, + ) : NestsSpeakerState() { + enum class ConnectStep { + ResolvingRoom, + OpeningTransport, + MoqHandshake, + } + } + + /** Connection live; ready for [NestsSpeaker.startBroadcasting]. */ + data class Connected( + val roomInfo: NestsRoomInfo, + val negotiatedMoqVersion: Long, + ) : NestsSpeakerState() + + /** Currently announcing + emitting OBJECT_DATAGRAMs for our track. */ + data class Broadcasting( + val roomInfo: NestsRoomInfo, + val negotiatedMoqVersion: Long, + val isMuted: Boolean, + ) : NestsSpeakerState() + + data class Failed( + val reason: String, + val cause: Throwable? = null, + ) : NestsSpeakerState() + + data object Closed : NestsSpeakerState() +} + +/** + * Default [NestsSpeaker] that wraps a connected [MoqSession] and plumbs an + * [AudioRoomBroadcaster] through it on [startBroadcasting]. + * + * Construction does NOT open the transport — call [connectNestsSpeaker] to + * walk the full HTTP + transport + MoQ handshake. + */ +class DefaultNestsSpeaker internal constructor( + private val session: MoqSession, + private val roomNamespace: TrackNamespace, + private val speakerTrackName: ByteArray, + private val captureFactory: () -> AudioCapture, + private val encoderFactory: () -> OpusEncoder, + private val scope: CoroutineScope, + private val mutableState: MutableStateFlow, +) : NestsSpeaker { + override val state: StateFlow = mutableState.asStateFlow() + + private val gate = Mutex() + private var activeHandle: DefaultBroadcastHandle? = null + + override suspend fun startBroadcasting(): BroadcastHandle { + gate.withLock { + val current = state.value + check(current is NestsSpeakerState.Connected) { + "startBroadcasting requires Connected state, was $current" + } + check(activeHandle == null) { "speaker is already broadcasting" } + + val announce = session.announce(roomNamespace) + val publisher = + try { + announce.openTrack(speakerTrackName) + } catch (t: Throwable) { + runCatching { announce.unannounce() } + throw t + } + val broadcaster = + AudioRoomBroadcaster( + capture = captureFactory(), + encoder = encoderFactory(), + publisher = publisher, + scope = scope, + ) + broadcaster.start() + mutableState.value = + NestsSpeakerState.Broadcasting( + roomInfo = current.roomInfo, + negotiatedMoqVersion = current.negotiatedMoqVersion, + isMuted = false, + ) + val handle = + DefaultBroadcastHandle( + broadcaster = broadcaster, + announce = announce, + parent = this, + ) + activeHandle = handle + return handle + } + } + + /** + * Lockless: callable from inside [close] (which already holds [gate]) and + * from [DefaultBroadcastHandle.close] (which doesn't). We compare-and-set + * on activeHandle and update state atomically via the StateFlow. + */ + internal fun broadcastClosed(handle: DefaultBroadcastHandle) { + if (activeHandle !== handle) return + activeHandle = null + val current = mutableState.value + if (current is NestsSpeakerState.Broadcasting) { + mutableState.value = + NestsSpeakerState.Connected(current.roomInfo, current.negotiatedMoqVersion) + } + } + + internal fun reportMuteState(muted: Boolean) { + val current = mutableState.value + if (current is NestsSpeakerState.Broadcasting) { + mutableState.value = current.copy(isMuted = muted) + } + } + + override suspend fun close() { + gate.withLock { + if (state.value is NestsSpeakerState.Closed) return + activeHandle?.let { runCatching { it.close() } } + activeHandle = null + runCatching { session.close() } + mutableState.value = NestsSpeakerState.Closed + } + } +} + +internal class DefaultBroadcastHandle( + private val broadcaster: AudioRoomBroadcaster, + private val announce: MoqSession.AnnounceHandle, + private val parent: DefaultNestsSpeaker, +) : BroadcastHandle { + @Volatile private var muted: Boolean = false + + @Volatile private var closed: Boolean = false + + override val isMuted: Boolean get() = muted + + override suspend fun setMuted(muted: Boolean) { + if (closed) return + this.muted = muted + broadcaster.setMuted(muted) + parent.reportMuteState(muted) + } + + override suspend fun close() { + if (closed) return + closed = true + runCatching { broadcaster.stop() } + runCatching { announce.unannounce() } + parent.broadcastClosed(this) + } +} diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/AudioRoomBroadcaster.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/AudioRoomBroadcaster.kt new file mode 100644 index 000000000..af98c46b0 --- /dev/null +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/AudioRoomBroadcaster.kt @@ -0,0 +1,141 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.nestsclient.audio + +import com.vitorpamplona.nestsclient.moq.MoqSession +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Job +import kotlinx.coroutines.launch + +/** + * Inverse of [AudioRoomPlayer]: pulls PCM from an [AudioCapture], runs it + * through an [OpusEncoder], and pushes the resulting Opus packets into a + * [MoqSession.TrackPublisher] as MoQ OBJECT_DATAGRAMs. + * + * One instance per outgoing track (typically one per local speaker, since + * Opus encoder state is per-stream). + * + * Lifecycle: + * - [start] opens the mic, starts the capture-encode-publish loop. + * - [setMuted] keeps the loop running but stops emitting frames so unmute + * is instant. Default unmuted. + * - [stop] cancels the loop, stops the mic, releases the encoder, and + * closes the publisher. + * + * Encoder warm-up frames (encoder returns empty) are dropped silently. + * Encode failures are reported via [onError] but do NOT tear the loop down — + * one bad frame shouldn't end the broadcast. + */ +class AudioRoomBroadcaster( + private val capture: AudioCapture, + private val encoder: OpusEncoder, + private val publisher: MoqSession.TrackPublisher, + private val scope: CoroutineScope, +) { + private var job: Job? = null + private var stopped = false + + @Volatile private var muted: Boolean = false + + /** + * Start capturing + encoding + publishing in the background. Returns + * immediately. Calling twice is an error. + */ + fun start(onError: (AudioException) -> Unit = { /* swallow */ }) { + check(!stopped) { "AudioRoomBroadcaster already stopped" } + check(job == null) { "AudioRoomBroadcaster.start already called" } + + capture.start() + job = + scope.launch { + try { + while (true) { + val pcm = capture.readFrame() ?: break + val opus = + try { + encoder.encode(pcm) + } catch (ce: CancellationException) { + throw ce + } catch (t: Throwable) { + onError( + AudioException( + AudioException.Kind.DecoderError, + "Opus encode failed for a frame", + t, + ), + ) + continue + } + if (opus.isEmpty()) continue + if (muted) continue + runCatching { publisher.send(opus) } + .onFailure { t -> + if (t is CancellationException) throw t + // Network drop on send is recoverable — log via onError but + // don't stop the loop; the next frame may go through. + onError( + AudioException( + AudioException.Kind.PlaybackFailed, + "publisher.send failed", + t, + ), + ) + } + } + } catch (ce: CancellationException) { + throw ce + } catch (t: Throwable) { + onError( + AudioException( + AudioException.Kind.PlaybackFailed, + "audio capture pipeline failed", + t, + ), + ) + } + } + } + + /** + * Toggle whether captured frames reach the wire. The mic stays open and + * the encoder keeps state consistent so unmute resumes mid-conversation + * with no glitch. No-op once [stop] has been called. + */ + fun setMuted(muted: Boolean) { + if (stopped) return + this.muted = muted + } + + /** + * Stop the loop, release the mic, release the encoder, close the MoQ + * publisher (which fires SUBSCRIBE_DONE to every attached subscriber). + * Idempotent. + */ + suspend fun stop() { + if (stopped) return + stopped = true + job?.cancel() + runCatching { capture.stop() } + runCatching { encoder.release() } + runCatching { publisher.close() } + } +} diff --git a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/NestsSpeakerTest.kt b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/NestsSpeakerTest.kt new file mode 100644 index 000000000..56870de6d --- /dev/null +++ b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/NestsSpeakerTest.kt @@ -0,0 +1,170 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.nestsclient + +import com.vitorpamplona.nestsclient.audio.AudioCapture +import com.vitorpamplona.nestsclient.audio.OpusEncoder +import com.vitorpamplona.nestsclient.moq.Announce +import com.vitorpamplona.nestsclient.moq.AnnounceOk +import com.vitorpamplona.nestsclient.moq.ClientSetup +import com.vitorpamplona.nestsclient.moq.MoqCodec +import com.vitorpamplona.nestsclient.moq.MoqSession +import com.vitorpamplona.nestsclient.moq.MoqVersion +import com.vitorpamplona.nestsclient.moq.ServerSetup +import com.vitorpamplona.nestsclient.moq.TrackNamespace +import com.vitorpamplona.nestsclient.transport.FakeWebTransport +import kotlinx.coroutines.async +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.first +import kotlinx.coroutines.test.runTest +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertIs +import kotlin.test.assertTrue + +class NestsSpeakerTest { + @Test + fun startBroadcasting_announces_then_state_is_Broadcasting() = + runTest { + val (speakerSide, peerSide) = FakeWebTransport.pair() + val speakerSession = MoqSession.client(speakerSide, backgroundScope) + val ns = TrackNamespace.of("nests", "test-room") + + val peer = + async { + val ctrl = peerSide.peerOpenedBidiStreams().first() + val cs = MoqCodec.decode(ctrl.incoming().first())!!.message as ClientSetup + ctrl.write(MoqCodec.encode(ServerSetup(cs.supportedVersions.first()))) + val ann = MoqCodec.decode(ctrl.incoming().first())!!.message as Announce + ctrl.write(MoqCodec.encode(AnnounceOk(ann.namespace))) + } + + speakerSession.setup(listOf(MoqVersion.DRAFT_17)) + + val state = + MutableStateFlow( + NestsSpeakerState.Connected( + roomInfo = NestsRoomInfo(endpoint = "https://relay.example/moq"), + negotiatedMoqVersion = speakerSession.selectedVersion!!, + ), + ) + val speaker = + DefaultNestsSpeaker( + session = speakerSession, + roomNamespace = ns, + speakerTrackName = "test-pubkey".encodeToByteArray(), + captureFactory = { ConstantCapture() }, + encoderFactory = { ConstantEncoder() }, + scope = backgroundScope, + mutableState = state, + ) + + val handle = speaker.startBroadcasting() + peer.await() + + val now = speaker.state.value + assertIs(now) + assertEquals(false, now.isMuted) + + // setMuted reflects through into the public state. + handle.setMuted(true) + assertEquals(true, (speaker.state.value as NestsSpeakerState.Broadcasting).isMuted) + assertEquals(true, handle.isMuted) + + handle.close() + // Should fall back to Connected after broadcast ends. + assertIs(speaker.state.value) + + speaker.close() + assertEquals(NestsSpeakerState.Closed, speaker.state.value) + } + + @Test + fun startBroadcasting_twice_throws_IllegalStateException() = + runTest { + val (speakerSide, peerSide) = FakeWebTransport.pair() + val speakerSession = MoqSession.client(speakerSide, backgroundScope) + val ns = TrackNamespace.of("nests", "test-room") + + val peer = + async { + val ctrl = peerSide.peerOpenedBidiStreams().first() + val cs = MoqCodec.decode(ctrl.incoming().first())!!.message as ClientSetup + ctrl.write(MoqCodec.encode(ServerSetup(cs.supportedVersions.first()))) + val ann = MoqCodec.decode(ctrl.incoming().first())!!.message as Announce + ctrl.write(MoqCodec.encode(AnnounceOk(ann.namespace))) + } + + speakerSession.setup(listOf(MoqVersion.DRAFT_17)) + val state = + MutableStateFlow( + NestsSpeakerState.Connected( + roomInfo = NestsRoomInfo(endpoint = "https://relay.example/moq"), + negotiatedMoqVersion = speakerSession.selectedVersion!!, + ), + ) + val speaker = + DefaultNestsSpeaker( + session = speakerSession, + roomNamespace = ns, + speakerTrackName = "test-pubkey".encodeToByteArray(), + captureFactory = { ConstantCapture() }, + encoderFactory = { ConstantEncoder() }, + scope = backgroundScope, + mutableState = state, + ) + + speaker.startBroadcasting() + peer.await() + + val ex = + runCatching { speaker.startBroadcasting() }.exceptionOrNull() + assertTrue(ex is IllegalStateException, "second startBroadcasting should throw IllegalStateException, got $ex") + + speaker.close() + } + + // ----- Fakes ----- + + private class ConstantCapture : AudioCapture { + // Yields a single dummy frame then drains; per-instance so tests + // don't share state through a singleton. + private val ch = Channel(capacity = 1) + + init { + ch.trySend(ShortArray(960) { 0 }) + ch.close() + } + + override fun start() {} + + override suspend fun readFrame(): ShortArray? = ch.receiveCatching().getOrNull() + + override fun stop() {} + } + + private class ConstantEncoder : OpusEncoder { + override fun encode(pcm: ShortArray): ByteArray = byteArrayOf(0xAA.toByte(), 0xBB.toByte()) + + override fun release() {} + } +} diff --git a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/audio/AudioRoomBroadcasterTest.kt b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/audio/AudioRoomBroadcasterTest.kt new file mode 100644 index 000000000..7fe42e86d --- /dev/null +++ b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/audio/AudioRoomBroadcasterTest.kt @@ -0,0 +1,210 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.nestsclient.audio + +import com.vitorpamplona.nestsclient.moq.MoqSession +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.test.runTest +import kotlin.test.Test +import kotlin.test.assertContentEquals +import kotlin.test.assertEquals +import kotlin.test.assertTrue + +class AudioRoomBroadcasterTest { + @Test + fun pcm_frames_flow_through_encoder_into_publisher_in_order() = + runTest { + val capture = ScriptedCapture(listOf(shortArrayOf(1), shortArrayOf(2), shortArrayOf(3))) + val encoder = ScriptedEncoder(prefix = byteArrayOf(0xAA.toByte())) + val publisher = RecordingPublisher("track") + val broadcaster = AudioRoomBroadcaster(capture, encoder, publisher, backgroundScope) + + broadcaster.start() + // Capture exhausts after 3 frames (readFrame returns null) so the + // loop terminates on its own; broadcaster.stop() cleans up the + // resources but the wire results are observable now. + capture.awaitDrained() + + assertEquals( + listOf(byteArrayOf(0xAA.toByte(), 1).toList(), byteArrayOf(0xAA.toByte(), 2).toList(), byteArrayOf(0xAA.toByte(), 3).toList()), + publisher.sent.map { it.toList() }, + ) + + broadcaster.stop() + assertTrue(capture.startedFlag, "capture should have started") + assertTrue(capture.stopCount > 0, "capture should have stopped") + assertTrue(encoder.released, "encoder should have been released") + assertTrue(publisher.closed, "publisher should have been closed") + } + + @Test + fun setMuted_true_keeps_capture_running_but_no_frames_published() = + runTest { + val capture = ScriptedCapture(listOf(shortArrayOf(1), shortArrayOf(2), shortArrayOf(3))) + val encoder = ScriptedEncoder(prefix = byteArrayOf(0xBB.toByte())) + val publisher = RecordingPublisher("track") + val broadcaster = AudioRoomBroadcaster(capture, encoder, publisher, backgroundScope) + + broadcaster.setMuted(true) + broadcaster.start() + capture.awaitDrained() + + // Mic was open + encoder ran; publisher saw nothing. + assertTrue(capture.startedFlag) + assertEquals(3, encoder.encodeCalls, "encoder still ran while muted (state preserved)") + assertTrue(publisher.sent.isEmpty(), "no frames should reach the wire while muted") + + broadcaster.stop() + } + + @Test + fun encoder_returning_empty_array_is_silently_skipped() = + runTest { + val capture = ScriptedCapture(listOf(shortArrayOf(1), shortArrayOf(2))) + val encoder = ScriptedEncoder(prefix = byteArrayOf(0xCC.toByte()), warmupSkips = 1) + val publisher = RecordingPublisher("track") + val broadcaster = AudioRoomBroadcaster(capture, encoder, publisher, backgroundScope) + + broadcaster.start() + capture.awaitDrained() + + // First frame was the warmup → encoder returned empty → skipped. + // Second frame should reach the publisher. + assertEquals(1, publisher.sent.size) + assertContentEquals(byteArrayOf(0xCC.toByte(), 2), publisher.sent.single()) + + broadcaster.stop() + } + + @Test + fun encoder_failure_does_not_stop_the_loop() = + runTest { + val capture = ScriptedCapture(listOf(shortArrayOf(1), shortArrayOf(2), shortArrayOf(3))) + val encoder = ScriptedEncoder(prefix = byteArrayOf(0xDD.toByte()), failOnNthCall = 1) + val publisher = RecordingPublisher("track") + val broadcaster = AudioRoomBroadcaster(capture, encoder, publisher, backgroundScope) + val errors = mutableListOf() + + broadcaster.start { errors.add(it) } + capture.awaitDrained() + + assertEquals(1, errors.size, "exactly one encode error should have been reported") + assertEquals(2, publisher.sent.size, "two non-failed frames should have reached the publisher") + broadcaster.stop() + } + + @Test + fun stop_is_idempotent() = + runTest { + val capture = ScriptedCapture(listOf(shortArrayOf(1))) + val encoder = ScriptedEncoder(prefix = byteArrayOf(0xEE.toByte())) + val publisher = RecordingPublisher("track") + val broadcaster = AudioRoomBroadcaster(capture, encoder, publisher, backgroundScope) + + broadcaster.start() + capture.awaitDrained() + broadcaster.stop() + broadcaster.stop() + broadcaster.stop() + + assertEquals(1, capture.stopCount) + assertTrue(encoder.released) + assertTrue(publisher.closed) + } + + // ---------- Fakes ---------- + + private class ScriptedCapture( + frames: List, + ) : AudioCapture { + private val queue = Channel(capacity = frames.size + 1) + private val drainSignal = kotlinx.coroutines.CompletableDeferred() + var startedFlag = false + private set + var stopCount = 0 + private set + + init { + frames.forEach { queue.trySend(it) } + queue.close() + } + + override fun start() { + startedFlag = true + } + + override suspend fun readFrame(): ShortArray? { + val next = queue.receiveCatching().getOrNull() + if (next == null && !drainSignal.isCompleted) drainSignal.complete(Unit) + return next + } + + override fun stop() { + stopCount++ + } + + suspend fun awaitDrained() { + drainSignal.await() + } + } + + private class ScriptedEncoder( + private val prefix: ByteArray, + private val warmupSkips: Int = 0, + private val failOnNthCall: Int = -1, // -1 = never fail + ) : OpusEncoder { + var encodeCalls: Int = 0 + private set + var released: Boolean = false + private set + + override fun encode(pcm: ShortArray): ByteArray { + val callIndex = encodeCalls + encodeCalls += 1 + if (callIndex == failOnNthCall) error("scripted encoder failure at call $callIndex") + if (callIndex < warmupSkips) return ByteArray(0) + // Append the first PCM sample byte so tests can verify ordering. + return prefix + byteArrayOf(pcm.first().toByte()) + } + + override fun release() { + released = true + } + } + + private class RecordingPublisher( + nameStr: String, + ) : MoqSession.TrackPublisher { + override val name: ByteArray = nameStr.encodeToByteArray() + val sent: MutableList = mutableListOf() + var closed: Boolean = false + private set + + override suspend fun send(payload: ByteArray): Boolean { + sent.add(payload) + return true + } + + override suspend fun close() { + closed = true + } + } +}