Keep video clock moving through audio underruns and drain final playback

tests/test_android_stream.c

@@ -38,13 +38,13 @@ static void *producer(void *arg) {
 
 int main(void) {
     debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN); debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_INFO);
-    size_t before = u_get_allocated_count(); test_clock_use_managed(); ua = uasync_create(); CHECK(ua);
+    size_t before = u_get_allocated_count(); test_clock_init(100); ua = uasync_create(); CHECK(ua);
     struct UTUN_INSTANCE *inst = utun_instance_create_from_str(ua,
         "[global]\nauto_sockets=no\n[chatserver]\ngroup_autoconnect=0\nstorage_autoload=0\nmember_files_enabled=0\n");
     CHECK(inst && !utun_core_start(inst) && !chat_service_start(inst));
     chat_core_create_channel_auto(inst, "Stream \"test\" 😀");
     sqlite3_stmt *statement = NULL;
-    CHECK(sqlite3_prepare_v2(chat_core_get_db(inst), "SELECT ch_id FROM channels LIMIT 1", -1, &statement, NULL) == SQLITE_OK);
+    CHECK(sqlite3_prepare_v2(chat_core_get_db(inst), "SELECT channel_id FROM channels LIMIT 1", -1, &statement, NULL) == SQLITE_OK);
     CHECK(sqlite3_step(statement) == SQLITE_ROW);
     uint64_t group = strtoull((const char *)sqlite3_column_text(statement, 0), NULL, 10); sqlite3_finalize(statement); CHECK(group);
     uasync_mark_running(ua); android_stream_attach(inst);
@@ -75,7 +75,7 @@ int main(void) {
     u_free(wait_event(4, 5, &size));
     android_stream_detach(inst); uasync_mark_stopped(ua); utun_instance_destroy(inst); uasync_destroy(ua, 0);
     CHECK(u_get_allocated_count() == before);
-    test_clock_use_real();
+    test_clock_finish();
     DEBUG_INFO(DEBUG_CATEGORY_STREAM, "PASS Android stream: preview, source snapshot, FIN, cancellation, concurrent detach and no leaks");
     return 0;
 }

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

@@ -56,6 +56,9 @@ class StreamPlayerEngine(private val group: Long, private val source: Long) {
         var report = 0L
         var decoded = 0L
         var discarded = 0L
+        var lastHead = -1L
+        var headChangedNs = 0L
+        var endDrainUntil = 0L
 
         fun resetVideo() {
             try { decoder?.stop(); decoder?.release() }
@@ -66,15 +69,18 @@ class StreamPlayerEngine(private val group: Long, private val source: Long) {
         fun resetAudio() {
             track?.pause(); track?.flush(); track?.release(); track = null
             if (opus != 0L) NativeLib.streamOpusClose(opus)
-            opus = 0; audioBase = -1; written = 0; audios.clear()
+            opus = 0; audioBase = -1; written = 0; lastHead = -1; audios.clear()
         }
         fun currentPts(now: Long): Long {
             val t = track
             if (t != null && audioBase >= 0) {
                 val played = if (t.getTimestamp(audioTimestamp)) {
                     (audioTimestamp.framePosition + (now - audioTimestamp.nanoTime).coerceAtLeast(0) * 48000 / 1_000_000_000)
-                        .coerceAtMost(written)
-                } else t.playbackHeadPosition.toLong() and 0xffffffffL
+                } else {
+                    val head = t.playbackHeadPosition.toLong() and 0xffffffffL
+                    if (head != lastHead) { lastHead = head; headChangedNs = now }
+                    head + if (head >= written) (now - headChangedNs) * 48000 / 1_000_000_000 else 0
+                }
                 return audioBase + played * 1_000_000 / 48000
             }
             return if (videoBase >= 0) videoBase + (now - videoStartNs) / 1000 else 0
@@ -198,7 +204,10 @@ class StreamPlayerEngine(private val group: Long, private val source: Long) {
                         }
                     }
                 }
-                if (ended && videos.isEmpty() && audios.isEmpty()) { mutableStatus.value = R.string.stream_ended; break }
+                if (ended && videos.isEmpty() && audios.isEmpty()) {
+                    if (endDrainUntil == 0L) endDrainUntil = now + 250_000_000
+                    if (now >= endDrainUntil) { mutableStatus.value = R.string.stream_ended; break }
+                }
                 if (now - report >= 1_000_000_000) {
                     report = now
                     LogManager.addLog("DEBUG", "stream", "viewer id=$id decoded=$decoded discarded=$discarded videoQ=${videos.size}/$videoBytes " +
@@ -214,7 +223,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long) {
             try { track?.stop(); track?.release() } catch (e: Exception) { LogManager.addLog("WARN", "stream", "audio release id=$id: ${e.message}") }
             if (opus != 0L) NativeLib.streamOpusClose(opus)
             val deadline = System.nanoTime() + 500_000_000
-            while (NativeLib.isReady() && System.nanoTime() < deadline) {
+            while (!ended && NativeLib.isReady() && System.nanoTime() < deadline) {
                 val event = NativeLib.streamPull(id)
                 if (event != null && event[0].toInt() in 5..6) break
                 Thread.sleep(5)