jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Files
b8a2d7a555c8be87f5d3025159f2ae551ee1fa54
utun2/tools/chatgui/transport/stream_bridge.cpp
T

290 lines
16 KiB
C++
Raw Blame History

#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;
bool wantRead = false, waiting = false, resync = false;
bool wantVideo = false;
int64_t position = -1, audioAhead = STREAM_MEDIA_READ_AHEAD_US;
stream_media_status status{};
uint64_t statusId = 0;
Viewer viewer, requested;
STREAM_HANDLE* handle = nullptr;
stream_publisher* publisher = nullptr;
std::deque<StreamRecord> output;
size_t bytes = 0;
} bridge;
void process(void*);
/* Вызывается под lock; резервирование делает отказ команды наблюдаемым до изменения состояния. */
bool schedule() {
if (!bridge.inst) {
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() >= 16 || 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.resync = 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 (kind == StreamRecordKind::Closed) {
for (auto i = bridge.output.begin(); i != bridge.output.end();) {
if (i->id == bridge.viewer.id && (i->kind == StreamRecordKind::Video || i->kind == StreamRecordKind::Audio)) {
bridge.bytes -= static_cast<size_t>(i->data.size()); i = bridge.output.erase(i);
} else ++i;
}
}
if (bridge.viewer.id) output({bridge.viewer.id, 0, kind, {}, 0, reason});
if (bridge.handle) stream_close(bridge.handle);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop viewer released id=%llu reason=%d",
(unsigned long long)bridge.viewer.id, reason);
bridge.handle = nullptr; bridge.viewer = {}; bridge.wantRead = bridge.waiting = bridge.resync = false;
}
void mediaFrame(void*, const stream_media_format* format, const stream_media_video* video, const stream_media_audio* audio) {
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));
}
/* Сеть только сообщает готовность индекса; ближайшие кадры читает запрос плеера из общего кольца. */
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_CLOSED || !stream_handle_active(bridge.handle))
closeViewer(StreamRecordKind::Closed, event->reason);
else if (event->type == STREAM_EVENT_AVAILABLE) { bridge.waiting = false; schedule(); }
}
/* Сериализованный 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) {
bridge.opening = false; bridge.viewer = bridge.requested; bridge.requested = {};
bridge.status = {}; bridge.statusId = bridge.viewer.id; bridge.position = -1; bridge.waiting = false;
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_enable(bridge.handle, STREAM_MEDIA_TRACK_VIDEO | STREAM_MEDIA_TRACK_AUDIO);
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);
bridge.wantRead = false;
}
if (!bridge.handle) return;
if (bridge.resync) {
bridge.resync = false; bridge.position = -1;
if (stream_media_resync(bridge.handle) != STREAM_OK) { closeViewer(StreamRecordKind::Closed, STREAM_CLOSE_FAILED); return; }
}
int statusResult = stream_media_get_status(bridge.handle, &bridge.status);
if (statusResult < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop media index failed result=%d", statusResult);
closeViewer(StreamRecordKind::Closed, STREAM_CLOSE_FAILED); return;
}
bool prepared = std::any_of(bridge.output.begin(), bridge.output.end(), [](const StreamRecord& r) { return r.id == bridge.viewer.id; });
if (!bridge.wantRead || prepared || bridge.bytes + STREAM_MEDIA_MAX_FRAME > 4u * 1024u * 1024u) return;
bridge.wantRead = false;
stream_media_packet packet{};
int result = stream_media_read(bridge.handle, bridge.position, bridge.wantVideo, bridge.audioAhead, &packet);
DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "desktop media read id=%llu PTS=%lld video=%d audio_ahead=%lld result=%d kind=%d records=%zu",
(unsigned long long)bridge.viewer.id, (long long)bridge.position, bridge.wantVideo,
(long long)bridge.audioAhead, result, result == STREAM_OK ? (int)packet.kind : -1, bridge.output.size());
if (stream_media_get_status(bridge.handle, &bridge.status) < 0) result = STREAM_ERR_FORMAT;
if (result == STREAM_OK) {
if (packet.kind == STREAM_MEDIA_READ_END) closeViewer(StreamRecordKind::End, 0);
else mediaFrame(nullptr, &packet.format, packet.kind == STREAM_MEDIA_READ_VIDEO ? &packet.video : nullptr,
packet.kind == STREAM_MEDIA_READ_AUDIO ? &packet.audio : nullptr);
} else if (result == STREAM_MORE) { bridge.wantRead = true; schedule(); }
else if (result == STREAM_WAIT) bridge.wantRead = bridge.waiting = true;
else if (result == STREAM_GAP || result == STREAM_EXPIRED) {
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop playback resync id=%llu result=%d", (unsigned long long)bridge.viewer.id, result);
stream_media_resync(bridge.handle); bridge.position = -1;
bridge.status = {};
output({bridge.viewer.id, 0, StreamRecordKind::Gap});
} else {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop playback read failed result=%d", result);
closeViewer(StreamRecordKind::Closed, STREAM_CLOSE_FAILED);
}
}
}
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;
}
if (stream_publisher_create(&bridge.publisher, inst) < 0) return false;
bridge.inst = inst; bridge.output.clear(); bridge.bytes = 0;
stream_set_event_callback(inst, mediaEvent, nullptr);
if (bridge.watchPending) {
bridge.watchPending = false;
if (stream_watch_sources(inst, bridge.watch) != STREAM_OK)
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop initial source watch failed group=%llx", (unsigned long long)bridge.watch);
}
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 (!inst || bridge.inst != inst) return;
if (bridge.pending) uasync_cancel_post(inst->ua, bridge.pending);
stream_watch_sources(inst, 0);
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});
stream_publisher_destroy(bridge.publisher); bridge.publisher = nullptr;
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 (!bridge.inst) {
bridge.watch = group; bridge.watchPending = true;
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "desktop source watch queued until attach group=%llx", (unsigned long long)group); return true;
}
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, int64_t position, bool wantVideo, int64_t audioAhead) {
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; });
bool found = i != bridge.output.end();
if (found) { bridge.bytes -= static_cast<size_t>(i->data.size()); record = std::move(*i); bridge.output.erase(i); }
if (!found && bridge.viewer.id == id && bridge.handle) {
bool changed = bridge.position != position || bridge.wantVideo != wantVideo || bridge.audioAhead != audioAhead;
bridge.position = position; bridge.wantVideo = wantVideo; bridge.audioAhead = audioAhead;
bridge.wantRead = wantVideo || audioAhead >= 0;
if (bridge.wantRead && (!bridge.waiting || changed)) { bridge.waiting = false; schedule(); }
}
return found;
}
bool gui_stream_status(uint64_t id, stream_media_status& status) {
std::lock_guard<std::mutex> guard(bridge.lock);
if (bridge.statusId != id) return false;
status = bridge.status; return true;
}
bool gui_stream_resync(uint64_t id) {
std::lock_guard<std::mutex> guard(bridge.lock);
if (bridge.viewer.id != id || !bridge.handle) return false;
bridge.resync = bridge.wantRead = true; bridge.waiting = false; bridge.position = -1;
bridge.status = {};
bridge.output.clear(); bridge.bytes = 0;
return schedule();
}
bool gui_stream_ready() { std::lock_guard<std::mutex> guard(bridge.lock); return bridge.inst != nullptr; }
bool gui_stream_publish_open(uint64_t id, uint64_t group, const stream_media_format& format) {
std::lock_guard<std::mutex> guard(bridge.lock);
return stream_publisher_open(bridge.publisher, id, group, &format) == 0;
}
int gui_stream_publish_feed(uint64_t id, bool video, uint64_t pts, bool key, const QByteArray& data,
const stream_media_format* format) {
std::lock_guard<std::mutex> guard(bridge.lock);
return stream_publisher_feed(bridge.publisher, id, video, pts, key,
reinterpret_cast<const uint8_t*>(data.constData()), static_cast<size_t>(data.size()), format);
}
void gui_stream_publish_stop(uint64_t id) {
std::lock_guard<std::mutex> guard(bridge.lock); stream_publisher_stop(bridge.publisher, id);
}
bool gui_stream_publish_status(uint64_t id, stream_publisher_status& status) {
std::lock_guard<std::mutex> guard(bridge.lock);
return stream_publisher_snapshot(bridge.publisher, id, &status) == 0;
}
Reference in new issue View Git Blame Copy Permalink