evgeny
7 days ago
3 changed files with 258 additions and 4 deletions
@ -0,0 +1,200 @@
|
||||
#include "stream_bridge.h" |
||||
#include <deque> |
||||
#include <mutex> |
||||
#include <algorithm> |
||||
|
||||
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<StreamRecord> 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<posted_task*>(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<size_t>(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<size_t>(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<const char*>(video->data), static_cast<int>(video->length)); |
||||
} else { |
||||
record.kind = StreamRecordKind::Audio; record.pts = audio->pts_us; |
||||
record.data = QByteArray(reinterpret_cast<const char*>(audio->data), static_cast<int>(audio->length)); |
||||
} |
||||
output(std::move(record)); |
||||
} |
||||
|
||||
/* Scratch receiver/stream не покидает callback; decoder получает собственные копии. */ |
||||
void mediaEvent(UTUN_INSTANCE* inst, const stream_event* event, void*) { |
||||
std::lock_guard<std::mutex> 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<std::mutex> 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<std::mutex> 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<std::mutex> 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<std::mutex> 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<std::mutex> 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<std::mutex> 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<size_t>(i->data.size()); i = bridge.output.erase(i); } |
||||
else ++i; |
||||
} |
||||
} |
||||
|
||||
bool gui_stream_pull(uint64_t id, StreamRecord& record) { |
||||
std::lock_guard<std::mutex> 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<size_t>(i->data.size()); record = std::move(*i); bridge.output.erase(i); return true; |
||||
} |
||||
|
||||
bool gui_stream_ready() { std::lock_guard<std::mutex> guard(bridge.lock); return bridge.inst != nullptr; } |
||||
@ -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 <QByteArray> |
||||
#include <cstdint> |
||||
#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(); |
||||
Loading…
Reference in new issue