GithubFollow/One-KVM
1
0
Fork 0
mirror of https://github.com/mofeng-git/One-KVM.git synced 2026-09-13 02:54:26 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

refactor: 简化运行时与服务协调

Browse Source 
提取运行时生命周期管理，集中远程访问与 USB 协调，并缩小 Web 路由状态依赖。
This commit is contained in:
mofeng-git
2026-08-26 10:53:08 +08:00
parent 827d24fde4
commit 37af369b43
24 changed files with 2016 additions and 1771 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/lib.rs
View File

@@ -36,6 +36,8 @@ pub mod redfish;
#[cfg(feature = "desktop")] #[cfg(feature = "desktop")]
pub mod rtsp; pub mod rtsp;
#[cfg(feature = "desktop")] #[cfg(feature = "desktop")]
pub mod runtime;
#[cfg(feature = "desktop")]
pub mod rustdesk; pub mod rustdesk;
#[cfg(feature = "desktop")] #[cfg(feature = "desktop")]
pub mod state; pub mod state;

832
src/main.rs
View File

@@ -3,43 +3,20 @@ use std::future::Future;
use std::io::Write; use std::io::Write;
use std::net::{IpAddr, SocketAddr}; use std::net::{IpAddr, SocketAddr};
use std::path::{Path, PathBuf}; use std::path::{Path, PathBuf};
use std::sync::Arc;
use axum_server::tls_rustls::RustlsConfig; use axum_server::tls_rustls::RustlsConfig;
use clap::{Args, Parser, Subcommand, ValueEnum}; use clap::{Args, Parser, Subcommand, ValueEnum};
use futures::{stream::FuturesUnordered, StreamExt}; use futures::{stream::FuturesUnordered, StreamExt};
use rustls::crypto::{ring, CryptoProvider}; use rustls::crypto::{ring, CryptoProvider};
use tokio::sync::{broadcast, mpsc};
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt}; use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
use one_kvm::atx::AtxController;
use one_kvm::audio::{AudioController, AudioControllerConfig, AudioQuality};
use one_kvm::auth::{SessionStore, TwoFactorService, UserStore}; use one_kvm::auth::{SessionStore, TwoFactorService, UserStore};
use one_kvm::computer_use::ComputerUseManager; use one_kvm::config;
use one_kvm::config::{self, AppConfig, ConfigStore};
use one_kvm::db::DatabasePool; use one_kvm::db::DatabasePool;
use one_kvm::events::EventBus;
use one_kvm::extensions::ExtensionManager;
use one_kvm::hid::{HidBackendType, HidController};
#[cfg(unix)]
use one_kvm::msd::MsdController;
#[cfg(unix)]
use one_kvm::otg::OtgService;
use one_kvm::platform::PlatformCapabilities; use one_kvm::platform::PlatformCapabilities;
use one_kvm::rtsp::RtspService; use one_kvm::runtime::{RuntimeBuilder, WebConfigOverrides};
use one_kvm::rustdesk::RustDeskService; use one_kvm::state::ShutdownAction;
use one_kvm::state::{AppState, ShutdownAction};
use one_kvm::update::UpdateService;
use one_kvm::utils::bind_tcp_listener; use one_kvm::utils::bind_tcp_listener;
use one_kvm::video::codec_constraints::{
enforce_constraints_with_stream_manager, validate_third_party_codec_compatibility,
StreamCodecConstraints,
};
use one_kvm::video::format::{PixelFormat, Resolution};
use one_kvm::video::{Streamer, VideoStreamManager};
use one_kvm::vnc::VncService;
use one_kvm::web;
use one_kvm::webrtc::{WebRtcStreamer, WebRtcStreamerConfig};
#[derive(Debug, Clone, Copy, Default, ValueEnum)] #[derive(Debug, Clone, Copy, Default, ValueEnum)]
enum LogLevel { enum LogLevel {
@@ -147,28 +124,20 @@ async fn main() -> anyhow::Result<()> {
return Ok(()); return Ok(());
} }
let (db, config_store, mut config) = load_runtime_config(&data_dir).await?; let overrides = WebConfigOverrides {
address: args.address,
if let Some(addr) = args.address { http_port: args.http_port,
config.web.bind_address = addr.clone(); https_port: args.https_port,
config.web.bind_addresses = vec![addr]; enable_https: args.enable_https,
} ssl_cert: args.ssl_cert,
if let Some(port) = args.http_port { ssl_key: args.ssl_key,
config.web.http_port = port; };
} let mut runtime = RuntimeBuilder::new(data_dir.clone())
if let Some(port) = args.https_port { .with_web_overrides(overrides)
config.web.https_port = port; .build()
} .await?;
if args.enable_https { let config = runtime.config();
config.web.https_enabled = true; let state = runtime.state().clone();
}
if let Some(cert_path) = args.ssl_cert {
config.web.ssl_cert_path = Some(cert_path.to_string_lossy().to_string());
}
if let Some(key_path) = args.ssl_key {
config.web.ssl_key_path = Some(key_path.to_string_lossy().to_string());
}
let bind_ips = resolve_bind_addresses(&config.web)?; let bind_ips = resolve_bind_addresses(&config.web)?;
let scheme = if config.web.https_enabled { let scheme = if config.web.https_enabled {
@@ -187,500 +156,12 @@ async fn main() -> anyhow::Result<()> {
tracing::info!("Server will listen on: {}://{}", scheme, addr); tracing::info!("Server will listen on: {}://{}", scheme, addr);
} }
let session_store = SessionStore::new(config.auth.session_timeout_secs as i64); let app = runtime.router();
let user_store = UserStore::new(db.clone_pool());
let two_factor = TwoFactorService::new(db.clone_pool());
let (shutdown_tx, _) = broadcast::channel::<ShutdownAction>(1);
let events = Arc::new(EventBus::new());
tracing::info!("Event bus initialized");
let (video_format, video_resolution) = parse_video_config(&config);
tracing::debug!(
"Parsed video config: {} @ {}x{}",
video_format,
video_resolution.width,
video_resolution.height
);
let streamer = Streamer::new();
streamer.set_event_bus(events.clone()).await;
if let Some(ref device_path) = config.video.device {
if let Err(e) = streamer
.apply_video_config(
device_path,
video_format,
video_resolution,
config.video.fps,
)
.await
{
tracing::warn!(
"Failed to initialize video with config: {}, will auto-detect",
e
);
} else {
tracing::info!(
"Video configured: {} @ {}x{} {}",
device_path,
video_resolution.width,
video_resolution.height,
video_format
);
}
}
let webrtc_streamer = {
let webrtc_config = WebRtcStreamerConfig {
resolution: video_resolution,
input_format: video_format,
fps: config.video.fps,
bitrate_preset: config.stream.bitrate_preset,
encoder_backend: one_kvm::stream_encoder::encoder_type_to_backend(
config.stream.encoder.clone(),
),
webrtc: {
let mut stun_servers = vec![];
let mut turn_servers = vec![];
let has_custom_stun = config
.stream
.stun_server
.as_ref()
.map(|s| !s.is_empty())
.unwrap_or(false);
let has_custom_turn = config
.stream
.turn_server
.as_ref()
.map(|s| !s.is_empty())
.unwrap_or(false);
if !has_custom_stun && !has_custom_turn {
use one_kvm::webrtc::config::public_ice;
let stun = public_ice::stun_server().to_string();
tracing::info!("Using public STUN server: {}", stun);
stun_servers.push(stun);
} else {
if let Some(ref stun) = config.stream.stun_server {
if !stun.is_empty() {
stun_servers.push(stun.clone());
tracing::info!("Using custom STUN server: {}", stun);
}
}
if let Some(ref turn) = config.stream.turn_server {
if !turn.is_empty() {
let username = config.stream.turn_username.clone().unwrap_or_default();
let credential =
config.stream.turn_password.clone().unwrap_or_default();
turn_servers.push(one_kvm::webrtc::config::TurnServer::new(
turn.clone(),
username.clone(),
credential,
));
tracing::info!(
"Using custom TURN server: {} (user: {})",
turn,
username
);
}
}
}
one_kvm::webrtc::config::WebRtcConfig {
stun_servers,
turn_servers,
..Default::default()
}
},
..Default::default()
};
WebRtcStreamer::with_config(webrtc_config)
};
tracing::info!("WebRTC streamer created");
#[cfg(unix)]
let otg_service = Arc::new(OtgService::new());
#[cfg(unix)]
tracing::info!("OTG Service created");
#[cfg(unix)]
if let Err(e) = otg_service
.apply_config(&config.hid, &config.msd, &config.otg_network, &config.uac)
.await
{
tracing::warn!("Failed to apply OTG config: {}", e);
}
let hid_backend = match config.hid.backend {
config::HidBackend::Otg => HidBackendType::Otg,
config::HidBackend::Ch9329 => HidBackendType::Ch9329 {
port: config.hid.ch9329_port.clone(),
baud_rate: config.hid.ch9329_baudrate,
hybrid_mouse: config.hid.ch9329_hybrid_mouse,
},
config::HidBackend::None => HidBackendType::None,
};
#[cfg(unix)]
let hid = Arc::new(HidController::new(hid_backend, Some(otg_service.clone())));
#[cfg(not(unix))]
let hid = Arc::new(HidController::new(hid_backend));
hid.set_event_bus(events.clone()).await;
if let Err(e) = hid.init().await {
tracing::warn!("Failed to initialize HID backend: {}", e);
}
#[cfg(unix)]
let msd = if config.msd.enabled {
let ventoy_resource_dir = data_dir.join("ventoy");
let controller = MsdController::new(otg_service.clone(), config.msd.msd_dir_path());
if let Err(e) = controller.init(&ventoy_resource_dir).await {
tracing::warn!("Failed to initialize MSD controller: {}", e);
None
} else {
controller.set_event_bus(events.clone()).await;
Some(controller)
}
} else {
tracing::info!("MSD disabled in configuration");
None
};
let atx = if config.atx.enabled {
let controller_config = config.atx.to_controller_config();
let controller = AtxController::new(controller_config);
if let Err(e) = controller.init().await {
tracing::warn!("Failed to initialize ATX controller: {}", e);
None
} else {
Some(controller)
}
} else {
tracing::info!("ATX disabled in configuration");
None
};
let audio = {
let audio_config = AudioControllerConfig {
enabled: config.audio.enabled,
device: config.audio.device.clone(),
quality: match config.audio.quality.parse::<AudioQuality>() {
Ok(q) => q,
Err(e) => {
tracing::warn!(
"Invalid audio quality in config (value={:?}): {}, using balanced",
config.audio.quality,
e
);
AudioQuality::Balanced
}
},
};
let controller = AudioController::new(audio_config);
controller.set_event_bus(events.clone()).await;
if config.audio.enabled {
tracing::info!(
"Audio enabled: {}, quality={}",
config.audio.device,
config.audio.quality
);
if let Err(e) = controller.start_streaming().await {
tracing::warn!("Failed to start audio streaming: {}", e);
}
} else {
tracing::info!("Audio disabled in configuration");
}
Arc::new(controller)
};
let extensions = Arc::new(ExtensionManager::new());
tracing::info!("Extension manager initialized");
webrtc_streamer.set_hid_controller(hid.clone()).await;
webrtc_streamer.set_audio_controller(audio.clone()).await;
if config.audio.enabled {
if let Err(e) = webrtc_streamer.set_audio_enabled(true).await {
tracing::warn!("Failed to enable WebRTC audio: {}", e);
} else {
tracing::debug!("WebRTC audio enabled");
}
}
let (device_path, actual_resolution, actual_format, actual_fps, jpeg_quality) =
streamer.current_capture_config().await;
tracing::debug!(
"Initial video config: {}x{} {:?} @ {}fps",
actual_resolution.width,
actual_resolution.height,
actual_format,
actual_fps
);
webrtc_streamer
.update_video_config(actual_resolution, actual_format, actual_fps)
.await;
if let Some(device_path) = device_path {
let device_info = streamer.current_device().await;
webrtc_streamer
.set_capture_device(device_path, jpeg_quality, device_info)
.await;
tracing::debug!("WebRTC streamer configured for direct capture");
} else {
tracing::warn!("No capture device configured for WebRTC");
}
let stream_manager = VideoStreamManager::with_webrtc_streamer(
streamer.clone(),
webrtc_streamer.clone() as std::sync::Arc<dyn one_kvm::video::traits::VideoOutput>,
);
stream_manager.set_event_bus(events.clone()).await;
stream_manager.set_config_store(config_store.clone()).await;
{
let stream_manager_weak = Arc::downgrade(&stream_manager);
audio
.set_recovered_callback(Arc::new(move || {
if let Some(stream_manager) = stream_manager_weak.upgrade() {
tokio::spawn(async move {
stream_manager.reconnect_webrtc_audio_sources().await;
});
}
}))
.await;
}
let initial_mode = config.stream.mode.clone();
if let Err(e) = stream_manager.init_with_mode(initial_mode.clone()).await {
tracing::warn!(
"Failed to initialize stream manager with mode {:?}: {}",
initial_mode,
e
);
} else {
tracing::info!(
"Video stream manager initialized with mode: {:?}",
initial_mode
);
}
let third_party_codec_config_valid = match validate_third_party_codec_compatibility(&config) {
Ok(()) => true,
Err(e) => {
tracing::warn!(
"Third-party access codec configuration is invalid; RustDesk/VNC/RTSP will not start: {}",
e
);
false
}
};
let rustdesk = if third_party_codec_config_valid && config.rustdesk.is_valid() {
tracing::info!(
"Initializing RustDesk service: ID={} -> {}",
config.rustdesk.device_id,
config.rustdesk.rendezvous_addr()
);
let service = RustDeskService::new(
config.rustdesk.clone(),
stream_manager.clone(),
hid.clone(),
audio.clone(),
);
Some(Arc::new(service))
} else {
if config.rustdesk.enabled {
tracing::warn!(
"RustDesk enabled but configuration is incomplete (missing server or credentials)"
);
} else {
tracing::info!("RustDesk disabled in configuration");
}
None
};
let rtsp = if third_party_codec_config_valid && config.rtsp.enabled {
tracing::info!(
"Initializing RTSP service: rtsp://{}:{}/{}",
config.rtsp.bind,
config.rtsp.port,
config.rtsp.path
);
let service = RtspService::new(config.rtsp.clone(), stream_manager.clone());
Some(Arc::new(service))
} else {
tracing::info!("RTSP disabled in configuration");
None
};
let vnc = if third_party_codec_config_valid && config.vnc.enabled {
tracing::info!(
"Initializing VNC service: {}:{} ({:?})",
config.vnc.bind,
config.vnc.port,
config.vnc.encoding
);
Some(Arc::new(VncService::new(
config.vnc.clone(),
stream_manager.clone(),
hid.clone(),
)))
} else {
tracing::info!("VNC disabled in configuration");
None
};
let update_service = Arc::new(UpdateService::new());
let computer_use = ComputerUseManager::new(config_store.clone(), hid.clone());
let state = AppState::new(
db.clone(),
config_store.clone(),
session_store,
user_store,
two_factor,
#[cfg(unix)]
otg_service,
stream_manager,
webrtc_streamer.clone(),
hid,
computer_use,
#[cfg(unix)]
msd,
atx,
audio,
rustdesk.clone(),
vnc.clone(),
rtsp.clone(),
extensions.clone(),
events.clone(),
update_service,
shutdown_tx.clone(),
data_dir.clone(),
);
#[cfg(unix)]
{
// Initialize UAC playback writer if UAC is enabled.
if config.uac.enabled {
let uac_cfg = one_kvm::audio::uac::UacPlaybackConfig {
sample_rate: config.uac.sample_rate,
channels: config.uac.channels as u16,
..Default::default()
};
match one_kvm::audio::uac::UacPlayback::start(uac_cfg) {
Ok(writer) => {
*state.uac_playback.write().await = Some(writer);
tracing::info!("UAC playback writer started");
}
Err(e) => {
tracing::warn!("Failed to start UAC playback writer: {}", e);
}
}
}
}
if config.watchdog.enabled {
if let Err(error) = state.watchdog.enable().await {
tracing::error!(
"Configured hardware watchdog failed to start; web service will continue: {}",
error
);
} else {
tracing::info!("Hardware watchdog started");
}
}
extensions.set_event_bus(events.clone()).await;
if let Some(ref service) = rustdesk {
if let Err(e) = service.start().await {
tracing::error!("Failed to start RustDesk service: {}", e);
} else {
if let Some(updated_config) = service.save_credentials() {
if let Err(e) = config_store
.update(|cfg| {
cfg.rustdesk.public_key = updated_config.public_key.clone();
cfg.rustdesk.private_key = updated_config.private_key.clone();
cfg.rustdesk.signing_public_key = updated_config.signing_public_key.clone();
cfg.rustdesk.signing_private_key =
updated_config.signing_private_key.clone();
cfg.rustdesk.uuid = updated_config.uuid.clone();
})
.await
{
tracing::warn!("Failed to save RustDesk credentials: {}", e);
}
}
tracing::info!("RustDesk service started");
}
}
if let Some(ref service) = vnc {
if let Err(e) = service.start().await {
tracing::error!("Failed to start VNC service: {}", e);
} else {
tracing::info!("VNC service started");
}
}
if let Some(ref service) = rtsp {
if let Err(e) = service.start().await {
tracing::error!("Failed to start RTSP service: {}", e);
} else {
tracing::info!("RTSP service started");
}
}
{
let runtime_config = state.runtime_third_party_config().await;
let constraints = StreamCodecConstraints::from_config(&runtime_config);
state
.stream_manager
.set_runtime_codec_constraints(constraints.clone())
.await;
match enforce_constraints_with_stream_manager(&state.stream_manager, &constraints).await {
Ok(result) if result.changed => {
if let Some(message) = result.message {
tracing::info!("{}", message);
}
}
Ok(_) => {}
Err(e) => tracing::warn!("Failed to enforce startup codec constraints: {}", e),
}
}
{
let ext_config = config_store.get();
extensions.start_enabled(&ext_config.extensions).await;
}
{
let extensions_clone = extensions.clone();
let config_store_clone = config_store.clone();
tokio::spawn(async move {
let mut interval = tokio::time::interval(std::time::Duration::from_secs(30));
loop {
interval.tick().await;
let config = config_store_clone.get();
extensions_clone.health_check(&config.extensions).await;
}
});
tracing::info!("Extension health check task started");
}
state.publish_device_info().await;
spawn_device_info_broadcaster(state.clone(), events);
let app = web::create_router(state.clone());
let listeners = bind_tcp_listeners(&bind_ips, bind_port)?; let listeners = bind_tcp_listeners(&bind_ips, bind_port)?;
let shutdown_signal = { let shutdown_signal = {
let shutdown_tx = state.shutdown_tx.clone();
let mut shutdown_rx = shutdown_tx.subscribe(); let mut shutdown_rx = shutdown_tx.subscribe();
async move { async move {
tokio::select! { tokio::select! {
@@ -740,7 +221,7 @@ async fn main() -> anyhow::Result<()> {
servers.push(server); servers.push(server);
} }
run_servers_until_shutdown(servers, shutdown_signal, &state, "HTTPS").await run_servers_until_shutdown(servers, shutdown_signal, "HTTPS").await
} else { } else {
let servers = FuturesUnordered::new(); let servers = FuturesUnordered::new();
for listener in listeners { for listener in listeners {
@@ -752,9 +233,10 @@ async fn main() -> anyhow::Result<()> {
servers.push(async move { server.await }); servers.push(async move { server.await });
} }
run_servers_until_shutdown(servers, shutdown_signal, &state, "HTTP").await run_servers_until_shutdown(servers, shutdown_signal, "HTTP").await
}; };
runtime.shutdown().await;
tracing::info!("Server shutdown complete"); tracing::info!("Server shutdown complete");
if let ShutdownAction::Restart { exe_path } = shutdown_action { if let ShutdownAction::Restart { exe_path } = shutdown_action {
restart_current_process(exe_path)?; restart_current_process(exe_path)?;
@@ -840,14 +322,13 @@ async fn open_database_pool(data_dir: &Path) -> anyhow::Result<DatabasePool> {
async fn run_servers_until_shutdown<F, E>( async fn run_servers_until_shutdown<F, E>(
mut servers: FuturesUnordered<F>, mut servers: FuturesUnordered<F>,
shutdown_signal: impl Future<Output = ShutdownAction>, shutdown_signal: impl Future<Output = ShutdownAction>,
state: &Arc<AppState>,
protocol: &'static str, protocol: &'static str,
) -> ShutdownAction ) -> ShutdownAction
where where
F: Future<Output = Result<(), E>> + Send, F: Future<Output = Result<(), E>> + Send,
E: std::fmt::Display, E: std::fmt::Display,
{ {
let action = tokio::select! { tokio::select! {
action = shutdown_signal => { action = shutdown_signal => {
action action
} }
@@ -857,9 +338,7 @@ where
} }
ShutdownAction::Exit ShutdownAction::Exit
} }
}; }
cleanup(state).await;
action
} }
fn restart_current_process(exe_path: Option<PathBuf>) -> anyhow::Result<()> { fn restart_current_process(exe_path: Option<PathBuf>) -> anyhow::Result<()> {
@@ -896,64 +375,6 @@ async fn run_cli_command(command: CliCommand, data_dir: PathBuf) -> anyhow::Resu
} }
} }
async fn load_runtime_config(
data_dir: &Path,
) -> anyhow::Result<(DatabasePool, ConfigStore, AppConfig)> {
tokio::fs::create_dir_all(data_dir).await?;
let db = open_database_pool(data_dir).await?;
let config_store = ConfigStore::new(db.clone_pool())?;
config_store.load().await?;
let mut config = (*config_store.get()).clone();
config.apply_platform_defaults();
prepare_linux_runtime_dirs(data_dir, &config_store, &mut config).await?;
Ok((db, config_store, config))
}
#[cfg(unix)]
async fn prepare_linux_runtime_dirs(
data_dir: &Path,
config_store: &ConfigStore,
config: &mut AppConfig,
) -> anyhow::Result<()> {
let mut msd_dir_updated = false;
if config.msd.msd_dir.trim().is_empty() {
let msd_dir = data_dir.join("msd");
config.msd.msd_dir = msd_dir.to_string_lossy().to_string();
msd_dir_updated = true;
} else if !PathBuf::from(&config.msd.msd_dir).is_absolute() {
let msd_dir = data_dir.join(&config.msd.msd_dir);
tracing::warn!(
"MSD directory is relative, rebasing to {}",
msd_dir.display()
);
config.msd.msd_dir = msd_dir.to_string_lossy().to_string();
msd_dir_updated = true;
}
if msd_dir_updated {
config_store.set(config.clone()).await?;
}
let msd_dir = PathBuf::from(&config.msd.msd_dir);
if let Err(e) = tokio::fs::create_dir_all(msd_dir.join("images")).await {
tracing::warn!("Failed to create MSD images directory: {}", e);
}
if let Err(e) = tokio::fs::create_dir_all(msd_dir.join("ventoy")).await {
tracing::warn!("Failed to create MSD ventoy directory: {}", e);
}
Ok(())
}
#[cfg(not(unix))]
async fn prepare_linux_runtime_dirs(
_data_dir: &Path,
_config_store: &ConfigStore,
_config: &mut AppConfig,
) -> anyhow::Result<()> {
Ok(())
}
async fn run_user_action( async fn run_user_action(
action: UserAction, action: UserAction,
users: &UserStore, users: &UserStore,
@@ -1065,17 +486,6 @@ fn bind_tcp_listeners(addrs: &[IpAddr], port: u16) -> anyhow::Result<Vec<std::ne
Ok(listeners) Ok(listeners)
} }
fn parse_video_config(config: &AppConfig) -> (PixelFormat, Resolution) {
let format = config
.video
.format
.as_ref()
.and_then(|f: &String| f.parse::<PixelFormat>().ok())
.unwrap_or(PixelFormat::Mjpeg);
let resolution = Resolution::new(config.video.width, config.video.height);
(format, resolution)
}
fn generate_self_signed_cert() -> anyhow::Result<rcgen::CertifiedKey<rcgen::KeyPair>> { fn generate_self_signed_cert() -> anyhow::Result<rcgen::CertifiedKey<rcgen::KeyPair>> {
use rcgen::generate_simple_self_signed; use rcgen::generate_simple_self_signed;
@@ -1088,197 +498,3 @@ fn generate_self_signed_cert() -> anyhow::Result<rcgen::CertifiedKey<rcgen::KeyP
let certified_key = generate_simple_self_signed(subject_alt_names)?; let certified_key = generate_simple_self_signed(subject_alt_names)?;
Ok(certified_key) Ok(certified_key)
} }
fn spawn_device_info_broadcaster(state: Arc<AppState>, events: Arc<EventBus>) {
use std::time::{Duration, Instant};
enum DeviceInfoTrigger {
Event,
Lagged { topic: &'static str, count: u64 },
}
const DEVICE_INFO_TOPICS: &[&str] = &[
"stream.state_changed",
"stream.config_applied",
"stream.mode_ready",
];
const DEBOUNCE_MS: u64 = 100;
let (trigger_tx, mut trigger_rx) = mpsc::unbounded_channel();
for topic in DEVICE_INFO_TOPICS {
let Some(mut rx) = events.subscribe_topic(topic) else {
tracing::warn!(
"DeviceInfo broadcaster missing topic subscription: {}",
topic
);
continue;
};
let trigger_tx = trigger_tx.clone();
let topic_name = *topic;
tokio::spawn(async move {
loop {
match rx.recv().await {
Ok(_) => {
if trigger_tx.send(DeviceInfoTrigger::Event).is_err() {
break;
}
}
Err(tokio::sync::broadcast::error::RecvError::Lagged(count)) => {
if trigger_tx
.send(DeviceInfoTrigger::Lagged {
topic: topic_name,
count,
})
.is_err()
{
break;
}
}
Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
}
}
});
}
{
let mut dirty_rx = events.subscribe_device_info_dirty();
let trigger_tx = trigger_tx.clone();
tokio::spawn(async move {
loop {
match dirty_rx.recv().await {
Ok(()) => {
if trigger_tx.send(DeviceInfoTrigger::Event).is_err() {
break;
}
}
Err(tokio::sync::broadcast::error::RecvError::Lagged(count)) => {
if trigger_tx
.send(DeviceInfoTrigger::Lagged {
topic: "device_info_dirty",
count,
})
.is_err()
{
break;
}
}
Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
}
}
});
}
tokio::spawn(async move {
let mut last_broadcast = Instant::now() - Duration::from_millis(DEBOUNCE_MS);
let mut pending_broadcast = false;
loop {
let recv_result = if pending_broadcast {
let remaining =
DEBOUNCE_MS.saturating_sub(last_broadcast.elapsed().as_millis() as u64);
tokio::time::timeout(Duration::from_millis(remaining), trigger_rx.recv()).await
} else {
Ok(trigger_rx.recv().await)
};
match recv_result {
Ok(Some(DeviceInfoTrigger::Event)) => {
pending_broadcast = true;
}
Ok(Some(DeviceInfoTrigger::Lagged { topic, count })) => {
tracing::warn!(
"DeviceInfo broadcaster lagged by {} events on topic {}",
count,
topic
);
pending_broadcast = true;
}
Ok(None) => {
tracing::info!("Event bus closed, stopping DeviceInfo broadcaster");
break;
}
Err(_timeout) => {}
}
if pending_broadcast && last_broadcast.elapsed() >= Duration::from_millis(DEBOUNCE_MS) {
state.publish_device_info().await;
tracing::trace!("Broadcasted DeviceInfo (debounced)");
last_broadcast = Instant::now();
pending_broadcast = false;
}
}
});
tracing::info!(
"DeviceInfo broadcaster task started (debounce: {}ms)",
DEBOUNCE_MS
);
}
async fn cleanup(state: &Arc<AppState>) {
state.extensions.stop_all().await;
tracing::info!("Extensions stopped");
if let Some(ref service) = *state.rustdesk.read().await {
if let Err(e) = service.stop().await {
tracing::warn!("Failed to stop RustDesk service: {}", e);
} else {
tracing::info!("RustDesk service stopped");
}
}
if let Some(ref service) = *state.vnc.read().await {
if let Err(e) = service.stop().await {
tracing::warn!("Failed to stop VNC service: {}", e);
} else {
tracing::info!("VNC service stopped");
}
}
if let Some(ref service) = *state.rtsp.read().await {
if let Err(e) = service.stop().await {
tracing::warn!("Failed to stop RTSP service: {}", e);
} else {
tracing::info!("RTSP service stopped");
}
}
if let Err(e) = state.stream_manager.stop().await {
tracing::warn!("Failed to stop streamer: {}", e);
}
if let Err(e) = state.hid.shutdown().await {
tracing::warn!("Failed to shutdown HID: {}", e);
}
#[cfg(unix)]
if let Some(msd) = state.msd.write().await.as_mut() {
if let Err(e) = msd.shutdown().await {
tracing::warn!("Failed to shutdown MSD: {}", e);
}
}
#[cfg(unix)]
if let Err(e) = state.otg_service.shutdown().await {
tracing::warn!("Failed to shutdown OTG: {}", e);
}
if let Some(atx) = state.atx.write().await.as_mut() {
if let Err(e) = atx.shutdown().await {
tracing::warn!("Failed to shutdown ATX: {}", e);
}
}
if let Err(e) = state.audio.shutdown().await {
tracing::warn!("Failed to shutdown audio: {}", e);
}
if let Err(error) = state.watchdog.disable().await {
tracing::error!(
"CRITICAL: failed to disable hardware watchdog during shutdown: {}",
error
);
}
}

598
src/runtime/builder.rs Normal file
View File

@@ -0,0 +1,598 @@
use std::path::{Path, PathBuf};
use std::sync::Arc;
use tokio::sync::broadcast;
use crate::atx::AtxController;
use crate::audio::{AudioController, AudioControllerConfig, AudioQuality};
use crate::auth::{SessionStore, TwoFactorService, UserStore};
use crate::computer_use::ComputerUseManager;
use crate::config::{self, AppConfig, ConfigStore};
use crate::db::DatabasePool;
use crate::events::EventBus;
use crate::extensions::ExtensionManager;
use crate::hid::{HidBackendType, HidController};
#[cfg(unix)]
use crate::msd::MsdController;
#[cfg(unix)]
use crate::otg::OtgService;
use crate::state::{AppState, ShutdownAction};
use crate::update::UpdateService;
use crate::video::format::{PixelFormat, Resolution};
use crate::video::{Streamer, VideoStreamManager};
use crate::webrtc::{WebRtcStreamer, WebRtcStreamerConfig};
use super::supervisor::RuntimeSupervisor;
#[derive(Debug, Clone, Default)]
pub struct WebConfigOverrides {
pub address: Option<String>,
pub http_port: Option<u16>,
pub https_port: Option<u16>,
pub enable_https: bool,
pub ssl_cert: Option<PathBuf>,
pub ssl_key: Option<PathBuf>,
}
impl WebConfigOverrides {
fn apply(self, config: &mut AppConfig) {
if let Some(address) = self.address {
config.web.bind_address = address.clone();
config.web.bind_addresses = vec![address];
}
if let Some(port) = self.http_port {
config.web.http_port = port;
}
if let Some(port) = self.https_port {
config.web.https_port = port;
}
if self.enable_https {
config.web.https_enabled = true;
}
if let Some(path) = self.ssl_cert {
config.web.ssl_cert_path = Some(path.to_string_lossy().to_string());
}
if let Some(path) = self.ssl_key {
config.web.ssl_key_path = Some(path.to_string_lossy().to_string());
}
}
}
pub struct RuntimeBuilder {
data_dir: PathBuf,
web_overrides: WebConfigOverrides,
}
impl RuntimeBuilder {
pub fn new(data_dir: PathBuf) -> Self {
Self {
data_dir,
web_overrides: WebConfigOverrides::default(),
}
}
pub fn with_web_overrides(mut self, overrides: WebConfigOverrides) -> Self {
self.web_overrides = overrides;
self
}
pub async fn build(self) -> anyhow::Result<ApplicationRuntime> {
let Self {
data_dir,
web_overrides,
} = self;
let (db, config_store, mut config) = load_runtime_config(&data_dir).await?;
web_overrides.apply(&mut config);
let sessions = SessionStore::new(config.auth.session_timeout_secs as i64);
let users = UserStore::new(db.clone_pool());
let two_factor = TwoFactorService::new(db.clone_pool());
let (shutdown_tx, _) = broadcast::channel::<ShutdownAction>(1);
let events = Arc::new(EventBus::new());
tracing::info!("Event bus initialized");
let (video_format, video_resolution) = parse_video_config(&config);
let streamer = build_streamer(&config, &events, video_format, video_resolution).await;
let webrtc = build_webrtc(&config, video_format, video_resolution);
#[cfg(unix)]
let otg_service = build_otg(&config).await;
let hid_backend = hid_backend_type(&config);
#[cfg(unix)]
let hid = Arc::new(HidController::new(hid_backend, Some(otg_service.clone())));
#[cfg(not(unix))]
let hid = Arc::new(HidController::new(hid_backend));
hid.set_event_bus(events.clone()).await;
if let Err(error) = hid.init().await {
tracing::warn!("Failed to initialize HID backend: {}", error);
}
#[cfg(unix)]
let msd = build_msd(&config, &data_dir, &otg_service, &events).await;
let atx = build_atx(&config).await;
let audio = build_audio(&config, &events).await;
let extensions = Arc::new(ExtensionManager::new());
tracing::info!("Extension manager initialized");
webrtc.set_hid_controller(hid.clone()).await;
webrtc.set_audio_controller(audio.clone()).await;
if config.audio.enabled {
if let Err(error) = webrtc.set_audio_enabled(true).await {
tracing::warn!("Failed to enable WebRTC audio: {}", error);
} else {
tracing::debug!("WebRTC audio enabled");
}
}
connect_capture_to_webrtc(&streamer, &webrtc).await;
let stream_manager = VideoStreamManager::with_webrtc_streamer(
streamer.clone(),
webrtc.clone() as Arc<dyn crate::video::traits::VideoOutput>,
);
stream_manager.set_event_bus(events.clone()).await;
stream_manager.set_config_store(config_store.clone()).await;
connect_audio_recovery(&audio, &stream_manager).await;
let initial_mode = config.stream.mode.clone();
if let Err(error) = stream_manager.init_with_mode(initial_mode.clone()).await {
tracing::warn!(
"Failed to initialize stream manager with mode {:?}: {}",
initial_mode,
error
);
} else {
tracing::info!(
"Video stream manager initialized with mode: {:?}",
initial_mode
);
}
let computer_use = ComputerUseManager::new(config_store.clone(), hid.clone());
let state = AppState::new(
db,
config_store.clone(),
sessions,
users,
two_factor,
#[cfg(unix)]
otg_service,
stream_manager,
webrtc,
hid,
computer_use,
#[cfg(unix)]
msd,
atx,
audio,
extensions.clone(),
events.clone(),
Arc::new(UpdateService::new()),
shutdown_tx,
data_dir.clone(),
);
start_uac_playback(&state, &config).await;
start_watchdog(&state, &config).await;
extensions.set_event_bus(events.clone()).await;
state.remote_access.start_configured(&config).await;
let extension_config = config_store.get();
extensions.start_enabled(&extension_config.extensions).await;
state.publish_device_info().await;
let supervisor = RuntimeSupervisor::start(state.clone(), events, extensions, config_store);
Ok(ApplicationRuntime {
state,
config,
data_dir,
supervisor,
})
}
}
pub struct ApplicationRuntime {
state: Arc<AppState>,
config: AppConfig,
data_dir: PathBuf,
supervisor: RuntimeSupervisor,
}
impl ApplicationRuntime {
pub fn state(&self) -> &Arc<AppState> {
&self.state
}
pub fn config(&self) -> &AppConfig {
&self.config
}
pub fn data_dir(&self) -> &Path {
&self.data_dir
}
pub fn router(&self) -> axum::Router {
crate::web::create_router(self.state.clone())
}
pub async fn shutdown(&mut self) {
self.supervisor.shutdown(&self.state).await;
}
}
async fn load_runtime_config(
data_dir: &Path,
) -> anyhow::Result<(DatabasePool, ConfigStore, AppConfig)> {
tokio::fs::create_dir_all(data_dir).await?;
let db_path = data_dir.join("one-kvm.db");
let db = DatabasePool::new(&db_path).await?;
db.init_schema().await?;
let config_store = ConfigStore::new(db.clone_pool())?;
config_store.load().await?;
let mut config = (*config_store.get()).clone();
config.apply_platform_defaults();
prepare_linux_runtime_dirs(data_dir, &config_store, &mut config).await?;
Ok((db, config_store, config))
}
#[cfg(unix)]
async fn prepare_linux_runtime_dirs(
data_dir: &Path,
config_store: &ConfigStore,
config: &mut AppConfig,
) -> anyhow::Result<()> {
let mut msd_dir_updated = false;
if config.msd.msd_dir.trim().is_empty() {
config.msd.msd_dir = data_dir.join("msd").to_string_lossy().to_string();
msd_dir_updated = true;
} else if !PathBuf::from(&config.msd.msd_dir).is_absolute() {
let msd_dir = data_dir.join(&config.msd.msd_dir);
tracing::warn!(
"MSD directory is relative, rebasing to {}",
msd_dir.display()
);
config.msd.msd_dir = msd_dir.to_string_lossy().to_string();
msd_dir_updated = true;
}
if msd_dir_updated {
config_store.set(config.clone()).await?;
}
let msd_dir = PathBuf::from(&config.msd.msd_dir);
if let Err(error) = tokio::fs::create_dir_all(msd_dir.join("images")).await {
tracing::warn!("Failed to create MSD images directory: {}", error);
}
if let Err(error) = tokio::fs::create_dir_all(msd_dir.join("ventoy")).await {
tracing::warn!("Failed to create MSD ventoy directory: {}", error);
}
Ok(())
}
#[cfg(not(unix))]
async fn prepare_linux_runtime_dirs(
_data_dir: &Path,
_config_store: &ConfigStore,
_config: &mut AppConfig,
) -> anyhow::Result<()> {
Ok(())
}
fn parse_video_config(config: &AppConfig) -> (PixelFormat, Resolution) {
let format = config
.video
.format
.as_ref()
.and_then(|format| format.parse::<PixelFormat>().ok())
.unwrap_or(PixelFormat::Mjpeg);
(
format,
Resolution::new(config.video.width, config.video.height),
)
}
async fn build_streamer(
config: &AppConfig,
events: &Arc<EventBus>,
format: PixelFormat,
resolution: Resolution,
) -> Arc<Streamer> {
tracing::debug!(
"Parsed video config: {} @ {}x{}",
format,
resolution.width,
resolution.height
);
let streamer = Streamer::new();
streamer.set_event_bus(events.clone()).await;
if let Some(device_path) = config.video.device.as_ref() {
if let Err(error) = streamer
.apply_video_config(device_path, format, resolution, config.video.fps)
.await
{
tracing::warn!(
"Failed to initialize video with config: {}, will auto-detect",
error
);
} else {
tracing::info!(
"Video configured: {} @ {}x{} {}",
device_path,
resolution.width,
resolution.height,
format
);
}
}
streamer
}
fn build_webrtc(
config: &AppConfig,
input_format: PixelFormat,
resolution: Resolution,
) -> Arc<WebRtcStreamer> {
let webrtc = WebRtcStreamer::with_config(WebRtcStreamerConfig {
resolution,
input_format,
fps: config.video.fps,
bitrate_preset: config.stream.bitrate_preset,
encoder_backend: crate::stream_encoder::encoder_type_to_backend(
config.stream.encoder.clone(),
),
webrtc: build_ice_config(config),
..Default::default()
});
tracing::info!("WebRTC streamer created");
webrtc
}
fn build_ice_config(config: &AppConfig) -> crate::webrtc::config::WebRtcConfig {
let mut stun_servers = Vec::new();
let mut turn_servers = Vec::new();
let has_custom_stun = config
.stream
.stun_server
.as_ref()
.is_some_and(|server| !server.is_empty());
let has_custom_turn = config
.stream
.turn_server
.as_ref()
.is_some_and(|server| !server.is_empty());
if !has_custom_stun && !has_custom_turn {
let stun = crate::webrtc::config::public_ice::stun_server().to_string();
tracing::info!("Using public STUN server: {}", stun);
stun_servers.push(stun);
} else {
if let Some(stun) = config
.stream
.stun_server
.as_ref()
.filter(|server| !server.is_empty())
{
tracing::info!("Using custom STUN server: {}", stun);
stun_servers.push(stun.clone());
}
if let Some(turn) = config
.stream
.turn_server
.as_ref()
.filter(|server| !server.is_empty())
{
let username = config.stream.turn_username.clone().unwrap_or_default();
let credential = config.stream.turn_password.clone().unwrap_or_default();
turn_servers.push(crate::webrtc::config::TurnServer::new(
turn.clone(),
username.clone(),
credential,
));
tracing::info!("Using custom TURN server: {} (user: {})", turn, username);
}
}
crate::webrtc::config::WebRtcConfig {
stun_servers,
turn_servers,
..Default::default()
}
}
#[cfg(unix)]
async fn build_otg(config: &AppConfig) -> Arc<OtgService> {
let service = Arc::new(OtgService::new());
tracing::info!("OTG Service created");
if let Err(error) = service
.apply_config(&config.hid, &config.msd, &config.otg_network, &config.uac)
.await
{
tracing::warn!("Failed to apply OTG config: {}", error);
}
service
}
fn hid_backend_type(config: &AppConfig) -> HidBackendType {
match config.hid.backend {
config::HidBackend::Otg => HidBackendType::Otg,
config::HidBackend::Ch9329 => HidBackendType::Ch9329 {
port: config.hid.ch9329_port.clone(),
baud_rate: config.hid.ch9329_baudrate,
hybrid_mouse: config.hid.ch9329_hybrid_mouse,
},
config::HidBackend::None => HidBackendType::None,
}
}
#[cfg(unix)]
async fn build_msd(
config: &AppConfig,
data_dir: &Path,
otg: &Arc<OtgService>,
events: &Arc<EventBus>,
) -> Option<MsdController> {
if !config.msd.enabled {
tracing::info!("MSD disabled in configuration");
return None;
}
let controller = MsdController::new(otg.clone(), config.msd.msd_dir_path());
if let Err(error) = controller.init(&data_dir.join("ventoy")).await {
tracing::warn!("Failed to initialize MSD controller: {}", error);
return None;
}
controller.set_event_bus(events.clone()).await;
Some(controller)
}
async fn build_atx(config: &AppConfig) -> Option<AtxController> {
if !config.atx.enabled {
tracing::info!("ATX disabled in configuration");
return None;
}
let controller = AtxController::new(config.atx.to_controller_config());
if let Err(error) = controller.init().await {
tracing::warn!("Failed to initialize ATX controller: {}", error);
return None;
}
Some(controller)
}
async fn build_audio(config: &AppConfig, events: &Arc<EventBus>) -> Arc<AudioController> {
let quality = config
.audio
.quality
.parse::<AudioQuality>()
.unwrap_or_else(|error| {
tracing::warn!(
"Invalid audio quality in config (value={:?}): {}, using balanced",
config.audio.quality,
error
);
AudioQuality::Balanced
});
let controller = Arc::new(AudioController::new(AudioControllerConfig {
enabled: config.audio.enabled,
device: config.audio.device.clone(),
quality,
}));
controller.set_event_bus(events.clone()).await;
if config.audio.enabled {
tracing::info!(
"Audio enabled: {}, quality={}",
config.audio.device,
config.audio.quality
);
if let Err(error) = controller.start_streaming().await {
tracing::warn!("Failed to start audio streaming: {}", error);
}
} else {
tracing::info!("Audio disabled in configuration");
}
controller
}
async fn connect_capture_to_webrtc(streamer: &Arc<Streamer>, webrtc: &Arc<WebRtcStreamer>) {
let (device_path, resolution, format, fps, jpeg_quality) =
streamer.current_capture_config().await;
tracing::debug!(
"Initial video config: {}x{} {:?} @ {}fps",
resolution.width,
resolution.height,
format,
fps
);
webrtc.update_video_config(resolution, format, fps).await;
if let Some(device_path) = device_path {
webrtc
.set_capture_device(device_path, jpeg_quality, streamer.current_device().await)
.await;
tracing::debug!("WebRTC streamer configured for direct capture");
} else {
tracing::warn!("No capture device configured for WebRTC");
}
}
async fn connect_audio_recovery(
audio: &Arc<AudioController>,
stream_manager: &Arc<VideoStreamManager>,
) {
let stream_manager = Arc::downgrade(stream_manager);
audio
.set_recovered_callback(Arc::new(move || {
if let Some(stream_manager) = stream_manager.upgrade() {
tokio::spawn(async move {
stream_manager.reconnect_webrtc_audio_sources().await;
});
}
}))
.await;
}
#[cfg(unix)]
async fn start_uac_playback(state: &Arc<AppState>, config: &AppConfig) {
if !config.uac.enabled {
return;
}
let playback_config = crate::audio::uac::UacPlaybackConfig {
sample_rate: config.uac.sample_rate,
channels: config.uac.channels as u16,
..Default::default()
};
match crate::audio::uac::UacPlayback::start(playback_config) {
Ok(writer) => {
*state.uac_playback.write().await = Some(writer);
tracing::info!("UAC playback writer started");
}
Err(error) => tracing::warn!("Failed to start UAC playback writer: {}", error),
}
}
#[cfg(not(unix))]
async fn start_uac_playback(_state: &Arc<AppState>, _config: &AppConfig) {}
async fn start_watchdog(state: &Arc<AppState>, config: &AppConfig) {
if !config.watchdog.enabled {
return;
}
if let Err(error) = state.watchdog.enable().await {
tracing::error!(
"Configured hardware watchdog failed to start; web service will continue: {}",
error
);
} else {
tracing::info!("Hardware watchdog started");
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn web_overrides_only_replace_explicit_values() {
let mut config = AppConfig::default();
let original_https_port = config.web.https_port;
WebConfigOverrides {
address: Some("127.0.0.1".to_string()),
http_port: Some(9000),
enable_https: true,
..Default::default()
}
.apply(&mut config);
assert_eq!(config.web.bind_address, "127.0.0.1");
assert_eq!(config.web.bind_addresses, ["127.0.0.1"]);
assert_eq!(config.web.http_port, 9000);
assert_eq!(config.web.https_port, original_https_port);
assert!(config.web.https_enabled);
}
}

44
src/runtime/config_apply.rs Normal file
View File

@@ -0,0 +1,44 @@
use std::sync::Arc;
use tokio::sync::{Mutex, OwnedMutexGuard};
use crate::error::{AppError, Result};
#[derive(Debug, Clone, Copy, Default)]
pub struct ConfigApplyOptions {
pub force: bool,
pub preserve_service_state: bool,
pub runtime_only: bool,
}
impl ConfigApplyOptions {
pub const fn forced() -> Self {
Self {
force: true,
preserve_service_state: false,
runtime_only: false,
}
}
pub const fn preserving_service_state() -> Self {
Self {
force: false,
preserve_service_state: true,
runtime_only: false,
}
}
pub const fn runtime_only() -> Self {
Self {
force: false,
preserve_service_state: false,
runtime_only: true,
}
}
}
pub fn try_apply_lock(lock: &Arc<Mutex<()>>, domain: &str) -> Result<OwnedMutexGuard<()>> {
lock.clone().try_lock_owned().map_err(|_| {
AppError::ServiceUnavailable(format!("{domain} configuration is already applying"))
})
}

10
src/runtime/mod.rs Normal file
View File

@@ -0,0 +1,10 @@
mod builder;
mod config_apply;
mod remote_access;
mod supervisor;
mod usb;
pub use builder::{ApplicationRuntime, RuntimeBuilder, WebConfigOverrides};
pub use config_apply::{try_apply_lock, ConfigApplyOptions};
pub use remote_access::{RemoteAccessCoordinator, RustDeskRuntimeStatus};
pub use usb::UsbCoordinator;

496
src/runtime/remote_access.rs Normal file
View File

@@ -0,0 +1,496 @@
use std::sync::Arc;
use tokio::sync::RwLock;
use crate::audio::AudioController;
use crate::config::{AppConfig, ConfigStore, RtspConfig, VncConfig};
use crate::error::{AppError, Result};
use crate::hid::HidController;
use crate::rtsp::{RtspService, RtspServiceStatus};
use crate::rustdesk::config::RustDeskConfig;
use crate::rustdesk::RustDeskService;
use crate::video::codec_constraints::{
enforce_constraints_with_stream_manager, validate_third_party_codec_compatibility,
StreamCodecConstraints,
};
use crate::video::VideoStreamManager;
use crate::vnc::{VncService, VncServiceStatus};
use super::ConfigApplyOptions;
#[derive(Debug, Clone)]
pub struct RustDeskRuntimeStatus {
pub service_status: String,
pub rendezvous_status: Option<String>,
}
pub struct RemoteAccessCoordinator {
config: ConfigStore,
stream_manager: Arc<VideoStreamManager>,
hid: Arc<HidController>,
audio: Arc<AudioController>,
rustdesk: RwLock<Option<Arc<RustDeskService>>>,
vnc: RwLock<Option<Arc<VncService>>>,
rtsp: RwLock<Option<Arc<RtspService>>>,
}
impl RemoteAccessCoordinator {
pub fn new(
config: ConfigStore,
stream_manager: Arc<VideoStreamManager>,
hid: Arc<HidController>,
audio: Arc<AudioController>,
) -> Arc<Self> {
Arc::new(Self {
config,
stream_manager,
hid,
audio,
rustdesk: RwLock::new(None),
vnc: RwLock::new(None),
rtsp: RwLock::new(None),
})
}
pub async fn start_configured(&self, config: &AppConfig) {
if let Err(error) = validate_third_party_codec_compatibility(config) {
tracing::warn!(
"Third-party access codec configuration is invalid; RustDesk/VNC/RTSP will not start: {}",
error
);
return;
}
if config.rustdesk.is_valid() {
if let Err(error) = self
.apply_rustdesk(
&RustDeskConfig::default(),
&config.rustdesk,
ConfigApplyOptions::default(),
)
.await
{
tracing::error!("Failed to start RustDesk service: {}", error);
}
} else if config.rustdesk.enabled {
tracing::warn!(
"RustDesk enabled but configuration is incomplete (missing server or credentials)"
);
} else {
tracing::info!("RustDesk disabled in configuration");
}
if config.vnc.enabled {
if let Err(error) = self
.apply_vnc(
&VncConfig::default(),
&config.vnc,
ConfigApplyOptions::default(),
)
.await
{
tracing::error!("Failed to start VNC service: {}", error);
}
} else {
tracing::info!("VNC disabled in configuration");
}
if config.rtsp.enabled {
if let Err(error) = self
.apply_rtsp(
&RtspConfig::default(),
&config.rtsp,
ConfigApplyOptions::default(),
)
.await
{
tracing::error!("Failed to start RTSP service: {}", error);
}
} else {
tracing::info!("RTSP disabled in configuration");
}
if let Err(error) = self.enforce_codec_constraints().await {
tracing::warn!("Failed to enforce startup codec constraints: {}", error);
}
}
pub async fn runtime_config(&self) -> AppConfig {
let mut config = self.config.get().as_ref().clone();
let rustdesk = self.rustdesk.read().await.clone();
let vnc = self.vnc.read().await.clone();
let rtsp = self.rtsp.read().await.clone();
config.rustdesk.enabled = rustdesk.is_some_and(|service| service.is_listening());
config.vnc.enabled = match vnc {
Some(service) => matches!(
service.status().await,
VncServiceStatus::Starting | VncServiceStatus::Running
),
None => false,
};
config.rtsp.enabled = match rtsp {
Some(service) => matches!(
service.status().await,
RtspServiceStatus::Starting | RtspServiceStatus::Running
),
None => false,
};
config
}
pub async fn rustdesk_status(&self) -> RustDeskRuntimeStatus {
let service = self.rustdesk.read().await.clone();
match service {
Some(service) => RustDeskRuntimeStatus {
service_status: service.status().to_string(),
rendezvous_status: service.rendezvous_status().map(|status| status.to_string()),
},
None => RustDeskRuntimeStatus {
service_status: "not_initialized".to_string(),
rendezvous_status: None,
},
}
}
pub async fn vnc_status(&self) -> (VncServiceStatus, usize) {
let service = self.vnc.read().await.clone();
match service {
Some(service) => (service.status().await, service.connection_count()),
None => (VncServiceStatus::Stopped, 0),
}
}
pub async fn rtsp_status(&self) -> RtspServiceStatus {
let service = self.rtsp.read().await.clone();
match service {
Some(service) => service.status().await,
None => RtspServiceStatus::Stopped,
}
}
pub async fn enforce_codec_constraints(&self) -> Result<Option<String>> {
let config = self.runtime_config().await;
let constraints = StreamCodecConstraints::from_config(&config);
self.stream_manager
.set_runtime_codec_constraints(constraints.clone())
.await;
let enforcement =
enforce_constraints_with_stream_manager(&self.stream_manager, &constraints).await?;
Ok(enforcement.message)
}
pub async fn apply_rustdesk(
&self,
old_config: &RustDeskConfig,
new_config: &RustDeskConfig,
options: ConfigApplyOptions,
) -> Result<()> {
tracing::info!("Applying RustDesk config changes...");
self.validate_rustdesk_candidate(new_config, options.runtime_only)
.await?;
let need_restart = options.force
|| old_config.codec != new_config.codec
|| old_config.rendezvous_server != new_config.rendezvous_server
|| old_config.device_id != new_config.device_id
|| old_config.device_password != new_config.device_password;
let current = self.rustdesk.read().await.clone();
let mut credentials_to_save = None;
if !options.preserve_service_state && !new_config.enabled {
if let Some(service) = current.as_ref() {
service.stop().await.map_err(|error| {
AppError::Config(format!("Failed to stop RustDesk service: {error}"))
})?;
tracing::info!("RustDesk service stopped");
}
*self.rustdesk.write().await = None;
} else if !options.preserve_service_state && new_config.enabled {
match current {
None => {
tracing::info!("Initializing RustDesk service...");
let service = Arc::new(RustDeskService::new(
new_config.clone(),
self.stream_manager.clone(),
self.hid.clone(),
self.audio.clone(),
));
*self.rustdesk.write().await = Some(service.clone());
service.start().await.map_err(|error| {
AppError::Config(format!("Failed to start RustDesk service: {error}"))
})?;
tracing::info!("RustDesk service started with ID: {}", new_config.device_id);
credentials_to_save = service.save_credentials();
}
Some(service) => {
if service.is_listening() {
if need_restart {
service.restart(new_config.clone()).await.map_err(|error| {
AppError::Config(format!(
"Failed to restart RustDesk service: {error}"
))
})?;
tracing::info!(
"RustDesk service restarted with ID: {}",
new_config.device_id
);
}
} else {
service.update_config(new_config.clone());
service.start().await.map_err(|error| {
AppError::Config(format!("Failed to start RustDesk service: {error}"))
})?;
}
credentials_to_save = service.save_credentials();
}
}
} else if options.preserve_service_state && need_restart {
if let Some(service) = current {
let mut runtime_config = new_config.clone();
runtime_config.enabled = true;
service.restart(runtime_config).await.map_err(|error| {
AppError::Config(format!("Failed to restart RustDesk service: {error}"))
})?;
credentials_to_save = service.save_credentials();
}
}
if let Some(updated) = credentials_to_save {
tracing::info!("Saving RustDesk credentials to config store...");
self.config
.update(|config| {
config.rustdesk.public_key = updated.public_key.clone();
config.rustdesk.private_key = updated.private_key.clone();
config.rustdesk.signing_public_key = updated.signing_public_key.clone();
config.rustdesk.signing_private_key = updated.signing_private_key.clone();
config.rustdesk.uuid = updated.uuid.clone();
})
.await?;
tracing::info!("RustDesk credentials saved successfully");
}
self.log_enforced_constraints().await?;
Ok(())
}
pub async fn apply_vnc(
&self,
old_config: &VncConfig,
new_config: &VncConfig,
options: ConfigApplyOptions,
) -> Result<()> {
tracing::info!("Applying VNC config changes...");
self.validate_vnc_candidate(new_config, options.runtime_only)
.await?;
let runtime_config = self.runtime_config().await;
let will_run = if options.preserve_service_state {
runtime_config.vnc.enabled
} else {
new_config.enabled
};
if will_run {
let mut candidate = runtime_config;
candidate.vnc = new_config.clone();
candidate.vnc.enabled = true;
let constraints = StreamCodecConstraints::from_config(&candidate);
match enforce_constraints_with_stream_manager(&self.stream_manager, &constraints).await
{
Ok(result) if result.changed => {
if let Some(message) = result.message {
tracing::info!("{}", message);
}
}
Ok(_) => {}
Err(error) => tracing::warn!(
"Failed to enforce VNC stream constraints before start: {}",
error
),
}
}
let need_restart = options.force
|| old_config.bind != new_config.bind
|| old_config.port != new_config.port
|| old_config.encoding != new_config.encoding
|| old_config.password != new_config.password
|| old_config.allow_one_client != new_config.allow_one_client;
let current = self.vnc.read().await.clone();
if !options.preserve_service_state && !new_config.enabled {
if let Some(service) = current.as_ref() {
service.stop().await?;
}
*self.vnc.write().await = None;
} else if !options.preserve_service_state && new_config.enabled {
match current {
None => {
let service = Arc::new(VncService::new(
new_config.clone(),
self.stream_manager.clone(),
self.hid.clone(),
));
*self.vnc.write().await = Some(service.clone());
service.start().await?;
tracing::info!("VNC service started");
}
Some(service) => {
if matches!(service.status().await, VncServiceStatus::Running) {
if need_restart {
service.restart(new_config.clone()).await?;
tracing::info!("VNC service restarted");
}
} else {
service.update_config(new_config.clone()).await;
service.start().await?;
}
}
}
} else if options.preserve_service_state && need_restart {
if let Some(service) = current {
let mut runtime_config = new_config.clone();
runtime_config.enabled = true;
service.restart(runtime_config).await?;
}
}
self.log_enforced_constraints().await?;
Ok(())
}
pub async fn apply_rtsp(
&self,
old_config: &RtspConfig,
new_config: &RtspConfig,
options: ConfigApplyOptions,
) -> Result<()> {
tracing::info!("Applying RTSP config changes...");
self.validate_rtsp_candidate(new_config, options.runtime_only)
.await?;
let need_restart = options.force
|| old_config.bind != new_config.bind
|| old_config.port != new_config.port
|| old_config.path != new_config.path
|| old_config.codec != new_config.codec
|| old_config.username != new_config.username
|| old_config.password != new_config.password
|| old_config.allow_one_client != new_config.allow_one_client;
let current = self.rtsp.read().await.clone();
if !options.preserve_service_state && !new_config.enabled {
if let Some(service) = current.as_ref() {
service.stop().await.map_err(|error| {
AppError::Config(format!("Failed to stop RTSP service: {error}"))
})?;
}
*self.rtsp.write().await = None;
} else if !options.preserve_service_state && new_config.enabled {
match current {
None => {
let service = Arc::new(RtspService::new(
new_config.clone(),
self.stream_manager.clone(),
));
*self.rtsp.write().await = Some(service.clone());
service.start().await?;
tracing::info!("RTSP service started");
}
Some(service) => {
if matches!(service.status().await, RtspServiceStatus::Running) {
if need_restart {
service.restart(new_config.clone()).await?;
tracing::info!("RTSP service restarted");
}
} else {
service.update_config(new_config.clone()).await;
service.start().await?;
}
}
}
} else if options.preserve_service_state && need_restart {
if let Some(service) = current {
let mut runtime_config = new_config.clone();
runtime_config.enabled = true;
service.restart(runtime_config).await?;
}
}
self.log_enforced_constraints().await?;
Ok(())
}
pub async fn shutdown(&self) {
let rustdesk = self.rustdesk.write().await.take();
let vnc = self.vnc.write().await.take();
let rtsp = self.rtsp.write().await.take();
if let Some(service) = rustdesk {
if let Err(error) = service.stop().await {
tracing::warn!("Failed to stop RustDesk service: {}", error);
} else {
tracing::info!("RustDesk service stopped");
}
}
if let Some(service) = vnc {
if let Err(error) = service.stop().await {
tracing::warn!("Failed to stop VNC service: {}", error);
} else {
tracing::info!("VNC service stopped");
}
}
if let Some(service) = rtsp {
if let Err(error) = service.stop().await {
tracing::warn!("Failed to stop RTSP service: {}", error);
} else {
tracing::info!("RTSP service stopped");
}
}
}
async fn validate_rustdesk_candidate(
&self,
new_config: &RustDeskConfig,
runtime_only: bool,
) -> Result<()> {
let mut candidate = self.candidate_config(runtime_only).await;
candidate.rustdesk = new_config.clone();
validate_third_party_codec_compatibility(&candidate)
}
async fn validate_vnc_candidate(
&self,
new_config: &VncConfig,
runtime_only: bool,
) -> Result<()> {
let mut candidate = self.candidate_config(runtime_only).await;
candidate.vnc = new_config.clone();
validate_third_party_codec_compatibility(&candidate)
}
async fn validate_rtsp_candidate(
&self,
new_config: &RtspConfig,
runtime_only: bool,
) -> Result<()> {
let mut candidate = self.candidate_config(runtime_only).await;
candidate.rtsp = new_config.clone();
validate_third_party_codec_compatibility(&candidate)
}
async fn candidate_config(&self, runtime_only: bool) -> AppConfig {
if runtime_only {
self.runtime_config().await
} else {
self.config.get().as_ref().clone()
}
}
async fn log_enforced_constraints(&self) -> Result<()> {
if let Some(message) = self.enforce_codec_constraints().await? {
tracing::info!("{}", message);
}
Ok(())
}
}

228
src/runtime/supervisor.rs Normal file
View File

@@ -0,0 +1,228 @@
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use crate::config::ConfigStore;
use crate::events::EventBus;
use crate::extensions::ExtensionManager;
use crate::state::AppState;
pub(super) struct RuntimeSupervisor {
tasks: Vec<JoinHandle<()>>,
}
impl RuntimeSupervisor {
pub(super) fn start(
state: Arc<AppState>,
events: Arc<EventBus>,
extensions: Arc<ExtensionManager>,
config: ConfigStore,
) -> Self {
let mut tasks = spawn_device_info_broadcaster(state, events);
tasks.push(spawn_extension_health_check(extensions, config));
Self { tasks }
}
pub(super) async fn shutdown(&mut self, state: &Arc<AppState>) {
for task in self.tasks.drain(..) {
task.abort();
}
cleanup(state).await;
}
}
fn spawn_extension_health_check(
extensions: Arc<ExtensionManager>,
config: ConfigStore,
) -> JoinHandle<()> {
let task = tokio::spawn(async move {
let mut interval = tokio::time::interval(Duration::from_secs(30));
loop {
interval.tick().await;
let config = config.get();
extensions.health_check(&config.extensions).await;
}
});
tracing::info!("Extension health check task started");
task
}
fn spawn_device_info_broadcaster(
state: Arc<AppState>,
events: Arc<EventBus>,
) -> Vec<JoinHandle<()>> {
enum DeviceInfoTrigger {
Event,
Lagged { topic: &'static str, count: u64 },
}
const DEVICE_INFO_TOPICS: &[&str] = &[
"stream.state_changed",
"stream.config_applied",
"stream.mode_ready",
];
const DEBOUNCE_MS: u64 = 100;
let (trigger_tx, mut trigger_rx) = mpsc::unbounded_channel();
let mut tasks = Vec::new();
for topic in DEVICE_INFO_TOPICS {
let Some(mut rx) = events.subscribe_topic(topic) else {
tracing::warn!(
"DeviceInfo broadcaster missing topic subscription: {}",
topic
);
continue;
};
let trigger_tx = trigger_tx.clone();
let topic_name = *topic;
tasks.push(tokio::spawn(async move {
loop {
match rx.recv().await {
Ok(_) => {
if trigger_tx.send(DeviceInfoTrigger::Event).is_err() {
break;
}
}
Err(tokio::sync::broadcast::error::RecvError::Lagged(count)) => {
if trigger_tx
.send(DeviceInfoTrigger::Lagged {
topic: topic_name,
count,
})
.is_err()
{
break;
}
}
Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
}
}
}));
}
{
let mut dirty_rx = events.subscribe_device_info_dirty();
let trigger_tx = trigger_tx.clone();
tasks.push(tokio::spawn(async move {
loop {
match dirty_rx.recv().await {
Ok(()) => {
if trigger_tx.send(DeviceInfoTrigger::Event).is_err() {
break;
}
}
Err(tokio::sync::broadcast::error::RecvError::Lagged(count)) => {
if trigger_tx
.send(DeviceInfoTrigger::Lagged {
topic: "device_info_dirty",
count,
})
.is_err()
{
break;
}
}
Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
}
}
}));
}
tasks.push(tokio::spawn(async move {
let mut last_broadcast = Instant::now() - Duration::from_millis(DEBOUNCE_MS);
let mut pending_broadcast = false;
loop {
let recv_result = if pending_broadcast {
let remaining =
DEBOUNCE_MS.saturating_sub(last_broadcast.elapsed().as_millis() as u64);
tokio::time::timeout(Duration::from_millis(remaining), trigger_rx.recv()).await
} else {
Ok(trigger_rx.recv().await)
};
match recv_result {
Ok(Some(DeviceInfoTrigger::Event)) => {
pending_broadcast = true;
}
Ok(Some(DeviceInfoTrigger::Lagged { topic, count })) => {
tracing::warn!(
"DeviceInfo broadcaster lagged by {} events on topic {}",
count,
topic
);
pending_broadcast = true;
}
Ok(None) => {
tracing::info!("Event bus closed, stopping DeviceInfo broadcaster");
break;
}
Err(_timeout) => {}
}
if pending_broadcast && last_broadcast.elapsed() >= Duration::from_millis(DEBOUNCE_MS) {
state.publish_device_info().await;
tracing::trace!("Broadcasted DeviceInfo (debounced)");
last_broadcast = Instant::now();
pending_broadcast = false;
}
}
}));
tracing::info!(
"DeviceInfo broadcaster task started (debounce: {}ms)",
DEBOUNCE_MS
);
tasks
}
async fn cleanup(state: &Arc<AppState>) {
state.extensions.stop_all().await;
tracing::info!("Extensions stopped");
state.remote_access.shutdown().await;
if let Err(error) = state.stream_manager.stop().await {
tracing::warn!("Failed to stop streamer: {}", error);
}
if let Err(error) = state.hid.shutdown().await {
tracing::warn!("Failed to shutdown HID: {}", error);
}
#[cfg(unix)]
{
let msd = state.msd.write().await.take();
if let Some(msd) = msd {
if let Err(error) = msd.shutdown().await {
tracing::warn!("Failed to shutdown MSD: {}", error);
}
}
if let Err(error) = state.otg_service.shutdown().await {
tracing::warn!("Failed to shutdown OTG: {}", error);
}
}
let atx = state.atx.write().await.take();
if let Some(atx) = atx {
if let Err(error) = atx.shutdown().await {
tracing::warn!("Failed to shutdown ATX: {}", error);
}
}
if let Err(error) = state.audio.shutdown().await {
tracing::warn!("Failed to shutdown audio: {}", error);
}
if let Err(error) = state.watchdog.disable().await {
tracing::error!(
"CRITICAL: failed to disable hardware watchdog during shutdown: {}",
error
);
}
}

335
src/runtime/usb.rs Normal file
View File

@@ -0,0 +1,335 @@
use std::path::PathBuf;
use std::sync::Arc;
#[cfg(unix)]
use tokio::sync::RwLock;
use crate::config::{AppConfig, HidBackend, HidConfig, MsdConfig, OtgNetworkConfig, UacConfig};
use crate::error::{AppError, Result};
use crate::events::EventBus;
use crate::hid::{HidBackendType, HidController};
#[cfg(unix)]
use crate::msd::MsdController;
#[cfg(unix)]
use crate::otg::OtgService;
use super::ConfigApplyOptions;
pub struct UsbCoordinator {
hid: Arc<HidController>,
#[cfg(unix)]
otg: Arc<OtgService>,
#[cfg(unix)]
msd: Arc<RwLock<Option<MsdController>>>,
#[cfg(unix)]
uac_playback: Arc<RwLock<Option<crate::audio::uac::UacPlayback>>>,
events: Arc<EventBus>,
data_dir: PathBuf,
}
impl UsbCoordinator {
#[allow(clippy::too_many_arguments)]
pub fn new(
hid: Arc<HidController>,
#[cfg(unix)] otg: Arc<OtgService>,
#[cfg(unix)] msd: Arc<RwLock<Option<MsdController>>>,
#[cfg(unix)] uac_playback: Arc<RwLock<Option<crate::audio::uac::UacPlayback>>>,
events: Arc<EventBus>,
data_dir: PathBuf,
) -> Arc<Self> {
Arc::new(Self {
hid,
#[cfg(unix)]
otg,
#[cfg(unix)]
msd,
#[cfg(unix)]
uac_playback,
events,
data_dir,
})
}
pub async fn apply_config(&self, old_config: &AppConfig, new_config: &AppConfig) -> Result<()> {
#[cfg(unix)]
{
let transitioning_away_from_otg = old_config.hid.backend == HidBackend::Otg
&& new_config.hid.backend != HidBackend::Otg;
let hid_unchanged = old_config.hid == new_config.hid;
let gadget_rebuilt = old_config.msd != new_config.msd
|| old_config.otg_network != new_config.otg_network
|| old_config.uac != new_config.uac
|| old_config.hid.otg_udc != new_config.hid.otg_udc
|| old_config.hid.otg_descriptor != new_config.hid.otg_descriptor
|| old_config.hid.backend != new_config.hid.backend
|| old_config.hid.constrained_otg_functions()
!= new_config.hid.constrained_otg_functions()
|| old_config.hid.effective_otg_keyboard_leds()
!= new_config.hid.effective_otg_keyboard_leds();
let restart_uac =
old_config.uac != new_config.uac || (new_config.uac.enabled && gadget_rebuilt);
if restart_uac {
let playback = self.uac_playback.write().await.take();
if let Some(playback) = playback {
playback.stop();
tracing::info!("UAC playback writer stopped before OTG reconcile");
}
}
if transitioning_away_from_otg {
self.apply_hid(
&old_config.hid,
&new_config.hid,
&new_config.msd,
&new_config.otg_network,
&new_config.uac,
ConfigApplyOptions::default(),
)
.await?;
} else {
self.reconcile_otg(
&new_config.hid,
&new_config.msd,
&new_config.otg_network,
&new_config.uac,
)
.await?;
self.apply_hid(
&old_config.hid,
&new_config.hid,
&new_config.msd,
&new_config.otg_network,
&new_config.uac,
ConfigApplyOptions::default(),
)
.await?;
}
if hid_unchanged && gadget_rebuilt && new_config.hid.backend == HidBackend::Otg {
tracing::info!("OTG gadget rebuilt, reloading HID backend for new devices");
self.hid
.reload(hid_backend_type(&new_config.hid))
.await
.map_err(|error| {
AppError::Config(format!("HID reload after gadget rebuild failed: {error}"))
})?;
}
self.apply_msd(
&old_config.msd,
&new_config.msd,
&new_config.hid,
&new_config.otg_network,
&new_config.uac,
ConfigApplyOptions::default(),
)
.await?;
if restart_uac && new_config.uac.enabled {
let config = crate::audio::uac::UacPlaybackConfig {
sample_rate: new_config.uac.sample_rate,
channels: new_config.uac.channels as u16,
..Default::default()
};
let writer = crate::audio::uac::UacPlayback::start(config).map_err(|error| {
AppError::Config(format!("Failed to start UAC playback: {error}"))
})?;
*self.uac_playback.write().await = Some(writer);
tracing::info!("UAC playback writer started after OTG reconcile");
} else if restart_uac {
tracing::info!("UAC playback remains disabled");
}
Ok(())
}
#[cfg(not(unix))]
{
self.apply_hid(
&old_config.hid,
&new_config.hid,
&new_config.msd,
&new_config.otg_network,
&new_config.uac,
ConfigApplyOptions::default(),
)
.await
}
}
async fn apply_hid(
&self,
old_config: &HidConfig,
new_config: &HidConfig,
msd_config: &MsdConfig,
network_config: &OtgNetworkConfig,
uac_config: &UacConfig,
options: ConfigApplyOptions,
) -> Result<()> {
new_config.validate_otg_functions()?;
let descriptor_changed = old_config.otg_descriptor != new_config.otg_descriptor;
let hid_functions_changed =
old_config.constrained_otg_functions() != new_config.constrained_otg_functions();
let keyboard_leds_changed =
old_config.effective_otg_keyboard_leds() != new_config.effective_otg_keyboard_leds();
let ch9329_runtime_changed =
old_config.ch9329_hybrid_mouse != new_config.ch9329_hybrid_mouse;
if old_config.backend == new_config.backend
&& old_config.ch9329_port == new_config.ch9329_port
&& old_config.ch9329_baudrate == new_config.ch9329_baudrate
&& !ch9329_runtime_changed
&& old_config.otg_udc == new_config.otg_udc
&& !descriptor_changed
&& !hid_functions_changed
&& !keyboard_leds_changed
&& !options.force
{
tracing::info!("HID config unchanged, skipping reload");
return Ok(());
}
tracing::info!("Applying HID config changes...");
let backend = hid_backend_type(new_config);
let transitioning_away_from_otg =
old_config.backend == HidBackend::Otg && new_config.backend != HidBackend::Otg;
let otg_changed = hid_otg_config_changed(old_config, new_config);
if transitioning_away_from_otg {
self.hid
.reload(backend.clone())
.await
.map_err(|error| AppError::Config(format!("HID reload failed: {error}")))?;
}
if otg_changed {
self.reconcile_otg(new_config, msd_config, network_config, uac_config)
.await?;
}
if !transitioning_away_from_otg {
self.hid
.reload(backend)
.await
.map_err(|error| AppError::Config(format!("HID reload failed: {error}")))?;
}
tracing::info!(
"HID backend reloaded successfully: {:?}",
new_config.backend
);
Ok(())
}
async fn reconcile_otg(
&self,
hid: &HidConfig,
msd: &MsdConfig,
network: &OtgNetworkConfig,
uac: &UacConfig,
) -> Result<()> {
#[cfg(unix)]
{
self.otg
.apply_config(hid, msd, network, uac)
.await
.map_err(|error| AppError::Config(format!("OTG reconcile failed: {error}")))
}
#[cfg(not(unix))]
{
let _ = (hid, msd, network, uac);
Ok(())
}
}
#[cfg(unix)]
async fn apply_msd(
&self,
old_config: &MsdConfig,
new_config: &MsdConfig,
hid_config: &HidConfig,
network_config: &OtgNetworkConfig,
uac_config: &UacConfig,
options: ConfigApplyOptions,
) -> Result<()> {
let old_enabled = old_config.enabled;
let new_enabled = new_config.enabled && hid_config.backend == HidBackend::Otg;
let directory_changed = old_config.msd_dir != new_config.msd_dir;
let inquiry_changed = old_config.flash_inquiry_string != new_config.flash_inquiry_string
|| old_config.cdrom_inquiry_string != new_config.cdrom_inquiry_string;
let msd_dir = new_config.msd_dir_path();
if let Err(error) = std::fs::create_dir_all(msd_dir.join("images")) {
tracing::warn!("Failed to create MSD images directory: {}", error);
}
if let Err(error) = std::fs::create_dir_all(msd_dir.join("ventoy")) {
tracing::warn!("Failed to create MSD ventoy directory: {}", error);
}
if !options.force && old_enabled == new_enabled && !directory_changed && !inquiry_changed {
tracing::info!("MSD configuration unchanged, no reload needed");
return Ok(());
}
if new_enabled {
tracing::info!("(Re)initializing MSD...");
self.reconcile_otg(hid_config, new_config, network_config, uac_config)
.await?;
let old_msd = self.msd.write().await.take();
if let Some(msd) = old_msd {
msd.shutdown()
.await
.map_err(|error| AppError::Config(format!("MSD shutdown failed: {error}")))?;
}
let msd = MsdController::new(self.otg.clone(), new_config.msd_dir_path());
msd.init(&self.data_dir.join("ventoy"))
.await
.map_err(|error| AppError::Config(format!("MSD initialization failed: {error}")))?;
msd.set_event_bus(self.events.clone()).await;
*self.msd.write().await = Some(msd);
tracing::info!("MSD initialized successfully");
} else {
tracing::info!("MSD disabled in config, shutting down...");
let old_msd = self.msd.write().await.take();
if let Some(msd) = old_msd {
msd.shutdown()
.await
.map_err(|error| AppError::Config(format!("MSD shutdown failed: {error}")))?;
}
tracing::info!("MSD shutdown complete");
self.reconcile_otg(hid_config, new_config, network_config, uac_config)
.await?;
}
if hid_config.backend == HidBackend::Otg && (options.force || old_enabled != new_enabled) {
self.hid
.reload(HidBackendType::Otg)
.await
.map_err(|error| AppError::Config(format!("OTG HID reload failed: {error}")))?;
}
Ok(())
}
}
fn hid_backend_type(config: &HidConfig) -> HidBackendType {
match config.backend {
HidBackend::Otg => HidBackendType::Otg,
HidBackend::Ch9329 => HidBackendType::Ch9329 {
port: config.ch9329_port.clone(),
baud_rate: config.ch9329_baudrate,
hybrid_mouse: config.ch9329_hybrid_mouse,
},
HidBackend::None => HidBackendType::None,
}
}
fn hid_otg_config_changed(old_config: &HidConfig, new_config: &HidConfig) -> bool {
old_config.backend == HidBackend::Otg
|| new_config.backend == HidBackend::Otg
|| old_config.otg_udc != new_config.otg_udc
|| old_config.otg_descriptor != new_config.otg_descriptor
|| old_config.constrained_otg_functions() != new_config.constrained_otg_functions()
|| old_config.effective_otg_keyboard_leds() != new_config.effective_otg_keyboard_leds()
}

70
src/state.rs
View File

@@ -19,11 +19,9 @@ use crate::hid::HidController;
use crate::msd::MsdController; use crate::msd::MsdController;
#[cfg(unix)] #[cfg(unix)]
use crate::otg::OtgService; use crate::otg::OtgService;
use crate::rtsp::RtspService; use crate::runtime::{RemoteAccessCoordinator, UsbCoordinator};
use crate::rustdesk::RustDeskService;
use crate::update::UpdateService; use crate::update::UpdateService;
use crate::video::VideoStreamManager; use crate::video::VideoStreamManager;
use crate::vnc::VncService;
use crate::watchdog::WatchdogController; use crate::watchdog::WatchdogController;
use crate::webrtc::WebRtcStreamer; use crate::webrtc::WebRtcStreamer;
@@ -81,9 +79,8 @@ pub struct AppState {
pub audio: Arc<AudioController>, pub audio: Arc<AudioController>,
#[cfg(unix)] #[cfg(unix)]
pub uac_playback: Arc<RwLock<Option<crate::audio::uac::UacPlayback>>>, pub uac_playback: Arc<RwLock<Option<crate::audio::uac::UacPlayback>>>,
pub rustdesk: Arc<RwLock<Option<Arc<RustDeskService>>>>, pub usb: Arc<UsbCoordinator>,
pub vnc: Arc<RwLock<Option<Arc<VncService>>>>, pub remote_access: Arc<RemoteAccessCoordinator>,
pub rtsp: Arc<RwLock<Option<Arc<RtspService>>>>,
pub extensions: Arc<ExtensionManager>, pub extensions: Arc<ExtensionManager>,
pub events: Arc<EventBus>, pub events: Arc<EventBus>,
device_info_tx: watch::Sender<Option<SystemEvent>>, device_info_tx: watch::Sender<Option<SystemEvent>>,
@@ -111,9 +108,6 @@ impl AppState {
#[cfg(unix)] msd: Option<MsdController>, #[cfg(unix)] msd: Option<MsdController>,
atx: Option<AtxController>, atx: Option<AtxController>,
audio: Arc<AudioController>, audio: Arc<AudioController>,
rustdesk: Option<Arc<RustDeskService>>,
vnc: Option<Arc<VncService>>,
rtsp: Option<Arc<RtspService>>,
extensions: Arc<ExtensionManager>, extensions: Arc<ExtensionManager>,
events: Arc<EventBus>, events: Arc<EventBus>,
update: Arc<UpdateService>, update: Arc<UpdateService>,
@@ -122,6 +116,28 @@ impl AppState {
) -> Arc<Self> { ) -> Arc<Self> {
let (device_info_tx, _device_info_rx) = watch::channel(None); let (device_info_tx, _device_info_rx) = watch::channel(None);
let remote_access = RemoteAccessCoordinator::new(
config.clone(),
stream_manager.clone(),
hid.clone(),
audio.clone(),
);
#[cfg(unix)]
let msd = Arc::new(RwLock::new(msd));
#[cfg(unix)]
let uac_playback = Arc::new(RwLock::new(None));
let usb = UsbCoordinator::new(
hid.clone(),
#[cfg(unix)]
otg_service.clone(),
#[cfg(unix)]
msd.clone(),
#[cfg(unix)]
uac_playback.clone(),
events.clone(),
data_dir.clone(),
);
Arc::new(Self { Arc::new(Self {
db, db,
config, config,
@@ -135,12 +151,11 @@ impl AppState {
hid, hid,
computer_use, computer_use,
#[cfg(unix)] #[cfg(unix)]
msd: Arc::new(RwLock::new(msd)), msd,
atx: Arc::new(RwLock::new(atx)), atx: Arc::new(RwLock::new(atx)),
audio, audio,
rustdesk: Arc::new(RwLock::new(rustdesk)), usb,
vnc: Arc::new(RwLock::new(vnc)), remote_access,
rtsp: Arc::new(RwLock::new(rtsp)),
extensions, extensions,
events, events,
device_info_tx, device_info_tx,
@@ -151,7 +166,7 @@ impl AppState {
config_apply_locks: ConfigApplyLocks::new(), config_apply_locks: ConfigApplyLocks::new(),
data_dir, data_dir,
#[cfg(unix)] #[cfg(unix)]
uac_playback: Arc::new(RwLock::new(None)), uac_playback,
}) })
} }
@@ -159,33 +174,6 @@ impl AppState {
&self.data_dir &self.data_dir
} }
pub async fn runtime_third_party_config(&self) -> crate::config::AppConfig {
let mut config = self.config.get().as_ref().clone();
config.rustdesk.enabled = self
.rustdesk
.read()
.await
.as_ref()
.is_some_and(|service| service.is_listening());
config.vnc.enabled = match self.vnc.read().await.as_ref() {
Some(service) => matches!(
service.status().await,
crate::vnc::VncServiceStatus::Starting | crate::vnc::VncServiceStatus::Running
),
None => false,
};
config.rtsp.enabled = match self.rtsp.read().await.as_ref() {
Some(service) => matches!(
service.status().await,
crate::rtsp::RtspServiceStatus::Starting | crate::rtsp::RtspServiceStatus::Running
),
None => false,
};
config
}
pub fn subscribe_device_info(&self) -> watch::Receiver<Option<SystemEvent>> { pub fn subscribe_device_info(&self) -> watch::Receiver<Option<SystemEvent>> {
self.device_info_tx.subscribe() self.device_info_tx.subscribe()
} }

721
src/web/handlers/config/apply.rs
View File

@@ -2,96 +2,9 @@ use std::sync::Arc;
use crate::config::*; use crate::config::*;
use crate::error::{AppError, Result}; use crate::error::{AppError, Result};
use crate::rtsp::RtspService; pub use crate::runtime::{try_apply_lock, ConfigApplyOptions};
use crate::state::AppState; use crate::state::AppState;
use crate::stream_encoder::encoder_type_to_backend; use crate::stream_encoder::encoder_type_to_backend;
use crate::video::codec_constraints::{
enforce_constraints_with_stream_manager, validate_third_party_codec_compatibility,
StreamCodecConstraints,
};
use tokio::sync::{Mutex, OwnedMutexGuard};
#[derive(Debug, Clone, Copy, Default)]
pub struct ConfigApplyOptions {
pub force: bool,
pub preserve_service_state: bool,
pub runtime_only: bool,
}
impl ConfigApplyOptions {
pub const fn forced() -> Self {
Self {
force: true,
preserve_service_state: false,
runtime_only: false,
}
}
pub const fn preserving_service_state() -> Self {
Self {
force: false,
preserve_service_state: true,
runtime_only: false,
}
}
pub const fn runtime_only() -> Self {
Self {
force: false,
preserve_service_state: false,
runtime_only: true,
}
}
}
pub fn try_apply_lock(lock: &Arc<Mutex<()>>, domain: &str) -> Result<OwnedMutexGuard<()>> {
lock.clone().try_lock_owned().map_err(|_| {
AppError::ServiceUnavailable(format!("{domain} configuration is already applying"))
})
}
fn hid_backend_type(config: &HidConfig) -> crate::hid::HidBackendType {
match config.backend {
HidBackend::Otg => crate::hid::HidBackendType::Otg,
HidBackend::Ch9329 => crate::hid::HidBackendType::Ch9329 {
port: config.ch9329_port.clone(),
baud_rate: config.ch9329_baudrate,
hybrid_mouse: config.ch9329_hybrid_mouse,
},
HidBackend::None => crate::hid::HidBackendType::None,
}
}
fn hid_otg_config_changed(old_config: &HidConfig, new_config: &HidConfig) -> bool {
old_config.backend == HidBackend::Otg
|| new_config.backend == HidBackend::Otg
|| old_config.otg_udc != new_config.otg_udc
|| old_config.otg_descriptor != new_config.otg_descriptor
|| old_config.constrained_otg_functions() != new_config.constrained_otg_functions()
|| old_config.effective_otg_keyboard_leds() != new_config.effective_otg_keyboard_leds()
}
async fn reconcile_otg_config(
state: &Arc<AppState>,
hid: &HidConfig,
msd: &MsdConfig,
network: &OtgNetworkConfig,
uac: &UacConfig,
) -> Result<()> {
#[cfg(not(unix))]
{
let _ = (state, hid, msd, network, uac);
Ok(())
}
#[cfg(unix)]
{
state
.otg_service
.apply_config(hid, msd, network, uac)
.await
.map_err(|e| AppError::Config(format!("OTG reconcile failed: {}", e)))
}
}
pub async fn apply_video_config( pub async fn apply_video_config(
state: &Arc<AppState>, state: &Arc<AppState>,
@@ -189,305 +102,6 @@ pub async fn apply_stream_config(
Ok(()) Ok(())
} }
pub async fn apply_hid_config(
state: &Arc<AppState>,
old_config: &HidConfig,
new_config: &HidConfig,
msd_config: &MsdConfig,
network_config: &OtgNetworkConfig,
uac_config: &UacConfig,
options: ConfigApplyOptions,
) -> Result<()> {
new_config.validate_otg_functions()?;
let descriptor_changed = old_config.otg_descriptor != new_config.otg_descriptor;
let old_hid_functions = old_config.constrained_otg_functions();
let new_hid_functions = new_config.constrained_otg_functions();
let hid_functions_changed = old_hid_functions != new_hid_functions;
let keyboard_leds_changed =
old_config.effective_otg_keyboard_leds() != new_config.effective_otg_keyboard_leds();
let ch9329_runtime_changed = old_config.ch9329_hybrid_mouse != new_config.ch9329_hybrid_mouse;
if old_config.backend == new_config.backend
&& old_config.ch9329_port == new_config.ch9329_port
&& old_config.ch9329_baudrate == new_config.ch9329_baudrate
&& !ch9329_runtime_changed
&& old_config.otg_udc == new_config.otg_udc
&& !descriptor_changed
&& !hid_functions_changed
&& !keyboard_leds_changed
&& !options.force
{
tracing::info!("HID config unchanged, skipping reload");
return Ok(());
}
tracing::info!("Applying HID config changes...");
let new_hid_backend = hid_backend_type(new_config);
let transitioning_away_from_otg =
old_config.backend == HidBackend::Otg && new_config.backend != HidBackend::Otg;
let otg_config_changed = hid_otg_config_changed(old_config, new_config);
if transitioning_away_from_otg {
state
.hid
.reload(new_hid_backend.clone())
.await
.map_err(|e| AppError::Config(format!("HID reload failed: {}", e)))?;
}
if otg_config_changed {
reconcile_otg_config(state, new_config, msd_config, network_config, uac_config).await?;
}
if !transitioning_away_from_otg {
state
.hid
.reload(new_hid_backend)
.await
.map_err(|e| AppError::Config(format!("HID reload failed: {}", e)))?;
}
tracing::info!(
"HID backend reloaded successfully: {:?}",
new_config.backend
);
Ok(())
}
#[cfg(unix)]
pub async fn apply_msd_config(
state: &Arc<AppState>,
old_config: &MsdConfig,
new_config: &MsdConfig,
hid_config: &HidConfig,
network_config: &OtgNetworkConfig,
uac_config: &UacConfig,
options: ConfigApplyOptions,
) -> Result<()> {
let hid_backend_is_otg = hid_config.backend == HidBackend::Otg;
let effective_new_msd_enabled = new_config.enabled && hid_backend_is_otg;
tracing::info!("MSD config sent, checking if reload needed...");
tracing::debug!("Old MSD config: {:?}", old_config);
tracing::debug!("New MSD config: {:?}", new_config);
let old_msd_enabled = old_config.enabled;
let new_msd_enabled = effective_new_msd_enabled;
let msd_dir_changed = old_config.msd_dir != new_config.msd_dir;
let inquiry_strings_changed = old_config.flash_inquiry_string
!= new_config.flash_inquiry_string
|| old_config.cdrom_inquiry_string != new_config.cdrom_inquiry_string;
tracing::info!(
"MSD enabled: old={}, new={}",
old_msd_enabled,
new_msd_enabled
);
if msd_dir_changed {
tracing::info!("MSD directory changed: {}", new_config.msd_dir);
}
if inquiry_strings_changed {
tracing::info!("MSD inquiry strings changed");
}
let msd_dir = new_config.msd_dir_path();
if let Err(e) = std::fs::create_dir_all(msd_dir.join("images")) {
tracing::warn!("Failed to create MSD images directory: {}", e);
}
if let Err(e) = std::fs::create_dir_all(msd_dir.join("ventoy")) {
tracing::warn!("Failed to create MSD ventoy directory: {}", e);
}
let needs_reload = options.force
|| old_msd_enabled != new_msd_enabled
|| msd_dir_changed
|| inquiry_strings_changed;
if !needs_reload {
tracing::info!("MSD configuration unchanged, no reload needed");
return Ok(());
}
if new_msd_enabled {
tracing::info!("(Re)initializing MSD...");
reconcile_otg_config(state, hid_config, new_config, network_config, uac_config).await?;
let mut msd_guard = state.msd.write().await;
if let Some(msd) = msd_guard.as_mut() {
msd.shutdown()
.await
.map_err(|e| AppError::Config(format!("MSD shutdown failed: {e}")))?;
}
*msd_guard = None;
drop(msd_guard);
let msd =
crate::msd::MsdController::new(state.otg_service.clone(), new_config.msd_dir_path());
let ventoy_resource_dir = state.data_dir().join("ventoy");
msd.init(&ventoy_resource_dir)
.await
.map_err(|e| AppError::Config(format!("MSD initialization failed: {}", e)))?;
let events = state.events.clone();
msd.set_event_bus(events).await;
*state.msd.write().await = Some(msd);
tracing::info!("MSD initialized successfully");
} else {
tracing::info!("MSD disabled in config, shutting down...");
let mut msd_guard = state.msd.write().await;
if let Some(msd) = msd_guard.as_mut() {
msd.shutdown()
.await
.map_err(|e| AppError::Config(format!("MSD shutdown failed: {e}")))?;
}
*msd_guard = None;
tracing::info!("MSD shutdown complete");
reconcile_otg_config(state, hid_config, new_config, network_config, uac_config).await?;
}
if hid_config.backend == HidBackend::Otg
&& (options.force || old_msd_enabled != new_msd_enabled)
{
state
.hid
.reload(crate::hid::HidBackendType::Otg)
.await
.map_err(|e| AppError::Config(format!("OTG HID reload failed: {}", e)))?;
}
Ok(())
}
pub async fn apply_usb_config(
state: &Arc<AppState>,
old_config: &AppConfig,
new_config: &AppConfig,
) -> Result<()> {
#[cfg(unix)]
{
let transitioning_away_from_otg =
old_config.hid.backend == HidBackend::Otg && new_config.hid.backend != HidBackend::Otg;
let hid_unchanged = old_config.hid == new_config.hid;
let otg_gadget_rebuilt = old_config.msd != new_config.msd
|| old_config.otg_network != new_config.otg_network
|| old_config.uac != new_config.uac
|| old_config.hid.otg_udc != new_config.hid.otg_udc
|| old_config.hid.otg_descriptor != new_config.hid.otg_descriptor
|| old_config.hid.backend != new_config.hid.backend
|| old_config.hid.constrained_otg_functions()
!= new_config.hid.constrained_otg_functions()
|| old_config.hid.effective_otg_keyboard_leds()
!= new_config.hid.effective_otg_keyboard_leds();
let restart_uac_playback =
old_config.uac != new_config.uac || (new_config.uac.enabled && otg_gadget_rebuilt);
// A bound ALSA handle refers to the old configfs function. Stop it
// before any gadget teardown so the worker cannot write through a
// disappearing PCM node. It is restarted only after every reconcile.
if restart_uac_playback {
let playback = state.uac_playback.write().await.take();
if let Some(playback) = playback {
playback.stop();
tracing::info!("UAC playback writer stopped before OTG reconcile");
}
}
if transitioning_away_from_otg {
apply_hid_config(
state,
&old_config.hid,
&new_config.hid,
&new_config.msd,
&new_config.otg_network,
&new_config.uac,
ConfigApplyOptions::default(),
)
.await?;
} else {
reconcile_otg_config(
state,
&new_config.hid,
&new_config.msd,
&new_config.otg_network,
&new_config.uac,
)
.await?;
apply_hid_config(
state,
&old_config.hid,
&new_config.hid,
&new_config.msd,
&new_config.otg_network,
&new_config.uac,
ConfigApplyOptions::default(),
)
.await?;
}
// When the OTG gadget was rebuilt due to MSD or network config changes
// while HID config stayed the same, the /dev/hidg* devices are new and
// the HID backend must be reloaded to reopen them.
if hid_unchanged && otg_gadget_rebuilt && new_config.hid.backend == HidBackend::Otg {
tracing::info!("OTG gadget rebuilt, reloading HID backend for new devices");
let hid_backend = hid_backend_type(&new_config.hid);
state.hid.reload(hid_backend).await.map_err(|e| {
AppError::Config(format!("HID reload after gadget rebuild failed: {}", e))
})?;
}
apply_msd_config(
state,
&old_config.msd,
&new_config.msd,
&new_config.hid,
&new_config.otg_network,
&new_config.uac,
ConfigApplyOptions::default(),
)
.await?;
// apply_msd_config may perform a second gadget reconcile. Resolve the
// new ALSA card only after that final rebuild, then publish the worker.
if restart_uac_playback && new_config.uac.enabled {
let config = crate::audio::uac::UacPlaybackConfig {
sample_rate: new_config.uac.sample_rate,
channels: new_config.uac.channels as u16,
..Default::default()
};
let writer = crate::audio::uac::UacPlayback::start(config).map_err(|error| {
AppError::Config(format!("Failed to start UAC playback: {error}"))
})?;
*state.uac_playback.write().await = Some(writer);
tracing::info!("UAC playback writer started after OTG reconcile");
} else if restart_uac_playback {
tracing::info!("UAC playback remains disabled");
}
Ok(())
}
#[cfg(not(unix))]
{
apply_hid_config(
state,
&old_config.hid,
&new_config.hid,
&new_config.msd,
&new_config.otg_network,
&new_config.uac,
ConfigApplyOptions::default(),
)
.await
}
}
pub async fn apply_atx_config( pub async fn apply_atx_config(
state: &Arc<AppState>, state: &Arc<AppState>,
_old_config: &AtxConfig, _old_config: &AtxConfig,
@@ -554,336 +168,3 @@ pub async fn apply_audio_config(
Ok(()) Ok(())
} }
pub async fn enforce_stream_codec_constraints(state: &Arc<AppState>) -> Result<Option<String>> {
let config = state.runtime_third_party_config().await;
let constraints = StreamCodecConstraints::from_config(&config);
state
.stream_manager
.set_runtime_codec_constraints(constraints.clone())
.await;
let enforcement =
enforce_constraints_with_stream_manager(&state.stream_manager, &constraints).await?;
Ok(enforcement.message)
}
async fn validate_runtime_candidate<T>(
state: &Arc<AppState>,
apply: impl FnOnce(&mut crate::config::AppConfig, T),
config: T,
) -> Result<()> {
let mut candidate = state.runtime_third_party_config().await;
apply(&mut candidate, config);
validate_third_party_codec_compatibility(&candidate)
}
fn validate_rustdesk_candidate(
state: &Arc<AppState>,
new_config: &crate::rustdesk::config::RustDeskConfig,
) -> Result<()> {
let mut candidate = state.config.get().as_ref().clone();
candidate.rustdesk = new_config.clone();
validate_third_party_codec_compatibility(&candidate)
}
fn validate_vnc_candidate(state: &Arc<AppState>, new_config: &VncConfig) -> Result<()> {
let mut candidate = state.config.get().as_ref().clone();
candidate.vnc = new_config.clone();
validate_third_party_codec_compatibility(&candidate)
}
fn validate_rtsp_candidate(state: &Arc<AppState>, new_config: &RtspConfig) -> Result<()> {
let mut candidate = state.config.get().as_ref().clone();
candidate.rtsp = new_config.clone();
validate_third_party_codec_compatibility(&candidate)
}
pub async fn apply_rustdesk_config(
state: &Arc<AppState>,
old_config: &crate::rustdesk::config::RustDeskConfig,
new_config: &crate::rustdesk::config::RustDeskConfig,
options: ConfigApplyOptions,
) -> Result<()> {
tracing::info!("Applying RustDesk config changes...");
if options.runtime_only {
validate_runtime_candidate(
state,
|candidate, config| candidate.rustdesk = config,
new_config.clone(),
)
.await?;
} else {
validate_rustdesk_candidate(state, new_config)?;
}
let mut rustdesk_guard = state.rustdesk.write().await;
let mut credentials_to_save = None;
let need_restart = options.force
|| old_config.codec != new_config.codec
|| old_config.rendezvous_server != new_config.rendezvous_server
|| old_config.device_id != new_config.device_id
|| old_config.device_password != new_config.device_password;
if !options.preserve_service_state && !new_config.enabled {
if let Some(ref service) = *rustdesk_guard {
service
.stop()
.await
.map_err(|e| AppError::Config(format!("Failed to stop RustDesk service: {}", e)))?;
tracing::info!("RustDesk service stopped");
}
*rustdesk_guard = None;
}
if !options.preserve_service_state && new_config.enabled {
if rustdesk_guard.is_none() {
tracing::info!("Initializing RustDesk service...");
let service = std::sync::Arc::new(crate::rustdesk::RustDeskService::new(
new_config.clone(),
state.stream_manager.clone(),
state.hid.clone(),
state.audio.clone(),
));
*rustdesk_guard = Some(service.clone());
service.start().await.map_err(|e| {
AppError::Config(format!("Failed to start RustDesk service: {}", e))
})?;
tracing::info!("RustDesk service started with ID: {}", new_config.device_id);
credentials_to_save = service.save_credentials();
} else {
if let Some(ref service) = *rustdesk_guard {
if service.is_listening() {
if need_restart {
service.restart(new_config.clone()).await.map_err(|e| {
AppError::Config(format!("Failed to restart RustDesk service: {}", e))
})?;
tracing::info!(
"RustDesk service restarted with ID: {}",
new_config.device_id
);
}
} else {
service.update_config(new_config.clone());
service.start().await.map_err(|e| {
AppError::Config(format!("Failed to start RustDesk service: {}", e))
})?;
}
credentials_to_save = service.save_credentials();
}
}
} else if options.preserve_service_state && need_restart {
if let Some(ref service) = *rustdesk_guard {
let mut runtime_config = new_config.clone();
runtime_config.enabled = true;
service.restart(runtime_config).await.map_err(|e| {
AppError::Config(format!("Failed to restart RustDesk service: {}", e))
})?;
credentials_to_save = service.save_credentials();
}
}
drop(rustdesk_guard);
if let Some(updated_config) = credentials_to_save {
tracing::info!("Saving RustDesk credentials to config store...");
state
.config
.update(|cfg| {
cfg.rustdesk.public_key = updated_config.public_key.clone();
cfg.rustdesk.private_key = updated_config.private_key.clone();
cfg.rustdesk.signing_public_key = updated_config.signing_public_key.clone();
cfg.rustdesk.signing_private_key = updated_config.signing_private_key.clone();
cfg.rustdesk.uuid = updated_config.uuid.clone();
})
.await?;
tracing::info!("RustDesk credentials saved successfully");
}
if let Some(message) = enforce_stream_codec_constraints(state).await? {
tracing::info!("{}", message);
}
Ok(())
}
pub async fn apply_vnc_config(
state: &Arc<AppState>,
old_config: &VncConfig,
new_config: &VncConfig,
options: ConfigApplyOptions,
) -> Result<()> {
tracing::info!("Applying VNC config changes...");
if options.runtime_only {
validate_runtime_candidate(
state,
|candidate, config| candidate.vnc = config,
new_config.clone(),
)
.await?;
} else {
validate_vnc_candidate(state, new_config)?;
}
let runtime_config = state.runtime_third_party_config().await;
let will_run = if options.preserve_service_state {
runtime_config.vnc.enabled
} else {
new_config.enabled
};
if will_run {
let mut candidate = runtime_config;
candidate.vnc = new_config.clone();
candidate.vnc.enabled = true;
let constraints = StreamCodecConstraints::from_config(&candidate);
match enforce_constraints_with_stream_manager(&state.stream_manager, &constraints).await {
Ok(result) if result.changed => {
if let Some(message) = result.message {
tracing::info!("{}", message);
}
}
Ok(_) => {}
Err(e) => tracing::warn!(
"Failed to enforce VNC stream constraints before start: {}",
e
),
}
}
let mut vnc_guard = state.vnc.write().await;
let need_restart = options.force
|| old_config.bind != new_config.bind
|| old_config.port != new_config.port
|| old_config.encoding != new_config.encoding
|| old_config.password != new_config.password
|| old_config.allow_one_client != new_config.allow_one_client;
if !options.preserve_service_state && !new_config.enabled {
if let Some(ref service) = *vnc_guard {
service.stop().await?;
}
*vnc_guard = None;
}
if !options.preserve_service_state && new_config.enabled {
if vnc_guard.is_none() {
let service = Arc::new(crate::vnc::VncService::new(
new_config.clone(),
state.stream_manager.clone(),
state.hid.clone(),
));
*vnc_guard = Some(service.clone());
service.start().await?;
tracing::info!("VNC service started");
} else {
if let Some(ref service) = *vnc_guard {
if matches!(
service.status().await,
crate::vnc::VncServiceStatus::Running
) {
if need_restart {
service.restart(new_config.clone()).await?;
tracing::info!("VNC service restarted");
}
} else {
service.update_config(new_config.clone()).await;
service.start().await?;
}
}
}
} else if options.preserve_service_state && need_restart {
if let Some(ref service) = *vnc_guard {
let mut runtime_config = new_config.clone();
runtime_config.enabled = true;
service.restart(runtime_config).await?;
}
}
drop(vnc_guard);
if let Some(message) = enforce_stream_codec_constraints(state).await? {
tracing::info!("{}", message);
}
Ok(())
}
pub async fn apply_rtsp_config(
state: &Arc<AppState>,
old_config: &RtspConfig,
new_config: &RtspConfig,
options: ConfigApplyOptions,
) -> Result<()> {
tracing::info!("Applying RTSP config changes...");
if options.runtime_only {
validate_runtime_candidate(
state,
|candidate, config| candidate.rtsp = config,
new_config.clone(),
)
.await?;
} else {
validate_rtsp_candidate(state, new_config)?;
}
let mut rtsp_guard = state.rtsp.write().await;
let need_restart = options.force
|| old_config.bind != new_config.bind
|| old_config.port != new_config.port
|| old_config.path != new_config.path
|| old_config.codec != new_config.codec
|| old_config.username != new_config.username
|| old_config.password != new_config.password
|| old_config.allow_one_client != new_config.allow_one_client;
if !options.preserve_service_state && !new_config.enabled {
if let Some(ref service) = *rtsp_guard {
service
.stop()
.await
.map_err(|e| AppError::Config(format!("Failed to stop RTSP service: {}", e)))?;
}
*rtsp_guard = None;
}
if !options.preserve_service_state && new_config.enabled {
if rtsp_guard.is_none() {
let service = Arc::new(RtspService::new(
new_config.clone(),
state.stream_manager.clone(),
));
*rtsp_guard = Some(service.clone());
service.start().await?;
tracing::info!("RTSP service started");
} else {
if let Some(ref service) = *rtsp_guard {
if matches!(
service.status().await,
crate::rtsp::RtspServiceStatus::Running
) {
if need_restart {
service.restart(new_config.clone()).await?;
tracing::info!("RTSP service restarted");
}
} else {
service.update_config(new_config.clone()).await;
service.start().await?;
}
}
}
} else if options.preserve_service_state && need_restart {
if let Some(ref service) = *rtsp_guard {
let mut runtime_config = new_config.clone();
runtime_config.enabled = true;
service.restart(runtime_config).await?;
}
}
drop(rtsp_guard);
if let Some(message) = enforce_stream_codec_constraints(state).await? {
tracing::info!("{}", message);
}
Ok(())
}

7
src/web/handlers/config/hid.rs
View File

@@ -1,19 +1,18 @@
use axum::{extract::State, Json}; use axum::{extract::State, Json};
use std::sync::Arc;
use crate::config::HidConfig; use crate::config::HidConfig;
use crate::error::Result; use crate::error::Result;
use crate::state::AppState; use crate::web::state::UsbApiState;
use super::types::HidConfigUpdate; use super::types::HidConfigUpdate;
use super::usb_update::{stage_hid_config_update, update_usb_config}; use super::usb_update::{stage_hid_config_update, update_usb_config};
pub async fn get_hid_config(State(state): State<Arc<AppState>>) -> Json<HidConfig> { pub async fn get_hid_config(State(state): State<UsbApiState>) -> Json<HidConfig> {
Json(state.config.get().hid.clone()) Json(state.config.get().hid.clone())
} }
pub async fn update_hid_config( pub async fn update_hid_config(
State(state): State<Arc<AppState>>, State(state): State<UsbApiState>,
Json(req): Json<HidConfigUpdate>, Json(req): Json<HidConfigUpdate>,
) -> Result<Json<HidConfig>> { ) -> Result<Json<HidConfig>> {
let config = update_usb_config(&state, move |staged| { let config = update_usb_config(&state, move |staged| {

7
src/web/handlers/config/msd.rs
View File

@@ -1,19 +1,18 @@
use axum::{extract::State, Json}; use axum::{extract::State, Json};
use std::sync::Arc;
use crate::config::MsdConfig; use crate::config::MsdConfig;
use crate::error::Result; use crate::error::Result;
use crate::state::AppState; use crate::web::state::UsbApiState;
use super::otg::update_otg_config_inner; use super::otg::update_otg_config_inner;
use super::types::{MsdConfigUpdate, OtgConfigUpdate}; use super::types::{MsdConfigUpdate, OtgConfigUpdate};
pub async fn get_msd_config(State(state): State<Arc<AppState>>) -> Json<MsdConfig> { pub async fn get_msd_config(State(state): State<UsbApiState>) -> Json<MsdConfig> {
Json(state.config.get().msd.clone()) Json(state.config.get().msd.clone())
} }
pub async fn update_msd_config( pub async fn update_msd_config(
State(state): State<Arc<AppState>>, State(state): State<UsbApiState>,
Json(req): Json<MsdConfigUpdate>, Json(req): Json<MsdConfigUpdate>,
) -> Result<Json<MsdConfig>> { ) -> Result<Json<MsdConfig>> {
let response = update_otg_config_inner( let response = update_otg_config_inner(

10
src/web/handlers/config/otg.rs
View File

@@ -1,5 +1,3 @@
use std::sync::Arc;
use axum::{extract::State, Json}; use axum::{extract::State, Json};
use serde::Serialize; use serde::Serialize;
use typeshare::typeshare; use typeshare::typeshare;
@@ -7,7 +5,7 @@ use typeshare::typeshare;
use crate::config::{HidConfig, MsdConfig, OtgNetworkConfig}; use crate::config::{HidConfig, MsdConfig, OtgNetworkConfig};
use crate::error::Result; use crate::error::Result;
use crate::otg::OtgNetworkStatus; use crate::otg::OtgNetworkStatus;
use crate::state::AppState; use crate::web::state::UsbApiState;
use super::types::OtgConfigUpdate; use super::types::OtgConfigUpdate;
use super::usb_update::{stage_hid_config_update, update_usb_config}; use super::usb_update::{stage_hid_config_update, update_usb_config};
@@ -22,14 +20,14 @@ pub struct OtgConfigResponse {
} }
pub async fn update_otg_config( pub async fn update_otg_config(
State(state): State<Arc<AppState>>, State(state): State<UsbApiState>,
Json(request): Json<OtgConfigUpdate>, Json(request): Json<OtgConfigUpdate>,
) -> Result<Json<OtgConfigResponse>> { ) -> Result<Json<OtgConfigResponse>> {
update_otg_config_inner(&state, request).await.map(Json) update_otg_config_inner(&state, request).await.map(Json)
} }
pub(super) async fn update_otg_config_inner( pub(super) async fn update_otg_config_inner(
state: &Arc<AppState>, state: &UsbApiState,
request: OtgConfigUpdate, request: OtgConfigUpdate,
) -> Result<OtgConfigResponse> { ) -> Result<OtgConfigResponse> {
let staged_config = update_usb_config(state, move |staged| { let staged_config = update_usb_config(state, move |staged| {
@@ -54,6 +52,6 @@ pub(super) async fn update_otg_config_inner(
hid: staged_config.hid, hid: staged_config.hid,
msd: staged_config.msd, msd: staged_config.msd,
network: staged_config.otg_network, network: staged_config.otg_network,
status: state.otg_service.network_status().await, status: state.otg.network_status().await,
}) })
} }

12
src/web/handlers/config/otg_network.rs
View File

@@ -1,21 +1,19 @@
use std::sync::Arc;
use axum::{extract::State, Json}; use axum::{extract::State, Json};
use crate::config::OtgNetworkConfig; use crate::config::OtgNetworkConfig;
use crate::error::Result; use crate::error::Result;
use crate::otg::OtgNetworkStatus; use crate::otg::OtgNetworkStatus;
use crate::state::AppState; use crate::web::state::UsbApiState;
use super::otg::update_otg_config_inner; use super::otg::update_otg_config_inner;
use super::types::{OtgConfigUpdate, OtgNetworkConfigUpdate}; use super::types::{OtgConfigUpdate, OtgNetworkConfigUpdate};
pub async fn get_otg_network_config(State(state): State<Arc<AppState>>) -> Json<OtgNetworkConfig> { pub async fn get_otg_network_config(State(state): State<UsbApiState>) -> Json<OtgNetworkConfig> {
Json(state.config.get().otg_network.clone()) Json(state.config.get().otg_network.clone())
} }
pub async fn update_otg_network_config( pub async fn update_otg_network_config(
State(state): State<Arc<AppState>>, State(state): State<UsbApiState>,
Json(request): Json<OtgNetworkConfigUpdate>, Json(request): Json<OtgNetworkConfigUpdate>,
) -> Result<Json<OtgNetworkConfig>> { ) -> Result<Json<OtgNetworkConfig>> {
let response = update_otg_config_inner( let response = update_otg_config_inner(
@@ -29,6 +27,6 @@ pub async fn update_otg_network_config(
Ok(Json(response.network)) Ok(Json(response.network))
} }
pub async fn get_otg_network_status(State(state): State<Arc<AppState>>) -> Json<OtgNetworkStatus> { pub async fn get_otg_network_status(State(state): State<UsbApiState>) -> Json<OtgNetworkStatus> {
Json(state.otg_service.network_status().await) Json(state.otg.network_status().await)
} }

60
src/web/handlers/config/rtsp.rs
View File

@@ -1,20 +1,22 @@
use axum::{extract::State, Json}; use axum::{extract::State, Json};
use std::sync::Arc;
use crate::error::Result; use crate::error::Result;
use crate::state::AppState; use crate::web::state::RemoteAccessApiState;
use super::apply::{apply_rtsp_config, try_apply_lock, ConfigApplyOptions};
use super::types::{RtspConfigResponse, RtspConfigUpdate, RtspStatusResponse}; use super::types::{RtspConfigResponse, RtspConfigUpdate, RtspStatusResponse};
use crate::runtime::{try_apply_lock, ConfigApplyOptions};
fn validate_candidate(state: &Arc<AppState>, config: &crate::config::RtspConfig) -> Result<()> { fn validate_candidate(
state: &RemoteAccessApiState,
config: &crate::config::RtspConfig,
) -> Result<()> {
let mut candidate = state.config.get().as_ref().clone(); let mut candidate = state.config.get().as_ref().clone();
candidate.rtsp = config.clone(); candidate.rtsp = config.clone();
crate::video::codec_constraints::validate_third_party_codec_compatibility(&candidate) crate::video::codec_constraints::validate_third_party_codec_compatibility(&candidate)
} }
async fn persist_and_apply( async fn persist_and_apply(
state: &Arc<AppState>, state: &RemoteAccessApiState,
old_config: crate::config::RtspConfig, old_config: crate::config::RtspConfig,
new_config: crate::config::RtspConfig, new_config: crate::config::RtspConfig,
) -> Result<crate::config::RtspConfig> { ) -> Result<crate::config::RtspConfig> {
@@ -26,8 +28,9 @@ async fn persist_and_apply(
}) })
.await?; .await?;
let stored_config = state.config.get().rtsp.clone(); let stored_config = state.config.get().rtsp.clone();
apply_rtsp_config( state
state, .coordinator
.apply_rtsp(
&old_config, &old_config,
&stored_config, &stored_config,
ConfigApplyOptions::preserving_service_state(), ConfigApplyOptions::preserving_service_state(),
@@ -36,21 +39,20 @@ async fn persist_and_apply(
Ok(stored_config) Ok(stored_config)
} }
async fn current_status(state: &Arc<AppState>) -> crate::rtsp::RtspServiceStatus { async fn current_status(state: &RemoteAccessApiState) -> crate::rtsp::RtspServiceStatus {
let guard = state.rtsp.read().await; state.coordinator.rtsp_status().await
if let Some(ref service) = *guard {
service.status().await
} else {
crate::rtsp::RtspServiceStatus::Stopped
}
} }
pub async fn get_rtsp_config(State(state): State<Arc<AppState>>) -> Json<RtspConfigResponse> { pub async fn get_rtsp_config(
State(state): State<RemoteAccessApiState>,
) -> Json<RtspConfigResponse> {
let config = state.config.get(); let config = state.config.get();
Json(RtspConfigResponse::from(&config.rtsp)) Json(RtspConfigResponse::from(&config.rtsp))
} }
pub async fn get_rtsp_status(State(state): State<Arc<AppState>>) -> Json<RtspStatusResponse> { pub async fn get_rtsp_status(
State(state): State<RemoteAccessApiState>,
) -> Json<RtspStatusResponse> {
let config = state.config.get().rtsp.clone(); let config = state.config.get().rtsp.clone();
let status = current_status(&state).await; let status = current_status(&state).await;
@@ -58,12 +60,12 @@ pub async fn get_rtsp_status(State(state): State<Arc<AppState>>) -> Json<RtspSta
} }
pub async fn update_rtsp_config( pub async fn update_rtsp_config(
State(state): State<Arc<AppState>>, State(state): State<RemoteAccessApiState>,
Json(req): Json<RtspConfigUpdate>, Json(req): Json<RtspConfigUpdate>,
) -> Result<Json<RtspConfigResponse>> { ) -> Result<Json<RtspConfigResponse>> {
req.validate()?; req.validate()?;
let _apply_guard = try_apply_lock(&state.config_apply_locks.rtsp, "rtsp")?; let _apply_guard = try_apply_lock(&state.rtsp_apply_lock, "rtsp")?;
let old_config = state.config.get().rtsp.clone(); let old_config = state.config.get().rtsp.clone();
let mut merged_config = old_config.clone(); let mut merged_config = old_config.clone();
req.apply_to(&mut merged_config); req.apply_to(&mut merged_config);
@@ -73,15 +75,16 @@ pub async fn update_rtsp_config(
} }
pub async fn start_rtsp_service( pub async fn start_rtsp_service(
State(state): State<Arc<AppState>>, State(state): State<RemoteAccessApiState>,
) -> Result<Json<RtspStatusResponse>> { ) -> Result<Json<RtspStatusResponse>> {
let _apply_guard = try_apply_lock(&state.config_apply_locks.rtsp, "rtsp")?; let _apply_guard = try_apply_lock(&state.rtsp_apply_lock, "rtsp")?;
let stored_config = state.config.get().rtsp.clone(); let stored_config = state.config.get().rtsp.clone();
let runtime_config = state.runtime_third_party_config().await.rtsp; let runtime_config = state.coordinator.runtime_config().await.rtsp;
let mut start_config = stored_config.clone(); let mut start_config = stored_config.clone();
start_config.enabled = true; start_config.enabled = true;
apply_rtsp_config( state
&state, .coordinator
.apply_rtsp(
&runtime_config, &runtime_config,
&start_config, &start_config,
ConfigApplyOptions::runtime_only(), ConfigApplyOptions::runtime_only(),
@@ -93,15 +96,16 @@ pub async fn start_rtsp_service(
} }
pub async fn stop_rtsp_service( pub async fn stop_rtsp_service(
State(state): State<Arc<AppState>>, State(state): State<RemoteAccessApiState>,
) -> Result<Json<RtspStatusResponse>> { ) -> Result<Json<RtspStatusResponse>> {
let _apply_guard = try_apply_lock(&state.config_apply_locks.rtsp, "rtsp")?; let _apply_guard = try_apply_lock(&state.rtsp_apply_lock, "rtsp")?;
let stored_config = state.config.get().rtsp.clone(); let stored_config = state.config.get().rtsp.clone();
let runtime_config = state.runtime_third_party_config().await.rtsp; let runtime_config = state.coordinator.runtime_config().await.rtsp;
let mut stop_config = stored_config.clone(); let mut stop_config = stored_config.clone();
stop_config.enabled = false; stop_config.enabled = false;
apply_rtsp_config( state
&state, .coordinator
.apply_rtsp(
&runtime_config, &runtime_config,
&stop_config, &stop_config,
ConfigApplyOptions::runtime_only(), ConfigApplyOptions::runtime_only(),

80
src/web/handlers/config/rustdesk.rs
View File

@@ -1,21 +1,20 @@
use axum::{extract::State, Json}; use axum::{extract::State, Json};
use std::sync::Arc;
use crate::error::Result; use crate::error::Result;
use crate::rustdesk::config::RustDeskConfig; use crate::rustdesk::config::RustDeskConfig;
use crate::state::AppState; use crate::web::state::RemoteAccessApiState;
use super::apply::{apply_rustdesk_config, try_apply_lock, ConfigApplyOptions};
use super::types::RustDeskConfigUpdate; use super::types::RustDeskConfigUpdate;
use crate::runtime::{try_apply_lock, ConfigApplyOptions};
fn validate_candidate(state: &Arc<AppState>, config: &RustDeskConfig) -> Result<()> { fn validate_candidate(state: &RemoteAccessApiState, config: &RustDeskConfig) -> Result<()> {
let mut candidate = state.config.get().as_ref().clone(); let mut candidate = state.config.get().as_ref().clone();
candidate.rustdesk = config.clone(); candidate.rustdesk = config.clone();
crate::video::codec_constraints::validate_third_party_codec_compatibility(&candidate) crate::video::codec_constraints::validate_third_party_codec_compatibility(&candidate)
} }
async fn persist_and_apply( async fn persist_and_apply(
state: &Arc<AppState>, state: &RemoteAccessApiState,
old_config: RustDeskConfig, old_config: RustDeskConfig,
new_config: RustDeskConfig, new_config: RustDeskConfig,
) -> Result<RustDeskConfig> { ) -> Result<RustDeskConfig> {
@@ -27,8 +26,9 @@ async fn persist_and_apply(
}) })
.await?; .await?;
let stored_config = state.config.get().rustdesk.clone(); let stored_config = state.config.get().rustdesk.clone();
apply_rustdesk_config( state
state, .coordinator
.apply_rustdesk(
&old_config, &old_config,
&stored_config, &stored_config,
ConfigApplyOptions::preserving_service_state(), ConfigApplyOptions::preserving_service_state(),
@@ -37,22 +37,16 @@ async fn persist_and_apply(
Ok(stored_config) Ok(stored_config)
} }
async fn current_status(state: &Arc<AppState>, config: RustDeskConfig) -> RustDeskStatusResponse { async fn current_status(
let (service_status, rendezvous_status) = { state: &RemoteAccessApiState,
let guard = state.rustdesk.read().await; config: RustDeskConfig,
if let Some(ref service) = *guard { ) -> RustDeskStatusResponse {
let status = format!("{}", service.status()); let runtime = state.coordinator.rustdesk_status().await;
let rv_status = service.rendezvous_status().map(|s| format!("{}", s));
(status, rv_status)
} else {
("not_initialized".to_string(), None)
}
};
RustDeskStatusResponse { RustDeskStatusResponse {
config: RustDeskConfigResponse::from(&config), config: RustDeskConfigResponse::from(&config),
service_status, service_status: runtime.service_status,
rendezvous_status, rendezvous_status: runtime.rendezvous_status,
} }
} }
@@ -91,25 +85,25 @@ pub struct RustDeskStatusResponse {
} }
pub async fn get_rustdesk_config( pub async fn get_rustdesk_config(
State(state): State<Arc<AppState>>, State(state): State<RemoteAccessApiState>,
) -> Json<RustDeskConfigResponse> { ) -> Json<RustDeskConfigResponse> {
Json(RustDeskConfigResponse::from(&state.config.get().rustdesk)) Json(RustDeskConfigResponse::from(&state.config.get().rustdesk))
} }
pub async fn get_rustdesk_status( pub async fn get_rustdesk_status(
State(state): State<Arc<AppState>>, State(state): State<RemoteAccessApiState>,
) -> Json<RustDeskStatusResponse> { ) -> Json<RustDeskStatusResponse> {
let config = state.config.get().rustdesk.clone(); let config = state.config.get().rustdesk.clone();
Json(current_status(&state, config).await) Json(current_status(&state, config).await)
} }
pub async fn update_rustdesk_config( pub async fn update_rustdesk_config(
State(state): State<Arc<AppState>>, State(state): State<RemoteAccessApiState>,
Json(req): Json<RustDeskConfigUpdate>, Json(req): Json<RustDeskConfigUpdate>,
) -> Result<Json<RustDeskConfigResponse>> { ) -> Result<Json<RustDeskConfigResponse>> {
req.validate()?; req.validate()?;
let _apply_guard = try_apply_lock(&state.config_apply_locks.rustdesk, "rustdesk")?; let _apply_guard = try_apply_lock(&state.rustdesk_apply_lock, "rustdesk")?;
let old_config = state.config.get().rustdesk.clone(); let old_config = state.config.get().rustdesk.clone();
let mut merged_config = old_config.clone(); let mut merged_config = old_config.clone();
req.apply_to(&mut merged_config); req.apply_to(&mut merged_config);
@@ -117,19 +111,11 @@ pub async fn update_rustdesk_config(
let new_config = persist_and_apply(&state, old_config, merged_config).await?; let new_config = persist_and_apply(&state, old_config, merged_config).await?;
let constraints = state.stream_manager.codec_constraints().await;
if constraints.rustdesk_enabled || constraints.rtsp_enabled {
tracing::info!(
"Stream codec constraints active after RustDesk update: {}",
constraints.reason
);
}
Ok(Json(RustDeskConfigResponse::from(&new_config))) Ok(Json(RustDeskConfigResponse::from(&new_config)))
} }
pub async fn regenerate_device_id( pub async fn regenerate_device_id(
State(state): State<Arc<AppState>>, State(state): State<RemoteAccessApiState>,
) -> Result<Json<RustDeskConfigResponse>> { ) -> Result<Json<RustDeskConfigResponse>> {
state state
.config .config
@@ -143,7 +129,7 @@ pub async fn regenerate_device_id(
} }
pub async fn regenerate_device_password( pub async fn regenerate_device_password(
State(state): State<Arc<AppState>>, State(state): State<RemoteAccessApiState>,
) -> Result<Json<RustDeskConfigResponse>> { ) -> Result<Json<RustDeskConfigResponse>> {
state state
.config .config
@@ -156,7 +142,9 @@ pub async fn regenerate_device_password(
Ok(Json(RustDeskConfigResponse::from(&new_config))) Ok(Json(RustDeskConfigResponse::from(&new_config)))
} }
pub async fn get_device_password(State(state): State<Arc<AppState>>) -> Json<serde_json::Value> { pub async fn get_device_password(
State(state): State<RemoteAccessApiState>,
) -> Json<serde_json::Value> {
let config = state.config.get().rustdesk.clone(); let config = state.config.get().rustdesk.clone();
Json(serde_json::json!({ Json(serde_json::json!({
"device_id": config.device_id, "device_id": config.device_id,
@@ -165,15 +153,16 @@ pub async fn get_device_password(State(state): State<Arc<AppState>>) -> Json<ser
} }
pub async fn start_rustdesk_service( pub async fn start_rustdesk_service(
State(state): State<Arc<AppState>>, State(state): State<RemoteAccessApiState>,
) -> Result<Json<RustDeskStatusResponse>> { ) -> Result<Json<RustDeskStatusResponse>> {
let _apply_guard = try_apply_lock(&state.config_apply_locks.rustdesk, "rustdesk")?; let _apply_guard = try_apply_lock(&state.rustdesk_apply_lock, "rustdesk")?;
let stored_config = state.config.get().rustdesk.clone(); let stored_config = state.config.get().rustdesk.clone();
let runtime_config = state.runtime_third_party_config().await.rustdesk; let runtime_config = state.coordinator.runtime_config().await.rustdesk;
let mut start_config = stored_config.clone(); let mut start_config = stored_config.clone();
start_config.enabled = true; start_config.enabled = true;
apply_rustdesk_config( state
&state, .coordinator
.apply_rustdesk(
&runtime_config, &runtime_config,
&start_config, &start_config,
ConfigApplyOptions::runtime_only(), ConfigApplyOptions::runtime_only(),
@@ -184,15 +173,16 @@ pub async fn start_rustdesk_service(
} }
pub async fn stop_rustdesk_service( pub async fn stop_rustdesk_service(
State(state): State<Arc<AppState>>, State(state): State<RemoteAccessApiState>,
) -> Result<Json<RustDeskStatusResponse>> { ) -> Result<Json<RustDeskStatusResponse>> {
let _apply_guard = try_apply_lock(&state.config_apply_locks.rustdesk, "rustdesk")?; let _apply_guard = try_apply_lock(&state.rustdesk_apply_lock, "rustdesk")?;
let stored_config = state.config.get().rustdesk.clone(); let stored_config = state.config.get().rustdesk.clone();
let runtime_config = state.runtime_third_party_config().await.rustdesk; let runtime_config = state.coordinator.runtime_config().await.rustdesk;
let mut stop_config = stored_config.clone(); let mut stop_config = stored_config.clone();
stop_config.enabled = false; stop_config.enabled = false;
apply_rustdesk_config( state
&state, .coordinator
.apply_rustdesk(
&runtime_config, &runtime_config,
&stop_config, &stop_config,
ConfigApplyOptions::runtime_only(), ConfigApplyOptions::runtime_only(),

2
src/web/handlers/config/stream.rs
View File

@@ -38,7 +38,7 @@ pub async fn update_stream_config(
) )
.await?; .await?;
super::apply::enforce_stream_codec_constraints(&state).await?; state.remote_access.enforce_codec_constraints().await?;
Ok(Json(StreamConfigResponse::from(&new_stream_config))) Ok(Json(StreamConfigResponse::from(&new_stream_config)))
} }

8
src/web/handlers/config/uac.rs
View File

@@ -1,19 +1,17 @@
use std::sync::Arc;
use axum::{extract::State, Json}; use axum::{extract::State, Json};
use crate::config::UacConfig; use crate::config::UacConfig;
use crate::error::Result; use crate::error::Result;
use crate::state::AppState; use crate::web::state::UsbApiState;
use super::usb_update::update_usb_config; use super::usb_update::update_usb_config;
pub async fn get_uac_config(State(state): State<Arc<AppState>>) -> Json<UacConfig> { pub async fn get_uac_config(State(state): State<UsbApiState>) -> Json<UacConfig> {
Json(state.config.get().uac.clone()) Json(state.config.get().uac.clone())
} }
pub async fn update_uac_config( pub async fn update_uac_config(
State(state): State<Arc<AppState>>, State(state): State<UsbApiState>,
Json(request): Json<UacConfig>, Json(request): Json<UacConfig>,
) -> Result<Json<UacConfig>> { ) -> Result<Json<UacConfig>> {
request.validate()?; request.validate()?;

29
src/web/handlers/config/usb_update.rs
View File

@@ -1,11 +1,9 @@
use std::sync::Arc;
use crate::config::{AppConfig, Ch9329DescriptorConfig, HidBackend, HidConfig}; use crate::config::{AppConfig, Ch9329DescriptorConfig, HidBackend, HidConfig};
use crate::error::{AppError, Result}; use crate::error::{AppError, Result};
use crate::state::AppState; use crate::web::state::UsbApiState;
use super::apply::{apply_usb_config, try_apply_lock};
use super::types::HidConfigUpdate; use super::types::HidConfigUpdate;
use crate::runtime::try_apply_lock;
pub(super) fn stage_hid_config_update( pub(super) fn stage_hid_config_update(
staged_hid: &mut HidConfig, staged_hid: &mut HidConfig,
@@ -27,14 +25,11 @@ pub(super) fn stage_hid_config_update(
Ok(requested_descriptor) Ok(requested_descriptor)
} }
pub(super) async fn update_usb_config<F>( pub(super) async fn update_usb_config<F>(state: &UsbApiState, stage_update: F) -> Result<AppConfig>
state: &Arc<AppState>,
stage_update: F,
) -> Result<AppConfig>
where where
F: FnOnce(&mut AppConfig) -> Result<Option<Ch9329DescriptorConfig>>, F: FnOnce(&mut AppConfig) -> Result<Option<Ch9329DescriptorConfig>>,
{ {
let _guard = try_apply_lock(&state.config_apply_locks.otg, "otg")?; let _guard = try_apply_lock(&state.apply_lock, "otg")?;
let old_config = state.config.get(); let old_config = state.config.get();
let mut staged_config = old_config.as_ref().clone(); let mut staged_config = old_config.as_ref().clone();
@@ -58,7 +53,11 @@ where
staged_config.uac.validate()?; staged_config.uac.validate()?;
} }
if let Err(error) = apply_usb_config(state, &old_config, &staged_config).await { if let Err(error) = state
.coordinator
.apply_config(&old_config, &staged_config)
.await
{
return Err(rollback_after_failure(state, &staged_config, &old_config, error, false).await); return Err(rollback_after_failure(state, &staged_config, &old_config, error, false).await);
} }
@@ -114,7 +113,7 @@ where
} }
async fn rollback_after_failure( async fn rollback_after_failure(
state: &Arc<AppState>, state: &UsbApiState,
failed_config: &AppConfig, failed_config: &AppConfig,
old_config: &AppConfig, old_config: &AppConfig,
primary_error: AppError, primary_error: AppError,
@@ -122,7 +121,11 @@ async fn rollback_after_failure(
) -> AppError { ) -> AppError {
let mut rollback_errors = Vec::new(); let mut rollback_errors = Vec::new();
if let Err(error) = apply_usb_config(state, failed_config, old_config).await { if let Err(error) = state
.coordinator
.apply_config(failed_config, old_config)
.await
{
rollback_errors.push(format!("runtime rollback failed: {error}")); rollback_errors.push(format!("runtime rollback failed: {error}"));
} }
if restore_descriptor && old_config.hid.backend == HidBackend::Ch9329 { if restore_descriptor && old_config.hid.backend == HidBackend::Ch9329 {
@@ -141,7 +144,7 @@ async fn rollback_after_failure(
let message = format!("{primary_error}; {}", rollback_errors.join("; ")); let message = format!("{primary_error}; {}", rollback_errors.join("; "));
#[cfg(unix)] #[cfg(unix)]
state.otg_service.mark_degraded(message.clone()).await; state.otg.mark_degraded(message.clone()).await;
AppError::Config(message) AppError::Config(message)
} }

56
src/web/handlers/config/vnc.rs
View File

@@ -1,20 +1,22 @@
use axum::{extract::State, Json}; use axum::{extract::State, Json};
use std::sync::Arc;
use crate::error::Result; use crate::error::Result;
use crate::state::AppState; use crate::web::state::RemoteAccessApiState;
use super::apply::{apply_vnc_config, try_apply_lock, ConfigApplyOptions};
use super::types::{VncConfigResponse, VncConfigUpdate, VncStatusResponse}; use super::types::{VncConfigResponse, VncConfigUpdate, VncStatusResponse};
use crate::runtime::{try_apply_lock, ConfigApplyOptions};
fn validate_candidate(state: &Arc<AppState>, config: &crate::config::VncConfig) -> Result<()> { fn validate_candidate(
state: &RemoteAccessApiState,
config: &crate::config::VncConfig,
) -> Result<()> {
let mut candidate = state.config.get().as_ref().clone(); let mut candidate = state.config.get().as_ref().clone();
candidate.vnc = config.clone(); candidate.vnc = config.clone();
crate::video::codec_constraints::validate_third_party_codec_compatibility(&candidate) crate::video::codec_constraints::validate_third_party_codec_compatibility(&candidate)
} }
async fn persist_and_apply( async fn persist_and_apply(
state: &Arc<AppState>, state: &RemoteAccessApiState,
old_config: crate::config::VncConfig, old_config: crate::config::VncConfig,
new_config: crate::config::VncConfig, new_config: crate::config::VncConfig,
) -> Result<crate::config::VncConfig> { ) -> Result<crate::config::VncConfig> {
@@ -26,8 +28,9 @@ async fn persist_and_apply(
}) })
.await?; .await?;
let stored_config = state.config.get().vnc.clone(); let stored_config = state.config.get().vnc.clone();
apply_vnc_config( state
state, .coordinator
.apply_vnc(
&old_config, &old_config,
&stored_config, &stored_config,
ConfigApplyOptions::preserving_service_state(), ConfigApplyOptions::preserving_service_state(),
@@ -36,20 +39,15 @@ async fn persist_and_apply(
Ok(stored_config) Ok(stored_config)
} }
async fn current_status(state: &Arc<AppState>) -> (crate::vnc::VncServiceStatus, usize) { async fn current_status(state: &RemoteAccessApiState) -> (crate::vnc::VncServiceStatus, usize) {
let guard = state.vnc.read().await; state.coordinator.vnc_status().await
if let Some(ref service) = *guard {
(service.status().await, service.connection_count())
} else {
(crate::vnc::VncServiceStatus::Stopped, 0)
}
} }
pub async fn get_vnc_config(State(state): State<Arc<AppState>>) -> Json<VncConfigResponse> { pub async fn get_vnc_config(State(state): State<RemoteAccessApiState>) -> Json<VncConfigResponse> {
Json(VncConfigResponse::from(&state.config.get().vnc)) Json(VncConfigResponse::from(&state.config.get().vnc))
} }
pub async fn get_vnc_status(State(state): State<Arc<AppState>>) -> Json<VncStatusResponse> { pub async fn get_vnc_status(State(state): State<RemoteAccessApiState>) -> Json<VncStatusResponse> {
let config = state.config.get().vnc.clone(); let config = state.config.get().vnc.clone();
let (status, connection_count) = current_status(&state).await; let (status, connection_count) = current_status(&state).await;
@@ -57,12 +55,12 @@ pub async fn get_vnc_status(State(state): State<Arc<AppState>>) -> Json<VncStatu
} }
pub async fn update_vnc_config( pub async fn update_vnc_config(
State(state): State<Arc<AppState>>, State(state): State<RemoteAccessApiState>,
Json(req): Json<VncConfigUpdate>, Json(req): Json<VncConfigUpdate>,
) -> Result<Json<VncConfigResponse>> { ) -> Result<Json<VncConfigResponse>> {
req.validate()?; req.validate()?;
let _apply_guard = try_apply_lock(&state.config_apply_locks.vnc, "vnc")?; let _apply_guard = try_apply_lock(&state.vnc_apply_lock, "vnc")?;
let old_config = state.config.get().vnc.clone(); let old_config = state.config.get().vnc.clone();
let mut merged_config = old_config.clone(); let mut merged_config = old_config.clone();
req.apply_to(&mut merged_config); req.apply_to(&mut merged_config);
@@ -73,18 +71,19 @@ pub async fn update_vnc_config(
} }
pub async fn start_vnc_service( pub async fn start_vnc_service(
State(state): State<Arc<AppState>>, State(state): State<RemoteAccessApiState>,
) -> Result<Json<VncStatusResponse>> { ) -> Result<Json<VncStatusResponse>> {
let _apply_guard = try_apply_lock(&state.config_apply_locks.vnc, "vnc")?; let _apply_guard = try_apply_lock(&state.vnc_apply_lock, "vnc")?;
let stored_config = state.config.get().vnc.clone(); let stored_config = state.config.get().vnc.clone();
let runtime_config = state.runtime_third_party_config().await.vnc; let runtime_config = state.coordinator.runtime_config().await.vnc;
let mut start_config = stored_config.clone(); let mut start_config = stored_config.clone();
start_config.enabled = true; start_config.enabled = true;
if start_config.password.as_deref().unwrap_or("").is_empty() { if start_config.password.as_deref().unwrap_or("").is_empty() {
start_config.password = stored_config.password.clone(); start_config.password = stored_config.password.clone();
} }
apply_vnc_config( state
&state, .coordinator
.apply_vnc(
&runtime_config, &runtime_config,
&start_config, &start_config,
ConfigApplyOptions::runtime_only(), ConfigApplyOptions::runtime_only(),
@@ -100,15 +99,16 @@ pub async fn start_vnc_service(
} }
pub async fn stop_vnc_service( pub async fn stop_vnc_service(
State(state): State<Arc<AppState>>, State(state): State<RemoteAccessApiState>,
) -> Result<Json<VncStatusResponse>> { ) -> Result<Json<VncStatusResponse>> {
let _apply_guard = try_apply_lock(&state.config_apply_locks.vnc, "vnc")?; let _apply_guard = try_apply_lock(&state.vnc_apply_lock, "vnc")?;
let stored_config = state.config.get().vnc.clone(); let stored_config = state.config.get().vnc.clone();
let runtime_config = state.runtime_third_party_config().await.vnc; let runtime_config = state.coordinator.runtime_config().await.vnc;
let mut stop_config = stored_config.clone(); let mut stop_config = stored_config.clone();
stop_config.enabled = false; stop_config.enabled = false;
apply_vnc_config( state
&state, .coordinator
.apply_vnc(
&runtime_config, &runtime_config,
&stop_config, &stop_config,
ConfigApplyOptions::runtime_only(), ConfigApplyOptions::runtime_only(),

2
src/web/handlers/mod.rs
View File

@@ -42,12 +42,12 @@ use serde::{Deserialize, Serialize};
use std::sync::Arc; use std::sync::Arc;
use tracing::{debug, info, warn}; use tracing::{debug, info, warn};
use self::config::apply::ConfigApplyOptions;
use crate::auth::{Session, SESSION_COOKIE}; use crate::auth::{Session, SESSION_COOKIE};
use crate::config::StreamMode; use crate::config::StreamMode;
use crate::diagnostics::{get_device_info, get_disk_space, DeviceInfo, DiskSpaceInfo}; use crate::diagnostics::{get_device_info, get_disk_space, DeviceInfo, DiskSpaceInfo};
use crate::error::{AppError, Result}; use crate::error::{AppError, Result};
use crate::platform::PlatformCapabilities; use crate::platform::PlatformCapabilities;
use crate::runtime::ConfigApplyOptions;
use crate::state::AppState; use crate::state::AppState;
use crate::update::{UpdateChannel, UpdateOverviewResponse, UpdateStatusResponse, UpgradeRequest}; use crate::update::{UpdateChannel, UpdateOverviewResponse, UpdateStatusResponse, UpgradeRequest};
use crate::utils::list_serial_ports; use crate::utils::list_serial_ports;

12
src/web/handlers/setup.rs
View File

@@ -175,7 +175,7 @@ pub async fn setup_init(
// Apply the complete USB runtime configuration, including the MSD controller. // Apply the complete USB runtime configuration, including the MSD controller.
let new_config = state.config.get(); let new_config = state.config.get();
if let Err(e) = config::apply::apply_usb_config(&state, &old_config, &new_config).await { if let Err(e) = state.usb.apply_config(&old_config, &new_config).await {
tracing::warn!("Failed to apply USB config during setup: {}", e); tracing::warn!("Failed to apply USB config during setup: {}", e);
} }
@@ -201,8 +201,9 @@ pub async fn setup_init(
// Start RustDesk if enabled // Start RustDesk if enabled
if new_config.rustdesk.enabled { if new_config.rustdesk.enabled {
let empty_config = crate::rustdesk::config::RustDeskConfig::default(); let empty_config = crate::rustdesk::config::RustDeskConfig::default();
if let Err(e) = config::apply::apply_rustdesk_config( if let Err(e) = state
&state, .remote_access
.apply_rustdesk(
&empty_config, &empty_config,
&new_config.rustdesk, &new_config.rustdesk,
ConfigApplyOptions::default(), ConfigApplyOptions::default(),
@@ -218,8 +219,9 @@ pub async fn setup_init(
// Start RTSP if enabled // Start RTSP if enabled
if new_config.rtsp.enabled { if new_config.rtsp.enabled {
let empty_config = crate::config::RtspConfig::default(); let empty_config = crate::config::RtspConfig::default();
if let Err(e) = config::apply::apply_rtsp_config( if let Err(e) = state
&state, .remote_access
.apply_rtsp(
&empty_config, &empty_config,
&new_config.rtsp, &new_config.rtsp,
ConfigApplyOptions::default(), ConfigApplyOptions::default(),

1
src/web/mod.rs
View File

@@ -2,6 +2,7 @@ mod audio_ws;
mod error; mod error;
mod handlers; mod handlers;
mod routes; mod routes;
pub(crate) mod state;
mod static_files; mod static_files;
#[cfg(unix)] #[cfg(unix)]
mod uac_ws; mod uac_ws;

55
src/web/state.rs Normal file
View File

@@ -0,0 +1,55 @@
use std::sync::Arc;
use axum::extract::FromRef;
use tokio::sync::Mutex;
use crate::config::ConfigStore;
use crate::hid::HidController;
#[cfg(unix)]
use crate::otg::OtgService;
use crate::runtime::{RemoteAccessCoordinator, UsbCoordinator};
use crate::state::AppState;
#[derive(Clone)]
pub(crate) struct RemoteAccessApiState {
pub config: ConfigStore,
pub coordinator: Arc<RemoteAccessCoordinator>,
pub rustdesk_apply_lock: Arc<Mutex<()>>,
pub vnc_apply_lock: Arc<Mutex<()>>,
pub rtsp_apply_lock: Arc<Mutex<()>>,
}
impl FromRef<Arc<AppState>> for RemoteAccessApiState {
fn from_ref(state: &Arc<AppState>) -> Self {
Self {
config: state.config.clone(),
coordinator: state.remote_access.clone(),
rustdesk_apply_lock: state.config_apply_locks.rustdesk.clone(),
vnc_apply_lock: state.config_apply_locks.vnc.clone(),
rtsp_apply_lock: state.config_apply_locks.rtsp.clone(),
}
}
}
#[derive(Clone)]
pub(crate) struct UsbApiState {
pub config: ConfigStore,
pub coordinator: Arc<UsbCoordinator>,
pub hid: Arc<HidController>,
pub apply_lock: Arc<Mutex<()>>,
#[cfg(unix)]
pub otg: Arc<OtgService>,
}
impl FromRef<Arc<AppState>> for UsbApiState {
fn from_ref(state: &Arc<AppState>) -> Self {
Self {
config: state.config.clone(),
coordinator: state.usb.clone(),
hid: state.hid.clone(),
apply_lock: state.config_apply_locks.otg.clone(),
#[cfg(unix)]
otg: state.otg_service.clone(),
}
}
}