551 lines
33 KiB
C++
551 lines
33 KiB
C++
#include "stream_publish_engine.h"
|
|
#include "sound_manager.h"
|
|
#include "audio_event_queue.h"
|
|
#include "../transport/stream_bridge.h"
|
|
#include "debug_config.h"
|
|
#include <QGuiApplication>
|
|
#include <QScreen>
|
|
#include <QRandomGenerator>
|
|
#include <QTransform>
|
|
#include <thread>
|
|
#include <mutex>
|
|
#include <condition_variable>
|
|
#include <chrono>
|
|
#include <atomic>
|
|
#include <cstring>
|
|
#include <climits>
|
|
#include <algorithm>
|
|
extern "C" {
|
|
#include "opus_codec.h"
|
|
#include "audio_compressor.h"
|
|
}
|
|
#ifdef HAVE_STREAM_CAPTURE
|
|
#include <QCamera>
|
|
#include <QMediaDevices>
|
|
#include <QMediaCaptureSession>
|
|
#include <QVideoSink>
|
|
#include <QScreenCapture>
|
|
#include <QWindowCapture>
|
|
#include <QVideoFrame>
|
|
#endif
|
|
|
|
namespace {
|
|
int64_t monotonicUs() { return std::chrono::duration_cast<std::chrono::microseconds>(
|
|
std::chrono::steady_clock::now().time_since_epoch()).count(); }
|
|
QString text(const char* value) { return QCoreApplication::translate("ChatGui", value); }
|
|
}
|
|
struct StreamPublishEngine::Impl {
|
|
struct Audio { int64_t pts = 0, created = 0; uint64_t sequence = 0; unsigned frames = 0; int16_t pcm[1920]{}; };
|
|
std::mutex lock;
|
|
std::condition_variable wake;
|
|
std::thread videoThread, audioThread, probeThread;
|
|
std::atomic<bool> running{false}, muted{false};
|
|
std::atomic<unsigned> audioDropped{0}, audioMissing{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 audioAnchor = -1;
|
|
uint64_t audioFrames = 0;
|
|
int channels = 2;
|
|
uint64_t capturedFrames = 0;
|
|
int64_t captureReport = 0;
|
|
bool probeReady = false, probeDelivered = false, hasGood = false, rollingBack = false;
|
|
StreamPublishConfig wanted, good;
|
|
QList<int> codecs;
|
|
stream_publisher_status stats{};
|
|
#ifdef HAVE_STREAM_CAPTURE
|
|
std::unique_ptr<QMediaCaptureSession> session;
|
|
std::unique_ptr<QVideoSink> sink;
|
|
std::unique_ptr<QCamera> camera;
|
|
std::unique_ptr<QScreenCapture> screen;
|
|
std::unique_ptr<QWindowCapture> window;
|
|
QVideoFrame frame;
|
|
QVideoFrame previewFrame;
|
|
QVideoFrame presentedPreview;
|
|
#endif
|
|
};
|
|
|
|
StreamPublishEngine::StreamPublishEngine(QObject* parent) : QObject(parent), m(std::make_unique<Impl>()) {
|
|
m_timer.setTimerType(Qt::PreciseTimer); m_timer.setInterval(100);
|
|
connect(&m_timer, &QTimer::timeout, this, &StreamPublishEngine::tick); m_timer.start();
|
|
#ifdef HAVE_STREAM_CAPTURE
|
|
m_previewTimer.setTimerType(Qt::PreciseTimer); m_previewTimer.setInterval(33);
|
|
connect(&m_previewTimer, &QTimer::timeout, this, [this] {
|
|
if (!m->running) return;
|
|
QVideoFrame frame;
|
|
{ std::lock_guard<std::mutex> guard(m->lock);
|
|
if (m->previewFrame == m->presentedPreview) return;
|
|
frame = m->presentedPreview = m->previewFrame; }
|
|
if (frame.isValid()) emit previewVideoChanged(QVariant::fromValue(frame));
|
|
});
|
|
m_previewTimer.start();
|
|
#endif
|
|
try {
|
|
m->probeThread = std::thread([this] {
|
|
auto codecs = supported() ? LiveVideoEncoder::available() : QList<int>();
|
|
std::lock_guard<std::mutex> guard(m->lock); m->codecs = std::move(codecs); m->probeReady = true;
|
|
});
|
|
} catch (const std::system_error& error) {
|
|
DEBUG_ERROR(DEBUG_CATEGORY_VIDEO, "desktop codec probe worker failed: %s", error.what()); m->probeReady = true;
|
|
}
|
|
auto* sound = SoundManager::instance();
|
|
connect(sound, &SoundManager::contextAboutToReset, this, [this] { m_capture.close(); });
|
|
connect(sound, &SoundManager::contextRestored, this, [this] { if (m->running && m->wanted.audio) openMicrophone(); });
|
|
}
|
|
StreamPublishEngine::~StreamPublishEngine() {
|
|
stop(); m_timer.stop(); m_previewTimer.stop();
|
|
if (m->probeThread.joinable()) m->probeThread.join();
|
|
}
|
|
bool StreamPublishEngine::supported() {
|
|
#ifdef HAVE_STREAM_CAPTURE
|
|
return true;
|
|
#else
|
|
return false;
|
|
#endif
|
|
}
|
|
bool StreamPublishEngine::active() const { return m->id != 0; }
|
|
bool StreamPublishEngine::running() const { return m->running.load(); }
|
|
uint64_t StreamPublishEngine::group() const { return m->group; }
|
|
StreamPublishConfig StreamPublishEngine::config() const { std::lock_guard<std::mutex> guard(m->lock); return m->wanted; }
|
|
QList<int> StreamPublishEngine::codecs() const { std::lock_guard<std::mutex> guard(m->lock); return m->codecs; }
|
|
void StreamPublishEngine::setMuted(bool muted) { m->muted = muted; }
|
|
|
|
QList<StreamVideoSource> StreamPublishEngine::sources() {
|
|
QList<StreamVideoSource> sources;
|
|
#ifdef HAVE_STREAM_CAPTURE
|
|
for (const auto& camera : QMediaDevices::videoInputs()) sources.push_back({StreamVideoSource::Camera, camera.description(), camera.id(), {}});
|
|
if (QGuiApplication::platformName().contains("wayland") || qEnvironmentVariable("XDG_SESSION_TYPE") == "wayland"
|
|
|| qEnvironmentVariableIsSet("WAYLAND_DISPLAY")) {
|
|
sources.push_back({StreamVideoSource::Screen, text("Screen (system selection)"), "portal", {}});
|
|
} else {
|
|
for (auto* screen : QGuiApplication::screens()) sources.push_back({StreamVideoSource::Screen, screen->name(), screen->name().toUtf8(), {}});
|
|
for (const auto& window : QWindowCapture::capturableWindows())
|
|
sources.push_back({StreamVideoSource::Window, window.description(), {}, QVariant::fromValue(window)});
|
|
}
|
|
#endif
|
|
return sources;
|
|
}
|
|
|
|
/* Source создаётся в GUI; callbacks старого поколения не меняют новый capture. */
|
|
bool StreamPublishEngine::openSource(const StreamPublishConfig& config) {
|
|
closeSource();
|
|
#ifdef HAVE_STREAM_CAPTURE
|
|
uint64_t generation;
|
|
{ std::lock_guard<std::mutex> guard(m->lock); generation = ++m->sourceGeneration;
|
|
m->backendOrigin = m->backendAnchor = -1; m->capturedFrames = 0; }
|
|
m->session = std::make_unique<QMediaCaptureSession>(); m->sink = std::make_unique<QVideoSink>(this);
|
|
m->session->setVideoSink(m->sink.get());
|
|
connect(m->sink.get(), &QVideoSink::videoFrameChanged, this, [this, generation](const QVideoFrame& frame) {
|
|
if (!frame.isValid()) { DEBUG_TRACE(DEBUG_CATEGORY_VIDEO, "desktop capture sink invalid frame generation=%llu", (unsigned long long)generation); return; }
|
|
if (!m->running) return;
|
|
std::lock_guard<std::mutex> guard(m->lock);
|
|
if (generation != m->sourceGeneration) return;
|
|
if (!m->capturedFrames++) DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "desktop capture first raw frame kind=%d size=%dx%d pixel=%d PTS=%lld",
|
|
m->wanted.source.kind, frame.width(), frame.height(), frame.pixelFormat(), (long long)frame.startTime());
|
|
int64_t now = monotonicUs(), capture = frame.startTime();
|
|
if (capture >= 0) {
|
|
if (m->backendOrigin < 0) {
|
|
m->backendOrigin = capture; m->backendAnchor = now;
|
|
DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "desktop capture timing generation=%llu source_PTS=%lld anchor=%lld calibration=arrival-estimate",
|
|
(unsigned long long)generation, (long long)capture, (long long)now);
|
|
}
|
|
now = m->backendAnchor + capture - m->backendOrigin;
|
|
} else if (m->backendAnchor < 0) {
|
|
m->backendAnchor = now;
|
|
DEBUG_WARN(DEBUG_CATEGORY_VIDEO, "desktop capture has no source PTS; using monotonic callback estimate");
|
|
}
|
|
if (m->framePts < 0 && m->wanted.source.id == "portal") m->started = monotonicUs();
|
|
m->frame = m->previewFrame = frame; m->framePts = std::max<int64_t>(1, now - m->origin); m->frameGeneration = m->generation;
|
|
m->wake.notify_all();
|
|
}, Qt::DirectConnection);
|
|
auto failed = [this, generation](const QString& error) {
|
|
if (generation == m->sourceGeneration) sourceFailed(m->generation, error);
|
|
};
|
|
if (config.source.kind == StreamVideoSource::Camera) {
|
|
QCameraDevice selected;
|
|
for (const auto& camera : QMediaDevices::videoInputs()) if (camera.id() == config.source.id) selected = camera;
|
|
if (selected.isNull()) { DEBUG_ERROR(DEBUG_CATEGORY_VIDEO, "desktop camera disappeared"); return false; }
|
|
m->camera = std::make_unique<QCamera>(selected);
|
|
QCameraFormat best; int score = INT_MAX;
|
|
for (const auto& format : selected.videoFormats()) {
|
|
if (format.maxFrameRate() + 0.1 < config.video.fps) continue;
|
|
int candidate = std::abs(format.resolution().width() - config.video.width)
|
|
+ std::abs(format.resolution().height() - config.video.height);
|
|
if (candidate < score) { score = candidate; best = format; }
|
|
}
|
|
if (best.isNull()) { DEBUG_ERROR(DEBUG_CATEGORY_VIDEO, "desktop camera lacks requested fps=%d", config.video.fps); return false; }
|
|
m->camera->setCameraFormat(best); m->session->setCamera(m->camera.get());
|
|
connect(m->camera.get(), &QCamera::activeChanged, this, [this, generation](bool active) {
|
|
DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "desktop camera active=%d generation=%llu", active, (unsigned long long)generation);
|
|
if (!active && generation == m->sourceGeneration) sourceFailed(m->generation, text("Video capture stopped"));
|
|
}, Qt::QueuedConnection);
|
|
connect(m->camera.get(), &QCamera::errorOccurred, this, [failed](QCamera::Error, const QString& error) { failed(error); }, Qt::QueuedConnection);
|
|
m->camera->start();
|
|
} else if (config.source.kind == StreamVideoSource::Screen) {
|
|
m->screen = std::make_unique<QScreenCapture>();
|
|
if (config.source.id != "portal") {
|
|
QScreen* selected = nullptr;
|
|
for (auto* screen : QGuiApplication::screens()) if (screen->name().toUtf8() == config.source.id) selected = screen;
|
|
if (!selected) { DEBUG_ERROR(DEBUG_CATEGORY_VIDEO, "desktop screen disappeared"); return false; }
|
|
m->screen->setScreen(selected);
|
|
}
|
|
m->session->setScreenCapture(m->screen.get());
|
|
connect(m->screen.get(), &QScreenCapture::activeChanged, this, [this, generation](bool active) {
|
|
DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "desktop screen capture active=%d generation=%llu", active, (unsigned long long)generation);
|
|
if (!active && generation == m->sourceGeneration) sourceFailed(m->generation, text("Video capture stopped"));
|
|
}, Qt::QueuedConnection);
|
|
connect(m->screen.get(), &QScreenCapture::errorOccurred, this, [failed](QScreenCapture::Error, const QString& error) { failed(error); }, Qt::QueuedConnection);
|
|
m->screen->start();
|
|
} else {
|
|
auto window = config.source.window.value<QCapturableWindow>();
|
|
if (!window.isValid()) { DEBUG_ERROR(DEBUG_CATEGORY_VIDEO, "desktop window is unavailable"); return false; }
|
|
m->window = std::make_unique<QWindowCapture>(); m->window->setWindow(window); m->session->setWindowCapture(m->window.get());
|
|
connect(m->window.get(), &QWindowCapture::activeChanged, this, [this, generation](bool active) {
|
|
DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "desktop window capture active=%d generation=%llu", active, (unsigned long long)generation);
|
|
if (!active && generation == m->sourceGeneration) sourceFailed(m->generation, text("Video capture stopped"));
|
|
}, Qt::QueuedConnection);
|
|
connect(m->window.get(), &QWindowCapture::errorOccurred, this, [failed](QWindowCapture::Error, const QString& error) { failed(error); }, Qt::QueuedConnection);
|
|
m->window->start();
|
|
}
|
|
DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "desktop capture started generation=%llu kind=%d source=%s target=%dx%d/%dfps",
|
|
(unsigned long long)generation, config.source.kind, qPrintable(config.source.name), config.video.width, config.video.height, config.video.fps);
|
|
return true;
|
|
#else
|
|
Q_UNUSED(config); return false;
|
|
#endif
|
|
}
|
|
void StreamPublishEngine::closeSource() {
|
|
#ifdef HAVE_STREAM_CAPTURE
|
|
emit previewVideoChanged(QVariant());
|
|
if (m->sink) disconnect(m->sink.get(), nullptr, this, nullptr);
|
|
if (m->camera) { disconnect(m->camera.get(), nullptr, this, nullptr); m->camera->stop(); }
|
|
if (m->screen) { disconnect(m->screen.get(), nullptr, this, nullptr); m->screen->stop(); }
|
|
if (m->window) { disconnect(m->window.get(), nullptr, this, nullptr); m->window->stop(); }
|
|
m->session.reset(); m->camera.reset(); m->screen.reset(); m->window.reset(); m->sink.reset();
|
|
std::lock_guard<std::mutex> guard(m->lock); m->frame = {}; m->previewFrame = {}; m->presentedPreview = {}; m->framePts = -1;
|
|
#endif
|
|
}
|
|
|
|
bool StreamPublishEngine::start(uint64_t group, const StreamPublishConfig& config) {
|
|
if (active()) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop publication already active"); emit errorOccurred(text("A stream is already running")); return false; }
|
|
if (!supported() || !codecs().contains(config.video.codec)) {
|
|
DEBUG_WARN(DEBUG_CATEGORY_VIDEO, "desktop capture/encoder unavailable codec=%d", config.video.codec);
|
|
emit errorOccurred(text("Video capture or encoder is unavailable")); return false;
|
|
}
|
|
if (config.audio && !m_microphone.acquire()) {
|
|
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop publication blocked: microphone owned by another operation");
|
|
emit errorOccurred(text("The microphone is used by another operation")); return false;
|
|
}
|
|
{ std::lock_guard<std::mutex> guard(m->lock);
|
|
m->wanted = config; m->good = {}; m->hasGood = m->rollingBack = false; m->generation++; m->confirmedGeneration = 0;
|
|
m->origin = m->started = monotonicUs(); m->stats = {}; }
|
|
m->id = QRandomGenerator::global()->generate64(); if (!m->id) m->id = 1;
|
|
m->group = group; m->audio.reset(); m->audioSequence = 0; m->muted = false;
|
|
stream_media_format format{(uint16_t)config.video.width, (uint16_t)config.video.height, (uint16_t)config.video.rotation,
|
|
(uint8_t)config.video.codec, (uint8_t)(config.audio ? STREAM_MEDIA_OPUS_STEREO : 0), 1};
|
|
if (!gui_stream_publish_open(m->id, group, format)) {
|
|
m->id = m->group = 0; m_microphone.release(); emit errorOccurred(text("Could not start the stream")); return false;
|
|
}
|
|
m->running = true;
|
|
try {
|
|
m->videoThread = std::thread(&StreamPublishEngine::videoLoop, this);
|
|
if (config.audio) m->audioThread = std::thread(&StreamPublishEngine::audioLoop, this);
|
|
} catch (const std::system_error& error) { publicationFailed(m->id, QString::fromUtf8(error.what())); return false; }
|
|
if (!openSource(config) || (config.audio && !openMicrophone())) {
|
|
sourceFailed(m->generation, text("Could not open the selected capture device")); return false;
|
|
}
|
|
emit stateChanged(config.source.id == "portal" ? text("Select a screen in the system dialog") : text("Starting stream…"));
|
|
emit configurationChanged(); return true;
|
|
}
|
|
|
|
bool StreamPublishEngine::reconfigure(const StreamPublishConfig& config) {
|
|
if (!m->running) return false;
|
|
StreamPublishConfig previous = this->config();
|
|
if (config.audio != previous.audio || config.microphone != previous.microphone) {
|
|
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop microphone selection changed during live publication");
|
|
emit errorOccurred(text("Change the microphone before starting the stream")); return false;
|
|
}
|
|
bool sourceChange = config.source.kind != previous.source.kind || config.source.id != previous.source.id
|
|
|| config.source.window != previous.source.window || (config.source.kind == StreamVideoSource::Camera
|
|
&& (config.video.width != previous.video.width || config.video.height != previous.video.height || config.video.fps != previous.video.fps));
|
|
{ std::lock_guard<std::mutex> guard(m->lock); m->wanted = config; m->generation++; m->started = monotonicUs();
|
|
#ifdef HAVE_STREAM_CAPTURE
|
|
if (!sourceChange) { m->frame = m->previewFrame; m->frameGeneration = m->generation; }
|
|
#endif
|
|
}
|
|
if (sourceChange && !openSource(config)) { sourceFailed(m->generation, text("Could not open the selected video source")); return false; }
|
|
m->wake.notify_all(); emit stateChanged(text("Applying video settings…")); return true;
|
|
}
|
|
|
|
void StreamPublishEngine::sourceFailed(uint64_t generation, const QString& error) {
|
|
StreamPublishConfig restore; bool rollback;
|
|
{ std::lock_guard<std::mutex> guard(m->lock);
|
|
if (generation != m->generation || !m->running) return;
|
|
rollback = m->hasGood && !m->rollingBack; restore = m->good; m->rollingBack = true; }
|
|
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop capture failed generation=%llu rollback=%d reason=%s",
|
|
(unsigned long long)generation, rollback, qPrintable(error));
|
|
emit errorOccurred(rollback ? text("Video settings were restored: %1").arg(error) : error);
|
|
if (rollback) reconfigure(restore); else stop();
|
|
}
|
|
|
|
/* Ошибку общего audio worker нельзя исправить заменой только video encoder. */
|
|
void StreamPublishEngine::publicationFailed(uint64_t id, const QString& error) {
|
|
if (id != m->id || !m->running) return;
|
|
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop publication failed id=%llu reason=%s", (unsigned long long)id, qPrintable(error));
|
|
emit errorOccurred(error); stop();
|
|
}
|
|
|
|
/* Сначала callbacks, затем workers; FIN не обгоняет уже принятые access units. */
|
|
void StreamPublishEngine::stop() {
|
|
if (!m->running.exchange(false)) return;
|
|
closeSource(); m_capture.close(); m->wake.notify_all();
|
|
if (m->videoThread.joinable()) m->videoThread.join();
|
|
if (m->audioThread.joinable()) m->audioThread.join();
|
|
m_microphone.release(); gui_stream_publish_stop(m->id);
|
|
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop capture stopped id=%llu; draining publisher", (unsigned long long)m->id);
|
|
emit stateChanged(text("Finishing stream…"));
|
|
}
|
|
|
|
bool StreamPublishEngine::openMicrophone() {
|
|
m_capture.close();
|
|
m->audioAnchor = -1; m->audioFrames = 0;
|
|
auto config = this->config(); auto* context = SoundManager::instance()->context();
|
|
if (!context) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop stream audio context unavailable"); m_capture.retryLater(); return false; }
|
|
ma_device_id device; bool selected = SoundManager::deviceId(config.microphone, &device);
|
|
if (!config.microphone.isEmpty() && !selected) {
|
|
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop selected microphone ID is unavailable"); m_capture.retryLater(); return false;
|
|
}
|
|
for (int channels : {2, 1}) {
|
|
ma_device_config capture = ma_device_config_init(ma_device_type_capture);
|
|
capture.capture.format = ma_format_s16; capture.capture.channels = channels; capture.sampleRate = 48000;
|
|
capture.capture.pDeviceID = selected ? &device : nullptr;
|
|
auto callback = [this](const void* in, void*, unsigned frames) { captureAudio(in, frames); };
|
|
if (!m_capture.init(context, capture, "stream-capture", m->id, callback)) continue;
|
|
m->channels = channels;
|
|
if (!m_capture.start()) continue;
|
|
int native = m_capture.device()->capture.internalChannels;
|
|
if (native == 1) DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop stream input is mono; stereo transport duplicates the input");
|
|
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop microphone opened requested=%d backend_channels=%d rate=%u", channels, native, m_capture.device()->sampleRate);
|
|
emit audioChannelsChanged(native == 1 ? 1 : channels); return true;
|
|
}
|
|
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop stream cannot open microphone: %s", qPrintable(config.microphone)); return false;
|
|
}
|
|
|
|
/* Hardware callback копирует максимум 20ms за элемент, без codec/DSP/allocations. */
|
|
void StreamPublishEngine::captureAudio(const void* input, unsigned frames) {
|
|
const auto* pcm = static_cast<const int16_t*>(input);
|
|
if (!m->running) return;
|
|
int64_t first = m_capture.callbackTimeUs() - m->origin;
|
|
if (m_capture.timingMeasured() || m->audioAnchor < 0 || m_capture.captureMissing()) {
|
|
m->audioAnchor = first; m->audioFrames = 0;
|
|
}
|
|
first = m->audioAnchor + (int64_t)m->audioFrames * 1000000 / 48000; m->audioFrames += frames;
|
|
if (!pcm) m->audioMissing++;
|
|
for (unsigned offset = 0; offset < frames;) {
|
|
Impl::Audio record; record.frames = std::min(960u, frames - offset);
|
|
record.pts = std::max<int64_t>(0, first + (int64_t)offset * 1000000 / 48000); record.created = monotonicUs();
|
|
record.sequence = ++m->audioSequence;
|
|
for (unsigned i = 0; i < record.frames; i++) {
|
|
record.pcm[2 * i] = pcm ? pcm[(offset + i) * m->channels] : 0;
|
|
record.pcm[2 * i + 1] = pcm ? pcm[(offset + i) * m->channels + (m->channels == 2 ? 1 : 0)] : 0;
|
|
}
|
|
if (!m->audio.push(record)) m->audioDropped++;
|
|
offset += record.frames;
|
|
}
|
|
m->wake.notify_all();
|
|
}
|
|
|
|
void StreamPublishEngine::audioLoop() {
|
|
const uint64_t id = m->id;
|
|
auto* encoder = opus_codec_encoder_create(48000, 2); auto* compressor = audio_compressor_create();
|
|
int result = encoder && compressor ? opus_codec_encoder_application_set(encoder, OPUS_CODEC_APP_AUDIO) : -1;
|
|
if (result >= 0) result = opus_codec_encoder_bitrate_set(encoder, 128000);
|
|
if (result < 0) {
|
|
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop stream audio encoder initialization failed");
|
|
QMetaObject::invokeMethod(this, [this, id] { publicationFailed(id, text("Audio encoder initialization failed")); });
|
|
} else {
|
|
int16_t pcm[1920]{}; unsigned used = 0; uint64_t sequence = 0;
|
|
int64_t pts = 0, expected = -1, lastSent = -1; int settings[3] = {-1, -1, -1};
|
|
while (m->running) {
|
|
Impl::Audio record;
|
|
if (!m->audio.pop(record)) { std::unique_lock<std::mutex> lock(m->lock); m->wake.wait_for(lock, std::chrono::milliseconds(5)); continue; }
|
|
if (monotonicUs() - record.created > 250000) { m->audioDropped++; used = 0; continue; }
|
|
if ((sequence && record.sequence != sequence + 1) || (expected >= 0 && std::abs(record.pts - expected) > 5000)) {
|
|
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop stream PCM discontinuity sequence=%llu->%llu PTS=%lld expected=%lld",
|
|
(unsigned long long)sequence, (unsigned long long)record.sequence, (long long)record.pts, (long long)expected);
|
|
used = 0; audio_compressor_reset(compressor);
|
|
}
|
|
sequence = record.sequence; expected = record.pts + (int64_t)record.frames * 1000000 / 48000;
|
|
unsigned offset = 0;
|
|
while (offset < record.frames) {
|
|
if (!used) pts = record.pts + (int64_t)offset * 1000000 / 48000;
|
|
unsigned take = std::min(960 - used, record.frames - offset);
|
|
memcpy(pcm + 2 * used, record.pcm + 2 * offset, take * 2 * sizeof(int16_t)); used += take; offset += take;
|
|
if (used < 960) continue;
|
|
stream_publisher_status stats{};
|
|
gui_stream_publish_status(m->id, stats);
|
|
int wanted[] = {stats.compressor_enabled, stats.compressor_max_gain_db ? stats.compressor_max_gain_db : 15,
|
|
stats.compressor_rise_rate ? stats.compressor_rise_rate : 2};
|
|
if (!std::equal(wanted, wanted + 3, settings)) {
|
|
audio_compressor_config_t config{48000, 2, 20, 200, 0, (float)wanted[1], (float)wanted[2], wanted[2] * 5.0f, 0.25f};
|
|
if (audio_compressor_configure(compressor, &config) < 0) {
|
|
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop stream compressor configuration failed"); result = -1; break;
|
|
}
|
|
audio_compressor_set_enabled(compressor, wanted[0]); std::copy(wanted, wanted + 3, settings);
|
|
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop stream compressor enabled=%d max_gain=%d rise=%d stereo=1 delay=0",
|
|
wanted[0], wanted[1], wanted[2]);
|
|
}
|
|
if (m->muted) memset(pcm, 0, sizeof(pcm));
|
|
uint8_t packet[STREAM_MEDIA_MAX_OPUS];
|
|
result = audio_compressor_process_frame(compressor, pcm, pcm, 1920);
|
|
if (!result) result = opus_codec_encode(encoder, pcm, 960, packet, sizeof(packet));
|
|
if (result <= 0) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop stream audio encoding failed result=%d", result); result = -1; break; }
|
|
if (pts > lastSent) {
|
|
gui_stream_publish_feed(m->id, false, pts, false, QByteArray(reinterpret_cast<const char*>(packet), result)); lastSent = pts;
|
|
} else DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop stream audio stale PTS=%lld last=%lld", (long long)pts, (long long)lastSent);
|
|
used = 0;
|
|
}
|
|
if (result < 0) {
|
|
QMetaObject::invokeMethod(this, [this, id] { publicationFailed(id, text("Audio encoding failed")); }); break;
|
|
}
|
|
}
|
|
}
|
|
opus_codec_encoder_destroy(encoder); audio_compressor_destroy(compressor);
|
|
}
|
|
|
|
void StreamPublishEngine::videoLoop() {
|
|
#ifdef HAVE_STREAM_CAPTURE
|
|
std::unique_ptr<LiveVideoEncoder> encoder;
|
|
StreamPublishConfig encoding;
|
|
QImage lastImage; uint64_t current = 0; uint32_t revision = 0;
|
|
int64_t lastPts = -1, lastKey = -1000000, next = 0, report = 0; unsigned count = 0, skipped = 0;
|
|
while (m->running) {
|
|
StreamPublishConfig config; uint64_t generation; QVideoFrame frame; int64_t capture;
|
|
{ std::unique_lock<std::mutex> lock(m->lock); config = m->wanted; generation = m->generation;
|
|
int64_t now = monotonicUs();
|
|
if (next > now) { m->wake.wait_for(lock, std::chrono::microseconds(next - now)); continue; }
|
|
if (m->frameGeneration == generation) { frame = m->frame; capture = m->framePts; m->frame = {}; }
|
|
else capture = -1; }
|
|
if (generation != current) {
|
|
auto comparable = config.video; comparable.bitrate = encoding.video.bitrate;
|
|
bool onlyBitrate = encoder && comparable == encoding.video && config.source.kind == encoding.source.kind
|
|
&& config.source.id == encoding.source.id && config.source.window == encoding.source.window;
|
|
if (onlyBitrate && encoder->setBitrate(config.video.bitrate)) {
|
|
current = generation; encoding = config;
|
|
{ std::lock_guard<std::mutex> guard(m->lock);
|
|
if (m->generation == generation) { m->good = config; m->hasGood = true; m->confirmedGeneration = generation; } }
|
|
QMetaObject::invokeMethod(this, [this, generation] {
|
|
if (generation == m->generation) { emit configurationChanged(); emit stateChanged(text("Live")); }
|
|
});
|
|
} else {
|
|
auto candidate = std::make_unique<LiveVideoEncoder>();
|
|
if (!candidate->open(config.video)) {
|
|
QString error = candidate->error();
|
|
QMetaObject::invokeMethod(this, [this, generation, error] { sourceFailed(generation, error); });
|
|
std::unique_lock<std::mutex> lock(m->lock);
|
|
m->wake.wait_for(lock, std::chrono::milliseconds(100), [this, generation] {
|
|
return !m->running || m->generation != generation;
|
|
});
|
|
continue;
|
|
}
|
|
encoder = std::move(candidate); current = generation; encoding = config; revision++; lastImage = {}; lastKey = -1000000;
|
|
DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "desktop video prepared generation=%llu revision=%u encoder=%s",
|
|
(unsigned long long)generation, revision, qPrintable(encoder->name()));
|
|
}
|
|
}
|
|
bool screen = config.source.kind != StreamVideoSource::Camera;
|
|
if (frame.isValid()) {
|
|
if (!screen && (capture <= lastPts || monotonicUs() - m->origin - capture > 250000)) { skipped++; continue; }
|
|
lastImage = frame.toImage();
|
|
if (lastImage.isNull()) {
|
|
DEBUG_ERROR(DEBUG_CATEGORY_VIDEO, "desktop captured frame conversion failed format=%d", frame.pixelFormat());
|
|
QMetaObject::invokeMethod(this, [this, generation] { sourceFailed(generation, text("Video frame conversion failed")); });
|
|
continue;
|
|
}
|
|
#if QT_VERSION >= QT_VERSION_CHECK(6, 7, 0)
|
|
if (frame.rotation() != QtVideo::Rotation::None) lastImage = lastImage.transformed(QTransform().rotate(static_cast<int>(frame.rotation())));
|
|
if (frame.mirrored()) lastImage = lastImage.transformed(QTransform().scale(-1, 1));
|
|
#else
|
|
if (frame.rotationAngle()) lastImage = lastImage.transformed(QTransform().rotate(frame.rotationAngle()));
|
|
#endif
|
|
} else if (!screen || lastImage.isNull()) {
|
|
std::unique_lock<std::mutex> lock(m->lock); m->wake.wait_for(lock, std::chrono::milliseconds(5)); continue;
|
|
}
|
|
int64_t now = monotonicUs(), pts = screen ? now - m->origin : capture;
|
|
next = now + 1000000 / config.video.fps;
|
|
stream_publisher_status status; bool needsKey = gui_stream_publish_status(m->id, status) && status.need_key;
|
|
bool key = pts - lastKey >= 1000000 || needsKey;
|
|
QList<LiveEncodedFrame> output;
|
|
if (!encoder->encode(lastImage, std::max(lastPts + 1, pts), key, output)) {
|
|
QString error = encoder->error(); QMetaObject::invokeMethod(this, [this, generation, error] { sourceFailed(generation, error); }); continue;
|
|
}
|
|
stream_media_format format{(uint16_t)config.video.width, (uint16_t)config.video.height, (uint16_t)config.video.rotation,
|
|
(uint8_t)config.video.codec, (uint8_t)(config.audio ? STREAM_MEDIA_OPUS_STEREO : 0), revision};
|
|
for (const auto& encoded : output) {
|
|
if (gui_stream_publish_feed(m->id, true, encoded.pts, encoded.key, encoded.data, &format) == 1) {
|
|
lastPts = encoded.pts; if (encoded.key) lastKey = encoded.pts; count++;
|
|
if (encoded.key) {
|
|
bool first;
|
|
{ std::lock_guard<std::mutex> guard(m->lock); first = !m->hasGood || m->good.video != config.video || m->rollingBack;
|
|
first |= m->confirmedGeneration != generation;
|
|
if (m->generation == generation) {
|
|
m->good = config; m->hasGood = true; m->rollingBack = false; m->confirmedGeneration = generation;
|
|
} }
|
|
if (first) {
|
|
DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "desktop first KEY committed generation=%llu revision=%u PTS=%llu bytes=%lld",
|
|
(unsigned long long)generation, revision, (unsigned long long)encoded.pts, (long long)encoded.data.size());
|
|
QMetaObject::invokeMethod(this, [this, generation] {
|
|
if (generation != m->generation) return;
|
|
emit configurationChanged(); emit stateChanged(text("Live"));
|
|
});
|
|
}
|
|
}
|
|
} else skipped++;
|
|
}
|
|
if (now - report >= 1000000) {
|
|
DEBUG_DEBUG(DEBUG_CATEGORY_VIDEO, "desktop capture generation=%llu encoded=%u skipped=%u frame_age=%lldus encoder=%s",
|
|
(unsigned long long)generation, count, skipped, (long long)(screen ? 0 : now - m->origin - capture), qPrintable(encoder->name()));
|
|
report = now; count = skipped = 0;
|
|
}
|
|
}
|
|
#endif
|
|
}
|
|
|
|
void StreamPublishEngine::tick() {
|
|
bool ready;
|
|
{ std::lock_guard<std::mutex> guard(m->lock); ready = m->probeReady && !m->probeDelivered; m->probeDelivered |= ready; }
|
|
if (ready) { if (m->probeThread.joinable()) m->probeThread.join(); emit codecsChanged(); }
|
|
if (!active()) return;
|
|
stream_publisher_status status;
|
|
if (!gui_stream_publish_status(m->id, status)) {
|
|
stop(); m->id = m->group = 0; emit stateChanged(text("Chat service is stopped")); return;
|
|
}
|
|
{ std::lock_guard<std::mutex> guard(m->lock); m->stats = status; }
|
|
emit telemetryChanged(status);
|
|
if (status.state == STREAM_PUBLISHER_ENDED || status.state == STREAM_PUBLISHER_FAILED) {
|
|
stop(); m->id = m->group = 0;
|
|
if (status.state == STREAM_PUBLISHER_FAILED) emit errorOccurred(text("Stream publication failed"));
|
|
emit stateChanged(text("Stopped")); emit configurationChanged(); return;
|
|
}
|
|
if (m->running) {
|
|
bool confirmed, waitingSelection; int64_t started;
|
|
{ std::lock_guard<std::mutex> guard(m->lock); confirmed = m->confirmedGeneration == m->generation; started = m->started;
|
|
waitingSelection = m->wanted.source.id == "portal" && m->framePts < 0; }
|
|
if (!confirmed && !waitingSelection && monotonicUs() - started > 8000000)
|
|
sourceFailed(m->generation, text("Timed out waiting for the first video frame"));
|
|
{ std::lock_guard<std::mutex> guard(m->lock);
|
|
if (monotonicUs() - m->captureReport >= 1000000) {
|
|
m->captureReport = monotonicUs();
|
|
DEBUG_DEBUG(DEBUG_CATEGORY_VIDEO, "desktop capture state source_generation=%llu video_generation=%llu confirmed=%llu raw_frames=%llu raw_PTS=%lld",
|
|
(unsigned long long)m->sourceGeneration, (unsigned long long)m->generation, (unsigned long long)m->confirmedGeneration,
|
|
(unsigned long long)m->capturedFrames, (long long)m->framePts);
|
|
} }
|
|
if (m->wanted.audio && m_capture.needsRecovery()) { m_capture.close(); m_capture.retryLater(); }
|
|
if (m->wanted.audio && m_capture.retryReady()) openMicrophone();
|
|
}
|
|
unsigned dropped = m->audioDropped.exchange(0);
|
|
if (dropped) DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop audio capture discarded blocks=%u", dropped);
|
|
unsigned missing = m->audioMissing.exchange(0);
|
|
if (missing) DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop microphone callbacks lack PCM count=%u; sent silence", missing);
|
|
}
|