From f91894aa1b9bb35d90ee1c7caa4274dea0080918 Mon Sep 17 00:00:00 2001 From: evgeny Date: Mon, 12 Oct 2026 21:57:20 +0200 Subject: [PATCH 01/22] Expose verified stream sources and channel watch events --- src/chat/chat_event.h | 1 + src/stream/stream.c | 118 +++++++++++++++++++++++++++++++++++++++++- src/stream/stream.h | 6 +++ 3 files changed, 124 insertions(+), 1 deletion(-) diff --git a/src/chat/chat_event.h b/src/chat/chat_event.h index 468500ca..1d3c9cf6 100644 --- a/src/chat/chat_event.h +++ b/src/chat/chat_event.h @@ -79,6 +79,7 @@ extern "C" { #define CHAT_EVT_RADIO_MUTE_RESULT 55 /* [group_id:8LE][node_id:8LE][success:1] — локальный переключатель рации */ #define CHAT_EVT_MSG_SUBMITTED 56 /* [request_id:8LE][success:1] — результат локального сохранения сообщения */ #define CHAT_EVT_CPU_LOAD 57 /* [valid:1][average_percent:1] — CPU потока uasync, среднее за секунду, 0..100 */ +#define CHAT_EVT_STREAM_SOURCES 58 /* [group_id:8BE][valid:1][JSON:utf8] — доступные публикации выбранного канала */ typedef void (*chat_event_handler_fn)(struct UTUN_INSTANCE* inst, int type, const uint8_t* data, int len); diff --git a/src/stream/stream.c b/src/stream/stream.c index 2565c0b2..d6cd9729 100644 --- a/src/stream/stream.c +++ b/src/stream/stream.c @@ -1,12 +1,16 @@ #include #include +#include #include "../../lib/debug_config.h" #include "../../lib/mem.h" #include "../../lib/platform_compat.h" #include "../../lib/u_async.h" +#include "../../lib/strbuf.h" +#include "../../lib/sqlite3.h" #include "../utun_instance.h" #include "../chat/chat_core.h" +#include "../chat/chat_event.h" #include "stream_member.h" #include "stream_source.h" #include "stream_priv.h" @@ -50,12 +54,112 @@ struct stream_service { stream_event_fn callback; void *arg; void *reconcile; + void *sources_pending; + uint64_t sources_group; + char *sources_snapshot; + int sources_valid; int stopping, dispatching, registered, initializing; }; static void session_deliver(void *arg); static void service_reconcile(void *arg); +/* Общий каталог не выделяет media-ресурсов: профиль проверяется тем же кодом, что и subscribe. */ +int stream_sources_json(struct UTUN_INSTANCE *inst, uint64_t group_id, char **out) { + if (!inst || !out || *out || !group_id) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source list invalid arguments"); return STREAM_ERR_ARGUMENT; + } + struct TOPO_GROUP *group = inst->topo_groups ? topo_groups_find(inst->topo_groups, group_id) : NULL; + if (!group || group->stopping || group->group_type != TOPO_GROUP_TYPE_CHAT) return STREAM_WAIT; + for (const char *p = group->channel_id; *p; p++) if (*p < '0' || *p > '9') { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source namespace invalid group=%llx", (unsigned long long)group_id); + return STREAM_ERR_STATE; + } + sqlite3 *db = chat_core_get_db(inst); + if (!db) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source list missing DB"); return STREAM_ERR_STATE; } + char query[320]; + snprintf(query, sizeof(query), "SELECT p.node_id,COALESCE(NULLIF(p.local_nick,''),n.name,'') " + "FROM \"peers_%s\" p LEFT JOIN nodes n ON p.node_id=n.node_id ORDER BY p.node_id", group->channel_id); + struct strbuf json = strbuf_new(); + sqlite3_stmt *statement = NULL; + int result = STREAM_OK, count = 0; + if (strbuf_addf(&json, "[") < 0) result = STREAM_ERR_MEMORY; + if (result == STREAM_OK && sqlite3_prepare_v2(db, query, -1, &statement, NULL) != SQLITE_OK) result = STREAM_ERR_STATE; + int step = SQLITE_DONE; + while (result == STREAM_OK && (step = sqlite3_step(statement)) == SQLITE_ROW) { + uint64_t node = (uint64_t)sqlite3_column_int64(statement, 0); + if (!stream_member_online(inst, group_id, node)) continue; + struct stream_member_state state; + int loaded = stream_member_load(inst, group_id, node, &state, NULL); + if (loaded == STREAM_WAIT || loaded == STREAM_ERR_NOT_MEMBER) continue; + if (loaded < 0) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "source excluded group=%llx node=%llx profile_result=%d", + (unsigned long long)group_id, (unsigned long long)node, loaded); continue; + } + if (!state.published) continue; + if (strbuf_addf(&json, "%s{\"id\":\"%llu\",\"epoch\":\"%016llx\",\"self\":%d,\"name\":", count ? "," : "", + (unsigned long long)node, (unsigned long long)state.source.source_epoch, node == inst->node_id) < 0 + || strbuf_add_json_string(&json, (const char *)sqlite3_column_text(statement, 1)) < 0 + || strbuf_addf(&json, "}") < 0) result = STREAM_ERR_MEMORY; + count++; + } + if (result == STREAM_OK && step != SQLITE_DONE) result = STREAM_ERR_STATE; + sqlite3_finalize(statement); + if (result == STREAM_OK && strbuf_addf(&json, "]") < 0) result = STREAM_ERR_MEMORY; + if (result == STREAM_OK) *out = strbuf_detach(&json); + strbuf_free(&json); + if (result != STREAM_OK) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source list failed group=%llx result=%d DB=%s", + (unsigned long long)group_id, result, sqlite3_errmsg(db)); + else DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream source list group=%llx count=%d", (unsigned long long)group_id, count); + return result; +} + +/* Один отложенный снимок объединяет Merkle/BGP-события; неизменный список повторно не отправляется. */ +static void sources_deliver(void *arg) { + struct stream_service *service = arg; + service->sources_pending = NULL; + if (service->stopping || !service->sources_group) return; + char *json = NULL; + int result = stream_sources_json(service->inst, service->sources_group, &json); + int valid = result == STREAM_OK || result == STREAM_WAIT; + const char *list = json ? json : "[]"; + if (service->sources_snapshot && valid == service->sources_valid && !strcmp(service->sources_snapshot, list)) { + u_free(json); return; + } + size_t length = strlen(list); + uint8_t *event = length <= INT_MAX - 9 ? u_malloc((uint32_t)(9 + length)) : NULL; + if (!event) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source event allocation failed bytes=%zu", length); u_free(json); return; + } + stream_write_be(event, service->sources_group, 8); event[8] = (uint8_t)valid; + memcpy(event + 9, list, length); + char *snapshot = json ? json : u_strdup(list); + if (!snapshot) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source snapshot allocation failed"); + u_free(service->sources_snapshot); service->sources_snapshot = snapshot; service->sources_valid = valid; + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream sources changed group=%llx valid=%d bytes=%zu", + (unsigned long long)service->sources_group, valid, length); + chat_event_post(service->inst, CHAT_EVT_STREAM_SOURCES, event, (int)(length + 9)); + u_free(event); +} + +/* Наблюдение принадлежит сервису и снимается до освобождения ядра. */ +int stream_watch_sources(struct UTUN_INSTANCE *inst, uint64_t group_id) { + struct stream_service *service = inst ? inst->stream : NULL; + if (!service || service->stopping) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream source watch without service group=%llx", (unsigned long long)group_id); + return STREAM_ERR_STATE; + } + service->sources_group = group_id; + u_free(service->sources_snapshot); service->sources_snapshot = NULL; + if (service->sources_pending) uasync_call_soon_cancel(inst->ua, service->sources_pending); + service->sources_pending = group_id ? uasync_call_soon(inst->ua, service, sources_deliver) : NULL; + DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream source watch group=%llx", (unsigned long long)group_id); + if (group_id && !service->sources_pending) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source watch scheduling failed"); return STREAM_ERR_MEMORY; + } + return STREAM_OK; +} + /* Observer выполняет только учёт; пользовательские callbacks и отправка остаются отложенными. */ static void cache_event(void *arg, const uint8_t *wire, size_t length, int added) { struct stream_session *s = arg; @@ -268,7 +372,13 @@ static void service_reconcile(void *arg) { static void member_changed(uint64_t node_id, const char *tags, const char *channel, void *arg) { (void)tags; struct stream_service *service = arg; - if (service->stopping || !service->sessions) return; + if (service->stopping) return; + struct TOPO_GROUP *watched = service->sources_group ? stream_group(service->inst, service->sources_group) : NULL; + if (watched && channel && !strcmp(watched->channel_id, channel) && !service->sources_pending) { + service->sources_pending = uasync_call_soon(service->inst->ua, service, sources_deliver); + if (!service->sources_pending) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source refresh scheduling failed"); + } + if (!service->sessions) return; stream_network_membership_changed(service->network, node_id, channel); for (struct stream_session *s = service->sessions; s; s = s->next) { int own = node_id == service->inst->node_id; @@ -344,6 +454,8 @@ void stream_service_destroy(struct UTUN_INSTANCE *inst) { stream_network_destroy(service->network); service->network = NULL; if (service->registered) member_sync_remove_props_cbk(inst, member_changed, service); if (service->reconcile) uasync_call_soon_cancel(inst->ua, service->reconcile); + if (service->sources_pending) uasync_call_soon_cancel(inst->ua, service->sources_pending); + u_free(service->sources_snapshot); while (service->sessions) { struct stream_session *s = service->sessions; if (!s->closing) session_close(s, STREAM_CLOSE_SERVICE, 1, 0); @@ -360,6 +472,10 @@ void stream_service_destroy(struct UTUN_INSTANCE *inst) { void stream_cancel_group(struct UTUN_INSTANCE *inst, uint64_t group_id) { struct stream_service *service = inst ? inst->stream : NULL; if (!service) return; + if (service->sources_group == group_id && !service->sources_pending) { + service->sources_pending = uasync_call_soon(inst->ua, service, sources_deliver); + if (!service->sources_pending) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream cancelled group source refresh failed"); + } struct stream_session *s = service->sessions; while (s) { struct stream_session *next = s->next; diff --git a/src/stream/stream.h b/src/stream/stream.h index 6f531ab5..3f6da37b 100644 --- a/src/stream/stream.h +++ b/src/stream/stream.h @@ -60,6 +60,12 @@ int stream_service_init(struct UTUN_INSTANCE *inst); void stream_service_destroy(struct UTUN_INSTANCE *inst); void stream_cancel_group(struct UTUN_INSTANCE *inst, uint64_t group_id); void stream_set_event_callback(struct UTUN_INSTANCE *inst, stream_event_fn callback, void *arg); +/* Проверенные online-публикации: JSON с id(decimal string), epoch(hex string), name, self. + * Только uasync; *out заранее NULL, результат освобождается u_free(). WAIT — группа недоступна. */ +int stream_sources_json(struct UTUN_INSTANCE *inst, uint64_t group_id, char **out); +/* Один выбранный канал frontend: начальный снимок и изменения через CHAT_EVT_STREAM_SOURCES. + * group_id=0 снимает наблюдение. Не создаёт подписку/буфер и не удерживает ACTIVE. Только uasync. */ +int stream_watch_sources(struct UTUN_INSTANCE *inst, uint64_t group_id); /* Локальный owner удерживает эффективный ACTIVE; желаемый frontend-режим ядро восстанавливает после последнего close. */ int stream_service_needs_activity(const struct UTUN_INSTANCE *inst); From ba3dfa9fcc6eb50f862017fdd307d962df59c0a2 Mon Sep 17 00:00:00 2001 From: evgeny Date: Mon, 12 Oct 2026 21:58:56 +0200 Subject: [PATCH 02/22] Watch Android stream sources and validate selected epochs --- .../jni_bridge/android_stream_jni.c | 9 +- .../libutun_lite/stream_bridge.c | 96 ++++++++++--------- .../libutun_lite/stream_bridge.h | 4 +- 3 files changed, 60 insertions(+), 49 deletions(-) diff --git a/tools/chatgui-android/jni_bridge/android_stream_jni.c b/tools/chatgui-android/jni_bridge/android_stream_jni.c index aef2002d..f4e688b4 100644 --- a/tools/chatgui-android/jni_bridge/android_stream_jni.c +++ b/tools/chatgui-android/jni_bridge/android_stream_jni.c @@ -11,9 +11,14 @@ struct android_stream_opus { opus_codec_encoder_t *encoder; opus_codec_decoder_t *decoder; }; JNIEXPORT jboolean JNICALL Java_com_utun_chat_data_NativeLib_nativeStreamOpen( - JNIEnv *env, jobject obj, jlong id, jlong group, jlong source, jint width, jint height, jint rotation, jboolean audio) { + JNIEnv *env, jobject obj, jlong id, jlong group, jlong source, jint width, jint height, jint rotation, jboolean audio, jlong epoch) { (void)env; (void)obj; - return android_stream_open((uint64_t)id, (uint64_t)group, (uint64_t)source, width, height, rotation, audio) == 0; + return android_stream_open((uint64_t)id, (uint64_t)group, (uint64_t)source, width, height, rotation, audio, (uint64_t)epoch) == 0; +} + +JNIEXPORT jboolean JNICALL Java_com_utun_chat_data_NativeLib_nativeStreamWatch(JNIEnv *env, jobject obj, jlong group) { + (void)env; (void)obj; + return android_stream_watch((uint64_t)group) == 0; } JNIEXPORT jint JNICALL Java_com_utun_chat_data_NativeLib_nativeStreamFeed( diff --git a/tools/chatgui-android/libutun_lite/stream_bridge.c b/tools/chatgui-android/libutun_lite/stream_bridge.c index d2a5526f..545f82bf 100644 --- a/tools/chatgui-android/libutun_lite/stream_bridge.c +++ b/tools/chatgui-android/libutun_lite/stream_bridge.c @@ -30,11 +30,11 @@ struct media_record { }; struct media_queue { struct media_record *head, *tail; size_t bytes; unsigned count; }; struct android_media_session { - uint64_t id, group, source, epoch, next_frame, last_pts[2], drain_until; + uint64_t id, group, source, epoch, expected_epoch, next_frame, last_pts[2], drain_until; struct STREAM_HANDLE *handle; struct stream_media_receiver *receiver; struct stream_media_format format; - int opening, stopping, finished, need_key, seen[2], rx_overflow; + int opening, stopping, finished, need_key, seen[2], rx_overflow, reason; }; static pthread_mutex_t media_lock = PTHREAD_MUTEX_INITIALIZER; @@ -44,6 +44,8 @@ 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); @@ -146,6 +148,7 @@ static void media_event(struct UTUN_INSTANCE *inst, const struct stream_event *e } else if (event->type == STREAM_EVENT_GAP) { stream_media_receiver_gap(viewer.receiver); media_output(viewer.id, 3, NULL, 0, 0, NULL, 0); } else { + viewer.reason = event->reason; media_close(&viewer, event->type == STREAM_EVENT_END ? 5 : 6); } } @@ -154,7 +157,7 @@ static void media_event(struct UTUN_INSTANCE *inst, const struct stream_event *e /* Publisher держит конечное окно выдачи после FIN; abrupt detach ничего не обещает получателям. */ static void media_close(struct android_media_session *s, int kind) { - if (kind) media_output(s->id, kind, NULL, 0, 0, NULL, 0); + if (kind) media_output(s->id, kind, NULL, 0, kind == 6 ? s->reason : 0, NULL, 0); if (s->handle) stream_close(s->handle); stream_media_receiver_destroy(s->receiver); if (s == &publisher) { queue_clear(&video_queue); queue_clear(&audio_queue); } @@ -199,44 +202,11 @@ static int media_publish(void) { /* Список содержит только проверенные публикации живых мемберов; SQL и BGP читаются в uasync. */ static void media_list_sources(void) { uint64_t group_id = list_group; list_group = 0; - struct TOPO_GROUP *group = media_inst->topo_groups ? topo_groups_find(media_inst->topo_groups, group_id) : NULL; - struct strbuf json = strbuf_new(); - sqlite3_stmt *statement = NULL; - int failed = strbuf_addf(&json, "[") < 0; - if (group && !failed) { - char query[256]; - for (const char *p = group->channel_id; *p; p++) if (*p < '0' || *p > '9') { - DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream source namespace invalid group=%llx", (unsigned long long)group_id); - failed = 1; break; - } - snprintf(query, sizeof(query), "SELECT p.node_id,COALESCE(NULLIF(p.local_nick,''),n.name,'') " - "FROM \"peers_%s\" p LEFT JOIN nodes n ON p.node_id=n.node_id", group->channel_id); - sqlite3 *db = chat_core_get_db(media_inst); - int result = failed ? SQLITE_ERROR : sqlite3_prepare_v2(db, query, -1, &statement, NULL), count = 0; - if (result == SQLITE_OK) { - while ((result = sqlite3_step(statement)) == SQLITE_ROW) { - uint64_t node = (uint64_t)sqlite3_column_int64(statement, 0); - if (!stream_member_online(media_inst, group_id, node)) continue; - struct stream_member_state state; - if (stream_member_load(media_inst, group_id, node, &state, NULL) != STREAM_OK || !state.published) continue; - if (count >= 256) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "Android stream source list truncated group=%llx", (unsigned long long)group_id); break; } - if (strbuf_addf(&json, "%s{\"id\":\"%llu\",\"epoch\":\"%016llx\",\"self\":%d,\"name\":", count ? "," : "", - (unsigned long long)node, (unsigned long long)state.source.source_epoch, node == media_inst->node_id) < 0 - || strbuf_add_json_string(&json, (const char *)sqlite3_column_text(statement, 1)) < 0 - || strbuf_addf(&json, "}") < 0) { failed = 1; break; } - count++; - } - } - if (result != SQLITE_DONE && !(result == SQLITE_ROW && (failed || count == 256))) { - DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream list query failed group=%llx: %s", - (unsigned long long)group_id, sqlite3_errmsg(db)); failed = 1; - } - } - sqlite3_finalize(statement); - if (strbuf_addf(&json, "]") < 0) failed = 1; - u_free(source_list); source_list = failed ? NULL : strbuf_detach(&json); - strbuf_free(&json); cached_group = group_id; list_updated = get_time_us(); - if (failed) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream source snapshot failed group=%llx", (unsigned long long)group_id); + 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 не будит мобильное ядро. */ @@ -245,6 +215,11 @@ static void media_process(void *arg) { pthread_mutex_lock(&media_lock); media_post = NULL; media_timer = NULL; if (!media_inst) { pthread_mutex_unlock(&media_lock); return; } + if (watch_pending) { + watch_pending = 0; + 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++) { @@ -252,13 +227,31 @@ static void media_process(void *arg) { if (s->opening) { if (s->stopping) { media_close(s, 5); continue; } s->opening = 0; - int result = s->source ? stream_subscribe_open(media_inst, s->group, s->source, &s->handle) - : stream_publish_open(media_inst, s->group, NULL, &s->handle); + int result = STREAM_OK; + s->reason = STREAM_CLOSE_FAILED; + if (s->source && s->expected_epoch) { + struct stream_member_state state; + result = stream_member_load(media_inst, s->group, s->source, &state, NULL); + if (result == STREAM_OK && (!state.published || state.source.source_epoch != s->expected_epoch)) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "Android selected source changed group=%llx source=%llx expected=%llx actual=%llx", + (unsigned long long)s->group, (unsigned long long)s->source, (unsigned long long)s->expected_epoch, + (unsigned long long)(state.published ? state.source.source_epoch : 0)); + result = STREAM_ERR_CONFLICT; s->reason = STREAM_CLOSE_SOURCE_CHANGED; + } + } + if (result == STREAM_OK) result = s->source ? stream_subscribe_open(media_inst, s->group, s->source, &s->handle) + : stream_publish_open(media_inst, s->group, NULL, &s->handle); if (result == STREAM_OK && s->source) result = stream_media_receiver_create(&s->receiver, media_frame, s); struct stream_info info; if (result == STREAM_OK) { result = stream_get_info(s->handle, &info); - if (result == STREAM_OK) s->epoch = info.descriptor.source_epoch; + if (result == STREAM_OK) { + s->epoch = info.descriptor.source_epoch; + if (s->expected_epoch && s->epoch != s->expected_epoch) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "Android viewer opened stale epoch id=%llu", (unsigned long long)s->id); + result = STREAM_ERR_CONFLICT; s->reason = STREAM_CLOSE_SOURCE_CHANGED; + } + } } if (result != STREAM_OK) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream open failed id=%llu group=%llx source=%llx result=%d", @@ -328,6 +321,7 @@ void android_stream_detach(struct UTUN_INSTANCE *inst) { 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"); } @@ -336,7 +330,7 @@ void android_stream_detach(struct UTUN_INSTANCE *inst) { /* Команда содержит идентификаторы, а не borrowed pointer ядра. */ int android_stream_open(uint64_t id, uint64_t group, uint64_t source, unsigned width, unsigned height, - unsigned rotation, int audio) { + unsigned rotation, int audio, uint64_t expected_epoch) { if (!id || !group || (!source && (!width || width > 4096 || !height || height > 4096 || rotation > 270 || rotation % 90))) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream open invalid arguments"); return -1; } @@ -346,13 +340,23 @@ int android_stream_open(uint64_t id, uint64_t group, uint64_t source, unsigned w if (s->id || id == publisher.id || id == viewer.id || !media_inst || !uasync_is_running(media_inst->ua)) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream open busy/detached id=%llu", (unsigned long long)id); } else if (media_schedule() == 0) { - *s = (struct android_media_session){.id = id, .group = group, .source = source, .opening = 1, .need_key = 1, + *s = (struct android_media_session){.id = id, .group = group, .source = source, .expected_epoch = expected_epoch, + .opening = 1, .need_key = 1, .format = {(uint16_t)width, (uint16_t)height, (uint16_t)rotation, STREAM_MEDIA_H264, audio ? STREAM_MEDIA_OPUS : 0}}; result = 0; } pthread_mutex_unlock(&media_lock); return result; } +/* Последняя выбранная группа заменяет ещё не обработанную команду, указатели ядра не покидают mutex. */ +int android_stream_watch(uint64_t group) { + pthread_mutex_lock(&media_lock); + int result = media_inst && uasync_is_running(media_inst->ua) ? media_schedule() : -1; + if (result == 0) { watch_group = group; watch_pending = 1; } + else DEBUG_WARN(DEBUG_CATEGORY_STREAM, "Android stream watch rejected group=%llx", (unsigned long long)group); + pthread_mutex_unlock(&media_lock); return result; +} + /* Перегрузка удаляет целые видео-кадры и блокирует P-кадры до нового KEY. */ int android_stream_feed(uint64_t id, int video, uint64_t pts, int key, const uint8_t *data, size_t length) { if (!data || !length || length > (video ? STREAM_MEDIA_MAX_FRAME : STREAM_MEDIA_MAX_OPUS) || pts > INT64_MAX) { diff --git a/tools/chatgui-android/libutun_lite/stream_bridge.h b/tools/chatgui-android/libutun_lite/stream_bridge.h index fbbcaa00..0963b381 100644 --- a/tools/chatgui-android/libutun_lite/stream_bridge.h +++ b/tools/chatgui-android/libutun_lite/stream_bridge.h @@ -16,7 +16,9 @@ struct UTUN_INSTANCE; int android_stream_attach(struct UTUN_INSTANCE *inst); void android_stream_detach(struct UTUN_INSTANCE *inst); int android_stream_open(uint64_t id, uint64_t group, uint64_t source, unsigned width, unsigned height, - unsigned rotation, int audio); + unsigned rotation, int audio, uint64_t expected_epoch); +/* Наблюдение выбранного канала через CHAT_EVT_STREAM_SOURCES; group=0 снимает его. */ +int android_stream_watch(uint64_t group); /* source=0 — publisher; source!=0 — viewer. 0 означает, что команда принята, результат приходит через pull. */ int android_stream_feed(uint64_t id, int video, uint64_t pts, int key, const uint8_t *data, size_t length); /* feed: 1 accepted, 0 dropped (video: запросить KEY), -1 detached/stopped/invalid. */ From ba3dd70f6cf1555526bea57bb48ef1595eda58c7 Mon Sep 17 00:00:00 2001 From: evgeny Date: Mon, 12 Oct 2026 22:00:12 +0200 Subject: [PATCH 03/22] Keep Android live sources in reactive channel state --- .../main/java/com/utun/chat/data/NativeLib.kt | 8 +++- .../java/com/utun/chat/data/StreamSource.kt | 21 ++++++++++ .../com/utun/chat/viewmodel/ChatViewModel.kt | 39 ++++++++++++++++++- 3 files changed, 65 insertions(+), 3 deletions(-) create mode 100644 tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamSource.kt diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/NativeLib.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/NativeLib.kt index 6654f487..eda6c8b5 100644 --- a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/NativeLib.kt +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/NativeLib.kt @@ -199,7 +199,9 @@ object NativeLib { /* Live media используют отдельную ограниченную очередь, минуя чатовый event bus. */ fun streamOpen(id: Long, group: Long, source: Long = 0, width: Int = 0, height: Int = 0, - rotation: Int = 0, audio: Boolean = true): Boolean = nativeStreamOpen(id, group, source, width, height, rotation, audio) + rotation: Int = 0, audio: Boolean = true, epoch: Long = 0): Boolean = + nativeStreamOpen(id, group, source, width, height, rotation, audio, epoch) + fun streamWatch(group: Long): Boolean = nativeStreamWatch(group) fun streamFeed(id: Long, video: Boolean, pts: Long, key: Boolean, data: ByteArray): Int = nativeStreamFeed(id, video, pts, key, data) fun streamStop(id: Long) { nativeStreamStop(id) } fun streamPull(id: Long): ByteArray? = nativeStreamPull(id) @@ -208,7 +210,9 @@ object NativeLib { fun streamOpusEncode(handle: Long, pcm: ShortArray): ByteArray? = nativeStreamOpusEncode(handle, pcm) fun streamOpusDecode(handle: Long, packet: ByteArray): ShortArray? = nativeStreamOpusDecode(handle, packet) fun streamOpusClose(handle: Long) { nativeStreamOpusClose(handle) } - private external fun nativeStreamOpen(id: Long, group: Long, source: Long, width: Int, height: Int, rotation: Int, audio: Boolean): Boolean + private external fun nativeStreamOpen(id: Long, group: Long, source: Long, width: Int, height: Int, rotation: Int, + audio: Boolean, epoch: Long): Boolean + private external fun nativeStreamWatch(group: Long): Boolean private external fun nativeStreamFeed(id: Long, video: Boolean, pts: Long, key: Boolean, data: ByteArray): Int private external fun nativeStreamStop(id: Long) private external fun nativeStreamPull(id: Long): ByteArray? diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamSource.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamSource.kt new file mode 100644 index 00000000..14c37097 --- /dev/null +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamSource.kt @@ -0,0 +1,21 @@ +package com.utun.chat.data + +import org.json.JSONArray + +/** Источник идентифицируется участником и эпохой; IDs сохраняют все 64 бита. */ +data class StreamSource(val id: Long, val name: String, val epoch: Long, val self: Boolean) { + val displayName: String get() = name.ifEmpty { id.toULong().toString() } + + companion object { + fun parse(json: String): List { + val array = JSONArray(json) + return (0 until array.length()).map { i -> + val item = array.getJSONObject(i) + val id = item.getString("id").toULong().toLong() + val epoch = item.getString("epoch").toULong(16).toLong() + require(id != 0L && epoch != 0L) { "Zero stream identity" } + StreamSource(id, item.getString("name"), epoch, item.getInt("self") == 1) + } + } + } +} diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/viewmodel/ChatViewModel.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/viewmodel/ChatViewModel.kt index 20488d8a..9dfb2065 100644 --- a/tools/chatgui-android/app/src/main/java/com/utun/chat/viewmodel/ChatViewModel.kt +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/viewmodel/ChatViewModel.kt @@ -220,6 +220,23 @@ class ChatViewModel : ViewModel() { private val _currentChannel = MutableStateFlow(null) val currentChannel: StateFlow = _currentChannel + private val _streamSources = MutableStateFlow>(emptyList()) + val streamSources: StateFlow> = _streamSources + private val _streamSourcesFailed = MutableStateFlow(false) + val streamSourcesFailed: StateFlow = _streamSourcesFailed + private val _streamSelection = MutableStateFlow(null) + val streamSelection: StateFlow = _streamSelection + + fun selectStream(source: StreamSource?) { + _streamSelection.value = source + LogManager.addLog("INFO", "stream", "selected channel=${_currentChannel.value?.id} source=${source?.id} epoch=${source?.epoch}") + } + + private fun watchStreams() { + if (!NativeLib.isReady()) return + val group = if (_foreground.value) _currentChannel.value?.id?.toULongOrNull()?.toLong() ?: 0L else 0L + if (!NativeLib.streamWatch(group)) LogManager.addLog("WARN", "stream", "source watch rejected group=$group") + } private val _soundOn = MutableStateFlow(true) val soundOn: StateFlow = _soundOn @@ -320,6 +337,7 @@ class ChatViewModel : ViewModel() { fun setForeground(fg: Boolean) { _foreground.value = fg + watchStreams() if (fg) { _isStuck.value = false if (dbReady && repo != null) { @@ -533,8 +551,9 @@ class ChatViewModel : ViewModel() { if (result == 0) refreshChannels() } 4 -> { if (repo != null) { LogManager.addLog("INFO", "VM", "CHANNEL_UPDATED — refreshChannels"); refreshChannels() } else { LogManager.addLog("WARN", "VM", "CHANNEL_UPDATED skipped — repo is NULL") } } - 12 -> LogManager.addLog("INFO", "VM", "service started") + 12 -> { LogManager.addLog("INFO", "VM", "service started"); watchStreams() } 13 -> { + _streamSources.value = emptyList(); _streamSourcesFailed.value = false; _streamSelection.value = null for (id in submissions.keys.toList()) submissionResult(id, false) _radioTalkState.value = RadioTalkState() _radioTalking.value = null @@ -641,6 +660,21 @@ class ChatViewModel : ViewModel() { val cur = _currentChannel.value if (cur != null) refreshMembers(cur.id) } + 58 -> { /* STREAM_SOURCES: group:8BE, valid:1, JSON. Старый канал не меняет новую шапку. */ + if (data == null || data.size < 11 || data[8].toInt() !in 0..1) { + LogManager.addLog("ERROR", "stream", "invalid source snapshot bytes=${data?.size}"); return + } + val group = java.nio.ByteBuffer.wrap(data).order(java.nio.ByteOrder.BIG_ENDIAN).long + if (_currentChannel.value?.id?.toULongOrNull()?.toLong() != group || !_foreground.value) return + try { + _streamSourcesFailed.value = data[8].toInt() == 0 + _streamSources.value = StreamSource.parse(String(data, 9, data.size - 9, Charsets.UTF_8)) + LogManager.addLog("INFO", "stream", "sources group=$group count=${_streamSources.value.size} valid=${!_streamSourcesFailed.value}") + } catch (e: Exception) { + _streamSources.value = emptyList(); _streamSourcesFailed.value = true + LogManager.addLog("ERROR", "stream", "source snapshot parse group=$group: ${e.message}") + } + } 19 -> { if (repo != null) refreshLocalSockets() } /* LOCAL_SOCKETS */ 23 -> { /* MEMBER_LIST: live-снимок дополняет расширенный JSON-список. */ if (data == null || data.size < 3) { @@ -1215,6 +1249,7 @@ class ChatViewModel : ViewModel() { fun selectChannel(channel: Channel) { val r = repo ?: return if (_currentChannel.value?.id != channel.id) { + _streamSources.value = emptyList(); _streamSourcesFailed.value = false; _streamSelection.value = null _historyAnchor.value = "" _messageJump.value = null pendingLatestJump = null @@ -1224,6 +1259,7 @@ class ChatViewModel : ViewModel() { memberDetailJob?.cancel(); memberDetailTarget = null; _selectedMember.value = null } _currentChannel.value = channel + watchStreams() _radioActive.value = (radioChannelId == channel.id) _radioTalking.value = if (_radioActive.value) _radioTalkState.value.speakers.firstOrNull() else null _soundOn.value = NativeLib.getSoundOnMessage(channel.id) @@ -1783,6 +1819,7 @@ class ChatViewModel : ViewModel() { _channels.value = emptyList() _messages.value = emptyList() _currentChannel.value = null + _streamSources.value = emptyList(); _streamSelection.value = null; _streamSourcesFailed.value = false _generatingKeys.value = false _localSockets.value = emptyList() _channelOwner.value = false From d5ed19ee349e3de16c7701cb293c6d99eb2b90ca Mon Sep 17 00:00:00 2001 From: evgeny Date: Mon, 12 Oct 2026 22:01:38 +0200 Subject: [PATCH 04/22] Open Android live streams from the channel header --- .../main/java/com/utun/chat/MainActivity.kt | 10 +++- .../com/utun/chat/ui/screens/ChatScreen.kt | 29 +++++++++- .../com/utun/chat/ui/screens/StreamScreen.kt | 53 +++++++------------ 3 files changed, 56 insertions(+), 36 deletions(-) diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/MainActivity.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/MainActivity.kt index 139de16e..4f379d38 100644 --- a/tools/chatgui-android/app/src/main/java/com/utun/chat/MainActivity.kt +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/MainActivity.kt @@ -394,7 +394,8 @@ class MainActivity : AppCompatActivity() { }, onInvite = { showInvite(channel) }, onBack = { screen = Screen.ChannelList }, - onStreams = { screen = Screen.Streams }, + onStreams = { vm.selectStream(null); screen = Screen.Streams }, + onWatchStream = { vm.selectStream(it); screen = Screen.Streams }, onPlayVideo = { msg -> videoMessage = msg }, onViewPhoto = { msg -> photoMessage = msg }, onRecordVideo = { @@ -412,7 +413,12 @@ class MainActivity : AppCompatActivity() { } Screen.Streams -> { vm.currentChannel.collectAsState().value?.let { channel -> - StreamScreen(channel.id, channel.name) { screen = Screen.Chat } + val sources by vm.streamSources.collectAsState() + val selected by vm.streamSelection.collectAsState() + val failed by vm.streamSourcesFailed.collectAsState() + StreamScreen(channel.id, channel.name, sources, selected, failed, vm::selectStream) { + vm.selectStream(null); screen = Screen.Chat + } } } Screen.MemberList -> { diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/ui/screens/ChatScreen.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/ui/screens/ChatScreen.kt index 9cf31f29..70fe6b3f 100644 --- a/tools/chatgui-android/app/src/main/java/com/utun/chat/ui/screens/ChatScreen.kt +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/ui/screens/ChatScreen.kt @@ -70,6 +70,8 @@ import com.utun.chat.ui.components.GroupAvatar import com.utun.chat.ui.components.GroupPropertiesDialog import com.utun.chat.data.ConversationTarget import com.utun.chat.viewmodel.ChatViewModel +import com.utun.chat.data.StreamSource +import androidx.compose.material.icons.filled.LiveTv import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.launch @@ -89,7 +91,8 @@ fun ChatScreen( onViewPhoto: (Message) -> Unit = {}, onRecordVideo: () -> Unit = {}, onCapturePhoto: () -> Unit = {}, - onStreams: () -> Unit = {} + onStreams: () -> Unit = {}, + onWatchStream: (StreamSource) -> Unit = {} ) { val listState = rememberLazyListState() val context = LocalContext.current @@ -98,6 +101,23 @@ fun ChatScreen( val scope = rememberCoroutineScope() var properties by remember(channel.id) { mutableStateOf(false) } var menu by remember(channel.id) { mutableStateOf(false) } + val streams by viewModel.streamSources.collectAsState() + var chooseStream by remember(channel.id) { mutableStateOf(false) } + if (chooseStream && streams.size > 1) AlertDialog( + onDismissRequest = { chooseStream = false }, + title = { Text(stringResource(R.string.stream_choose_source)) }, + text = { + Column { + streams.forEach { source -> + TextButton(onClick = { chooseStream = false; onWatchStream(source) }) { + Text("${source.displayName} · ${source.id.toULong()}") + } + } + } + }, + confirmButton = { TextButton(onClick = { chooseStream = false }) { Text(stringResource(R.string.cancel)) } } + ) + LaunchedEffect(streams) { if (streams.size <= 1) chooseStream = false } /* рация */ val radioActive by viewModel.radioActive.collectAsState() @@ -172,6 +192,13 @@ fun ChatScreen( IconButton(onClick = onBack) { Icon(Icons.AutoMirrored.Filled.ArrowBack, stringResource(R.string.back)) } }, actions = { + if (streams.isNotEmpty()) IconButton(onClick = { + if (streams.size == 1) onWatchStream(streams.single()) else chooseStream = true + }) { + BadgedBox(badge = { if (streams.size > 1) Badge { Text(streams.size.toString()) } }) { + Icon(Icons.Filled.LiveTv, stringResource(R.string.stream_available, streams.size), tint = Color(0xFFE53935)) + } + } IconButton( onClick = { viewModel.toggleRadio() }, modifier = Modifier.background( diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/ui/screens/StreamScreen.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/ui/screens/StreamScreen.kt index d8cdef33..3e22c3e6 100644 --- a/tools/chatgui-android/app/src/main/java/com/utun/chat/ui/screens/StreamScreen.kt +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/ui/screens/StreamScreen.kt @@ -32,23 +32,21 @@ import com.utun.chat.data.NativeLib import com.utun.chat.data.StreamController import com.utun.chat.data.StreamPhase import com.utun.chat.data.StreamPlayerEngine +import com.utun.chat.data.StreamSource import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.delay import kotlinx.coroutines.withContext import org.json.JSONArray -private data class LiveSource(val id: Long, val name: String, val epoch: String, val self: Boolean) - /** Выход с экрана закрывает только reader; foreground publisher останавливает явная кнопка/notification. */ @OptIn(ExperimentalMaterial3Api::class) @Composable -fun StreamScreen(channelId: String, channelName: String, onBack: () -> Unit) { +fun StreamScreen(channelId: String, channelName: String, sources: List, selected: StreamSource?, + sourcesFailed: Boolean, onSelect: (StreamSource?) -> Unit, onBack: () -> Unit) { val context = LocalContext.current val group = remember(channelId) { channelId.toULongOrNull()?.toLong() ?: 0L } val state by StreamController.state.collectAsState() val local = state?.takeIf { it.group == group } - var sources by remember(group) { mutableStateOf>(emptyList()) } - var selected by remember(group) { mutableStateOf(null) } var previewedSession by remember(group) { mutableStateOf(null) } var audio by remember { mutableStateOf(true) } var camera by remember { mutableStateOf("") } @@ -64,24 +62,10 @@ fun StreamScreen(channelId: String, channelName: String, onBack: () -> Unit) { val cameraGranted = permissionVersion.let { ContextCompat.checkSelfPermission(context, Manifest.permission.CAMERA) == PackageManager.PERMISSION_GRANTED } val micGranted = permissionVersion.let { ContextCompat.checkSelfPermission(context, Manifest.permission.RECORD_AUDIO) == PackageManager.PERMISSION_GRANTED } - LaunchedEffect(group) { - while (true) { - if (NativeLib.isReady()) sources = withContext(Dispatchers.IO) { - try { - val json = JSONArray(NativeLib.streamSources(group)) - (0 until json.length()).map { i -> - val member = json.getJSONObject(i) - LiveSource(member.getString("id").toULong().toLong(), member.getString("name"), member.getString("epoch"), member.optInt("self") == 1) - } - } catch (e: Exception) { LogManager.addLog("WARN", "stream", "stream list: ${e.message}"); emptyList() } - } else sources = emptyList() - delay(1000) - } - } LaunchedEffect(sources, local?.id, local?.phase) { - if (local != null && !local.ongoing && selected?.self == true) selected = null + if (local != null && !local.ongoing && selected?.self == true) onSelect(null) if (local?.phase == StreamPhase.LIVE && previewedSession != local.id) sources.firstOrNull { it.self }?.let { - selected = it; previewedSession = local.id + onSelect(it); previewedSession = local.id } } @@ -92,6 +76,14 @@ fun StreamScreen(channelId: String, channelName: String, onBack: () -> Unit) { }) { padding -> Column(Modifier.fillMaxSize().padding(padding).padding(12.dp), verticalArrangement = Arrangement.spacedBy(8.dp)) { Text(channelName, style = MaterialTheme.typography.titleMedium) + if (selected != null) { + Text(selected.displayName, style = MaterialTheme.typography.titleSmall) + Box(Modifier.weight(1f).fillMaxWidth()) { + key(group, selected.id, selected.epoch) { LiveStreamView(group, selected.id, selected.epoch, selected.self) } + } + TextButton(onClick = { onSelect(null); onBack() }) { Text(stringResource(R.string.stream_close_view)) } + return@Column + } if (local?.ongoing == true) { Text(stringResource(when (local.phase) { StreamPhase.LIVE -> R.string.stream_live @@ -117,24 +109,19 @@ fun StreamScreen(channelId: String, channelName: String, onBack: () -> Unit) { }, enabled = camera.isNotEmpty() && group != 0L && state?.ongoing != true) { Text(stringResource(R.string.stream_start)) } } local?.error?.let { Text(stringResource(it), color = MaterialTheme.colorScheme.error) } - if (sources.isEmpty() && local?.ongoing != true) Text(stringResource(R.string.stream_none)) + if (sourcesFailed) Text(stringResource(R.string.stream_list_failed), color = MaterialTheme.colorScheme.error) + else if (sources.isEmpty() && local?.ongoing != true) Text(stringResource(R.string.stream_none)) Column(Modifier.fillMaxWidth().heightIn(max = 160.dp).verticalScroll(rememberScrollState())) { sources.forEach { source -> - TextButton(onClick = { selected = source }) { Text(stringResource(R.string.stream_watch, source.name.ifEmpty { source.id.toULong().toString() })) } - } - } - selected?.let { source -> - Box(Modifier.weight(1f).fillMaxWidth()) { - key(group, source.id, source.epoch) { LiveStreamView(group, source.id, source.self) } + TextButton(onClick = { onSelect(source) }) { Text(stringResource(R.string.stream_watch, source.displayName)) } } - TextButton(onClick = { selected = null }) { Text(stringResource(R.string.stream_close_view)) } } } } } @Composable -private fun LiveStreamView(group: Long, source: Long, localPreview: Boolean) { +private fun LiveStreamView(group: Long, source: Long, epoch: Long, localPreview: Boolean) { val owner = LocalLifecycleOwner.current var visible by remember(owner) { mutableStateOf(owner.lifecycle.currentState.isAtLeast(Lifecycle.State.STARTED)) } DisposableEffect(owner) { @@ -145,12 +132,12 @@ private fun LiveStreamView(group: Long, source: Long, localPreview: Boolean) { owner.lifecycle.addObserver(observer) onDispose { owner.lifecycle.removeObserver(observer) } } - if (visible) LiveStreamSurface(group, source, localPreview) + if (visible) LiveStreamSurface(group, source, epoch, localPreview) } @Composable -private fun LiveStreamSurface(group: Long, source: Long, localPreview: Boolean) { - val engine = remember(group, source, localPreview) { StreamPlayerEngine(group, source, localPreview) } +private fun LiveStreamSurface(group: Long, source: Long, epoch: Long, localPreview: Boolean) { + val engine = remember(group, source, epoch, localPreview) { StreamPlayerEngine(group, source, epoch, localPreview) } val aspect by engine.aspect.collectAsState() val status by engine.status.collectAsState() var surface by remember { mutableStateOf(null) } From e3d9ac9d31822df60c5bcaa08ed0fea4afce0db5 Mon Sep 17 00:00:00 2001 From: evgeny Date: Mon, 12 Oct 2026 22:01:38 +0200 Subject: [PATCH 05/22] Report Android stream close reasons and epoch changes --- .../com/utun/chat/data/StreamPlayerEngine.kt | 19 +++++++++++++++---- .../app/src/main/res/values-ru/strings.xml | 6 ++++++ .../app/src/main/res/values/strings.xml | 6 ++++++ 3 files changed, 27 insertions(+), 4 deletions(-) diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamPlayerEngine.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamPlayerEngine.kt index 262debcb..75f69a54 100644 --- a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamPlayerEngine.kt +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamPlayerEngine.kt @@ -20,7 +20,8 @@ import kotlin.math.roundToInt /** Один worker владеет декодерами и AudioTrack. Hardware audio clock управляет временем показа видео. * Просроченные P-кадры декодируются без показа, чтобы сохранить references; при переполнении ждём KEY. * viewerLock сериализует stop/open разных экранов до подтверждения освобождения native owner. */ -class StreamPlayerEngine(private val group: Long, private val source: Long, private val muteAudio: Boolean = false) { +class StreamPlayerEngine(private val group: Long, private val source: Long, private val epoch: Long, + private val muteAudio: Boolean = false) { companion object { private val viewerLock = ReentrantLock() } private data class Frame(val pts: Long, val key: Boolean, val width: Int, val height: Int, val rotation: Int, val data: ByteArray) private val mutableAspect = MutableStateFlow(16f / 9f) @@ -120,7 +121,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv } } try { - check(NativeLib.streamOpen(id, group, source)) { "Viewer open refused" } + check(NativeLib.streamOpen(id, group, source, epoch = epoch)) { "Viewer open refused" } val openDeadline = System.nanoTime() + 10_000_000_000 var opened = false var prefillUntil = 0L @@ -131,7 +132,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv if (raw != null) { check(raw.size >= 20) val b = ByteBuffer.wrap(raw).order(ByteOrder.BIG_ENDIAN) - val kind = b.get().toInt(); val key = b.get().toInt() and 1 != 0 + val kind = b.get().toInt(); val flags = b.get().toInt() and 255; val key = flags and 1 != 0 val width = b.short.toInt() and 65535; val height = b.short.toInt() and 65535 val rotation = b.short.toInt() and 65535; val pts = b.long; val size = b.int check(size == b.remaining()) @@ -161,7 +162,17 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv 3 -> { LogManager.addLog("INFO", "stream", "viewer GAP id=$id"); resetVideo(); resetAudio(); videoBase = -1; prefillUntil = 0 } 4 -> opened = true 5 -> ended = true - 6 -> error("Native viewer closed") + 6 -> { + mutableStatus.value = when (flags) { + 1 -> R.string.stream_group_closed + 2 -> R.string.chat_service_stopped + 3 -> R.string.stream_source_changed + 4 -> R.string.stream_member_removed + else -> R.string.stream_failed + } + LogManager.addLog("WARN", "stream", "native viewer closed id=$id group=$group source=$source epoch=$epoch reason=$flags") + ended = true; running = false + } else -> error("Unknown media event=$kind") } } diff --git a/tools/chatgui-android/app/src/main/res/values-ru/strings.xml b/tools/chatgui-android/app/src/main/res/values-ru/strings.xml index 63b76696..4ae65ff1 100644 --- a/tools/chatgui-android/app/src/main/res/values-ru/strings.xml +++ b/tools/chatgui-android/app/src/main/res/values-ru/strings.xml @@ -1,6 +1,12 @@ Трансляции + Трансляции: %1$d + Выберите трансляцию + Не удалось получить список трансляций + Трансляция перезапущена или завершена. Выберите источник заново. + Источник больше не является участником канала + Канал закрыт Начать трансляцию Остановить трансляцию Прямая трансляция diff --git a/tools/chatgui-android/app/src/main/res/values/strings.xml b/tools/chatgui-android/app/src/main/res/values/strings.xml index e09b866d..c451e585 100644 --- a/tools/chatgui-android/app/src/main/res/values/strings.xml +++ b/tools/chatgui-android/app/src/main/res/values/strings.xml @@ -1,6 +1,12 @@ Live streams + Live streams: %1$d + Choose a stream + Could not load live streams + The stream has restarted or ended. Select a source again. + The source is no longer a channel member + The channel is closed Start stream Stop stream Streaming live From d9821fa89cb36284c212a124ad564d3aa7e2d4ac Mon Sep 17 00:00:00 2001 From: evgeny Date: Mon, 12 Oct 2026 22:07:35 +0200 Subject: [PATCH 06/22] Add bounded desktop live media bridge and source lifecycle tests --- tests/test_android_stream.c | 36 +++- tools/chatgui/transport/stream_bridge.cpp | 200 ++++++++++++++++++++++ tools/chatgui/transport/stream_bridge.h | 26 +++ 3 files changed, 258 insertions(+), 4 deletions(-) create mode 100644 tools/chatgui/transport/stream_bridge.cpp create mode 100644 tools/chatgui/transport/stream_bridge.h diff --git a/tests/test_android_stream.c b/tests/test_android_stream.c index 0f8ee5bc..2826ab62 100644 --- a/tests/test_android_stream.c +++ b/tests/test_android_stream.c @@ -7,6 +7,9 @@ #include "../lib/mem.h" #include "../lib/sqlite3.h" #include "../src/chat/chat_core.h" +#include "../src/chat/chat_event.h" +#include "../src/stream/stream.h" +#include "../src/stream/stream_member.h" #include "../src/utun_instance.h" #include "../tools/chatgui-android/libutun_lite/stream_bridge.h" #include "../src/stream/stream_proto.h" @@ -16,6 +19,18 @@ static uint8_t video[9000]; static struct UASYNC *ua; static int producer_started, producer_stop; +static unsigned source_events; +static uint64_t source_group; +static char *source_snapshot; + +static void chat_event(struct UTUN_INSTANCE *inst, int type, const uint8_t *data, int length) { + (void)inst; + if (type != CHAT_EVT_STREAM_SOURCES) return; + CHECK(length >= 11 && data[8] == 1); + source_group = stream_read_be(data, 8); source_events++; + u_free(source_snapshot); source_snapshot = u_malloc((uint32_t)(length - 8)); CHECK(source_snapshot); + memcpy(source_snapshot, data + 9, (size_t)(length - 9)); source_snapshot[length - 9] = '\0'; +} static uint8_t *wait_event(uint64_t id, int kind, size_t *size) { for (unsigned i = 0; i < 1000; i++) { @@ -48,11 +63,22 @@ int main(void) { 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); - CHECK(android_stream_open(1, group, 0, 1280, 720, 90, 1) == 0); + chat_event_set_handler(inst, chat_event); + CHECK(android_stream_watch(group) == 0); test_clock_poll(ua, 100); + CHECK(source_events == 1 && source_group == group && !strcmp(source_snapshot, "[]")); + CHECK(!stream_service_needs_activity(inst)); + CHECK(android_stream_open(1, group, 0, 1280, 720, 90, 1, 0) == 0); size_t size = 0; u_free(wait_event(1, 4, &size)); char *sources = android_stream_sources(group); u_free(sources); test_clock_poll(ua, 100); sources = android_stream_sources(group); CHECK(strstr(sources, "epoch") && strstr(sources, "name")); u_free(sources); - CHECK(android_stream_open(2, group, inst->node_id, 0, 0, 0, 1) == 0); u_free(wait_event(2, 4, &size)); + CHECK(source_events == 2 && strstr(source_snapshot, "epoch")); + struct stream_member_state member; + CHECK(stream_member_load(inst, group, inst->node_id, &member, NULL) == STREAM_OK && member.published); + CHECK(android_stream_open(5, group, inst->node_id, 0, 0, 0, 1, member.source.source_epoch ^ 1) == 0); + uint8_t *error = wait_event(5, 6, &size); CHECK(error[1] == STREAM_CLOSE_SOURCE_CHANGED); u_free(error); + CHECK(android_stream_open(2, group, inst->node_id, 0, 0, 0, 1, member.source.source_epoch) == 0); + u_free(wait_event(2, 4, &size)); + CHECK(source_events == 2); /* Подписка не меняет каталог публикаций. */ for (unsigned i = 0; i < sizeof(video); i++) video[i] = (uint8_t)i; CHECK(android_stream_feed(1, 1, 100000, 0, video, sizeof(video)) == 0); CHECK(android_stream_feed(1, 1, 100000, 1, video, sizeof(video)) == 1); @@ -69,7 +95,8 @@ int main(void) { CHECK(video_seen == 1 && audio_seen == 1); android_stream_stop(1); u_free(wait_event(2, 5, &size)); android_stream_stop(2); u_free(wait_event(1, 5, &size)); while ((output = android_stream_pull(2, &size))) u_free(output); - CHECK(android_stream_open(3, group, 0, 1280, 720, 0, 0) == 0); u_free(wait_event(3, 4, &size)); + CHECK(!strcmp(source_snapshot, "[]")); + CHECK(android_stream_open(3, group, 0, 1280, 720, 0, 0, 0) == 0); u_free(wait_event(3, 4, &size)); pthread_t thread; CHECK(pthread_create(&thread, NULL, producer, NULL) == 0); while (!__atomic_load_n(&producer_started, __ATOMIC_ACQUIRE)) {} android_stream_detach(inst); @@ -77,9 +104,10 @@ int main(void) { CHECK(android_stream_feed(3, 1, 900000, 1, video, sizeof(video)) == -1); while ((output = android_stream_pull(3, &size))) u_free(output); android_stream_attach(inst); - CHECK(android_stream_open(4, group, 0, 1280, 720, 0, 0) == 0); android_stream_stop(4); + CHECK(android_stream_open(4, group, 0, 1280, 720, 0, 0, 0) == 0); android_stream_stop(4); u_free(wait_event(4, 5, &size)); android_stream_detach(inst); uasync_mark_stopped(ua); utun_instance_destroy(inst); uasync_destroy(ua, 0); + u_free(source_snapshot); CHECK(u_get_allocated_count() == before); test_clock_finish(); DEBUG_INFO(DEBUG_CATEGORY_STREAM, "PASS Android stream: preview, source snapshot, FIN, cancellation, concurrent detach and no leaks"); diff --git a/tools/chatgui/transport/stream_bridge.cpp b/tools/chatgui/transport/stream_bridge.cpp new file mode 100644 index 00000000..a6208eef --- /dev/null +++ b/tools/chatgui/transport/stream_bridge.cpp @@ -0,0 +1,200 @@ +#include "stream_bridge.h" +#include +#include +#include + +extern "C" { +#include "../../../lib/debug_config.h" +#include "../../../lib/mem.h" +#include "../../../lib/u_async.h" +#include "../../../src/utun_instance.h" +#include "../../../src/stream/stream.h" +#include "../../../src/stream/stream_member.h" +} + +namespace { +struct Viewer { uint64_t id = 0, group = 0, source = 0, epoch = 0; }; +struct Bridge { + std::mutex lock; + UTUN_INSTANCE* inst = nullptr; + posted_task* pending = nullptr; + uint64_t watch = 0; + bool watchPending = false, opening = false, stopping = false, overflow = false; + Viewer viewer, requested; + STREAM_HANDLE* handle = nullptr; + stream_media_receiver* receiver = nullptr; + std::deque output; + size_t bytes = 0; +} bridge; + +void process(void*); + +/* Вызывается под lock; резервирование делает отказ команды наблюдаемым до изменения состояния. */ +bool schedule() { + if (!bridge.inst || !uasync_is_running(bridge.inst->ua)) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop stream command rejected: core unavailable"); return false; + } + if (bridge.pending) return true; + auto* task = static_cast(u_calloc(1, sizeof(posted_task))); + if (!task) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop stream task allocation failed"); return false; } + task->callback = process; bridge.pending = task; + uasync_post_reserved(bridge.inst->ua, task); + return true; +} + +/* Управляющие записи не теряются при media pressure; старые readers не накапливают очередь. */ +void output(StreamRecord record) { + if (bridge.output.size() >= 256 || bridge.bytes + static_cast(record.data.size()) > 4u * 1024u * 1024u) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop stream RX pressure id=%llu records=%zu bytes=%zu; wait KEY", + (unsigned long long)record.id, bridge.output.size(), bridge.bytes); + bridge.output.clear(); bridge.bytes = 0; bridge.overflow = true; + bridge.output.push_back({record.id, 0, StreamRecordKind::Gap}); + if (record.kind == StreamRecordKind::Video || record.kind == StreamRecordKind::Audio) return; + } + bridge.bytes += static_cast(record.data.size()); bridge.output.push_back(std::move(record)); +} + +/* Только uasync под lock; снимает собственное владение подпиской, не транспорт других владельцев. */ +void closeViewer(StreamRecordKind kind, int reason) { + if (bridge.viewer.id) output({bridge.viewer.id, 0, kind, {}, 0, reason}); + if (bridge.handle) stream_close(bridge.handle); + stream_media_receiver_destroy(bridge.receiver); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop viewer released id=%llu reason=%d", + (unsigned long long)bridge.viewer.id, reason); + bridge.handle = nullptr; bridge.receiver = nullptr; bridge.viewer = {}; bridge.overflow = false; +} + +void mediaFrame(void*, const stream_media_format* format, const stream_media_video* video, const stream_media_audio* audio) { + if (bridge.overflow) return; + StreamRecord record; record.id = bridge.viewer.id; record.format = *format; + if (video) { + record.kind = StreamRecordKind::Video; record.pts = video->pts_us; record.flags = video->flags; + record.data = QByteArray(reinterpret_cast(video->data), static_cast(video->length)); + } else { + record.kind = StreamRecordKind::Audio; record.pts = audio->pts_us; + record.data = QByteArray(reinterpret_cast(audio->data), static_cast(audio->length)); + } + output(std::move(record)); +} + +/* Scratch receiver/stream не покидает callback; decoder получает собственные копии. */ +void mediaEvent(UTUN_INSTANCE* inst, const stream_event* event, void*) { + std::lock_guard guard(bridge.lock); + if (inst != bridge.inst || !bridge.handle || event->descriptor.group_id != bridge.viewer.group + || event->descriptor.source_node_id != bridge.viewer.source || event->descriptor.source_epoch != bridge.viewer.epoch) return; + if (event->type == STREAM_EVENT_DATA) { + int result = stream_media_receiver_feed(bridge.receiver, event->data, event->length); + if (result != STREAM_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop viewer media rejected result=%d", result); + closeViewer(StreamRecordKind::Closed, STREAM_CLOSE_FAILED); + } else if (bridge.overflow) { + stream_media_receiver_gap(bridge.receiver); bridge.overflow = false; + } + } else if (event->type == STREAM_EVENT_GAP) { + stream_media_receiver_gap(bridge.receiver); output({bridge.viewer.id, 0, StreamRecordKind::Gap}); + } else if (event->type == STREAM_EVENT_END || !stream_handle_active(bridge.handle)) { + closeViewer(event->type == STREAM_EVENT_END ? StreamRecordKind::End : StreamRecordKind::Closed, event->reason); + } +} + +/* Сериализованный actor: старый owner закрывается перед открытием нового выбранного источника. */ +void process(void*) { + std::lock_guard guard(bridge.lock); + bridge.pending = nullptr; + if (!bridge.inst) return; + if (bridge.watchPending) { + bridge.watchPending = false; + if (stream_watch_sources(bridge.inst, bridge.watch) != STREAM_OK) + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop stream source watch failed group=%llx", (unsigned long long)bridge.watch); + } + if (bridge.stopping || bridge.opening) { + if (bridge.viewer.id) closeViewer(StreamRecordKind::Closed, STREAM_CLOSE_STOPPED); + bridge.stopping = false; + } + if (!bridge.opening) return; + bridge.opening = false; bridge.viewer = bridge.requested; bridge.requested = {}; + stream_member_state member{}; + int reason = STREAM_CLOSE_FAILED; + int result = stream_member_load(bridge.inst, bridge.viewer.group, bridge.viewer.source, &member, nullptr); + if (result == STREAM_OK && (!member.published || member.source.source_epoch != bridge.viewer.epoch)) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop selected source changed source=%llx expected=%llx actual=%llx", + (unsigned long long)bridge.viewer.source, (unsigned long long)bridge.viewer.epoch, + (unsigned long long)(member.published ? member.source.source_epoch : 0)); + result = STREAM_ERR_CONFLICT; reason = STREAM_CLOSE_SOURCE_CHANGED; + } + if (result == STREAM_OK) result = stream_subscribe_open(bridge.inst, bridge.viewer.group, bridge.viewer.source, &bridge.handle); + stream_info info{}; + if (result == STREAM_OK) result = stream_get_info(bridge.handle, &info); + if (result == STREAM_OK && info.descriptor.source_epoch != bridge.viewer.epoch) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop reader encountered stale local epoch"); + result = STREAM_ERR_CONFLICT; reason = STREAM_CLOSE_SOURCE_CHANGED; + } + if (result == STREAM_OK) result = stream_media_receiver_create(&bridge.receiver, mediaFrame, nullptr); + if (result != STREAM_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop viewer open failed id=%llu group=%llx source=%llx result=%d", + (unsigned long long)bridge.viewer.id, (unsigned long long)bridge.viewer.group, + (unsigned long long)bridge.viewer.source, result); + closeViewer(StreamRecordKind::Closed, reason); return; + } + output({bridge.viewer.id, 0, StreamRecordKind::Opened}); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop viewer opened id=%llu group=%llx source=%llx epoch=%llx", + (unsigned long long)bridge.viewer.id, (unsigned long long)bridge.viewer.group, + (unsigned long long)bridge.viewer.source, (unsigned long long)bridge.viewer.epoch); +} +} + +bool gui_stream_attach(UTUN_INSTANCE* inst) { + std::lock_guard guard(bridge.lock); + if (!inst || !inst->stream || bridge.inst) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop stream attach rejected"); return false; + } + bridge.inst = inst; bridge.output.clear(); bridge.bytes = 0; + stream_set_event_callback(inst, mediaEvent, nullptr); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop stream bridge attached"); return true; +} + +void gui_stream_detach(UTUN_INSTANCE* inst) { + std::lock_guard guard(bridge.lock); + if (bridge.inst != inst) return; + if (bridge.pending) uasync_cancel_post(inst->ua, bridge.pending); + bridge.pending = nullptr; stream_set_event_callback(inst, nullptr, nullptr); + if (bridge.viewer.id) closeViewer(StreamRecordKind::Closed, STREAM_CLOSE_SERVICE); + if (bridge.opening) output({bridge.requested.id, 0, StreamRecordKind::Closed, {}, 0, STREAM_CLOSE_SERVICE}); + bridge.inst = nullptr; bridge.requested = {}; bridge.opening = bridge.stopping = bridge.watchPending = false; + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop stream bridge detached"); +} + +bool gui_stream_watch(uint64_t group) { + std::lock_guard guard(bridge.lock); + if (!schedule()) return false; + bridge.watch = group; bridge.watchPending = true; return true; +} + +bool gui_stream_open(uint64_t id, uint64_t group, uint64_t source, uint64_t epoch) { + std::lock_guard guard(bridge.lock); + if (!id || !group || !source || !epoch) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop viewer invalid identity"); return false; } + if (!schedule()) return false; + if (bridge.opening) output({bridge.requested.id, 0, StreamRecordKind::Closed, {}, 0, STREAM_CLOSE_STOPPED}); + bridge.requested = {id, group, source, epoch}; bridge.opening = true; return true; +} + +void gui_stream_stop(uint64_t id) { + std::lock_guard guard(bridge.lock); + if (bridge.opening && bridge.requested.id == id) { + bridge.opening = false; bridge.requested = {}; output({id, 0, StreamRecordKind::Closed, {}, 0, STREAM_CLOSE_STOPPED}); + } + if (bridge.viewer.id == id && schedule()) bridge.stopping = true; + for (auto i = bridge.output.begin(); i != bridge.output.end();) { + if (i->id == id) { bridge.bytes -= static_cast(i->data.size()); i = bridge.output.erase(i); } + else ++i; + } +} + +bool gui_stream_pull(uint64_t id, StreamRecord& record) { + std::lock_guard guard(bridge.lock); + auto i = std::find_if(bridge.output.begin(), bridge.output.end(), [id](const StreamRecord& r) { return r.id == id; }); + if (i == bridge.output.end()) return false; + bridge.bytes -= static_cast(i->data.size()); record = std::move(*i); bridge.output.erase(i); return true; +} + +bool gui_stream_ready() { std::lock_guard guard(bridge.lock); return bridge.inst != nullptr; } diff --git a/tools/chatgui/transport/stream_bridge.h b/tools/chatgui/transport/stream_bridge.h new file mode 100644 index 00000000..8bcf828f --- /dev/null +++ b/tools/chatgui/transport/stream_bridge.h @@ -0,0 +1,26 @@ +/* Desktop live viewer: attach/detach в uasync до destroy ядра; остальные вызовы потокобезопасны. + * Один viewer, команды содержат IDs; RX ограничен 4MB/256 записей, borrowed данные копируются. + * Каталог идёт через CHAT_EVT_STREAM_SOURCES, медиаданные — через отдельную очередь pull. */ +#pragma once + +#include +#include +#include "../../../src/stream/stream_media.h" + +struct UTUN_INSTANCE; +enum class StreamRecordKind { Video, Audio, Gap, Opened, End, Closed }; +struct StreamRecord { + uint64_t id = 0, pts = 0; + StreamRecordKind kind = StreamRecordKind::Gap; + stream_media_format format{}; + int flags = 0, reason = 0; + QByteArray data; +}; + +bool gui_stream_attach(UTUN_INSTANCE* inst); +void gui_stream_detach(UTUN_INSTANCE* inst); +bool gui_stream_watch(uint64_t group); +bool gui_stream_open(uint64_t id, uint64_t group, uint64_t source, uint64_t epoch); +void gui_stream_stop(uint64_t id); +bool gui_stream_pull(uint64_t id, StreamRecord& record); +bool gui_stream_ready(); From 0a5b9b50f1c7c26b3d246b80540ae1377d800252 Mon Sep 17 00:00:00 2001 From: evgeny Date: Mon, 12 Oct 2026 22:09:39 +0200 Subject: [PATCH 07/22] Connect desktop stream bridge to core lifecycle and events --- tools/chatgui/transport/gui_bridge.h | 1 + tools/chatgui/transport/gui_bridge_impl.cpp | 5 +++++ tools/chatgui/transport/utun_node.cpp | 6 ++++++ 3 files changed, 12 insertions(+) diff --git a/tools/chatgui/transport/gui_bridge.h b/tools/chatgui/transport/gui_bridge.h index b5337e5f..a963ebc9 100644 --- a/tools/chatgui/transport/gui_bridge.h +++ b/tools/chatgui/transport/gui_bridge.h @@ -253,6 +253,7 @@ void gui_bridge_set_feedback_changed_cb(gui_dm_messages_fn cb); void gui_bridge_set_feedback_result_cb(gui_dm_messages_fn cb); void gui_bridge_set_msg_submitted_cb(gui_dm_messages_fn cb); void gui_bridge_set_radio_mute_result_cb(gui_dm_messages_fn cb); +void gui_bridge_set_stream_sources_cb(gui_dm_messages_fn cb); int gui_bridge_toggle_member_radio_mute(const char* channel_id, uint64_t node_id); /* CHAT_EVT_ATTACHMENT_PREPARATION: target + job UUID + state. */ void gui_bridge_set_attachment_preparation_cb(gui_dm_messages_fn cb); diff --git a/tools/chatgui/transport/gui_bridge_impl.cpp b/tools/chatgui/transport/gui_bridge_impl.cpp index e55d8352..c6d5275e 100644 --- a/tools/chatgui/transport/gui_bridge_impl.cpp +++ b/tools/chatgui/transport/gui_bridge_impl.cpp @@ -77,6 +77,7 @@ static gui_dm_messages_fn g_feedback_changed_cb = nullptr; static gui_dm_messages_fn g_feedback_result_cb = nullptr; static gui_dm_messages_fn g_msg_submitted_cb = nullptr; static gui_dm_messages_fn g_radio_mute_result_cb = nullptr; +static gui_dm_messages_fn g_stream_sources_cb = nullptr; static gui_network_changed_fn g_network_changed_cb = nullptr; static struct UASYNC* g_ua = nullptr; static struct UTUN_INSTANCE* g_inst = nullptr; @@ -91,6 +92,9 @@ void GuiBridgeReceiver::processPost(int eventType, QByteArray data) { eventType == 16 || eventType == 17 || eventType == 24 || eventType == 25 || eventType == 31 || eventType == 33)) g_network_changed_cb(eventType); switch (eventType) { + case CHAT_EVT_STREAM_SOURCES: + if (g_stream_sources_cb) g_stream_sources_cb(d, dlen); + break; case CHAT_EVT_CPU_LOAD: if (dlen != 2 || d[0] > 1 || d[1] > 100) { DEBUG_WARN(DEBUG_CATEGORY_SYS, "invalid CPU load response bytes=%d", dlen); @@ -615,6 +619,7 @@ void gui_bridge_set_dm_msg_received_cb(gui_dm_msg_received_fn cb) { } void gui_bridge_set_network_changed_cb(gui_network_changed_fn cb) { g_network_changed_cb = cb; } +void gui_bridge_set_stream_sources_cb(gui_dm_messages_fn cb) { g_stream_sources_cb = cb; } void gui_bridge_set_dm_call_route_cb(gui_dm_messages_fn cb) { g_dm_call_route_cb = cb; } void gui_bridge_set_dm_invite_result_cb(gui_dm_messages_fn cb) { g_dm_invite_result_cb = cb; } void gui_bridge_set_member_files_response_cb(gui_dm_messages_fn cb) { g_member_files_response_cb = cb; } diff --git a/tools/chatgui/transport/utun_node.cpp b/tools/chatgui/transport/utun_node.cpp index 68348cc6..2da8c79c 100644 --- a/tools/chatgui/transport/utun_node.cpp +++ b/tools/chatgui/transport/utun_node.cpp @@ -1,5 +1,6 @@ // utun_node.cpp — embedded uTun node running in a dedicated thread #include "utun_node.h" +#include "stream_bridge.h" #include #include #include "../../lib/socket_compat.h" @@ -246,6 +247,10 @@ void UtunNode::runLoop() { } call_audio_init(m_instance); radio_audio_init(m_instance); + if (!gui_stream_attach(m_instance)) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop stream bridge initialization failed"); + QMetaObject::invokeMethod(this, [this] { emit error("stream bridge initialization failed"); }); + } chat_core_sync_my_addresses(m_instance); DEBUG_DEBUG(DEBUG_CATEGORY_GENERAL, "chat_core + chat_sync initialized"); @@ -262,6 +267,7 @@ void UtunNode::runLoop() { if (m_instance) { DEBUG_DEBUG(DEBUG_CATEGORY_GENERAL, "runLoop: destroying instance in worker thread"); + gui_stream_detach(m_instance); call_audio_destroy(m_instance); radio_audio_destroy(m_instance); utun_instance_destroy(m_instance); From 10f197c817071ae2c79900ab7a95b2eafac300df Mon Sep 17 00:00:00 2001 From: evgeny Date: Mon, 12 Oct 2026 22:11:35 +0200 Subject: [PATCH 08/22] Decode and play desktop live H264 and Opus streams --- tools/chatgui/src/streamplayer_window.cpp | 303 ++++++++++++++++++++++ tools/chatgui/src/streamplayer_window.h | 39 +++ 2 files changed, 342 insertions(+) create mode 100644 tools/chatgui/src/streamplayer_window.cpp create mode 100644 tools/chatgui/src/streamplayer_window.h diff --git a/tools/chatgui/src/streamplayer_window.cpp b/tools/chatgui/src/streamplayer_window.cpp new file mode 100644 index 00000000..4a3df57d --- /dev/null +++ b/tools/chatgui/src/streamplayer_window.cpp @@ -0,0 +1,303 @@ +#include "streamplayer_window.h" +#include "videoplayer_window.h" +#include "sound_manager.h" +#include "../transport/stream_bridge.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +extern "C" { +#include "../../../lib/debug_config.h" +#include "../../../lib/opus_codec.h" +#include "../../../src/stream/stream.h" +#ifdef HAVE_FFMPEG +#include +#include +#endif +} + +namespace { +int64_t nowUs() { + return std::chrono::duration_cast(std::chrono::steady_clock::now().time_since_epoch()).count(); +} +std::atomic nextId{0}; +QString text(const char* value) { return QCoreApplication::translate("ChatGui", value); } +} + +struct StreamPlayerWindow::Playback { + struct Video { QImage image; int64_t pts; }; + struct Audio { std::array pcm; int64_t pts; unsigned offset = 0; }; + uint64_t id = ++nextId, group = 0, source = 0, epoch = 0; + std::atomic running{true}, audioAvailable{false}; + std::atomic volume{1.0f}; + std::thread worker; + std::mutex lock; + std::deque