Test bounded batches, track budgets, notifications and FIN

tools/chatgui-android/libutun_lite/stream_bridge.c

@@ -174,7 +174,7 @@ int android_stream_read_request(uint64_t id, int64_t position, int want_video, i
                     want_video, (long long)audio_ahead); return -1;
     }
     pthread_mutex_lock(&media_lock);
-    if (viewers && id != publisher_id) stream_viewer_request(viewers, id, position, want_video, audio_ahead);
+    if (viewers && id != publisher_id) stream_viewer_request(viewers, id, position, want_video, audio_ahead, 1);
     /* CLOSED/END остаются доступны pull даже после закрытия cursor; неизвестный ID даёт пустой pull. */
     pthread_mutex_unlock(&media_lock); return 0;
 }

tools/chatgui/tests/test_stream_view.cpp

@@ -31,6 +31,7 @@ extern "C" {
 #include "chat/chat_event.h"
 #include "stream/stream.h"
 #include "stream/stream_member.h"
+#include "stream/stream_viewer.h"
 #include "debug_config.h"
 #include "mem.h"
 #include "sqlite3.h"
@@ -543,6 +544,76 @@ private slots:
 #endif
     }
 
+    void boundedBatchesAndNotifications() {
+        QCOMPARE(stream_publish_finish(publisher), STREAM_OK); stream_close(publisher); publisher = nullptr;
+        QCOMPARE(stream_publish_open(inst, group, nullptr, &publisher), STREAM_OK);
+        stream_info info{}; QCOMPARE(stream_get_info(publisher, &info), STREAM_OK); epoch = info.descriptor.source_epoch;
+        mediaPts = frameId = 0; revision = 1;
+        unsigned notifications = 0;
+        struct Readers { ~Readers() { gui_stream_stop(301); gui_stream_stop(302); } } readers;
+        QVERIFY(gui_stream_open(301, group, inst->node_id, epoch, [](void* arg) { ++*static_cast<unsigned*>(arg); }, &notifications));
+        QVERIFY(gui_stream_open(302, group, inst->node_id, epoch));
+        StreamRecord opened;
+        QVERIFY(receive(301, opened)); QCOMPARE(opened.kind, StreamRecordKind::Opened);
+        QVERIFY(receive(302, opened)); QCOMPARE(opened.kind, StreamRecordKind::Opened);
+        const QByteArray frame(1500, 'k'); const QList<QByteArray> audio(5, QByteArray("opus"));
+        publish(frame, audio, 3);
+        stream_media_status status{};
+        QTRY_VERIFY(gui_stream_status(301, status) && status.initialized);
+        QTRY_VERIFY(gui_stream_status(302, status) && status.initialized);
+        for (unsigned i = 1; i < 10; i++) { publish(frame, audio, 3); uasync_poll(ua, 0); }
+        auto batch = [&](uint64_t id, unsigned video, int64_t ahead, std::deque<StreamRecord>& out) {
+            QElapsedTimer elapsed; elapsed.start();
+            while (elapsed.elapsed() < 1000) {
+                if (gui_stream_pull_batch(id, out, 100000, video, ahead)) return true;
+                QTest::qWait(1);
+            }
+            return false;
+        };
+        std::deque<StreamRecord> records;
+        QVERIFY(batch(301, 2, 1000000, records));
+        QVERIFY(records.size() > 2); QVERIFY(records.size() <= STREAM_VIEWER_BATCH_RECORDS);
+        unsigned videos = 0, audios = 0, batches = 1;
+        uint64_t lastVideo = 0, lastAudio = 0;
+        auto consume = [&] {
+            for (const auto& record : records) {
+                if (record.kind == StreamRecordKind::Video) {
+                    QCOMPARE(record.pts, uint64_t(100000 + videos / 3 * 100000 + videos % 3 * 100000 / 3));
+                    QCOMPARE(record.data, frame); lastVideo = record.pts; videos++;
+                } else if (record.kind == StreamRecordKind::Audio) {
+                    QCOMPARE(record.pts, uint64_t(100000 + audios * 20000)); lastAudio = record.pts; audios++;
+                }
+            }
+            records.clear();
+        };
+        consume(); QCOMPARE(videos, 2u); QVERIFY(audios > 0);
+        unsigned before = notifications; QTest::qWait(20); QCOMPARE(notifications, before);
+        // Видео приостановлено потребителем: звук продолжается, video cursor остаётся на третьем кадре.
+        QVERIFY(batch(301, 0, 1000000, records)); batches++;
+        for (const auto& record : records) QCOMPARE(record.kind, StreamRecordKind::Audio);
+        consume(); QCOMPARE(videos, 2u);
+        QVERIFY(batch(301, 2, -1, records)); batches++;
+        for (const auto& record : records) QCOMPARE(record.kind, StreamRecordKind::Video);
+        consume(); QCOMPARE(videos, 4u);
+        QVERIFY(batch(302, 2, -1, records)); QCOMPARE(records.front().pts, uint64_t(100000)); records.clear();
+        QCOMPARE(stream_publish_finish(publisher), STREAM_OK);
+        bool ended = false;
+        for (unsigned i = 0; i < 50 && !ended; i++) {
+            QVERIFY(batch(301, 2, 1000000, records)); batches++;
+            QVERIFY(records.size() <= STREAM_VIEWER_BATCH_RECORDS);
+            for (const auto& record : records) {
+                QVERIFY(record.kind != StreamRecordKind::Closed && record.kind != StreamRecordKind::Gap);
+                if (record.kind == StreamRecordKind::End) { QCOMPARE(&record, &records.back()); ended = true; }
+            }
+            consume();
+        }
+        QVERIFY(ended); QCOMPARE(videos, 30u); QCOMPARE(audios, 50u); QVERIFY(batches < 30);
+        gui_stream_stop(301); before = notifications; QTest::qWait(20); QCOMPARE(notifications, before);
+        DEBUG_INFO(DEBUG_CATEGORY_STREAM,
+            "fixture batches: video=%u audio=%u batches=%u last_video=%llu last_audio=%llu signals=%u independent_readers=1",
+            videos, audios, batches, (unsigned long long)lastVideo, (unsigned long long)lastAudio, notifications);
+    }
+
     void realH264RenderingAndClose() {
 #ifdef HAVE_FFMPEG
         /* Фикстура протокола выше содержит фиктивный H264; реальный decoder получает отдельную эпоху. */