Embed stream players and verify grid and concurrent playback

tools/chatgui/src/streamplayer_window.cpp

@@ -50,6 +50,7 @@ struct StreamPlayerWindow::Playback {
     } audioGap;
     uint64_t id = ++nextId, group = 0, source = 0, epoch = 0;
     std::atomic<bool> running{true}, audioAvailable{false};
+    std::atomic<bool> complete{false}, failed{false};
     std::atomic<float> volume{1.0f};
     float outputVolume = 0.0f;
     uint64_t audioGeneration = 0;
@@ -86,11 +87,12 @@ struct StreamPlayerWindow::Playback {
     }
 };
 
-StreamPlayerWindow::StreamPlayerWindow(uint64_t group, uint64_t source, uint64_t epoch, const QString& name, QWidget* parent)
+StreamPlayerWindow::StreamPlayerWindow(uint64_t group, uint64_t source, uint64_t epoch, const QString& name, QWidget* parent, bool embedded)
     : QWidget(parent), m_playback(std::make_unique<Playback>()) {
     m_playback->group = group; m_playback->source = source; m_playback->epoch = epoch;
     setWindowTitle(text("Live stream: %1").arg(name)); resize(800, 500);
-    auto* layout = new QVBoxLayout(this); layout->addWidget(new QLabel(name, this));
+    setObjectName(QStringLiteral("streamPlayer_%1").arg(source));
+    auto* layout = new QVBoxLayout(this); auto* title = new QLabel(name, this); layout->addWidget(title);
     m_status = new QLabel(text("Connecting…"), this); layout->addWidget(m_status);
     m_render = new VideoRenderWidget(this, true); layout->addWidget(m_render, 1);
     auto* controls = new QHBoxLayout;
@@ -98,6 +100,11 @@ StreamPlayerWindow::StreamPlayerWindow(uint64_t group, uint64_t source, uint64_t
     volume->setAccessibleName(text("Stream volume")); controls->addWidget(new QLabel(text("Volume"), this)); controls->addWidget(volume);
     auto* fullscreen = new QPushButton(text("Fullscreen"), this); controls->addWidget(fullscreen);
     auto* close = new QPushButton(text("Close player"), this); controls->addWidget(close); layout->addLayout(controls);
+    if (embedded) {
+        fullscreen->hide(); close->hide(); layout->setContentsMargins(3, 3, 3, 3); layout->setSpacing(2);
+        title->setSizePolicy(QSizePolicy::Ignored, QSizePolicy::Preferred); title->setToolTip(name);
+        m_status->setSizePolicy(QSizePolicy::Ignored, QSizePolicy::Preferred); setMinimumSize(0, 0);
+    }
     connect(volume, &QSlider::valueChanged, this, [this](int value) { m_playback->volume = value / 100.0f; });
     connect(fullscreen, &QPushButton::clicked, this, [this] { if (isFullScreen()) showNormal(); else showFullScreen(); });
     connect(close, &QPushButton::clicked, this, &QWidget::close);
@@ -109,6 +116,13 @@ StreamPlayerWindow::StreamPlayerWindow(uint64_t group, uint64_t source, uint64_t
 
 StreamPlayerWindow::~StreamPlayerWindow() { stop(); }
 
+bool StreamPlayerWindow::finished() const { return m_playback->complete; }
+bool StreamPlayerWindow::failed() const { return m_playback->failed; }
+int StreamPlayerWindow::drainTimeoutMs() const {
+    auto& p = *m_playback; std::lock_guard<std::mutex> guard(p.lock);
+    return p.complete ? 0 : static_cast<int>(std::clamp<int64_t>((p.ready - p.clock(nowUs())) / 1000 + 2000, 2000, 30000));
+}
+
 /* Hardware callback заполняет только ближайшую порцию; source time тишины остаётся неподвижным. */
 void StreamPlayerWindow::attachAudio() {
     if (!m_playback->running || m_audioAttached) return;
@@ -173,8 +187,9 @@ void StreamPlayerWindow::tick() {
                               (unsigned long long)p.id, image.width(), image.height(), (long long)image.bytesPerLine());
         emit framePresented();
     }
-    if (finished) {
-        m_status->setText(text("Stream ended"));
+    if (finished || p.failed) {
+        if (finished) m_status->setText(text("Stream ended"));
+        p.complete = true;
         SoundManager::instance()->detachStream(p.id); m_audioAttached = false; m_timer.stop();
     }
 }
@@ -200,6 +215,7 @@ void StreamPlayerWindow::decode() {
     auto& p = *m_playback;
 #ifndef HAVE_FFMPEG
     DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop live decoder unavailable: FFmpeg disabled");
+    p.failed = p.complete = true;
     emit statusChanged(text("Live video playback requires FFmpeg"));
 #else
     AVCodecContext* decoder = nullptr; AVFrame* frame = av_frame_alloc(); AVPacket* packet = av_packet_alloc();
@@ -227,6 +243,7 @@ void StreamPlayerWindow::decode() {
         deadline = nowUs() + 10000000; emit statusChanged(text("Waiting for a key frame…"));
     };
     auto fail = [&](const QString& reason) {
+        p.failed = p.complete = true;
         DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop viewer failed id=%llu: %s", (unsigned long long)p.id, qPrintable(reason));
         emit statusChanged(reason);
     };

tools/chatgui/src/streamplayer_window.h

@@ -16,9 +16,13 @@ class QKeyEvent;
 class StreamPlayerWindow : public QWidget {
     Q_OBJECT
 public:
-    explicit StreamPlayerWindow(uint64_t group, uint64_t source, uint64_t epoch, const QString& name, QWidget* parent = nullptr);
+    explicit StreamPlayerWindow(uint64_t group, uint64_t source, uint64_t epoch, const QString& name,
+                                QWidget* parent = nullptr, bool embedded = false);
     ~StreamPlayerWindow() override;
     void stop();
+    bool finished() const;
+    bool failed() const;
+    int drainTimeoutMs() const; // Остаток готового media + 2s для FIN; максимум 30s при исчезновении источника.
 signals:
     void statusChanged(const QString& status);
     void framePresented();

tools/chatgui/tests/test_stream_view.cpp

@@ -1,11 +1,14 @@
 /* Настоящее ядро/actor, подписанный медиаблок и H.264 decoder: каталог до подписки, эпоха, показ и teardown. */
 #include "../src/streamplayer_window.h"
+#include "../src/streamgrid_window.h"
 #include "../src/videoplayer_window.h"
 #include "../src/sound_manager.h"
 #include "../transport/stream_bridge.h"
 #include <QtTest>
 #include <QJsonDocument>
 #include <QJsonArray>
+#include <QJsonObject>
+#include <QPointer>
 #include <QLabel>
 #include <QSlider>
 #include <QPainter>
@@ -386,6 +389,44 @@ private slots:
         QVERIFY(gui_stream_watch(0, 77)); QVERIFY(gui_stream_watch(group)); QTest::qWait(20);
     }
 
+    void gridKeepsPlayersAndWaitsForSources() {
+#if defined(TEST_NULL_AUDIO) && defined(HAVE_FFMPEG)
+        QVERIFY(SoundManager::instance()->init());
+        StreamGridWindow grid(group); grid.show();
+        auto snapshot = [&](int count, bool valid = true) {
+            QJsonArray sources;
+            for (int i = 0; i < count; ++i) sources.append(QJsonObject{
+                {"id", QString::number(inst->node_id ^ static_cast<uint64_t>(i))},
+                {"epoch", QString::number(epoch, 16)}, {"name", QStringLiteral("Source %1").arg(i)}, {"self", i == 0 ? 1 : 0}});
+            QByteArray bytes(9, '\0'); stream_write_be(reinterpret_cast<uint8_t*>(bytes.data()), group, 8); bytes[8] = valid;
+            bytes += QJsonDocument(sources).toJson(QJsonDocument::Compact);
+            grid.onSources(reinterpret_cast<const uint8_t*>(bytes.constData()), bytes.size());
+        };
+        snapshot(1);
+        QPointer<StreamPlayerWindow> first = grid.findChild<StreamPlayerWindow*>(QStringLiteral("streamPlayer_%1").arg(inst->node_id));
+        QVERIFY(first);
+        for (int count : {2, 4, 5, 9, 10, 16}) {
+            snapshot(count); QCOMPARE(grid.tileCount(), count);
+            QCOMPARE(grid.gridSide(), static_cast<int>(std::ceil(std::sqrt(count))));
+            QVERIFY(first); QCOMPARE(grid.findChild<StreamPlayerWindow*>(first->objectName()), first.data());
+        }
+        snapshot(0, false); QCOMPARE(grid.tileCount(), 16); // Недоступность каталога не снимает readers.
+        snapshot(1); QTRY_COMPARE_WITH_TIMEOUT(grid.tileCount(), 1, 3000);
+        QCOMPARE(grid.gridSide(), 1); QVERIFY(first);
+        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;
+        snapshot(1); QVERIFY(first.isNull()); QCOMPARE(grid.tileCount(), 1);
+        snapshot(0); QTRY_COMPARE_WITH_TIMEOUT(grid.tileCount(), 0, 3000);
+        QVERIFY(grid.isVisible()); QCOMPARE(grid.gridSide(), 0);
+        snapshot(1); QCOMPARE(grid.tileCount(), 1);
+        grid.close();
+        QTRY_VERIFY(stream_get_info(publisher, &info) == STREAM_OK && info.owners == 1);
+#else
+        QSKIP("Grid lifetime fixture requires null audio and FFmpeg");
+#endif
+    }
+
     void realH264RenderingAndClose() {
 #ifdef HAVE_FFMPEG
         /* Фикстура протокола выше содержит фиктивный H264; реальный decoder получает отдельную эпоху. */
@@ -459,6 +500,16 @@ private slots:
 #ifdef TEST_NULL_AUDIO
         QCOMPARE(silenceGaps.load(), gapsBefore);
 #endif
+        {
+            StreamPlayerWindow second(group, inst->node_id, epoch, "Second reader"); second.show();
+            QSignalSpy secondShown(&second, &StreamPlayerWindow::framePresented);
+            int beforeSecond = shown.count();
+            QTRY_VERIFY_WITH_TIMEOUT(secondShown.count() > 3 && shown.count() > beforeSecond + 3, 5000);
+            stream_info together{}; QCOMPARE(stream_get_info(publisher, &together), STREAM_OK); QCOMPARE(together.owners, 3u);
+            second.close();
+            QTRY_VERIFY(stream_get_info(publisher, &together) == STREAM_OK && together.owners == 2);
+            beforeSecond = shown.count(); QTRY_VERIFY(shown.count() > beforeSecond + 3);
+        }
         int beforeSwitch = shown.count(); revision++; rotation = 270;
         QTRY_VERIFY_WITH_TIMEOUT(shown.count() > beforeSwitch + 2, 5000);
         stream_info switched{}; QCOMPARE(stream_get_info(publisher, &switched), STREAM_OK);