Add bounded Android microphone subscriptions with independent ownership

2026-10-16 01:13:23 +02:00
1 parent d4eb1da233
commit 4a49c830d1
@@ -0,0 +1,67 @@
+package com.utun.chat.data
+
+import android.media.AudioDeviceInfo
+import android.media.AudioFormat
+import android.media.AudioRecord
+import android.media.AudioTimestamp
+import android.media.MediaRecorder
+
+/** Реестр трансляционного захвата. Рация заимствует его только при совпадении выбранного входа.
+ * Системный default при отсутствии явного выбора означает тот же вход, который уже использует трансляция. */
+object AndroidMicrophoneCapture {
+    private var stream: MicrophoneCapture? = null
+
+    private class Record(private val record: AudioRecord) : MicrophoneCapture.Device {
+        override val channels = record.channelCount
+        override val inputId: Int? get() = record.routedDevice?.id
+        private val stamp = AudioTimestamp()
+        private var total = 0L
+        @Volatile private var running = true
+        override fun read(): MicrophoneCapture.Frame {
+            val pcm = ShortArray(960 * channels)
+            var used = 0
+            while (running && used < pcm.size) {
+                val count = record.read(pcm, used, pcm.size - used, AudioRecord.READ_BLOCKING)
+                check(count > 0) { "AudioRecord read=$count input=$inputId" }
+                used += count
+            }
+            check(running) { "AudioRecord stopped" }
+            val first = total; total += 960
+            val measured = record.getTimestamp(stamp, AudioTimestamp.TIMEBASE_MONOTONIC) == AudioRecord.SUCCESS
+            val time = if (measured) stamp.nanoTime - (stamp.framePosition - first) * 1_000_000_000 / 48000 else 0L
+            return MicrophoneCapture.Frame(pcm, time, first)
+        }
+        override fun stop() { running = false; record.stop() }
+        override fun close() { record.release() }
+    }
+
+    fun own(record: AudioRecord, released: () -> Unit = {}): MicrophoneCapture {
+        try {
+            record.startRecording()
+            check(record.recordingState == AudioRecord.RECORDSTATE_RECORDING) { "Microphone did not start" }
+            return MicrophoneCapture(Record(record), { LogManager.addLog("WARN", "radio", it) }, released)
+        } catch (e: Exception) { record.release(); throw e }
+    }
+
+    @Synchronized fun openStream(released: () -> Unit): MicrophoneCapture.Subscription {
+        check(stream == null || stream!!.closed) { "Stream microphone already owned" }
+        val min = AudioRecord.getMinBufferSize(48000, AudioFormat.CHANNEL_IN_STEREO, AudioFormat.ENCODING_PCM_16BIT)
+        check(min > 0) { "CAMCORDER stereo format unavailable" }
+        val record = AudioRecord(MediaRecorder.AudioSource.CAMCORDER, 48000, AudioFormat.CHANNEL_IN_STEREO,
+            AudioFormat.ENCODING_PCM_16BIT, (min * 2).coerceAtLeast(1920 * 8))
+        if (record.state != AudioRecord.STATE_INITIALIZED || record.channelCount != 2) {
+            record.release(); error("CAMCORDER stereo capture unavailable")
+        }
+        val capture = own(record, released)
+        stream = capture
+        LogManager.addLog("INFO", "stream", "capture opened source=CAMCORDER channels=2 rate=48000 input=${capture.inputId}")
+        return capture.subscribe("stream")
+    }
+
+    @Synchronized fun borrowRadio(input: AudioDeviceInfo?): MicrophoneCapture.Subscription? {
+        val capture = stream ?: return null
+        if (capture.closed || (input != null && input.id != capture.inputId)) return null
+        LogManager.addLog("INFO", "radio", "shared stream capture input=${capture.inputId} selected=${input?.id} channels=${capture.channels}")
+        return capture.subscribe("radio")
+    }
+}
@@ -0,0 +1,88 @@
+package com.utun.chat.data
+
+import java.util.concurrent.ArrayBlockingQueue
+import java.util.concurrent.TimeUnit
+
+/** Один аппаратный reader, независимые ограниченные очереди потребителей. PCM каждого потребителя — своя копия.
+ * Закрытие подписки не останавливает соседей; последний владелец останавливает read, делает join и release.
+ * PTT попадает в ту же очередь, что PCM, поэтому native TX всегда имеет одного последовательного владельца. */
+class MicrophoneCapture(private val device: Device, private val log: (String) -> Unit, private val released: () -> Unit = {}) {
+    interface Device {
+        val channels: Int
+        val inputId: Int?
+        fun read(): Frame
+        fun stop()
+        fun close()
+    }
+    sealed interface Event
+    data class Frame(val pcm: ShortArray, val timeNs: Long, val sequence: Long) : Event
+    data class Ptt(val pressed: Boolean) : Event
+    private val consumers = mutableListOf<Subscription>()
+    @Volatile private var reading = true
+    @Volatile private var failure: Exception? = null
+    private var worker: Thread? = null
+    val closed: Boolean get() = !reading
+    val inputId: Int? get() = device.inputId
+    val channels: Int get() = device.channels
+
+    inner class Subscription internal constructor(val name: String) : AutoCloseable {
+        private val events = ArrayBlockingQueue<Event>(32)
+        @Volatile private var attached = true
+        private var dropped = 0L
+        val channels: Int get() = device.channels
+
+        @Synchronized internal fun deliver(frame: Frame) {
+            if (!attached) return
+            if (events.size >= 24 || !events.offer(frame.copy(pcm = frame.pcm.copyOf()))) {
+                dropped++
+                if (dropped == 1L || dropped % 50L == 0L) log("capture queue overflow consumer=$name dropped=$dropped")
+            }
+        }
+        /** Только управляющий поток; восемь мест зарезервированы под команды. */
+        @Synchronized fun ptt(pressed: Boolean): Boolean {
+            if (!attached) return false
+            if (events.offer(Ptt(pressed))) return true
+            log("capture control queue full consumer=$name pressed=$pressed")
+            return false
+        }
+        fun poll(): Event? {
+            failure?.let { throw IllegalStateException("Microphone read failed", it) }
+            return if (attached) events.poll(20, TimeUnit.MILLISECONDS) else null
+        }
+        override fun close() {
+            synchronized(this) { if (!attached) return; attached = false; events.clear() }
+            detach(this)
+        }
+    }
+
+    @Synchronized fun subscribe(name: String): Subscription {
+        check(reading && failure == null) { "Microphone is closed or failed" }
+        val subscription = Subscription(name)
+        consumers.add(subscription)
+        if (worker == null) worker = Thread({
+            try {
+                while (reading) {
+                    val frame = device.read()
+                    val targets = synchronized(this) { consumers.toList() }
+                    for (target in targets) target.deliver(frame)
+                }
+            } catch (e: Exception) {
+                if (reading) { failure = e; log("capture reader failed: ${e.message}") }
+            }
+        }, "microphone-capture").apply { start() }
+        log("capture attached consumer=$name channels=$channels consumers=${consumers.size}")
+        return subscription
+    }
+
+    private fun detach(subscription: Subscription) {
+        val last = synchronized(this) {
+            consumers.remove(subscription)
+            log("capture detached consumer=${subscription.name} remaining=${consumers.size}")
+            consumers.isEmpty().also { if (it) reading = false }
+        }
+        if (!last) return
+        try { device.stop() } catch (e: Exception) { log("capture stop failed: ${e.message}") }
+        worker?.join()
+        try { device.close() } finally { released() }
+    }
+}
@@ -0,0 +1,69 @@
+package com.utun.chat.data
+
+import org.junit.Assert.*
+import org.junit.Test
+import java.util.concurrent.LinkedBlockingQueue
+import java.util.concurrent.atomic.AtomicInteger
+
+class MicrophoneCaptureTest {
+    private class Device : MicrophoneCapture.Device {
+        override val channels = 2
+        override val inputId = 7
+        val input = LinkedBlockingQueue<MicrophoneCapture.Frame>()
+        val reads = AtomicInteger()
+        var closes = 0
+        override fun read(): MicrophoneCapture.Frame = input.take().also { reads.incrementAndGet() }
+        override fun stop() { input.offer(MicrophoneCapture.Frame(shortArrayOf(), 0, 0)) }
+        override fun close() { closes++ }
+    }
+    private fun frame(subscription: MicrophoneCapture.Subscription): MicrophoneCapture.Frame {
+        repeat(50) { (subscription.poll() as? MicrophoneCapture.Frame)?.let { return it } }
+        error("Capture frame timeout")
+    }
+
+    @Test fun consumersShareCaptureFormatAndTimeButOwnTheirSamples() {
+        val device = Device(); val capture = MicrophoneCapture(device, {})
+        val stream = capture.subscribe("stream"); val radio = capture.subscribe("radio")
+        try {
+            device.input.add(MicrophoneCapture.Frame(shortArrayOf(10, 20), 987654321, 960))
+            val first = frame(stream); val second = frame(radio)
+            assertEquals(2, radio.channels); assertEquals(first.timeNs, second.timeNs); assertEquals(first.sequence, second.sequence)
+            first.pcm[0] = 99
+            assertArrayEquals(shortArrayOf(10, 20), second.pcm)
+            stream.close()
+            device.input.add(MicrophoneCapture.Frame(shortArrayOf(30, 40), 987674321, 1920))
+            assertArrayEquals(shortArrayOf(30, 40), frame(radio).pcm); assertEquals(0, device.closes)
+        } finally { stream.close(); radio.close() }
+        assertEquals(1, device.closes); assertTrue(capture.closed)
+    }
+
+    @Test fun stoppingRadioKeepsStreamAndPttIsOrderedWithPcm() {
+        val device = Device(); val capture = MicrophoneCapture(device, {})
+        val stream = capture.subscribe("stream"); val radio = capture.subscribe("radio")
+        try {
+            assertTrue(radio.ptt(true))
+            device.input.add(MicrophoneCapture.Frame(shortArrayOf(1, 2), 1000, 0))
+            assertEquals(MicrophoneCapture.Ptt(true), radio.poll())
+            assertArrayEquals(shortArrayOf(1, 2), frame(radio).pcm)
+            assertTrue(radio.ptt(false)); assertEquals(MicrophoneCapture.Ptt(false), radio.poll())
+            radio.close(); assertEquals(0, device.closes)
+            assertArrayEquals(shortArrayOf(1, 2), frame(stream).pcm)
+        } finally { radio.close(); stream.close() }
+        assertEquals(1, device.closes)
+    }
+
+    @Test fun slowConsumerHasBoundedQueueAndReservedControlCapacity() {
+        val device = Device(); val logs = LinkedBlockingQueue<String>(); val capture = MicrophoneCapture(device, { logs.add(it) })
+        val stream = capture.subscribe("stream"); val radio = capture.subscribe("radio")
+        try {
+            repeat(40) {
+                device.input.add(MicrophoneCapture.Frame(shortArrayOf(1, 2), it * 20000000L, it * 960L))
+                frame(stream) // Трансляция успевает, рация намеренно не читает.
+            }
+            assertTrue(logs.any { it.contains("consumer=radio dropped=") })
+            assertTrue(radio.ptt(true)); assertTrue(radio.ptt(false))
+            repeat(24) { assertTrue(radio.poll() is MicrophoneCapture.Frame) }
+            assertEquals(MicrophoneCapture.Ptt(true), radio.poll()); assertEquals(MicrophoneCapture.Ptt(false), radio.poll())
+        } finally { radio.close(); stream.close() }
+    }
+}