Deliver bounded media batches to desktop consumers

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<const char*>(r->data), static_cast<int>(r->length));
+    u_free(r); return record;
+}
+
 void watch(void*) {
     std::lock_guard<std::mutex> 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<std::mutex> 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<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);
+        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<const char*>(r->data), static_cast<int>(r->length));
-    u_free(r); return true;
+    record = takeRecord(id, r); return true;
+}
+
+bool gui_stream_pull_batch(uint64_t id, std::deque<StreamRecord>& records, int64_t position,
+                            unsigned videoBudget, 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, 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) {

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 <QByteArray>
 #include <cstdint>
+#include <deque>
 #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<StreamRecord>& 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();