diff --git a/tools/chatgui/transport/stream_bridge.cpp b/tools/chatgui/transport/stream_bridge.cpp index e68d1532..645eeb4d 100644 --- a/tools/chatgui/transport/stream_bridge.cpp +++ b/tools/chatgui/transport/stream_bridge.cpp @@ -23,6 +23,22 @@ struct Bridge { stream_publisher* publisher = nullptr; } bridge; +/* Забираем payload из owned записи ядра; next остаётся у обходящего батч caller. */ +StreamRecord takeRecord(uint64_t id, stream_viewer_record* r) { + StreamRecord record; + 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(r->data), static_cast(r->length)); + u_free(r); return record; +} + void watch(void*) { std::lock_guard guard(bridge.lock); bridge.pending = nullptr; @@ -77,9 +93,13 @@ bool gui_stream_watch(uint64_t group, uint64_t owner) { bridge.watches[owner] = group; bridge.watchChanges.insert(owner); return true; } -bool gui_stream_open(uint64_t id, uint64_t group, uint64_t source, uint64_t epoch) { +bool gui_stream_open(uint64_t id, uint64_t group, uint64_t source, uint64_t epoch, void (*notify)(void*), void* arg) { std::lock_guard guard(bridge.lock); - return stream_viewer_open(bridge.viewers, id, group, source, epoch, STREAM_MEDIA_TRACK_VIDEO | STREAM_MEDIA_TRACK_AUDIO) == 0; + if (stream_viewer_open(bridge.viewers, id, group, source, epoch, STREAM_MEDIA_TRACK_VIDEO | STREAM_MEDIA_TRACK_AUDIO) < 0) return false; + if (notify && stream_viewer_set_notify(bridge.viewers, id, notify, arg) < 0) { + stream_viewer_stop(bridge.viewers, id); return false; + } + return true; } void gui_stream_stop(uint64_t id) { @@ -90,20 +110,26 @@ bool gui_stream_pull(uint64_t id, StreamRecord& record, int64_t position, bool w std::lock_guard 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); + if (bridge.viewers) stream_viewer_request(bridge.viewers, id, position, wantVideo, audioAhead, 1); 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; + if (r->next) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop single reader received batch id=%llu", (unsigned long long)id); + stream_viewer_records_free(r); stream_viewer_resync(bridge.viewers, id); return false; } - record.id = id; record.pts = r->pts; record.format = r->format; record.flags = r->flags; record.reason = r->reason; - record.data = QByteArray(reinterpret_cast(r->data), static_cast(r->length)); - u_free(r); return true; + record = takeRecord(id, r); return true; +} + +bool gui_stream_pull_batch(uint64_t id, std::deque& records, int64_t position, + unsigned videoBudget, int64_t audioAhead) { + std::lock_guard guard(bridge.lock); + auto* r = stream_viewer_pull(bridge.viewers, id); + if (!r) { + if (bridge.viewers) stream_viewer_request(bridge.viewers, id, position, videoBudget, audioAhead, STREAM_VIEWER_BATCH_RECORDS); + return false; + } + while (r) { auto* next = r->next; records.push_back(takeRecord(id, r)); r = next; } + return true; } bool gui_stream_status(uint64_t id, stream_media_status& status) { diff --git a/tools/chatgui/transport/stream_bridge.h b/tools/chatgui/transport/stream_bridge.h index 5c92a8ed..712f4464 100644 --- a/tools/chatgui/transport/stream_bridge.h +++ b/tools/chatgui/transport/stream_bridge.h @@ -1,12 +1,13 @@ /* Desktop live media: attach/detach в uasync до destroy ядра; остальные вызовы потокобезопасны. * Один общий publisher и независимые viewers; команды содержат IDs, не borrowed pointers ядра. - * Один ответ на запрос каждого плеера; следующий кадр не читается автоматически. + * Плеер запрашивает ограниченный батч; каждый трек имеет свой бюджет, новые данные будят worker. * Watch до attach запоминает выбранную группу. Каталог идёт через CHAT_EVT_STREAM_SOURCES, * media pull читает ближайшие кадры из общего RS-кольца; копия нужна только для передачи между потоками. */ #pragma once #include #include +#include #include "../../../src/stream/stream_media.h" #include "../../../src/stream/stream_media_buffer.h" #include "../../../src/stream/stream_publisher.h" @@ -25,10 +26,16 @@ bool gui_stream_attach(UTUN_INSTANCE* inst); void gui_stream_detach(UTUN_INSTANCE* inst); /* owner=0 — выбранный чат; окно многопросмотра держит свой ненулевой owner. group=0 снимает только его. */ bool gui_stream_watch(uint64_t group, uint64_t owner = 0); -bool gui_stream_open(uint64_t id, uint64_t group, uint64_t source, uint64_t epoch); +/* notify только будит worker; arg живёт до gui_stream_stop/detach. Callback не вызывает bridge/API ядра. */ +bool gui_stream_open(uint64_t id, uint64_t group, uint64_t source, uint64_t epoch, + void (*notify)(void*) = nullptr, void* arg = nullptr); void gui_stream_stop(uint64_t id); bool gui_stream_pull(uint64_t id, StreamRecord& record, int64_t position = -1, bool wantVideo = true, int64_t audioAhead = STREAM_MEDIA_READ_AHEAD_US); +/* Добавляет готовый батч в records; при пустом ответе запрашивает до videoBudget кадров и audioAhead звука. + * Вызывать после расходования нужного трека: удерживаемые видеокадры не мешают запросить только звук. */ +bool gui_stream_pull_batch(uint64_t id, std::deque& records, int64_t position, + unsigned videoBudget, int64_t audioAhead); bool gui_stream_status(uint64_t id, stream_media_status& status); bool gui_stream_resync(uint64_t id); bool gui_stream_ready();