From d9821fa89cb36284c212a124ad564d3aa7e2d4ac Mon Sep 17 00:00:00 2001 From: evgeny Date: Mon, 12 Oct 2026 22:07:35 +0200 Subject: [PATCH] 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();