Remove obsolete Android stream polling and simplify lifecycle tests

tools/chatgui-android/jni_bridge/android_stream_jni.c

@@ -37,15 +37,6 @@ JNIEXPORT void JNICALL Java_com_utun_chat_data_NativeLib_nativeStreamStop(JNIEnv
     (void)env; (void)obj; android_stream_stop((uint64_t)id);
 }
 
-JNIEXPORT jstring JNICALL Java_com_utun_chat_data_NativeLib_nativeStreamSources(JNIEnv *env, jobject obj, jlong group) {
-    (void)obj;
-    char *json = android_stream_sources((uint64_t)group);
-    if (!json) return NULL;
-    jstring output = (*env)->NewStringUTF(env, json); u_free(json);
-    if (!output) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "JNI stream source list string allocation failed");
-    return output;
-}
-
 JNIEXPORT jbyteArray JNICALL Java_com_utun_chat_data_NativeLib_nativeStreamPull(JNIEnv *env, jobject obj, jlong id) {
     (void)obj;
     size_t length = 0; uint8_t *data = android_stream_pull((uint64_t)id, &length);

tools/chatgui-android/libutun_lite/stream_bridge.c

@@ -1,15 +1,11 @@
 #include <pthread.h>
-#include <stdio.h>
 #include <string.h>
 
 #include "../../../lib/debug_config.h"
 #include "../../../lib/mem.h"
 #include "../../../lib/u_async.h"
-#include "../../../lib/sqlite3.h"
-#include "../../../lib/strbuf.h"
 #include "../../../src/utun_instance.h"
 #include "../../../src/chat/chat_core.h"
-#include "../../../src/routing_layer/topo_group.h"
 #include "../../../src/stream/stream.h"
 #include "../../../src/stream/stream_member.h"
 #include "../../../src/stream/stream_media.h"
@@ -43,10 +39,8 @@ static struct posted_task *media_post;
 static void *media_timer;
 static struct android_media_session publisher, viewer;
 static struct media_queue video_queue, audio_queue, output_queue;
-static uint64_t list_group, cached_group, list_updated;
 static uint64_t watch_group;
 static int watch_pending;
-static char *source_list;
 static void media_process(void *arg);
 static void media_close(struct android_media_session *s, int kind);
 
@@ -199,16 +193,6 @@ static int media_publish(void) {
     return STREAM_OK;
 }
 
-/* Список содержит только проверенные публикации живых мемберов; SQL и BGP читаются в uasync. */
-static void media_list_sources(void) {
-    uint64_t group_id = list_group; list_group = 0;
-    u_free(source_list); source_list = NULL;
-    int result = stream_sources_json(media_inst, group_id, &source_list);
-    cached_group = group_id; list_updated = get_time_us();
-    if (result < 0) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream source snapshot failed group=%llx result=%d",
-                              (unsigned long long)group_id, result);
-}
-
 /* Один actor обслуживает команды и не больше двух FEC-пачек за вызов; idle не будит мобильное ядро. */
 static void media_process(void *arg) {
     (void)arg;
@@ -220,7 +204,6 @@ static void media_process(void *arg) {
         if (stream_watch_sources(media_inst, watch_group) != STREAM_OK)
             DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream source watch failed group=%llx", (unsigned long long)watch_group);
     }
-    if (list_group) media_list_sources();
     struct android_media_session *sessions[] = {&publisher, &viewer};
     for (unsigned i = 0; i < 2; i++) {
         struct android_media_session *s = sessions[i];
@@ -316,13 +299,13 @@ void android_stream_detach(struct UTUN_INSTANCE *inst) {
         if (media_post) uasync_cancel_post(inst->ua, media_post);
         if (media_timer) uasync_cancel_timeout(inst->ua, media_timer);
         media_post = NULL; media_timer = NULL;
+        stream_watch_sources(inst, 0);
         stream_set_event_callback(inst, NULL, NULL);
         queue_clear(&output_queue);
         if (publisher.id) media_close(&publisher, 0);
         if (viewer.id) media_close(&viewer, 0);
         media_inst = NULL;
         watch_pending = 0; watch_group = 0;
-        u_free(source_list); source_list = NULL; list_group = cached_group = list_updated = 0;
         DEBUG_INFO(DEBUG_CATEGORY_STREAM, "Android stream bridge detached; producers rejected");
     }
     pthread_mutex_unlock(&media_lock);
@@ -425,15 +408,3 @@ uint8_t *android_stream_pull(uint64_t id, size_t *length) {
     pthread_mutex_unlock(&media_lock); return data;
 }
 
-/* Внешний поток не читает ни borrowed instance, ни SQLite; snapshot строится следующим actor-вызовом. */
-char *android_stream_sources(uint64_t group) {
-    pthread_mutex_lock(&media_lock);
-    if (group && media_inst && uasync_is_running(media_inst->ua)
-        && (group != cached_group || get_time_us() - list_updated >= 500000)) {
-        list_group = group;
-        if (media_schedule() < 0) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android source snapshot schedule failed");
-    }
-    char *copy = u_strdup(group == cached_group && source_list ? source_list : "[]");
-    if (!copy) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android source snapshot copy allocation failed");
-    pthread_mutex_unlock(&media_lock); return copy;
-}

tools/chatgui-android/libutun_lite/stream_bridge.h

@@ -24,7 +24,5 @@ int android_stream_feed(uint64_t id, int video, uint64_t pts, int key, const uin
 /* feed: 1 accepted, 0 dropped (video: запросить KEY), -1 detached/stopped/invalid. */
 void android_stream_stop(uint64_t id);
 uint8_t *android_stream_pull(uint64_t id, size_t *length);
-/* Кеш live-снимка источников: запрос в actor, чтение только копии под mutex; caller освобождает u_free(). */
-char *android_stream_sources(uint64_t group);
 
 #endif

tools/chatgui/tests/test_stream_view.cpp

@@ -35,6 +35,15 @@ class StreamViewTest : public QObject {
     QTimer poll;
     size_t allocations = 0;
 
+    bool receive(uint64_t id, StreamRecord& record) {
+        QElapsedTimer elapsed; elapsed.start();
+        while (elapsed.elapsed() < 1000) {
+            if (gui_stream_pull(id, record)) return true;
+            QTest::qWait(1);
+        }
+        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop test missing record id=%llu", (unsigned long long)id); return false;
+    }
+
     void publish(const QByteArray& video, const QByteArray& audio = {}) {
         stream_media_block media{}; media.format = {160, 90, 90, STREAM_MEDIA_H264, static_cast<uint8_t>(audio.isEmpty() ? 0 : STREAM_MEDIA_OPUS)};
         media.first_frame_offset = 0; media.video_count = 1;
@@ -53,6 +62,7 @@ class StreamViewTest : public QObject {
 private slots:
     void initTestCase() {
         debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN); debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_INFO);
+        debug_set_category_level(DEBUG_CATEGORY_SYS, DEBUG_LEVEL_ERROR);
         allocations = u_get_allocated_count(); ua = uasync_create(); QVERIFY(ua);
         inst = utun_instance_create_from_str(ua,
             "[global]\nauto_sockets=no\n[chatserver]\ngroup_autoconnect=0\nstorage_autoload=0\nmember_files_enabled=0\n");
@@ -81,7 +91,7 @@ private slots:
     void staleSelectionAndCancellation() {
         QVERIFY(gui_stream_open(100, group, inst->node_id, epoch ^ 1));
         StreamRecord event;
-        QTRY_VERIFY(gui_stream_pull(100, event)); QCOMPARE(event.kind, StreamRecordKind::Closed);
+        QVERIFY(receive(100, event)); QCOMPARE(event.kind, StreamRecordKind::Closed);
         QCOMPARE(event.reason, STREAM_CLOSE_SOURCE_CHANGED);
         QVERIFY(gui_stream_open(101, group, inst->node_id, epoch)); gui_stream_stop(101); QTest::qWait(20);
         stream_info info{}; QCOMPARE(stream_get_info(publisher, &info), STREAM_OK); QCOMPARE(info.owners, 1u);
@@ -89,16 +99,16 @@ private slots:
 
     void mediaAndPressure() {
         QVERIFY(gui_stream_open(102, group, inst->node_id, epoch)); StreamRecord event;
-        QTRY_VERIFY(gui_stream_pull(102, event)); QCOMPARE(event.kind, StreamRecordKind::Opened);
+        QVERIFY(receive(102, event)); QCOMPARE(event.kind, StreamRecordKind::Opened);
         const QByteArray frame(1500, 'k'); const QByteArray audio("opus"); publish(frame, audio);
-        QTRY_VERIFY(gui_stream_pull(102, event)); QCOMPARE(event.kind, StreamRecordKind::Video); QCOMPARE(event.data, frame);
+        QVERIFY(receive(102, event)); QCOMPARE(event.kind, StreamRecordKind::Video); QCOMPARE(event.data, frame);
         QCOMPARE(event.format.rotation, uint16_t(90));
-        QTRY_VERIFY(gui_stream_pull(102, event)); QCOMPARE(event.kind, StreamRecordKind::Audio); QCOMPARE(event.data, audio);
+        QVERIFY(receive(102, event)); QCOMPARE(event.kind, StreamRecordKind::Audio); QCOMPARE(event.data, audio);
         for (int i = 0; i < 270; i++) { publish(frame); uasync_poll(ua, 0); }
         bool gap = false; unsigned records = 0;
         while (gui_stream_pull(102, event)) { gap |= event.kind == StreamRecordKind::Gap; records++; }
         QVERIFY(gap); QVERIFY(records <= 256);
-        publish(frame); QTRY_VERIFY(gui_stream_pull(102, event)); QCOMPARE(event.kind, StreamRecordKind::Video);
+        publish(frame); QVERIFY(receive(102, event)); QCOMPARE(event.kind, StreamRecordKind::Video);
         gui_stream_stop(102); QTest::qWait(20);
         stream_info info{}; QCOMPARE(stream_get_info(publisher, &info), STREAM_OK); QCOMPARE(info.owners, 1u);
     }
@@ -137,7 +147,7 @@ private slots:
         QCOMPARE(stream_publish_open(inst, group, nullptr, &publisher), STREAM_OK);
         stream_info info{}; QCOMPARE(stream_get_info(publisher, &info), STREAM_OK); QVERIFY(info.descriptor.source_epoch != epoch);
         QVERIFY(gui_stream_open(103, group, inst->node_id, epoch)); StreamRecord event;
-        QTRY_VERIFY(gui_stream_pull(103, event)); QCOMPARE(event.reason, STREAM_CLOSE_SOURCE_CHANGED);
+        QVERIFY(receive(103, event)); QCOMPARE(event.reason, STREAM_CLOSE_SOURCE_CHANGED);
         QVERIFY(gui_stream_open(104, group, inst->node_id, info.descriptor.source_epoch));
         gui_stream_detach(inst); QVERIFY(!gui_stream_ready()); QTest::qWait(20);
         QCOMPARE(stream_get_info(publisher, &info), STREAM_OK); QCOMPARE(info.owners, 1u);