[feat] negotiate per-device preview and audio subscriptions

This commit is contained in:
dijunkun
2026-10-12 03:07:10 +08:00
parent c9e0078bfb
commit c8907a00f9
10 changed files with 251 additions and 3 deletions
@@ -49,7 +49,14 @@ SessionDeviceManager::SessionDeviceManager(GuiRuntime &owner)
frame.data = reinterpret_cast<const char *>(data);
frame.size = size;
frame.captured_timestamp = GetSystemTimeMicros(owner_.peer_);
SendAudioFrame(owner_.peer_, &frame, owner_.audio_label_.c_str());
std::shared_lock lock(owner_.connection_status_mutex_);
for (const auto& [id, status] : owner_.connection_status_) {
if (status == ConnectionStatus::Connected &&
owner_.preview_subscriptions_.ShouldSendAudio(id))
SendAudioFrameToPeer(owner_.peer_, &frame,
owner_.audio_label_.c_str(), id.data(),
id.size());
}
}) {}
SessionDeviceManager::~SessionDeviceManager() { speaker_capture_.Shutdown(); }
@@ -245,6 +252,8 @@ int SessionDeviceManager::InitializeScreenCapturer() {
const auto send_started = std::chrono::steady_clock::now();
uint64_t send_errors = 0;
for (const std::string &remote_id : connected_remote_ids) {
if (!owner_.preview_subscriptions_.ShouldSend(remote_id, now_time,
stream_id)) continue;
if (SendVideoFrameToPeer(owner_.peer_, &frame, stream_id.c_str(),
remote_id.data(), remote_id.size()) != 0)
++send_errors;
@@ -130,6 +130,7 @@ void GuiRuntime::HandlePresenceProbeTimeout() {
void GuiRuntime::HandleServerControllerDisconnected(
const std::string& remote_id, const char* reason) {
preview_subscriptions_.Remove(remote_id);
keyboard_.ReleaseRemotePressedKeys(remote_id, reason);
#ifdef _WIN32
devices_.ReleaseRemoteMouseButtons();
@@ -151,14 +151,67 @@ void PeerEventHandler::OnReceiveDataBuffer(
return;
}
const std::string remote_id(user_id, user_id_size);
std::string json_str(data, size);
const auto preview_json =
json_str.find("\"deskhub_preview") != std::string::npos
? nlohmann::json::parse(json_str, nullptr, false)
: nlohmann::json{};
if (preview_json.is_object() &&
preview_json.value("type", nlohmann::json{}) == "deskhub_preview") {
if (source_id != runtime->control_data_label_ ||
!runtime->IsAuthorizedController(remote_id))
return;
const auto subscription = PreviewSubscription::Parse(preview_json);
if (!subscription) return;
bool accepted = true;
// Apply encoder quality to this controller only; other controllers retain
// their own demand. Frame gating happens before encoding and transmission.
if (subscription->fps > 0 && runtime->preview_subscriptions_.IsNewer(
remote_id, subscription->request_id))
accepted = UpdateVideoSettings(
runtime->peer_, remote_id.data(), remote_id.size(),
static_cast<VideoQuality>(subscription->quality),
subscription->frame_rate,
static_cast<VideoDegradationPreference>(
subscription->EncoderPreference())) == 0;
if (accepted && runtime->preview_subscriptions_.Set(remote_id, *subscription) &&
subscription->fps > 0)
RequestAllVideoKeyFrames(runtime->peer_);
const auto reply = nlohmann::json{
{"type", "deskhub_preview_status"},
{"version", 1},
{"request_id", subscription->request_id},
{"accepted",
accepted}}.dump();
SendReliableDataFrameToPeer(runtime->peer_, reply.data(), reply.size(),
runtime->control_data_label_.c_str(),
remote_id.data(), remote_id.size());
return;
}
if (preview_json.is_object() &&
preview_json.value("type", nlohmann::json{}) ==
"deskhub_preview_status") {
if (auto props = runtime->FindRemoteSession(remote_id);
props && source_id == props->control_data_label_) {
try {
if (preview_json.at("version") == 1 &&
preview_json.at("request_id").is_number_unsigned() &&
preview_json.at("accepted").is_boolean()) {
props->preview_accepted_ = preview_json.at("accepted").get<bool>();
props->preview_ack_ = preview_json.at("request_id").get<uint32_t>();
}
} catch (...) {
}
}
return;
}
RemoteAction remote_action{};
if (!remote_action.from_json(json_str)) {
LOG_ERROR("Failed to parse RemoteAction JSON payload");
return;
}
std::string remote_id(user_id, user_id_size);
if (remote_action.type == ControlType::video_settings_status) {
if (auto props = runtime->FindRemoteSession(remote_id)) {
if (source_id != props->control_data_label_) return;
@@ -342,6 +395,8 @@ void PeerEventHandler::OnReceiveDataBuffer(
if (props && source_id == props->control_data_label_) {
std::lock_guard lock(props->video_settings_mutex_);
const bool first_advertisement = !props->video_settings_supported_;
props->preview_subscription_supported_ =
remote_action.i.supports_deskhub_preview;
props->video_settings_supported_ =
remote_action.i.supports_video_settings;
if (first_advertisement && props->video_settings_supported_) {
@@ -288,6 +288,7 @@ void PeerEventHandler::OnConnectionStatus(ConnectionStatus status,
/*supports_privacy_screen=*/!runtime->unattended_host_,
CROSSDESK_VERSION, kClientPlatform);
remote_action.i.supports_video_settings = true;
remote_action.i.supports_deskhub_preview = true;
for (std::size_t i = 0; i < remote_action.i.display_num; i++) {
LOG_INFO("Local display [{}:{}]", i + 1,
remote_action.i.display_list[i]);
@@ -327,6 +328,7 @@ void PeerEventHandler::OnConnectionStatus(ConnectionStatus status,
}
props->net_traffic_stats_.Reset();
props->enable_mouse_control_ = false;
props->preview_subscription_supported_ = false;
runtime->ResetRemoteServiceStatus(*props);
{
std::lock_guard lock(props->video_settings_mutex_);
@@ -414,6 +416,7 @@ void PeerEventHandler::OnConnectionStatus(ConnectionStatus status,
switch (status) {
case ConnectionStatus::Connected: {
runtime->preview_subscriptions_.Remove(remote_id);
#if _WIN32
runtime->last_windows_service_status_tick_ = 0;
#endif
@@ -423,6 +426,7 @@ void PeerEventHandler::OnConnectionStatus(ConnectionStatus status,
/*supports_privacy_screen=*/!runtime->unattended_host_,
CROSSDESK_VERSION, kClientPlatform);
remote_action.i.supports_video_settings = true;
remote_action.i.supports_deskhub_preview = true;
for (std::size_t i = 0; i < remote_action.i.display_num; i++) {
LOG_INFO("Local display [{}:{}]", i + 1,
remote_action.i.display_list[i]);
@@ -0,0 +1,78 @@
/*
* @Author: DI JUNKUN
* @Date: 2026-10-12
* Copyright (c) 2026 by DI JUNKUN, All Rights Reserved.
*/
#ifndef _PREVIEW_SUBSCRIPTIONS_H_
#define _PREVIEW_SUBSCRIPTIONS_H_
#include <preview_subscription.h>
#include <chrono>
#include <mutex>
#include <string>
#include <unordered_map>
namespace crossdesk {
// Accessed by the network callback and screen-capture callback. Legacy
// controllers have no entry and continue receiving their normal stream.
class PreviewSubscriptions {
public:
using Clock = std::chrono::steady_clock;
bool IsNewer(const std::string &id, uint32_t request_id) {
std::lock_guard lock(mutex_);
const auto found = states_.find(id);
return found == states_.end() ||
request_id > found->second.request.request_id;
}
bool Set(const std::string &id, const PreviewSubscription &request) {
std::lock_guard lock(mutex_);
auto &state = states_[id];
if (request.request_id <= state.request.request_id)
return false;
state.request = request;
state.next.clear();
return true;
}
bool ShouldSend(const std::string &id, Clock::time_point now = Clock::now(),
const std::string &stream = {}) {
std::lock_guard lock(mutex_);
const auto found = states_.find(id);
if (found == states_.end())
return true;
auto &state = found->second;
if (!state.request.fps)
return false;
// Interactive streams use the encoder's own 30/60 FPS cadence. A second
// time gate here can drop valid frames because capture timing varies.
if (state.request.fps == 60)
return true;
auto &next = state.next[stream];
if (now < next)
return false;
next = now + std::chrono::milliseconds(1000 / state.request.fps);
return true;
}
bool ShouldSendAudio(const std::string &id) {
std::lock_guard lock(mutex_);
const auto found = states_.find(id);
return found == states_.end() || found->second.request.audio;
}
void Remove(const std::string &id) {
std::lock_guard lock(mutex_);
states_.erase(id);
}
private:
struct State {
PreviewSubscription request;
// One display must not consume another display's preview frame budget.
std::unordered_map<std::string, Clock::time_point> next;
};
std::mutex mutex_;
std::unordered_map<std::string, State> states_;
};
} // namespace crossdesk
#endif
@@ -20,6 +20,7 @@
#include <vector>
#include <remote_action.h>
#include <preview_subscription.h>
#include <stream_names.h>
#include "display_info.h"
#include "cursor_presentation.h"
@@ -70,6 +71,12 @@ struct FileTransferState {
// Runtime state for one connected or connecting remote endpoint. It groups the
// connection, media, input, window and transfer data that share one lifetime.
struct RemoteSession {
std::atomic<bool> preview_subscription_supported_{false};
std::atomic<uint32_t> preview_ack_{0};
std::atomic<bool> preview_accepted_{false};
uint32_t preview_request_ = 0;
PreviewSubscription preview_requested_;
std::chrono::steady_clock::time_point preview_request_at_{};
Params params_;
PeerPtr* peer_ = nullptr;
std::shared_ptr<PeerEventHandler> peer_events_;
@@ -25,6 +25,7 @@
#include "common/task_queue_lock_free.h"
#include "path_manager.h"
#include "runtime/device_presence_cache.h"
#include "runtime/preview_subscriptions.h"
#include "runtime/remote_session.h"
#include "runtime/session_lifecycle.h"
#include "thumbnail.h"
@@ -36,6 +37,7 @@
namespace crossdesk::gui_detail {
struct InfrastructureState {
PreviewSubscriptions preview_subscriptions_;
std::unique_ptr<ConfigCenter> config_center_;
ConfigCenter::LANGUAGE localization_language_ =
ConfigCenter::LANGUAGE::CHINESE;