diff --git a/doc/stream.txt b/doc/stream.txt index 653e3220..15aa0088 100644 --- a/doc/stream.txt +++ b/doc/stream.txt @@ -755,7 +755,7 @@ Android: - Camera/codec callbacks — HandlerThread, Opus encoder — audio capture thread, decoder/AudioTrack — viewer worker. - Native bridge держит pthread mutex, ограниченные очереди и один pending post либо timer. attach/detach выполняются в uasync вне callbacks. detach отменяет post/timer, закрывает handles, - очищает media, snapshots и запрещает producers до уничтожения instance/ua. + очищает media, снимает наблюдение каталога и запрещает producers до уничтожения instance/ua. UI не сохраняет borrowed inst/ua. Idle без pending data не будит мобильное ядро. - TX<=2MiB/128 records, не более 250ms ожидания. Actor публикует <=2 пачек за вызов, неполную пачку объединяет с аудио/видео до 20ms; продолжение работы через timer 5ms. @@ -771,8 +771,16 @@ Android: своего микрофона, чтобы исключить акустическую обратную связь. Камера при закрытии preview продолжает работать. Viewer закрывается на Activity ON_STOP и создаётся заново на ON_START; publisher от UI не зависит. Видимый player удерживает экран включённым. Preview ограничен оставшимся местом, кнопки управления не перекрывает. -- Source list строится в uasync по stream_member_load и stream_member_online, выдаётся UI копией JSON. - Свежий snapshot содержит source ID, epoch, self и экранированное имя. Увиденный источник не создаёт подписку. +- Source list общий для Android/desktop: stream_sources_json в uasync использует stream_member_load + и stream_member_online. stream_watch_sources наблюдает один выбранный канал; group=0 снимает наблюдение. + Начальный snapshot и изменения после Merkle/BGP передаются через CHAT_EVT_STREAM_SOURCES=58: + group_id:8BE, valid:1, JSON с id(decimal string), epoch(hex string), self, name. + События объединяются отложенным callback, одинаковый список повторно не отправляется. Таймера опроса нет. + Обнаружение не создаёт подписку, media-буфер и не удерживает ядро в ACTIVE. +- В шапке канала значок виден при наличии источников; один источник открывается сразу, + несколько выбираются из списка по имени/ID. Пункт меню «Трансляции» также открывает управление публикацией. + Выбранная эпоха проверяется actor до подписки и по фактически открытому handle. Перезапуск источника + требует нового выбора. Ошибка viewer kind=6 передаёт stream_close_reason в flags. - Stop сначала завершает capture producers и encoder EOS (максимум 1s), затем FIN. Publisher сохраняет конечный буфер ещё 2s для сетевой выдачи, затем закрывается. Это ограниченный drain, не подтверждение воспроизведения всеми viewers. Abort/core restart очищает очередь сразу. @@ -785,3 +793,24 @@ Kotlin progress — DEBUG раз в секунду; C packet-level диагно Проверки: test_stream_media (все пары потерь 6+2, parser/границы/GAP), test_android_stream (живое ядро, preview, verified snapshots, FIN/cancel, concurrent producer/detach и leak count), Android H264AnnexBTest (Annex B/length-prefixed NAL, SPS/PPS/IDR и усечённые данные). + +Desktop: +- Значок трансляций расположен в шапке MessageList. Источники берутся из того же каталога ядра; + ответы предыдущего канала не меняют текущую шапку. Личные беседы снимают наблюдение. +- Один StreamPlayerWindow на процесс. Он содержит имя источника, состояние подключения, громкость, + полноэкранный режим и закрытие. Выбор нового источника закрывает предыдущего reader. +- transport/stream_bridge владеет одним viewer и ограничивает RX 4MiB/256 записей. + Attach/detach выполняются в uasync; команды и копирование медиаданных защищены mutex. + Post отменяется до destroy ядра; GUI/decoder не хранят заимствованные instance/handle pointers. +- stream_media_receiver собирает H.264 access units и выдаёт Opus 48kHz mono/20ms. + FFmpeg/Opus декодируются отдельным worker; Qt-очередь не накапливает кадры. + Очереди ограничены шестью декодированными видеокадрами и 25 аудиопакетами (500ms). + Видео декодируется без задержки frame-threading. При GAP сбрасываются decoder refs и timeline до KEY. +- После prebuffer 150ms показ следует физическим audio callback timestamps; без звука — монотонным часам. + Короткие пропуски аудио заполняются Opus PLC; большой разрыв/pressure сбрасывает timeline. + Закрытие окна снимает только reader, собственная публикация сохраняется. + Удаление канала сообщает STREAM_CLOSE_GROUP; остановка ядра снимает media и catalogue owners. +- Без FFmpeg окно сообщает, что декодирование недоступно. Нужна desktop-сборка с HAVE_FFMPEG. +Проверки: test_stream_view — настоящее ядро, старые эпохи/cancel, RX pressure, H.264 с поворотом, +Opus/громкость на настоящем null audio backend, закрытие канала и полный баланс tracked allocations. +test_stream_service также проверяет общий каталог двух независимых источников одного канала. 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..e57f7c30 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,10 +472,14 @@ 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; - if (!s->closing && s->descriptor.group_id == group_id) session_close(s, STREAM_CLOSE_GROUP, 1, 0); + if (!s->closing && s->descriptor.group_id == group_id) session_close(s, STREAM_CLOSE_GROUP, 1, 1); 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); diff --git a/tests/test_android_stream.c b/tests/test_android_stream.c index 0f8ee5bc..49416a27 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)); + char *sources = NULL; CHECK(stream_sources_json(inst, group, &sources) == STREAM_OK); + CHECK(strstr(sources, "epoch") && strstr(sources, "name")); u_free(sources); test_clock_poll(ua, 100); + 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/tests/test_stream_service.c b/tests/test_stream_service.c index 6838ecc8..8fe9b852 100644 --- a/tests/test_stream_service.c +++ b/tests/test_stream_service.c @@ -420,6 +420,14 @@ static void test_network(void) { CHECK(stream_publish_open(nodes[3], GROUP1, ¶meters, ¶llel_publisher) == STREAM_OK); struct stream_info parallel_info; CHECK(stream_get_info(parallel_publisher, ¶llel_info) == STREAM_OK); wait_state(1, 3, GROUP1, parallel_info.descriptor.source_epoch, 0); + /* Две независимые публикации одного канала видны через общий каталог; подписчики не становятся источниками. */ + char *catalog = NULL, origin_id[64], parallel_id[64], subscriber_id[64]; + CHECK(stream_sources_json(nodes[1], GROUP1, &catalog) == STREAM_OK); + snprintf(origin_id, sizeof(origin_id), "\"id\":\"%llu\"", (unsigned long long)nodes[0]->node_id); + snprintf(parallel_id, sizeof(parallel_id), "\"id\":\"%llu\"", (unsigned long long)nodes[3]->node_id); + snprintf(subscriber_id, sizeof(subscriber_id), "\"id\":\"%llu\"", (unsigned long long)nodes[1]->node_id); + CHECK(strstr(catalog, origin_id) && strstr(catalog, parallel_id) && !strstr(catalog, subscriber_id)); u_free(catalog); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] shared catalogue exposes two live publishers in the same channel"); CHECK(stream_subscribe_manual_open(nodes[1], GROUP1, nodes[3]->node_id, ¶llel_subscriber) == STREAM_OK); size_t accepted = 0; CHECK(stream_publish_write(parallel_publisher, expected, sizeof(expected), &accepted) == STREAM_OK && accepted == sizeof(expected)); 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/data/NativeLib.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/NativeLib.kt index 6654f487..a29cadab 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,20 +199,22 @@ 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) - fun streamSources(group: Long): String = nativeStreamSources(group) ?: "[]" fun streamOpusCreate(encode: Boolean): Long = nativeStreamOpusCreate(encode) 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? - private external fun nativeStreamSources(group: Long): String? private external fun nativeStreamOpusCreate(encode: Boolean): Long private external fun nativeStreamOpusEncode(handle: Long, pcm: ShortArray): ByteArray? private external fun nativeStreamOpusDecode(handle: Long, packet: ByteArray): ShortArray? 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/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/ui/screens/ChatScreen.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/ui/screens/ChatScreen.kt index 9cf31f29..bbbd53ab 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 @@ -15,6 +15,7 @@ import androidx.compose.foundation.combinedClickable import androidx.compose.foundation.ExperimentalFoundationApi import androidx.compose.foundation.gestures.awaitFirstDown import androidx.compose.foundation.horizontalScroll +import androidx.compose.foundation.verticalScroll import androidx.compose.foundation.layout.* import androidx.compose.foundation.lazy.rememberLazyListState import androidx.compose.foundation.rememberScrollState @@ -70,6 +71,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 +92,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 +102,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(Modifier.heightIn(max = 360.dp).verticalScroll(rememberScrollState())) { + 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 +193,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..5df6a9ae 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 @@ -28,27 +28,20 @@ import androidx.lifecycle.Lifecycle import androidx.lifecycle.LifecycleEventObserver import com.utun.chat.R import com.utun.chat.data.LogManager -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 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) +import com.utun.chat.data.StreamSource /** Выход с экрана закрывает только 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 +57,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 +71,16 @@ 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)) } + if (selected.self && local?.ongoing == true) Button(onClick = StreamController::stop, + enabled = local.phase != StreamPhase.STOPPING) { Text(stringResource(R.string.stream_stop)) } + return@Column + } if (local?.ongoing == true) { Text(stringResource(when (local.phase) { StreamPhase.LIVE -> R.string.stream_live @@ -117,24 +106,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() })) } + TextButton(onClick = { onSelect(source) }) { Text(stringResource(R.string.stream_watch, source.displayName)) } } } - selected?.let { source -> - Box(Modifier.weight(1f).fillMaxWidth()) { - key(group, source.id, source.epoch) { LiveStreamView(group, source.id, source.self) } - } - 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 +129,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) } 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 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 diff --git a/tools/chatgui-android/jni_bridge/android_stream_jni.c b/tools/chatgui-android/jni_bridge/android_stream_jni.c index aef2002d..e0a23961 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( @@ -32,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 d2a5526f..30c9d5b0 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" @@ -30,11 +26,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; @@ -43,8 +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 char *source_list; +static uint64_t watch_group; +static int watch_pending; static void media_process(void *arg); static void media_close(struct android_media_session *s, int kind); @@ -146,6 +142,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 +151,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); } @@ -196,69 +193,48 @@ 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; - 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); -} - /* Один actor обслуживает команды и не больше двух FEC-пачек за вызов; idle не будит мобильное ядро. */ static void media_process(void *arg) { (void)arg; pthread_mutex_lock(&media_lock); media_post = NULL; media_timer = NULL; if (!media_inst) { pthread_mutex_unlock(&media_lock); return; } - if (list_group) media_list_sources(); + 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); + } struct android_media_session *sessions[] = {&publisher, &viewer}; for (unsigned i = 0; i < 2; i++) { struct android_media_session *s = sessions[i]; 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", @@ -323,12 +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; - u_free(source_list); source_list = NULL; list_group = cached_group = list_updated = 0; + watch_pending = 0; watch_group = 0; DEBUG_INFO(DEBUG_CATEGORY_STREAM, "Android stream bridge detached; producers rejected"); } pthread_mutex_unlock(&media_lock); @@ -336,9 +313,12 @@ 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) { - 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; + unsigned rotation, int audio, uint64_t expected_epoch) { + if (!id || !group || (source && !expected_epoch) + || (!source && (!width || width > 4096 || !height || height > 4096 || rotation > 270 || rotation % 90))) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream open invalid id=%llu group=%llx source=%llx epoch=%llx dimensions=%ux%u", + (unsigned long long)id, (unsigned long long)group, (unsigned long long)source, + (unsigned long long)expected_epoch, width, height); return -1; } pthread_mutex_lock(&media_lock); struct android_media_session *s = source ? &viewer : &publisher; @@ -346,13 +326,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) { @@ -420,16 +410,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 fbbcaa00..d7dc2278 100644 --- a/tools/chatgui-android/libutun_lite/stream_bridge.h +++ b/tools/chatgui-android/libutun_lite/stream_bridge.h @@ -3,7 +3,7 @@ * Один publisher и один viewer, TX <=2MB/128 records, RX <=4MB/256 records. * Data принимает полные H.264 Annex B кадры (KEY содержит SPS/PPS/IDR) и Opus 48k mono/20ms. * pull передаёт владение u_malloc-буфером: kind:1,flags:1,width:2,height:2,rotation:2,PTS_us:8,length:4,data. - * kinds: video=1,audio=2,gap=3,opened=4,ended=5,error=6. Числа big-endian. + * kinds: video=1,audio=2,gap=3,opened=4,ended=5,error=6; error.flags = stream_close_reason. Числа big-endian. * Медиаданные передаются через pull, не через общий чатовый event bus. */ #ifndef ANDROID_STREAM_BRIDGE_H #define ANDROID_STREAM_BRIDGE_H @@ -15,14 +15,14 @@ struct UTUN_INSTANCE; /* 0 attached, -1 invalid/busy: нельзя передать singleton другому ядру до detach старого. */ int android_stream_attach(struct UTUN_INSTANCE *inst); void android_stream_detach(struct UTUN_INSTANCE *inst); +/* source=0 — publisher; source!=0 — viewer с обязательным expected_epoch. Команда принята=0, результат через pull. */ int android_stream_open(uint64_t id, uint64_t group, uint64_t source, unsigned width, unsigned height, - unsigned rotation, int audio); -/* source=0 — publisher; source!=0 — viewer. 0 означает, что команда принята, результат приходит через pull. */ + unsigned rotation, int audio, uint64_t expected_epoch); +/* Наблюдение выбранного канала через CHAT_EVT_STREAM_SOURCES; group=0 снимает его. */ +int android_stream_watch(uint64_t group); 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. */ 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/CMakeLists.txt b/tools/chatgui/CMakeLists.txt index 31d7756a..5ff17ebd 100644 --- a/tools/chatgui/CMakeLists.txt +++ b/tools/chatgui/CMakeLists.txt @@ -195,6 +195,7 @@ add_executable(vibechat src/media_blocks.cpp src/videoplayer_engine.cpp src/videoplayer_window.cpp + src/streamplayer_window.cpp src/imageviewer_window.cpp src/member_files/browser.cpp src/member_files/batch.cpp @@ -216,6 +217,7 @@ add_executable(vibechat transport/node_config.cpp transport/config_updater.cpp transport/gui_bridge_impl.cpp + transport/stream_bridge.cpp transport/miniaudio_impl.c transport/audio_diagnostics.cpp db/db_manager.cpp @@ -435,7 +437,7 @@ target_link_libraries(test_radio_member_mute PRIVATE ${QT_CORE} utun_voice utun add_test(NAME test_radio_member_mute COMMAND test_radio_member_mute) set(RADIO_MEMBER_TEST_SOURCES src/memberlistmodel.cpp src/avatars/avatarcache.cpp src/avatars/avatarmanager.cpp src/avatars/avatarcrop.cpp src/avatars/avatargallery.cpp src/imageviewer_window.cpp db/db_manager.cpp) -add_executable(test_radio_mute_bridge tests/test_radio_mute_bridge.cpp transport/utun_node.cpp transport/gui_bridge_impl.cpp +add_executable(test_radio_mute_bridge tests/test_radio_mute_bridge.cpp transport/utun_node.cpp transport/gui_bridge_impl.cpp transport/stream_bridge.cpp src/accountdelegate.cpp src/flagpainter.cpp ${RADIO_MEMBER_TEST_SOURCES}) target_include_directories(test_radio_mute_bridge PRIVATE ${UTUN_INCLUDE_DIRS}) target_link_libraries(test_radio_mute_bridge PRIVATE ${QT_LIBS} ${QT_TEST_LIB} utun_voice utun pthread) @@ -500,6 +502,21 @@ else() endif() add_executable(test_ptt_key_chord tests/test_ptt_key_chord.cpp) +add_executable(test_stream_view tests/test_stream_view.cpp src/streamplayer_window.cpp transport/stream_bridge.cpp + src/videoplayer_window.cpp src/videoplayer_engine.cpp src/sound_manager.cpp src/voice_audio_io.cpp src/audio_device.cpp + transport/miniaudio_impl.c transport/audio_diagnostics.cpp) +target_include_directories(test_stream_view PRIVATE ${UTUN_INCLUDE_DIRS} ${FFMPEG_INCLUDE_DIRS}) +target_link_libraries(test_stream_view PRIVATE ${QT_LIBS} ${QT_TEST_LIB} utun_voice utun pthread ${CMAKE_DL_LIBS} ${FFMPEG_TARGET} ${FFMPEG_LIBRARIES}) +if(FFMPEG_FOUND) + target_compile_definitions(test_stream_view PRIVATE HAVE_FFMPEG) +endif() +if(CMAKE_SYSTEM_NAME STREQUAL "Linux") + target_compile_definitions(test_stream_view PRIVATE TEST_NULL_AUDIO) + target_link_options(test_stream_view PRIVATE "-Wl,--wrap=ma_context_init" "-Wl,--wrap=ma_device_init") +endif() +add_test(NAME test_stream_view COMMAND test_stream_view) +set_tests_properties(test_stream_view PROPERTIES TIMEOUT 30 ENVIRONMENT "QT_QPA_PLATFORM=offscreen") + target_link_libraries(test_ptt_key_chord PRIVATE ${QT_CORE}) add_test(NAME test_ptt_key_chord COMMAND test_ptt_key_chord) diff --git a/tools/chatgui/src/mainwindow.cpp b/tools/chatgui/src/mainwindow.cpp index e98a8c25..1f0dd8f0 100644 --- a/tools/chatgui/src/mainwindow.cpp +++ b/tools/chatgui/src/mainwindow.cpp @@ -1,5 +1,6 @@ #include #include "mainwindow.h" +#include "streamplayer_window.h" #include "channellist.h" #include "messagelist.h" #include "messagedelegate.h" @@ -197,6 +198,8 @@ MainWindow::MainWindow(QWidget *parent, DbManager* db, const QString& cfgPath, } MainWindow::~MainWindow() { + if (m_streamPlayer) delete m_streamPlayer.data(); + gui_bridge_set_stream_sources_cb(nullptr); s_mainWindow = nullptr; m_cpuLoadTimer->stop(); gui_bridge_set_cpu_load_cb(nullptr); @@ -585,8 +588,11 @@ void MainWindow::setupBridgeCallbacks() { s_mainWindow->m_channelList->setCpuLoad(0); s_mainWindow->m_accountList->model()->setRadioState(0, false, false); s_mainWindow->m_messageList->clearDmCallRoute(); + s_mainWindow->m_messageList->clearStreams(); + if (s_mainWindow->m_streamPlayer) s_mainWindow->m_streamPlayer->close(); s_mainWindow->m_channelList->clearDmPresence(); } else { + if (type == 12) s_mainWindow->m_messageList->watchStreams(); s_mainWindow->m_channelList->refreshDmPresence(); } }); @@ -599,12 +605,21 @@ void MainWindow::setupBridgeCallbacks() { if (s_mainWindow) s_mainWindow->m_messageList->onDmMessages(data, len); }); + gui_bridge_set_stream_sources_cb([](const uint8_t* data, int len) { + if (s_mainWindow) s_mainWindow->m_messageList->onStreamSources(data, len); + }); gui_bridge_set_attachment_preparation_cb([](const uint8_t* data, int len) { if (s_mainWindow) s_mainWindow->m_messageList->onAttachmentPreparation(data, len); }); } void MainWindow::setupConnects() { + connect(m_messageList, &MessageList::streamRequested, this, + [this](quint64 group, quint64 source, quint64 epoch, const QString& name) { + if (m_streamPlayer) delete m_streamPlayer.data(); + m_streamPlayer = new StreamPlayerWindow(group, source, epoch, name); + m_streamPlayer->setAttribute(Qt::WA_DeleteOnClose); m_streamPlayer->show(); + }); connect(RadioAudioEngine::instance(), &RadioAudioEngine::stateChanged, this, [this] { auto* engine = RadioAudioEngine::instance(); m_accountList->model()->setRadioState(engine->groupId(), engine->isActive(), engine->isTransmitting()); diff --git a/tools/chatgui/src/mainwindow.h b/tools/chatgui/src/mainwindow.h index 2cc23eb5..dda791d8 100644 --- a/tools/chatgui/src/mainwindow.h +++ b/tools/chatgui/src/mainwindow.h @@ -21,6 +21,7 @@ class UtunNode; class AudioRecorder; class ConnMonitorWindow; class VideoPlayerWindow; +class StreamPlayerWindow; class ImageViewerWindow; class CallWindow; class CallAudioEngine; @@ -124,6 +125,7 @@ private: AudioRecorder *m_recorder = nullptr; ConnMonitorWindow *m_connMonitor = nullptr; VideoPlayerWindow *m_videoPlayer = nullptr; + QPointer m_streamPlayer; ImageViewerWindow *m_imageViewer = nullptr; QPointer m_fileBrowser; diff --git a/tools/chatgui/src/messagelist.cpp b/tools/chatgui/src/messagelist.cpp index 0decf11f..e9f35119 100644 --- a/tools/chatgui/src/messagelist.cpp +++ b/tools/chatgui/src/messagelist.cpp @@ -1,5 +1,6 @@ #include #include "messagelist.h" +#include "../transport/stream_bridge.h" #include "debug_ui.h" #include "chatview.h" #include "messagedelegate.h" @@ -426,6 +427,30 @@ MessageList::MessageList(DbManager* db, AudioRecorder* recorder, QWidget *parent connect(m_radioBtn, &QPushButton::toggled, this, &MessageList::onRadioToggled); hl->addWidget(m_radioBtn); + m_streamBtn = new QPushButton(header); m_streamBtn->setObjectName("streamButton"); + QPixmap streamIcon(24, 24); streamIcon.fill(Qt::transparent); + { QPainter painter(&streamIcon); painter.setRenderHint(QPainter::Antialiasing); painter.setPen(QPen(QColor("#e53935"), 2)); + painter.drawRoundedRect(QRectF(2, 6, 20, 14), 2, 2); painter.drawLine(8, 1, 12, 5); painter.drawLine(16, 1, 12, 5); + painter.setBrush(QColor("#e53935")); painter.drawEllipse(QPointF(12, 13), 3, 3); } + m_streamBtn->setIcon(QIcon(streamIcon)); m_streamBtn->setIconSize(QSize(24, 24)); m_streamBtn->hide(); + m_streamBtn->setAccessibleName(QCoreApplication::translate("ChatGui", "Live streams")); + m_streamBtn->setCursor(Qt::PointingHandCursor); + connect(m_streamBtn, &QPushButton::clicked, this, [this] { + auto watch = [this](const LiveSource& source) { + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop selected stream group=%s source=%llx epoch=%llx", + qPrintable(m_currentChannelId), (unsigned long long)source.id, (unsigned long long)source.epoch); + emit streamRequested(m_currentChannelId.toULongLong(), source.id, source.epoch, source.name); + }; + if (m_streamSources.size() == 1) { watch(m_streamSources.front()); return; } + QMenu menu(this); + for (const auto& source : m_streamSources) { + auto* action = menu.addAction(QStringLiteral("%1 · %2").arg(source.name).arg(source.id)); + connect(action, &QAction::triggered, this, [watch, source] { watch(source); }); + } + menu.exec(m_streamBtn->mapToGlobal(QPoint(0, m_streamBtn->height()))); + }); + hl->addWidget(m_streamBtn); + m_groupMenuButton = new QPushButton(QStringLiteral("…"), header); m_groupMenuButton->setObjectName("groupMenuButton"); m_groupMenuButton->setFixedSize(34, 30); @@ -749,6 +774,7 @@ void MessageList::loadChannel(const QString& channelId) { m_avatarCache.clear(); if (m_currentChannelId != channelId) { m_feedback.clear(); m_pendingPoll.clear(); } m_currentChannelId = channelId; + clearStreams(); watchStreams(); m_groupMenuButton->setVisible(channelId.toULongLong() != 0); clearDmCallRoute(); m_callBtn->setVisible(isDmChannel(channelId)); @@ -1451,6 +1477,43 @@ void MessageList::onRadioTalk(quint64 groupId, quint64 srcNodeId, bool on) { } } +/* Каталог выбранного канала не удерживает media owner; DM/пустой target снимает наблюдение. */ +void MessageList::watchStreams() { + gui_stream_watch(isDmChannel(m_currentChannelId) ? 0 : m_currentChannelId.toULongLong()); +} + +void MessageList::clearStreams() { m_streamSources.clear(); m_streamBtn->hide(); } + +/* Полный снимок применяется атомарно; ответы предыдущего канала не меняют текущую шапку. */ +void MessageList::onStreamSources(const uint8_t* data, int len) { + if (!data || len < 11 || data[8] > 1) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop source event invalid len=%d", len); return; } + quint64 group = stream_read_be(data, 8); + if (isDmChannel(m_currentChannelId) || group != m_currentChannelId.toULongLong()) return; + QJsonParseError error; + auto document = QJsonDocument::fromJson(QByteArray(reinterpret_cast(data + 9), len - 9), &error); + if (!data[8] || error.error != QJsonParseError::NoError || !document.isArray()) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop source list failed group=%llx valid=%u parse=%s", + (unsigned long long)group, data[8], qPrintable(error.errorString())); + clearStreams(); return; + } + QList sources; + for (const auto& value : document.array()) { + auto member = value.toObject(); bool idOk = false, epochOk = false; + quint64 id = member.value("id").toString().toULongLong(&idOk); + quint64 epoch = member.value("epoch").toString().toULongLong(&epochOk, 16); + if (!idOk || !epochOk || !id || !epoch || !member.value("name").isString()) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop source identity malformed group=%llx", (unsigned long long)group); + clearStreams(); return; + } + QString name = member.value("name").toString(); if (name.isEmpty()) name = QString::number(id); + sources.push_back({id, epoch, name}); + } + m_streamSources = std::move(sources); m_streamBtn->setVisible(!m_streamSources.isEmpty()); + m_streamBtn->setText(m_streamSources.size() > 1 ? QString::number(m_streamSources.size()) : QString()); + m_streamBtn->setToolTip(QCoreApplication::translate("ChatGui", "Live streams: %1").arg(m_streamSources.size())); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop stream header group=%llx sources=%d", (unsigned long long)group, (int)m_streamSources.size()); +} + void MessageList::onRadioToggled(bool on) { if (m_currentChannelId.isEmpty() || isDmChannel(m_currentChannelId)) return; diff --git a/tools/chatgui/src/messagelist.h b/tools/chatgui/src/messagelist.h index deeb1236..110e5a12 100644 --- a/tools/chatgui/src/messagelist.h +++ b/tools/chatgui/src/messagelist.h @@ -46,11 +46,15 @@ public: void onFeedbackResult(const uint8_t* data, int len); void onMessageSubmitted(const uint8_t* data, int len); void onRadioTalk(quint64 groupId, quint64 srcNodeId, bool on); /* индикация говорящего */ + void onStreamSources(const uint8_t* data, int len); + void clearStreams(); + void watchStreams(); signals: void readPositionChanged(const QString& channelId, qint64 lastReadMsgId); void callRequested(const QString& target, quint64 peer, const QString& name); void openVideoRequested(const QString& filePath); + void streamRequested(quint64 group, quint64 source, quint64 epoch, const QString& name); void openImageRequested(const QString& filePath); void inviteRequested(const QString& sourceChannel, quint64 peer, const QString& name); void ownFilesRequested(const QString& channelId); @@ -83,6 +87,9 @@ private: quint64 m_callPeer = 0; quint64 m_callGroup = 0; QPushButton* m_radioBtn = nullptr; /* шапка: переключатель рации */ + QPushButton* m_streamBtn = nullptr; + struct LiveSource { quint64 id, epoch; QString name; }; + QList m_streamSources; RadioPanel* m_radioPanel = nullptr; /* верхняя панель устройств, VAD и PTT */ bool m_radioOn = false; QThreadPool m_thumbnailPool; diff --git a/tools/chatgui/src/streamplayer_window.cpp b/tools/chatgui/src/streamplayer_window.cpp new file mode 100644 index 00000000..72a45244 --- /dev/null +++ b/tools/chatgui/src/streamplayer_window.cpp @@ -0,0 +1,305 @@ +#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