Mix Android streams on one AudioTrack with independent clocks

2026-10-20 16:58:32 +02:00
1 parent 57d28376b6
commit 4e386fc8e3
@@ -0,0 +1,53 @@
+package com.utun.chat.data
+
+import kotlin.math.abs
+import kotlin.math.max
+import kotlin.math.min
+import kotlin.math.roundToInt
+
+/** Начало отдельного источника на шкале общего AudioTrack. PTS источников здесь не сравниваются. */
+internal class StreamMixTimeline {
+    var origin = -1L
+        private set
+    var frames = 0L
+        private set
+
+    fun append(firstOutputFrame: Long, count: Int) {
+        require(firstOutputFrame >= 0 && count > 0)
+        if (origin < 0) origin = firstOutputFrame
+        check(firstOutputFrame == origin + frames) { "Non-contiguous mixer output $firstOutputFrame != ${origin + frames}" }
+        frames += count
+    }
+
+    fun played(outputHead: Long): Long = if (origin < 0) 0 else (outputHead - origin).coerceIn(0, frames)
+}
+
+/** Одна порция stereo 48k. Тихий новый участник не снижает громкость остальных.
+ * Буфер output переиспользуется только после полной записи предыдущей порции устройством. */
+internal class StreamMixBuffer(val frames: Int = 480) {
+    private val sum = FloatArray(frames * 2)
+    private val output = ShortArray(frames * 2)
+    private var gain = 1f
+    fun clear() = sum.fill(0f)
+
+    fun add(pcm: ShortArray, fadeFrames: Int = 480): Int {
+        require(pcm.size <= sum.size && pcm.size % 2 == 0)
+        for (i in 0 until pcm.size / 2) {
+            val volume = min(1f, (fadeFrames + i + 1) / 480f) / 32768f
+            sum[2 * i] += pcm[2 * i] * volume
+            sum[2 * i + 1] += pcm[2 * i + 1] * volume
+        }
+        return min(480, fadeFrames + pcm.size / 2)
+    }
+
+    fun finish(): ShortArray {
+        for (i in 0 until frames) {
+            val peak = max(abs(sum[i * 2]), abs(sum[i * 2 + 1]))
+            val target = if (peak > 1f) 1f / peak else 1f
+            gain = min(target, gain + (1f - gain) / 4800f)
+            output[i * 2] = (sum[i * 2] * gain * 32768f).roundToInt().coerceIn(-32768, 32767).toShort()
+            output[i * 2 + 1] = (sum[i * 2 + 1] * gain * 32768f).roundToInt().coerceIn(-32768, 32767).toShort()
+        }
+        return output
+    }
+}
@@ -0,0 +1,131 @@
+package com.utun.chat.data
+
+/** Один AudioTrack для всех трансляций. Worker не удерживает общий lock при обращении к decoder.
+ * Input закрывается после исключения из списка; его собственный lock защищает native DSP от release. */
+internal object StreamAudioMixer {
+    data class OutputState(val played: Long = 0, val leadUs: Long = 0)
+    @Volatile var output = OutputState()
+        private set
+    private val lock = Any()
+    private val inputs = linkedSetOf<StreamMixedPlayback>()
+    private var worker: Thread? = null
+
+    fun attach(input: StreamMixedPlayback) = synchronized(lock) {
+        check(inputs.add(input))
+        if (worker == null) {
+            output = OutputState()
+            worker = Thread(::run, "stream-mixer").apply { start() }
+        }
+        LogManager.addLog("INFO", "stream", "mixer input attached id=${input.id} count=${inputs.size}")
+    }
+
+    fun detach(input: StreamMixedPlayback) = synchronized(lock) {
+        inputs.remove(input)
+        LogManager.addLog("INFO", "stream", "mixer input detached id=${input.id} count=${inputs.size}")
+    }
+
+    private fun run() {
+        val audio = StreamAudioOutput(log = { LogManager.addLog("INFO", "stream", "mixer $it") })
+        val mix = StreamMixBuffer()
+        var generated = 0L
+        try {
+            while (true) {
+                val current = synchronized(lock) {
+                    if (inputs.isEmpty()) {
+                        // Закрытие под lock исключает открытие второго AudioTrack новым attach.
+                        audio.close(); output = OutputState(); worker = null
+                        return
+                    }
+                    inputs.toList()
+                }
+                audio.step(System.nanoTime(), hold = false, pull = {
+                    mix.clear()
+                    var active = false
+                    for (input in current) active = input.mix(mix, generated) || active
+                    if (active) {
+                        generated += mix.frames
+                        mix.finish()
+                    } else null
+                }, finished = { false })
+                output = OutputState(audio.played, audio.leadUs)
+                Thread.sleep(5)
+            }
+        } catch (e: Exception) {
+            LogManager.addLog("ERROR", "stream", "shared audio output failed: ${e.message}; ${audio.description}")
+            synchronized(lock) {
+                for (input in inputs) input.outputFailure = e.message ?: "Audio output failed"
+                inputs.clear()
+                try { audio.close() }
+                catch (close: Exception) { LogManager.addLog("WARN", "stream", "mixer release: ${close.message}") }
+                output = OutputState(); worker = null
+            }
+        }
+    }
+}
+
+/** Независимые Opus/SoundTouch, темп и PTS. Video worker и mixer сериализуют только свой DSP.
+ * Каждый 10ms блок общего выхода отражён в своей шкале, включая тишину при starvation/hold.
+ * Reset создаёт новый Input с новой origin; AudioTrack и остальные источники продолжают играть. */
+internal class StreamMixedPlayback(val id: Long, pts: Long, private val audioEnabled: Boolean) {
+    private val lock = Any()
+    private var handle = NativeLib.streamPlaybackCreate(id, pts, audioEnabled).also { check(it != 0L) }
+    private val timeline = StreamMixTimeline()
+    private var started = false
+    private var hold = false
+    private var ready = -1L
+    private var ended = false
+    private var endFrame: Long? = null
+    private var fadeFrames = 0
+    @Volatile var outputFailure: String? = null
+    init { if (audioEnabled) StreamAudioMixer.attach(this) }
+
+    val leadUs: Long get() = if (audioEnabled) StreamAudioMixer.output.leadUs else 0
+    val empty: Boolean get() = synchronized(lock) { handle == 0L || NativeLib.streamPlaybackEmpty(handle) }
+    val drained: Boolean get() = synchronized(lock) {
+        !audioEnabled || (ended && (endFrame?.let { timeline.played(StreamAudioMixer.output.played) >= it }
+            ?: (timeline.frames == 0L && NativeLib.streamPlaybackEmpty(handle))))
+    }
+    val description: String get() = synchronized(lock) {
+        "mixed=$audioEnabled origin=${timeline.origin} written=${timeline.frames} " +
+            "played=${timeline.played(StreamAudioMixer.output.played)} hold=$hold leadUs=$leadUs"
+    }
+
+    fun feed(pts: Long, packet: ByteArray): Int = synchronized(lock) {
+        check(handle != 0L); NativeLib.streamPlaybackFeed(handle, pts, packet)
+    }
+    fun finish() = synchronized(lock) {
+        check(handle != 0L); ended = true; NativeLib.streamPlaybackFinish(handle)
+    }
+
+    fun update(now: Long, available: Long, guard: Long, videoHeld: Boolean): LongArray? = synchronized(lock) {
+        check(handle != 0L); check(outputFailure == null) { outputFailure!! }
+        ready = available; hold = videoHeld; started = true
+        NativeLib.streamPlaybackUpdate(handle, now, timeline.played(StreamAudioMixer.output.played), available, guard, videoHeld)
+    }
+
+    /** Mixer не блокирует чужой input ожиданием media; недостающее заменяется тишиной только этого источника. */
+    fun mix(buffer: StreamMixBuffer, firstOutputFrame: Long): Boolean = synchronized(lock) {
+        if (handle == 0L || !started || !audioEnabled || outputFailure != null) return false
+        try {
+            val pcm = if (hold) null else NativeLib.streamPlaybackPull(handle, buffer.frames, if (ended) Long.MAX_VALUE else ready)
+            val count = (pcm?.size ?: 0) / 2
+            if (pcm != null) fadeFrames = buffer.add(pcm, fadeFrames)
+            if (ended && endFrame == null && NativeLib.streamPlaybackEmpty(handle)) endFrame = timeline.frames + count
+            if (count < buffer.frames) NativeLib.streamPlaybackSilence(handle, buffer.frames - count)
+            timeline.append(firstOutputFrame, buffer.frames)
+            true
+        } catch (e: Exception) {
+            outputFailure = e.message ?: "Audio mixing failed"
+            LogManager.addLog("ERROR", "stream", "mixer input failed id=$id: ${e.message}")
+            false
+        }
+    }
+
+    fun close() {
+        if (audioEnabled) StreamAudioMixer.detach(this)
+        synchronized(lock) {
+            if (handle != 0L) NativeLib.streamPlaybackClose(handle)
+            handle = 0
+        }
+    }
+}
@@ -0,0 +1,60 @@
+package com.utun.chat.data
+
+import org.junit.Assert.*
+import org.junit.Test
+import kotlin.math.PI
+import kotlin.math.abs
+import kotlin.math.sin
+
+class StreamAudioMixTest {
+    @Test fun joiningAndResetUseTheirOwnOutputOrigin() {
+        val first = StreamMixTimeline()
+        first.append(0, 480); first.append(480, 480)
+        val joined = StreamMixTimeline()
+        joined.append(960, 480); first.append(960, 480)
+        assertEquals(1200, first.played(1200))
+        assertEquals(240, joined.played(1200))
+        assertEquals(0, joined.played(700)) // Устройство ещё до подключения второго источника.
+        val reset = StreamMixTimeline()
+        reset.append(1440, 480); first.append(1440, 480)
+        assertEquals(1600, first.played(1600))
+        assertEquals(160, reset.played(1600))
+        assertEquals(480, joined.played(1600))
+    }
+
+    @Test fun partialDeviceWritesAdvanceOnlyHeardFrames() {
+        val clock = StreamMixTimeline()
+        clock.append(2000, 480)
+        for (head in listOf(1900L, 2000L, 2120L, 2240L, 2480L, 3000L))
+            assertEquals((head - 2000).coerceIn(0, 480), clock.played(head))
+    }
+
+    @Test fun differentTonesAreSummedAndSilentInputDoesNotReduceVolume() {
+        val a = ShortArray(960) { (4000 * sin(2 * PI * 1000 * (it / 2) / 48000)).toInt().toShort() }
+        val b = ShortArray(960) { (2000 * sin(2 * PI * 500 * (it / 2) / 48000)).toInt().toShort() }
+        val mix = StreamMixBuffer()
+        mix.clear(); mix.add(a); mix.add(b)
+        val sum = mix.finish().copyOf()
+        for (i in sum.indices) assertTrue(abs(sum[i].toInt() - a[i] - b[i]) <= 1)
+        mix.clear(); mix.add(a); mix.add(ShortArray(960))
+        assertArrayEquals(a, mix.finish())
+    }
+
+    @Test fun limiterKeepsStereoBalanceAndReleasesSmoothly() {
+        val mix = StreamMixBuffer()
+        val loud = ShortArray(960) { if (it % 2 == 0) 24000 else 12000 }
+        mix.clear(); mix.add(loud); mix.add(loud)
+        val limited = mix.finish().copyOf()
+        assertEquals(32767, limited[0].toInt()); assertEquals(16384, limited[1].toInt())
+        mix.clear(); mix.add(loud)
+        val quieter = mix.finish()
+        assertTrue(quieter.first() < 18000); assertTrue(quieter.last() < 12000)
+        assertTrue(quieter[958] > quieter[0])
+    }
+
+    @Test fun accidentalHoleInOutputMappingIsRejected() {
+        val clock = StreamMixTimeline(); clock.append(1000, 480)
+        try { clock.append(1500, 480); fail("Gap must not silently move this source's clock") }
+        catch (_: IllegalStateException) {}
+    }
+}