diff --git a/nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/AudioTrackPlayer.kt b/nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/AudioTrackPlayer.kt new file mode 100644 index 0000000000..31b26bfa9b --- /dev/null +++ b/nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/AudioTrackPlayer.kt @@ -0,0 +1,136 @@ +/* + * 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 android.media.AudioAttributes +import android.media.AudioManager +import android.media.AudioTrack +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.withContext +import android.media.AudioFormat as AndroidAudioFormat + +/** + * [AudioPlayer] backed by Android's [AudioTrack] in `MODE_STREAM`. Targets the + * voice-call usage stream so the OS treats audio-room playback like a phone + * call (volume rocker controls call volume, ducks notifications, etc.). + * + * 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. + */ +class AudioTrackPlayer( + private val usage: Int = AudioAttributes.USAGE_VOICE_COMMUNICATION, + private val contentType: Int = AudioAttributes.CONTENT_TYPE_SPEECH, +) : AudioPlayer { + private var track: AudioTrack? = null + + override fun start() { + if (track != null) return + + val channelMask = + when (AudioFormat.CHANNELS) { + 1 -> AndroidAudioFormat.CHANNEL_OUT_MONO + 2 -> AndroidAudioFormat.CHANNEL_OUT_STEREO + else -> error("unsupported channel count ${AudioFormat.CHANNELS}") + } + + val minBuffer = + AudioTrack.getMinBufferSize( + AudioFormat.SAMPLE_RATE_HZ, + channelMask, + AndroidAudioFormat.ENCODING_PCM_16BIT, + ) + if (minBuffer <= 0) { + throw AudioException( + AudioException.Kind.DeviceUnavailable, + "AudioTrack.getMinBufferSize returned $minBuffer for ${AudioFormat.SAMPLE_RATE_HZ} Hz", + ) + } + val bufferBytes = minBuffer * 4 + + val newTrack = + try { + AudioTrack + .Builder() + .setAudioAttributes( + AudioAttributes + .Builder() + .setUsage(usage) + .setContentType(contentType) + .build(), + ).setAudioFormat( + AndroidAudioFormat + .Builder() + .setEncoding(AndroidAudioFormat.ENCODING_PCM_16BIT) + .setSampleRate(AudioFormat.SAMPLE_RATE_HZ) + .setChannelMask(channelMask) + .build(), + ).setBufferSizeInBytes(bufferBytes) + .setTransferMode(AudioTrack.MODE_STREAM) + .build() + } catch (t: Throwable) { + throw AudioException( + AudioException.Kind.DeviceUnavailable, + "Failed to construct AudioTrack", + t, + ) + } + + try { + newTrack.play() + } catch (t: Throwable) { + runCatching { newTrack.release() } + throw AudioException( + AudioException.Kind.DeviceUnavailable, + "AudioTrack.play() rejected start", + t, + ) + } + 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 written = t.write(pcm, 0, pcm.size, AudioTrack.WRITE_BLOCKING) + if (written < 0) { + throw AudioException( + AudioException.Kind.PlaybackFailed, + "AudioTrack.write returned error code $written", + ) + } + } + } + + override fun stop() { + val t = track ?: return + track = null + runCatching { t.pause() } + runCatching { t.flush() } + runCatching { t.stop() } + runCatching { t.release() } + } + + @Suppress("unused") + val voiceCallUsage: Int get() = AudioManager.STREAM_VOICE_CALL // kept for documentation +} diff --git a/nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/MediaCodecOpusDecoder.kt b/nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/MediaCodecOpusDecoder.kt new file mode 100644 index 0000000000..23c9e3bca6 --- /dev/null +++ b/nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/MediaCodecOpusDecoder.kt @@ -0,0 +1,164 @@ +/* + * 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 android.media.MediaCodec +import android.media.MediaFormat +import java.nio.ByteBuffer +import java.nio.ByteOrder + +/** + * [OpusDecoder] backed by Android's [MediaCodec] (`audio/opus`, available on + * API 21+). One instance per track — Opus carries forward predictor state + * across packets, so sharing a decoder across speakers would cause clicks. + * + * Configuration: + * - 48 kHz mono signed 16-bit PCM output (matches [AudioFormat]). + * - CSD-0: Opus identification header per RFC 7845 §5.1, 19 bytes. + * - CSD-1 / CSD-2: pre-skip + seek pre-roll, both zero (we don't seek). + */ +class MediaCodecOpusDecoder : OpusDecoder { + private val codec: MediaCodec = + try { + MediaCodec.createDecoderByType(MediaFormat.MIMETYPE_AUDIO_OPUS).apply { + configure(buildFormat(), null, null, 0) + start() + } + } catch (t: Throwable) { + throw AudioException( + AudioException.Kind.DeviceUnavailable, + "Failed to allocate MediaCodec audio/opus decoder", + t, + ) + } + + private val bufferInfo = MediaCodec.BufferInfo() + private var presentationTimeUs: Long = 0L + private var released = false + + override fun decode(opusPacket: ByteArray): ShortArray { + check(!released) { "decoder released" } + + val inputIndex = codec.dequeueInputBuffer(DEQUEUE_TIMEOUT_US) + if (inputIndex < 0) return ShortArray(0) + val inputBuffer = + codec.getInputBuffer(inputIndex) + ?: throw AudioException( + AudioException.Kind.DecoderError, + "MediaCodec returned null input buffer at index $inputIndex", + ) + inputBuffer.clear() + inputBuffer.put(opusPacket) + codec.queueInputBuffer(inputIndex, 0, opusPacket.size, presentationTimeUs, 0) + // 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. + val collected = ArrayList(AudioFormat.FRAME_SIZE_SAMPLES) + while (true) { + val outputIndex = codec.dequeueOutputBuffer(bufferInfo, DEQUEUE_TIMEOUT_US) + when { + outputIndex >= 0 -> { + val outputBuffer = + codec.getOutputBuffer(outputIndex) + ?: continue + if (bufferInfo.size > 0) { + outputBuffer.position(bufferInfo.offset) + outputBuffer.limit(bufferInfo.offset + bufferInfo.size) + val shorts = outputBuffer.order(ByteOrder.nativeOrder()).asShortBuffer() + val tmp = ShortArray(shorts.remaining()) + shorts.get(tmp) + for (s in tmp) collected.add(s) + } + codec.releaseOutputBuffer(outputIndex, false) + if (bufferInfo.flags and MediaCodec.BUFFER_FLAG_END_OF_STREAM != 0) break + // No more buffered output for this packet. + if (bufferInfo.size == 0) break + if (collected.size >= AudioFormat.FRAME_SIZE_SAMPLES) break + } + + outputIndex == MediaCodec.INFO_OUTPUT_FORMAT_CHANGED -> { + // The format change carries no audio data; loop to read + // the actual PCM frame. + continue + } + + outputIndex == MediaCodec.INFO_TRY_AGAIN_LATER -> { + break + } + + else -> { + break + } + } + } + return ShortArray(collected.size) { collected[it] } + } + + override fun release() { + if (released) return + released = true + runCatching { codec.stop() } + runCatching { codec.release() } + } + + companion object { + private const val DEQUEUE_TIMEOUT_US = 10_000L // 10 ms + private const val FRAME_DURATION_US = 20_000L // 20 ms + + private fun buildFormat(): MediaFormat { + val format = + MediaFormat.createAudioFormat( + MediaFormat.MIMETYPE_AUDIO_OPUS, + AudioFormat.SAMPLE_RATE_HZ, + AudioFormat.CHANNELS, + ) + format.setByteBuffer("csd-0", ByteBuffer.wrap(buildOpusIdHeader())) + // Pre-skip + seek pre-roll: both zero, encoded as little-endian + // 64-bit nanoseconds per Android's MediaCodec contract. + format.setByteBuffer("csd-1", ByteBuffer.wrap(zeroLongLe())) + format.setByteBuffer("csd-2", ByteBuffer.wrap(zeroLongLe())) + return format + } + + private fun buildOpusIdHeader(): ByteArray { + // RFC 7845 §5.1 — 19 bytes for mono, mapping family 0. + val buf = ByteBuffer.allocate(19).order(ByteOrder.LITTLE_ENDIAN) + buf.put("OpusHead".encodeToByteArray()) // 8 bytes magic + buf.put(1.toByte()) // version + buf.put(AudioFormat.CHANNELS.toByte()) // channel count + buf.putShort(0) // pre-skip + buf.putInt(AudioFormat.SAMPLE_RATE_HZ) // input sample rate + buf.putShort(0) // output gain (Q7.8 dB) + buf.put(0.toByte()) // mapping family 0 + return buf.array() + } + + private fun zeroLongLe(): ByteArray = + ByteBuffer + .allocate(8) + .order(ByteOrder.LITTLE_ENDIAN) + .putLong(0L) + .array() + } +} diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/Audio.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/Audio.kt new file mode 100644 index 0000000000..79c8496f4e --- /dev/null +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/Audio.kt @@ -0,0 +1,100 @@ +/* + * 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 + +/** + * PCM audio format the audio pipeline produces and consumes. + * + * Listener-only flow runs at 48 kHz mono signed-16-bit, matching the nests + * Opus profile (RFC 6716 wideband at the codec's native rate). The whole + * pipeline is hardcoded to this format for now — when nests starts varying + * codec settings, this becomes a per-room negotiated value out of the + * `/api/v1/nests/` response. + */ +object AudioFormat { + const val SAMPLE_RATE_HZ: Int = 48_000 + const val CHANNELS: Int = 1 + + /** 20 ms at 48 kHz. */ + const val FRAME_SIZE_SAMPLES: Int = 960 + + /** Bytes per PCM 16-bit sample. */ + const val BYTES_PER_SAMPLE: Int = 2 +} + +/** + * Decoder for one Opus frame at a time. + * + * Implementations are stateful — Opus carries forward predictor state across + * frames — so a single decoder must be used by a single track for the + * lifetime of that subscription. Call [release] when the track ends. + */ +interface OpusDecoder { + /** + * Decode one Opus packet (the bytes of an OBJECT_DATAGRAM payload from + * MoQ) into PCM 16-bit signed mono samples. Returns an empty array if + * the decoder needs more input before producing output (some codec + * pipelines have a ramp-up frame), but never throws on a well-formed + * packet. + */ + fun decode(opusPacket: ByteArray): ShortArray + + fun release() +} + +/** + * Sink for PCM audio playback. Implementations buffer internally — [enqueue] + * may suspend if the device's playback buffer is full. + */ +interface AudioPlayer { + /** Allocate underlying audio resources and begin playback. */ + fun start() + + /** + * Feed one PCM frame (any length, but typically [AudioFormat.FRAME_SIZE_SAMPLES] + * samples) into the playback queue. + */ + suspend fun enqueue(pcm: ShortArray) + + /** Stop playback and release resources. After this, the player is unusable. */ + fun stop() +} + +/** + * Audio-pipeline exception type that lets UI code distinguish between + * recoverable codec/IO errors and fatal device-resource failures. + */ +class AudioException( + val kind: Kind, + message: String, + cause: Throwable? = null, +) : RuntimeException(message, cause) { + enum class Kind { + /** Decoder rejected an Opus packet (corrupted bytes, unsupported config). */ + DecoderError, + + /** Audio device resource (AudioTrack/AudioRecord) couldn't be allocated. */ + DeviceUnavailable, + + /** Underlying audio device threw mid-playback. */ + PlaybackFailed, + } +} diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/AudioRoomPlayer.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/AudioRoomPlayer.kt new file mode 100644 index 0000000000..6601171fae --- /dev/null +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/AudioRoomPlayer.kt @@ -0,0 +1,112 @@ +/* + * 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.MoqObject +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Job +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.launch + +/** + * Bridges a track's `Flow` (from [com.vitorpamplona.nestsclient.moq.SubscribeHandle.objects]) + * through an [OpusDecoder] into an [AudioPlayer]. + * + * Single-track. To play multiple speakers in a room, instantiate one + * [AudioRoomPlayer] per [com.vitorpamplona.nestsclient.moq.SubscribeHandle]; + * each owns its own decoder (Opus state is per-track). + * + * Lifecycle: + * - [play] starts the player and the decode loop. Returns immediately. + * - [stop] cancels the decode loop, stops the player, and releases the + * decoder. Idempotent. + */ +class AudioRoomPlayer( + private val decoder: OpusDecoder, + private val player: AudioPlayer, + private val scope: CoroutineScope, +) { + private var job: Job? = null + private var stopped = false + + /** + * Start consuming [objects] in the background. Each MoQ object's payload + * is fed to the Opus decoder; the resulting PCM frame is enqueued to the + * player. + * + * Decoder errors are reported via [onError] but do NOT stop the loop — + * one bad packet shouldn't tear down the room. Player errors are fatal. + */ + fun play( + objects: Flow, + onError: (AudioException) -> Unit = { /* swallow */ }, + ) { + check(!stopped) { "AudioRoomPlayer already stopped" } + check(job == null) { "AudioRoomPlayer.play already called" } + + player.start() + job = + scope.launch { + try { + objects.collect { obj -> + val pcm = + try { + decoder.decode(obj.payload) + } catch (ce: CancellationException) { + throw ce + } catch (t: Throwable) { + onError( + AudioException( + AudioException.Kind.DecoderError, + "Opus decode failed for object ${obj.objectId}", + t, + ), + ) + return@collect + } + if (pcm.isNotEmpty()) { + player.enqueue(pcm) + } + } + } catch (ce: CancellationException) { + throw ce + } catch (t: Throwable) { + onError( + AudioException( + AudioException.Kind.PlaybackFailed, + "audio pipeline failed", + t, + ), + ) + } + } + } + + /** Stop playback, cancel the decode loop, release the decoder. Idempotent. */ + fun stop() { + if (stopped) return + stopped = true + job?.cancel() + runCatching { player.stop() } + runCatching { decoder.release() } + } +} diff --git a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/audio/AudioRoomPlayerTest.kt b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/audio/AudioRoomPlayerTest.kt new file mode 100644 index 0000000000..f2918f07cd --- /dev/null +++ b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/audio/AudioRoomPlayerTest.kt @@ -0,0 +1,225 @@ +/* + * 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.MoqObject +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.flow.flowOf +import kotlinx.coroutines.flow.receiveAsFlow +import kotlinx.coroutines.test.runTest +import kotlin.test.Test +import kotlin.test.assertContentEquals +import kotlin.test.assertEquals +import kotlin.test.assertFailsWith +import kotlin.test.assertTrue + +class AudioRoomPlayerTest { + @Test + fun every_object_payload_is_decoded_and_enqueued_in_order() = + runTest { + val decoder = FakeOpusDecoder { byteToShorts(it) } + val player = FakeAudioPlayer() + + val objects = + flowOf( + moqObject(byteArrayOf(0x01, 0x02)), + moqObject(byteArrayOf(0x03)), + moqObject(byteArrayOf(0x04, 0x05, 0x06)), + ) + + val sut = AudioRoomPlayer(decoder, player, this) + sut.play(objects) + testScheduler.advanceUntilIdle() + + assertTrue(player.started) + assertContentEquals( + expected = + listOf( + byteToShorts(byteArrayOf(0x01, 0x02)), + byteToShorts(byteArrayOf(0x03)), + byteToShorts(byteArrayOf(0x04, 0x05, 0x06)), + ).flatten(), + actual = player.queued.flatten(), + ) + sut.stop() + assertTrue(player.stopped) + assertTrue(decoder.released) + } + + @Test + fun decoder_failure_invokes_onError_but_loop_continues() = + runTest { + val errors = mutableListOf() + val decoder = + FakeOpusDecoder { bytes -> + if (bytes.contentEquals(byteArrayOf(0xFF.toByte()))) { + throw IllegalStateException("synthetic decoder error") + } + byteToShorts(bytes) + } + val player = FakeAudioPlayer() + val objects = + flowOf( + moqObject(byteArrayOf(0x01)), + moqObject(byteArrayOf(0xFF.toByte())), + moqObject(byteArrayOf(0x02)), + ) + + val sut = AudioRoomPlayer(decoder, player, this) + sut.play(objects, onError = { errors.add(it) }) + testScheduler.advanceUntilIdle() + + assertEquals(1, errors.size) + assertEquals(AudioException.Kind.DecoderError, errors.single().kind) + // The good packets either side of the bad one still made it through. + assertContentEquals( + listOf(byteToShorts(byteArrayOf(0x01)), byteToShorts(byteArrayOf(0x02))).flatten(), + player.queued.flatten(), + ) + sut.stop() + } + + @Test + fun stop_is_idempotent_and_releases_decoder_only_once() = + runTest { + val decoder = FakeOpusDecoder { byteToShorts(it) } + val player = FakeAudioPlayer() + val sut = AudioRoomPlayer(decoder, player, this) + sut.play(flowOf()) + testScheduler.advanceUntilIdle() + sut.stop() + sut.stop() // second call must not double-release + assertEquals(1, decoder.releaseCount) + assertEquals(1, player.stopCount) + } + + @Test + fun play_cannot_be_called_twice_on_the_same_instance() = + runTest { + val sut = + AudioRoomPlayer( + FakeOpusDecoder { byteToShorts(it) }, + FakeAudioPlayer(), + this, + ) + sut.play(flowOf()) + assertFailsWith { sut.play(flowOf()) } + sut.stop() + } + + @Test + fun play_after_stop_is_rejected() = + runTest { + val sut = + AudioRoomPlayer( + FakeOpusDecoder { byteToShorts(it) }, + FakeAudioPlayer(), + this, + ) + sut.stop() + assertFailsWith { sut.play(flowOf()) } + } + + @Test + fun decoder_emitting_empty_pcm_does_not_call_player_enqueue() = + runTest { + val decoder = FakeOpusDecoder { ShortArray(0) } + val player = FakeAudioPlayer() + val sut = AudioRoomPlayer(decoder, player, this) + sut.play(flowOf(moqObject(byteArrayOf(0x01)))) + testScheduler.advanceUntilIdle() + assertEquals(0, player.queued.size) + sut.stop() + } + + @Test + fun objects_arriving_after_play_are_streamed_through_the_pipeline() = + runTest { + val channel = Channel(capacity = 8) + val decoder = FakeOpusDecoder { byteToShorts(it) } + val player = FakeAudioPlayer() + + val sut = AudioRoomPlayer(decoder, player, this) + sut.play(channel.receiveAsFlow()) + testScheduler.runCurrent() + + channel.send(moqObject(byteArrayOf(0x10))) + channel.send(moqObject(byteArrayOf(0x20))) + testScheduler.advanceUntilIdle() + + assertEquals(2, player.queued.size) + assertContentEquals(byteToShorts(byteArrayOf(0x10)), player.queued[0]) + assertContentEquals(byteToShorts(byteArrayOf(0x20)), player.queued[1]) + + sut.stop() + } + + // -- helpers ----------------------------------------------------------- + + private fun moqObject(payload: ByteArray): MoqObject = + MoqObject( + trackAlias = 1, + groupId = 0, + objectId = 0, + publisherPriority = 0x80, + payload = payload, + ) + + private fun byteToShorts(b: ByteArray): ShortArray = ShortArray(b.size) { b[it].toShort() } + + private class FakeOpusDecoder( + private val transform: (ByteArray) -> ShortArray, + ) : OpusDecoder { + var releaseCount = 0 + private set + val released: Boolean get() = releaseCount > 0 + + override fun decode(opusPacket: ByteArray): ShortArray = transform(opusPacket) + + override fun release() { + releaseCount++ + } + } + + private class FakeAudioPlayer : AudioPlayer { + var started = false + private set + var stopCount = 0 + private set + val stopped: Boolean get() = stopCount > 0 + val queued = mutableListOf() + + override fun start() { + started = true + } + + override suspend fun enqueue(pcm: ShortArray) { + queued.add(pcm) + } + + override fun stop() { + stopCount++ + } + } +} + +/** Small helper so test assertions can flatten lists of ShortArrays. */ +private fun List.flatten(): List = flatMap { sa -> sa.toList() }