diff --git a/tools/chatgui-android/jni_bridge/android_stream_jni.c b/tools/chatgui-android/jni_bridge/android_stream_jni.c index f4e688b4..e0a23961 100644 --- a/tools/chatgui-android/jni_bridge/android_stream_jni.c +++ b/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); diff --git a/tools/chatgui-android/libutun_lite/stream_bridge.c b/tools/chatgui-android/libutun_lite/stream_bridge.c index 545f82bf..9cb32b62 100644 --- a/tools/chatgui-android/libutun_lite/stream_bridge.c +++ b/tools/chatgui-android/libutun_lite/stream_bridge.c @@ -1,15 +1,11 @@ #include -#include #include #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; -} diff --git a/tools/chatgui-android/libutun_lite/stream_bridge.h b/tools/chatgui-android/libutun_lite/stream_bridge.h index 0963b381..255de96b 100644 --- a/tools/chatgui-android/libutun_lite/stream_bridge.h +++ b/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 diff --git a/tools/chatgui/tests/test_stream_view.cpp b/tools/chatgui/tests/test_stream_view.cpp index 0855fb1f..182c82c1 100644 --- a/tools/chatgui/tests/test_stream_view.cpp +++ b/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(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);