utun2/tools/chatgui/transport/stream_bridge.cpp

#include "stream_bridge.h"
#include <mutex>

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_viewer.h"
}

namespace {
struct Bridge {
    std::mutex lock;                  // Ограничивает lifetime managers для callers GUI/workers.
    UTUN_INSTANCE* inst = nullptr;
    posted_task* pending = nullptr;
    uint64_t watch = 0;
    bool watchPending = false;
    stream_viewers* viewers = nullptr;
    stream_publisher* publisher = nullptr;
} bridge;

void watch(void*) {
    std::lock_guard<std::mutex> guard(bridge.lock);
    bridge.pending = nullptr; bridge.watchPending = false;
    if (stream_watch_sources(bridge.inst, bridge.watch) != STREAM_OK)
        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop source watch failed group=%llx", (unsigned long long)bridge.watch);
}
}

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;
    if (stream_viewers_create(&bridge.viewers, inst) < 0) {
        stream_publisher_destroy(bridge.publisher); bridge.publisher = nullptr; return false;
    }
    bridge.inst = inst;
    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_viewers_destroy(bridge.viewers); bridge.viewers = nullptr;
    stream_publisher_destroy(bridge.publisher); bridge.publisher = nullptr;
    bridge.inst = nullptr; bridge.watchPending = bridge.watch != 0;
    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.pending) {
        auto* task = static_cast<posted_task*>(u_calloc(1, sizeof(posted_task)));
        if (!task) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop watch task allocation failed"); return false; }
        task->callback = watch; bridge.pending = task;
        uasync_post_reserved(bridge.inst->ua, task);
    }
    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);
    return stream_viewer_open(bridge.viewers, id, group, source, epoch, STREAM_MEDIA_TRACK_VIDEO | STREAM_MEDIA_TRACK_AUDIO) == 0;
}

void gui_stream_stop(uint64_t id) {
    std::lock_guard<std::mutex> guard(bridge.lock); stream_viewer_stop(bridge.viewers, id);
}

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* r = stream_viewer_pull(bridge.viewers, id);
    if (!r) {
        if (bridge.viewers) stream_viewer_request(bridge.viewers, id, position, wantVideo, audioAhead);
        return false;
    }
    switch (r->kind) {
    case STREAM_VIEWER_VIDEO: record.kind = StreamRecordKind::Video; break;
    case STREAM_VIEWER_AUDIO: record.kind = StreamRecordKind::Audio; break;
    case STREAM_VIEWER_GAP: record.kind = StreamRecordKind::Gap; break;
    case STREAM_VIEWER_OPENED: record.kind = StreamRecordKind::Opened; break;
    case STREAM_VIEWER_END: record.kind = StreamRecordKind::End; break;
    case STREAM_VIEWER_CLOSED: record.kind = StreamRecordKind::Closed; break;
    }
    record.id = id; record.pts = r->pts; record.format = r->format; record.flags = r->flags; record.reason = r->reason;
    record.data = QByteArray(reinterpret_cast<const char*>(r->data), static_cast<int>(r->length));
    u_free(r); return true;
}

bool gui_stream_status(uint64_t id, stream_media_status& status) {
    std::lock_guard<std::mutex> guard(bridge.lock);
    return stream_viewer_status(bridge.viewers, id, &status) == 0;
}

bool gui_stream_resync(uint64_t id) {
    std::lock_guard<std::mutex> guard(bridge.lock);
    return stream_viewer_resync(bridge.viewers, id) == 0;
}

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);
}
bool gui_stream_publish_audio(uint64_t id, bool enabled) {
    std::lock_guard<std::mutex> guard(bridge.lock);
    return stream_publisher_set_audio_enabled(bridge.publisher, id, enabled) == 0;
}
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;
}