Render Android call video against the audio playback clock

tools/chatgui-android/app/src/main/java/com/utun/chat/data/CallVideoPlayer.kt

@@ -0,0 +1,94 @@
+package com.utun.chat.data
+
+import android.media.MediaCodec
+import android.view.Surface
+
+/** Один RX worker; удерживаем output buffer до фактического audio PTS. Новое поколение ждёт drain старого. */
+internal class CallVideoPlayer(private val id: Long, private val failure: (String) -> Unit) {
+    @Volatile private var running = true
+    @Volatile private var target: Surface? = null
+    private val worker = Thread(::loop, "call-video-rx").apply { start() }
+    fun surface(surface: Surface?) { target = surface }
+    fun stop() { running = false; worker.join() }
+
+    private fun loop() {
+        var decoder: MediaCodec? = null; var surface: Surface? = null; var format: CallVideoFormat? = null
+        var pending: ByteArray? = null; var meta = LongArray(12)
+        var held = -1; var heldPts = -1L; var inflight = 0; var draining = false
+        var sourceEpoch = 0L; var lastOutput = -1L; var lastProgress = System.nanoTime()
+        val clock = LongArray(8); val info = MediaCodec.BufferInfo()
+        fun close() {
+            try { decoder?.stop() } catch (e: Exception) { LogManager.addLog("WARN", "call", "video decoder stop: ${e.message}") }
+            decoder?.release(); decoder = null; format = null; held = -1; inflight = 0; draining = false; lastOutput = -1
+        }
+        try {
+            while (running) {
+                try {
+                    val now = System.nanoTime()
+                    if (surface !== target || surface?.isValid == false) {
+                        close(); surface = target; pending = null; CallVideoNative.requestKey(id)
+                    }
+                    val valid = NativeLib.callAudioClock(id, now / 1000, clock) && clock[5] != 0L
+                    if (valid && sourceEpoch != clock[1]) {
+                        if (sourceEpoch != 0L) { close(); pending = null; CallVideoNative.requestKey(id) }
+                        sourceEpoch = clock[1]
+                    }
+                    if (held >= 0 && valid && heldPts <= clock[4]) {
+                        decoder!!.releaseOutputBuffer(held, heldPts >= clock[4] - 500_000); held = -1
+                        lastProgress = now
+                    }
+                    if (decoder != null && held < 0) {
+                        val out = decoder!!.dequeueOutputBuffer(info, 0)
+                        if (out >= 0) {
+                            if (info.flags and MediaCodec.BUFFER_FLAG_END_OF_STREAM != 0) {
+                                decoder!!.releaseOutputBuffer(out, false); close()
+                            } else if (info.size == 0) decoder!!.releaseOutputBuffer(out, false)
+                            else {
+                                check(info.presentationTimeUs > lastOutput) { "Video decoder reordered PTS" }
+                                lastOutput = info.presentationTimeUs; inflight = (inflight - 1).coerceAtLeast(0)
+                                held = out; heldPts = info.presentationTimeUs; lastProgress = now
+                            }
+                        }
+                    }
+                    if (pending == null && held < 0 && inflight < 8) pending = CallVideoNative.pull(id, meta)
+                    val data = pending
+                    if (data != null && surface?.isValid == true && held < 0) {
+                        val next = CallVideoFormat.from(meta)
+                        val key = meta[11] != 0L
+                        if (format != next && decoder != null) {
+                            if (!draining) {
+                                val input = decoder!!.dequeueInputBuffer(0)
+                                if (input >= 0) { decoder!!.queueInputBuffer(input, 0, 0, meta[10], MediaCodec.BUFFER_FLAG_END_OF_STREAM); draining = true }
+                            }
+                        } else {
+                            if (decoder == null) {
+                                if (!key) { pending = null; CallVideoNative.requestKey(id); continue }
+                                CallVideoCodec.validateKey(data, next)
+                                val type = StreamVideoCodec.fromId(next.codec); val config = type.unpackKey(data).first
+                                decoder = MediaCodec.createDecoderByType(type.mime)
+                                decoder!!.configure(type.decoderFormat(next.width, next.height, next.rotation, config), surface, null, 0)
+                                decoder!!.start(); format = next; lastProgress = now
+                                LogManager.addLog("INFO", "call", "video decoder started call=$id format=$next")
+                            }
+                            val index = decoder!!.dequeueInputBuffer(0)
+                            if (index >= 0) {
+                                val bytes = if (key) {
+                                    CallVideoCodec.validateKey(data, next)
+                                    val parts = StreamVideoCodec.fromId(next.codec).unpackKey(data); parts.first + parts.second
+                                } else data
+                                val buffer = checkNotNull(decoder!!.getInputBuffer(index)); check(bytes.size <= buffer.capacity())
+                                buffer.put(bytes); decoder!!.queueInputBuffer(index, 0, bytes.size, meta[10], 0)
+                                pending = null; inflight++
+                            }
+                        }
+                    } else if (data != null && surface == null) { pending = null; CallVideoNative.requestKey(id) }
+                    if (decoder != null && held < 0 && inflight > 0 && now - lastProgress > 2_000_000_000) error("Video decoder stalled")
+                } catch (e: Exception) {
+                    LogManager.addLog("ERROR", "call", "video receive call=$id: ${e.message}")
+                    close(); pending = null; CallVideoNative.requestKey(id); failure(e.message ?: "Video decoder failed")
+                }
+                Thread.sleep(5)
+            }
+        } finally { close() }
+    }
+}