diff --git a/tools/chatgui-android/libutun_lite/stream_bridge.c b/tools/chatgui-android/libutun_lite/stream_bridge.c index d5606921..500db089 100644 --- a/tools/chatgui-android/libutun_lite/stream_bridge.c +++ b/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; } diff --git a/tools/chatgui/tests/test_stream_view.cpp b/tools/chatgui/tests/test_stream_view.cpp index 2a8cc2d8..2174b78d 100644 --- a/tools/chatgui/tests/test_stream_view.cpp +++ b/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(arg); }, ¬ifications)); + 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 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& 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 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 получает отдельную эпоху. */