Keep capture completion observable and retain session ownership through stop

tools/chatgui/CMakeLists.txt

@@ -1024,7 +1024,7 @@ target_link_libraries(test_stream_audio PRIVATE utun_voice utun pthread ${CMAKE_
 add_test(NAME test_stream_audio COMMAND test_stream_audio)
 set_tests_properties(test_stream_audio PROPERTIES TIMEOUT 30)
 
-add_executable(test_stream_publish tests/test_stream_publish.cpp transport/stream_bridge.cpp
+add_executable(test_stream_publish tests/test_stream_publish.cpp transport/stream_bridge.cpp src/videoplayer_window.cpp src/videoplayer_engine.cpp
     src/sound_manager.cpp src/voice_audio_io.cpp src/audio_device.cpp transport/miniaudio_impl.c transport/audio_diagnostics.cpp)
 add_dependencies(test_stream_publish chatgui_translations)
 target_link_libraries(test_stream_publish PRIVATE chatgui_stream_publish ${QT_TEST_LIB} utun_voice pthread ${CMAKE_DL_LIBS})

tools/chatgui/src/stream_publish_engine.cpp

@@ -53,14 +53,14 @@ struct StreamPublishEngine::Impl {
     std::condition_variable wake;
     std::thread videoThread, audioThread, joinThread;
     std::atomic<bool> workersJoined{false}, videoExited{false}, audioExited{false};
-    bool stopping = false, finSent = false, destroying = false;
+    bool stopping = false, destroying = false;
     int64_t stopStarted = 0, stopReport = 0;
     std::atomic<bool> running{false}, muted{false};
     std::atomic<unsigned> audioDropped{0}, audioMissing{0};
     std::atomic<int64_t> lastAudioAt{0};
     AudioEventQueue<Audio, 32> audio;
     uint64_t id = 0, group = 0, generation = 0, sourceGeneration = 0, confirmedGeneration = 0, frameGeneration = 0, audioSequence = 0;
-    int64_t origin = 0, started = 0, framePts = -1, backendOrigin = -1, backendAnchor = -1;
+    int64_t origin = 0, started = 0, framePts = -1;
     int64_t audioAnchor = -1;
     uint64_t audioFrames = 0;
     int channels = 2;
@@ -89,7 +89,6 @@ struct StreamPublishEngine::Impl {
     std::unique_ptr<VideoCapture> capture;
     QList<StreamVideoSource> sources;
     QImage frame, latestImage, previewFrame, presentedPreview;
-    int64_t lastImageCaptureAt = 0;
     int64_t previewTickAt = 0, previewReportAt = 0, previewGapMax = 0, previewDispatchMax = 0;
     unsigned previewUpdates = 0; // Диагностика GUI-предпросмотра; только GUI-поток.
 #endif
@@ -354,8 +353,9 @@ bool StreamPublishEngine::reconfigure(const StreamPublishConfig& config) {
     }
     if (config.showCursor != previous.showCursor)
         DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "desktop cursor selected enabled=%d supported=%d", config.showCursor, cursorSupported(config.source));
-    bool sourceChange = !config.source.sameSource(previous.source) || (config.source.kind == StreamVideoSource::Camera
-        && (config.video.width != previous.video.width || config.video.height != previous.video.height || config.video.fps != previous.video.fps));
+    bool sourceChange = !config.source.sameSource(previous.source) || config.video.fps != previous.video.fps
+        || config.showCursor != previous.showCursor || (config.source.kind == StreamVideoSource::Camera
+        && (config.video.width != previous.video.width || config.video.height != previous.video.height));
     { std::lock_guard<std::mutex> guard(m->lock); m->wanted = config; m->generation++; m->started = monotonicUs(); m->softwareRequest = 0;
 #ifdef HAVE_STREAM_CAPTURE
       if (!sourceChange) { m->frame = m->latestImage; m->frameGeneration = m->generation; }
@@ -392,7 +392,7 @@ void StreamPublishEngine::publicationFailed(uint64_t id, const QString& error) {
 void StreamPublishEngine::stop() {
     if (!m->running.exchange(false)) return;
     int64_t started = monotonicUs();
-    m->stopping = true; m->finSent = false; m->stopStarted = m->stopReport = started;
+    m->stopping = true; m->stopStarted = m->stopReport = started;
     DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "desktop stop begin id=%llu video_stage=%s stage_age_ms=%lld converted=%llu encoded=%llu delivered=%llu",
         (unsigned long long)m->id, m->videoStage.load(), (long long)((started - m->videoStageAt.load()) / 1000),
         (unsigned long long)m->convertedFrames.load(), (unsigned long long)m->encodedFrames.load(),
@@ -400,7 +400,7 @@ void StreamPublishEngine::stop() {
     { std::lock_guard<std::mutex> guard(m->lock); m->softwareRequest = 0; }
     m->wake.notify_all();
     DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "desktop stop closing video source");
-    gui_stream_publish_stop(m->id); m->finSent = true;
+    gui_stream_publish_stop(m->id);
     DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop publisher admission closed id=%llu; accepted data drains independently of capture",
         (unsigned long long)m->id);
     closeSource();
@@ -423,16 +423,13 @@ bool StreamPublishEngine::finishStop() {
         if (!m->workersJoined) return false;
         m->joinThread.join();
     }
-    if (!m->finSent) {
-        gui_stream_publish_stop(m->id); m->finSent = true;
-        DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop workers stopped id=%llu elapsed_ms=%lld microphone_closing=%d; draining publisher",
-            (unsigned long long)m->id, (long long)((monotonicUs() - m->stopStarted) / 1000), m_capture.closing());
-    }
     if (m_capture.closing()) return false;
 #ifdef HAVE_STREAM_CAPTURE
     if (m->capture && m->capture->closing()) return false;
 #endif
     m_microphone.release(); m->stopping = false;
+    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop publisher producers released id=%llu elapsed_ms=%lld; actor drains independently",
+        (unsigned long long)m->id, (long long)((monotonicUs() - m->stopStarted) / 1000));
     return true;
 }
 

tools/chatgui/src/video_capture.cpp

@@ -31,7 +31,7 @@ struct VideoCapture::Session : QObject {
     QByteArray device;
     int width = 640, height = 480, fps = 30;
     QString phase = QStringLiteral("launch");
-    qint64 progressAt = 0, frameAt = 0, stopAt = 0, stopStarted = 0, stageAt = 0;
+    qint64 progressAt = 0, frameAt = 0, stopAt = 0, stopStarted = 0, stageAt = 0, workerPid = 0;
 
     explicit Session(VideoCapture* target, Usage use) : owner(target), usage(use) {
         server.setSocketOptions(QLocalServer::UserAccessOption);
@@ -53,6 +53,11 @@ struct VideoCapture::Session : QObject {
             }
         });
         process.setProcessChannelMode(QProcess::MergedChannels);
+        connect(&process, &QProcess::started, this, [this] {
+            workerPid = process.processId();
+            DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "capture process started pid=%lld generation=%llu", (long long)workerPid,
+                (unsigned long long)generation);
+        });
         connect(&process, &QProcess::readyReadStandardOutput, this, [this] {
             const auto log = process.readAllStandardOutput().trimmed();
             for (const auto& line : log.split('\n')) {
@@ -66,12 +71,13 @@ struct VideoCapture::Session : QObject {
         connect(&process, &QProcess::errorOccurred, this, [this](QProcess::ProcessError error) {
             if (error == QProcess::FailedToStart) {
                 fail(QStringLiteral("Cannot start camera worker: %1").arg(process.errorString()));
+                if (stopping) { stopping = false; stopAt = 0; if (owner) emit owner->stopped(); }
                 if (retiring) deleteLater();
             }
         });
         connect(&process, QOverload<int, QProcess::ExitStatus>::of(&QProcess::finished), this, [this](int code, QProcess::ExitStatus status) {
             DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "capture process exited pid=%lld generation=%llu code=%d status=%d expected=%d forced=%d "
-                "last_stage=%s stop_ms=%lld", (long long)process.processId(), (unsigned long long)generation, code, int(status),
+                "last_stage=%s stop_ms=%lld", (long long)workerPid, (unsigned long long)generation, code, int(status),
                 stopping, forced, qPrintable(phase), stopStarted ? (long long)((call_video_playout_now_us() - stopStarted) / 1000) : 0);
             if (socket) { socket->disconnect(this); socket->deleteLater(); socket = nullptr; }
             server.close(); watchdog.stop();
@@ -109,7 +115,7 @@ struct VideoCapture::Session : QObject {
     void launch() {
         if (exiting || process.state() != QProcess::NotRunning) return;
         server.close();
-        reader = {}; failureReported = false; phase = QStringLiteral("launch"); progressAt = call_video_playout_now_us();
+        reader = {}; failureReported = false; phase = QStringLiteral("launch"); progressAt = stageAt = call_video_playout_now_us();
         if (!pixels) {
             if (!buffer.create(usage == Usage::Stream ? MaxFrameBytes : FrameBytes)) {
                 fail(QStringLiteral("Cannot allocate camera frame buffer: %1").arg(buffer.errorString())); return;

tools/chatgui/src/video_capture_worker.cpp

@@ -8,6 +8,8 @@
 #include "debug_config.h"
 #include <QGuiApplication>
 #include <QTimer>
+#include <QTransform>
+#include <QUuid>
 #include <mutex>
 #include <climits>
 #include <algorithm>
@@ -49,6 +51,7 @@ class QtVideoCaptureSource : public VideoCaptureSource {
     QPointer<QScreen> selectedScreen;
     QHash<QByteArray, QCapturableWindow> windows;
     quint64 nextWindow = 0;
+    const QByteArray windowPrefix = QUuid::createUuid().toByteArray(QUuid::Id128) + '-';
     CallCameraRate rate;
 #ifdef UTUN_STREAM_PORTAL
     std::unique_ptr<StreamPortalCapture> portal;
@@ -103,7 +106,7 @@ public:
         for (const auto& item : QWindowCapture::capturableWindows()) {
             QByteArray id;
             for (auto it = windows.cbegin(); it != windows.cend(); ++it) if (it.value() == item) { id = it.key(); break; }
-            if (id.isEmpty()) id = "window-" + QByteArray::number(++nextWindow);
+            if (id.isEmpty()) id = windowPrefix + QByteArray::number(++nextWindow);
             current.insert(id, item);
             result.append({item.description(), id, {}, VideoCaptureKind::Window, {}});
         }