Consume Android batches with independent track completion

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

@@ -101,6 +101,12 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
         var audioEnabled = false
         var opened = false
         var ended = false
+        var videoInputEnded = false
+        var audioInputEnded = false
+        val records = StreamReadQueue()
+        var signal = 0L
+        var batches = 0L
+        var batchMax = 0
         var videoRevision = 0L
         var position = -1L
         var ready = -1L
@@ -235,7 +241,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
             ready = meta[0]
             if (playback != null) return
             audioEnabled = meta[5] and 2 != 0L
-            playback = StreamMixedPlayback(id, meta[3], audioEnabled)
+            playback = StreamMixedPlayback(id, meta[3], audioEnabled) { NativeLib.streamSignalNotify(signal) }
             position = meta[3]; generation = meta[4]
             LogManager.addLog("INFO", "stream", "playback starts id=$id PTS=$position generation=$generation fixedPrefill=0 " +
                 "elapsedUs=${System.nanoTime() / 1000 - requestedAtUs} readyUs=$ready videoUs=${meta[1]} audioUs=${meta[2]} " +
@@ -253,22 +259,29 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
             try { playback?.close() }
             catch (e: Exception) { LogManager.addLog("WARN", "stream", "audio reset id=$id: ${e.message}") }
             playback = null; position = -1; generation = 0; needsAudio = true; clockState = null; playbackStarted = false
+            records.clear(); ended = false; videoInputEnded = false; audioInputEnded = false
             resetVideo(); deadline = System.nanoTime() + 10_000_000_000
             if (request) check(NativeLib.streamResync(id)) { "Media resync refused" }
         }
         video = createVideoPump()
         try {
             check(NativeLib.streamOpen(id, group, source, audio = true, epoch = epoch)) { "Viewer open refused" }
+            signal = NativeLib.streamSignalCreate(id).also { check(it != 0L) { "Viewer notification initialization failed" } }
             while (running && surface.isValid) {
                 check(NativeLib.isReady()) { "Core stopped" }
                 observeStatus(NativeLib.streamStatus(id))
                 drainTimeoutMs = ((ready - position) / 1000 + 2000).coerceIn(2000, 30000)
                 val deviceLead = playback?.leadUs ?: 0
-                for (work in 0 until 4) {
+                var consumed = 0
+                for (work in 0 until 32) {
                     video.step(position, System.nanoTime() / 1000)
-                    val audioAhead = if (audioEnabled && needsAudio) deviceLead + 80000 else -1L
-                    val raw = if (ended) null else NativeLib.streamPull(id, position, video.wantsInput, audioAhead)
-                    if (raw == null) break
+                    needsAudio = !audioInputEnded && (playback?.needsInput ?: true)
+                    val audioAhead = if (!audioInputEnded && (!audioEnabled || needsAudio) && records.audios == 0) deviceLead + 80000 else -1L
+                    val videoBudget = if (!videoInputEnded && video.wantsInput && records.videos == 0) 1 else 0
+                    val batch = if (ended) null else NativeLib.streamPull(id, position, videoBudget, audioAhead, 16)
+                    if (batch != null) { batches++; batchMax = maxOf(batchMax, records.add(batch)) }
+                    val raw = records.take(!videoInputEnded && video.wantsInput, !audioEnabled || needsAudio) ?: break
+                    consumed++
                     check(raw.size >= 28)
                     val b = ByteBuffer.wrap(raw).order(ByteOrder.BIG_ENDIAN)
                     val kind = b.get().toInt(); val flags = b.get().toInt() and 255
@@ -308,7 +321,13 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
                         }
                         3 -> { LogManager.addLog("INFO", "stream", "playback GAP id=$id"); resetTimeline(false) }
                         4 -> opened = true
-                        5 -> { ended = true; playback?.finish() }
+                        5, 7, 8 -> {
+                            if (kind != 8) videoInputEnded = true
+                            if (kind != 7 && !audioInputEnded) { audioInputEnded = true; playback?.finish() }
+                            if (kind == 5) ended = true
+                            LogManager.addLog("INFO", "stream", "viewer track end id=$id kind=$kind videoEnd=$videoInputEnded " +
+                                "audioEnd=$audioInputEnded pending=${records.size}")
+                        }
                         6 -> {
                             mutableStatus.value = when (flags) {
                                 1 -> R.string.stream_group_closed; 2 -> R.string.chat_service_stopped
@@ -316,6 +335,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
                                 else -> R.string.stream_failed
                             }
                             LogManager.addLog("WARN", "stream", "native viewer closed id=$id reason=$flags"); running = false
+                            break
                         }
                         else -> error("Unknown media event=$kind")
                     }
@@ -333,12 +353,12 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
                 }
                 val codec = decoder
                 if (codec != null) {
-                    if (ended && video.input == null && video.output == null && !eosQueued) {
+                    if (videoInputEnded && video.input == null && video.output == null && !eosQueued) {
                         val index = codec.dequeueInputBuffer(0)
                         if (index >= 0) { codec.queueInputBuffer(index, 0, 0, video.lastInputPts.coerceAtLeast(0),
                             MediaCodec.BUFFER_FLAG_END_OF_STREAM); eosQueued = true }
                     }
-                    video.step(position, System.nanoTime() / 1000); video.checkProgress(System.nanoTime() / 1000, ended)
+                    video.step(position, System.nanoTime() / 1000); video.checkProgress(System.nanoTime() / 1000, videoInputEnded)
                     check(!video.ended || video.firstOutputPts != null || video.inputCount == 0L) { "Decoder ended without producing an image" }
                 }
                 if (ended && video.input == null && (decoder == null || video.ended) && playback?.empty != false
@@ -354,9 +374,15 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
                         "clock=$position ready=$ready audio=$audioEnabled audioInputs=$audioInputs audioPTS=$lastAudioPts needsAudio=$needsAudio " +
                         "reserve=${clockState?.get(1)} minimum=${clockState?.get(2)} " +
                         "tempo=${clockState?.get(3)} ${playback?.description} waitingKey=$waitingKey " +
-                        "shownPTS=$lastShownPts displayLagUs=${if (lastShownPts >= 0) position - lastShownPts else -1}")
+                        "shownPTS=$lastShownPts displayLagUs=${if (lastShownPts >= 0) position - lastShownPts else -1} " +
+                        "batches=$batches batchMax=$batchMax pending=${records.size} videoEnd=$videoInputEnded audioEnd=$audioInputEnded")
+                    batches = 0; batchMax = 0
+                }
+                // Синхронный MediaCodec требует проверки готовности; media/PCM/stop будят без ожидания deadline.
+                if (running && consumed < 32) {
+                    val codecPending = video.input != null || video.output != null || video.inflight > 0 || (videoInputEnded && !video.ended)
+                    check(NativeLib.streamSignalWait(signal, if (codecPending) 5 else 50) >= 0) { "Viewer notification wait failed" }
                 }
-                Thread.sleep(5)
             }
         } catch (e: Exception) {
             if (running) {
@@ -369,6 +395,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
             catch (e: Exception) { LogManager.addLog("WARN", "video", "decoder release id=$id: ${e.message}") }
             try { playback?.close() }
             catch (e: Exception) { LogManager.addLog("WARN", "stream", "audio release id=$id: ${e.message}") }
+            NativeLib.streamSignalClose(signal)
             running = false
             LogManager.addLog("INFO", "stream", "viewer released id=$id inputs=${video.inputCount} " +
                 "decoded=${video.outputCount} shown=${video.shown} discarded=${video.discarded}")

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

@@ -0,0 +1,47 @@
+package com.utun.chat.data
+
+import java.nio.ByteBuffer
+import java.nio.ByteOrder
+
+/** Ближайшие записи viewer: порядок каждого трека независим, EOF не ждёт чужой трек.
+ * GAP/CLOSED отменяют прежние данные. Только player worker; ошибки логирует его владелец. */
+internal class StreamReadQueue {
+    private val records = mutableListOf<ByteArray>()
+    val size: Int get() = records.size
+    val videos: Int get() = records.count { it[0].toInt() == 1 }
+    val audios: Int get() = records.count { it[0].toInt() == 2 }
+    fun clear() = records.clear()
+
+    fun add(batch: ByteArray): Int {
+        val input = ByteBuffer.wrap(batch).order(ByteOrder.BIG_ENDIAN)
+        var count = 0
+        while (input.hasRemaining()) {
+            check(input.remaining() >= 28) { "Truncated stream record header" }
+            val start = input.position()
+            val kind = batch[start].toInt()
+            val length = input.getInt(start + 16)
+            check(kind in 1..8 && length >= 0 && length <= input.remaining() - 28) { "Invalid stream record kind=$kind length=$length" }
+            if (kind == 3 || kind == 6) records.clear()
+            records.add(batch.copyOfRange(start, start + 28 + length))
+            input.position(start + 28 + length)
+            check(++count <= 16) { "Stream batch exceeds record budget" }
+        }
+        return count
+    }
+
+    fun take(video: Boolean, audio: Boolean): ByteArray? {
+        var videoBefore = false
+        var audioBefore = false
+        for (i in records.indices) {
+            val eligible = when (records[i][0].toInt()) {
+                1 -> video.also { videoBefore = true }
+                2 -> audio.also { audioBefore = true }
+                7 -> !videoBefore
+                8 -> !audioBefore
+                else -> i == 0
+            }
+            if (eligible) return records.removeAt(i)
+        }
+        return null
+    }
+}

tools/chatgui-android/app/src/test/java/com/utun/chat/data/StreamReadQueueTest.kt

@@ -0,0 +1,55 @@
+package com.utun.chat.data
+
+import org.junit.Assert.*
+import org.junit.Test
+import java.nio.ByteBuffer
+
+class StreamReadQueueTest {
+    private fun record(kind: Int, pts: Long = 0, bytes: ByteArray = byteArrayOf()): ByteArray =
+        ByteBuffer.allocate(28 + bytes.size).apply {
+            put(0, kind.toByte()); putLong(8, pts); putInt(16, bytes.size); position(28); put(bytes)
+        }.array()
+
+    @Test fun videoEndPassesBlockedAudioButPreservesVideoOrder() {
+        val queue = StreamReadQueue()
+        queue.add(record(2, 20) + record(1, 40) + record(7) + record(8) + record(5))
+        assertNull(queue.take(video = false, audio = false))
+        assertEquals(1, queue.take(video = true, audio = false)!![0].toInt())
+        assertEquals(7, queue.take(video = false, audio = false)!![0].toInt())
+        assertNull(queue.take(video = false, audio = false))
+        assertEquals(2, queue.take(video = false, audio = true)!![0].toInt())
+        assertEquals(8, queue.take(video = false, audio = false)!![0].toInt())
+        assertEquals(5, queue.take(video = false, audio = false)!![0].toInt())
+        assertEquals(0, queue.size)
+    }
+
+    @Test fun audioEndPassesBlockedVideo() {
+        val queue = StreamReadQueue()
+        queue.add(record(1) + record(2) + record(8) + record(7) + record(5))
+        assertEquals(2, queue.take(video = false, audio = true)!![0].toInt())
+        assertEquals(8, queue.take(video = false, audio = false)!![0].toInt())
+        assertNull(queue.take(video = false, audio = false))
+        assertEquals(1, queue.take(video = true, audio = false)!![0].toInt())
+        assertEquals(7, queue.take(video = false, audio = false)!![0].toInt())
+        assertEquals(5, queue.take(video = false, audio = false)!![0].toInt())
+    }
+
+    @Test fun gapAndCloseImmediatelyCancelBlockedTracks() {
+        for (kind in listOf(3, 6)) {
+            val queue = StreamReadQueue()
+            queue.add(record(1) + record(2))
+            queue.add(record(kind))
+            assertEquals(kind, queue.take(video = false, audio = false)!![0].toInt())
+            assertEquals(0, queue.size)
+        }
+    }
+
+    @Test fun variablePayloadsKeepRecordBoundaries() {
+        val queue = StreamReadQueue()
+        val video = record(1, 42, byteArrayOf(1, 2, 3))
+        val audio = record(2, 43, byteArrayOf(4))
+        assertEquals(2, queue.add(video + audio))
+        assertArrayEquals(audio, queue.take(video = false, audio = true))
+        assertArrayEquals(video, queue.take(video = true, audio = false))
+    }
+}