GithubFollow/One-KVM
1
0
Fork 0
mirror of https://github.com/mofeng-git/One-KVM.git synced 2026-01-28 16:41:52 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

feat(webrtc): 添加公共ICE服务器支持和优化HID延迟

Browse Source 
- 重构ICE配置：将TURN配置改为统一的ICE配置，支持STUN和多TURN URL
- 添加公共ICE服务器：类似RustDesk，用户留空时使用编译时配置的公共服务器
- 优化DataChannel HID消息：使用tokio::spawn立即处理，避免依赖webrtc-rs轮询
- 添加WebRTCReady事件：客户端等待此事件后再建立连接
- 初始化时启动音频流，确保WebRTC可订阅
- 移除多余的trace/debug日志减少开销
- 更新前端配置界面支持公共ICE服务器显示
This commit is contained in:
mofeng-git
2026-01-04 15:06:08 +08:00
parent 0c82d1a840
commit 9ab3d052f9
24 changed files with 766 additions and 258 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
src/audio/encoder.rs
View File

@@ -4,7 +4,7 @@ use audiopus::coder::GenericCtl;
use audiopus::{coder::Encoder, Application, Bitrate, Channels, SampleRate};
use bytes::Bytes;
use std::time::Instant;
use tracing::{info, trace};
use tracing::info;
use super::capture::AudioFrame;
use crate::error::{AppError, Result};
@@ -187,12 +187,6 @@ impl OpusEncoder {
self.frame_count += 1;
trace!(
"Encoded {} samples to {} bytes Opus",
pcm_data.len(),
encoded_len
);
Ok(OpusFrame {
data: Bytes::copy_from_slice(&self.output_buffer[..encoded_len]),
duration_ms,

15
src/config/schema.rs
View File

@@ -383,8 +383,10 @@ pub struct StreamConfig {
/// Bitrate preset (Speed/Balanced/Quality)
pub bitrate_preset: BitratePreset,
/// Custom STUN server (e.g., "stun:stun.l.google.com:19302")
/// If empty, uses public ICE servers from secrets.toml
pub stun_server: Option<String>,
/// Custom TURN server (e.g., "turn:turn.example.com:3478")
/// If empty, uses public ICE servers from secrets.toml
pub turn_server: Option<String>,
/// TURN username
pub turn_username: Option<String>,
@@ -407,7 +409,8 @@ impl Default for StreamConfig {
mode: StreamMode::Mjpeg,
encoder: EncoderType::Auto,
bitrate_preset: BitratePreset::Balanced,
stun_server: Some("stun:stun.l.google.com:19302".to_string()),
// Empty means use public ICE servers (like RustDesk)
stun_server: None,
turn_server: None,
turn_username: None,
turn_password: None,
@@ -418,6 +421,16 @@ impl Default for StreamConfig {
}
}
impl StreamConfig {
/// Check if using public ICE servers (user left fields empty)
pub fn is_using_public_ice_servers(&self) -> bool {
use crate::webrtc::config::public_ice;
self.stun_server.as_ref().map(|s| s.is_empty()).unwrap_or(true)
&& self.turn_server.as_ref().map(|s| s.is_empty()).unwrap_or(true)
&& public_ice::is_configured()
}
}
/// Web server configuration
#[typeshare]
#[derive(Debug, Clone, Serialize, Deserialize)]

13
src/events/types.rs
View File

@@ -187,6 +187,18 @@ pub enum SystemEvent {
device: String,
},
/// WebRTC is ready to accept connections
///
/// Sent after video frame source is connected to WebRTC pipeline.
/// Clients should wait for this event before attempting to create WebRTC sessions.
#[serde(rename = "stream.webrtc_ready")]
WebRTCReady {
/// Current video codec
codec: String,
/// Whether hardware encoding is being used
hardware: bool,
},
/// Stream statistics update (sent periodically for client stats)
#[serde(rename = "stream.stats_update")]
StreamStatsUpdate {
@@ -485,6 +497,7 @@ impl SystemEvent {
Self::StreamDeviceLost { .. } => "stream.device_lost",
Self::StreamReconnecting { .. } => "stream.reconnecting",
Self::StreamRecovered { .. } => "stream.recovered",
Self::WebRTCReady { .. } => "stream.webrtc_ready",
Self::StreamStatsUpdate { .. } => "stream.stats_update",
Self::StreamModeChanged { .. } => "stream.mode_changed",
Self::HidStateChanged { .. } => "hid.state_changed",

14
src/hid/datachannel.rs
View File

@@ -34,7 +34,7 @@
//! Consumer control event (type 0x03):
//! - Bytes 1-2: Usage code (u16 LE)
use tracing::{debug, warn};
use tracing::warn;
use super::types::ConsumerEvent;
use super::{
@@ -115,11 +115,6 @@ fn parse_keyboard_message(data: &[u8]) -> Option<HidChannelEvent> {
right_meta: modifiers_byte & 0x80 != 0,
};
debug!(
"Parsed keyboard: {:?} key=0x{:02X} modifiers=0x{:02X}",
event_type, key, modifiers_byte
);
Some(HidChannelEvent::Keyboard(KeyboardEvent {
event_type,
key,
@@ -168,11 +163,6 @@ fn parse_mouse_message(data: &[u8]) -> Option<HidChannelEvent> {
_ => (None, 0i8),
};
debug!(
"Parsed mouse: {:?} x={} y={} button={:?} scroll={}",
event_type, x, y, button, scroll
);
Some(HidChannelEvent::Mouse(MouseEvent {
event_type,
x,
@@ -191,8 +181,6 @@ fn parse_consumer_message(data: &[u8]) -> Option<HidChannelEvent> {
let usage = u16::from_le_bytes([data[0], data[1]]);
debug!("Parsed consumer: usage=0x{:04X}", usage);
Some(HidChannelEvent::Consumer(ConsumerEvent { usage }))
}

59
src/main.rs
View File

@@ -184,25 +184,44 @@ async fn main() -> anyhow::Result<()> {
let mut stun_servers = vec![];
let mut turn_servers = vec![];
// Add STUN server from config
if let Some(ref stun) = config.stream.stun_server {
if !stun.is_empty() {
stun_servers.push(stun.clone());
tracing::info!("WebRTC STUN server configured: {}", stun);
}
}
// Check if user configured custom servers
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);
// Add TURN server from config
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 {
url: turn.clone(),
username: username.clone(),
credential,
});
tracing::info!("WebRTC TURN server configured: {} (user: {})", turn, username);
// If no custom servers, use public ICE servers (like RustDesk)
if !has_custom_stun && !has_custom_turn {
use one_kvm::webrtc::config::public_ice;
if public_ice::is_configured() {
if let Some(stun) = public_ice::stun_server() {
stun_servers.push(stun.clone());
tracing::info!("Using public STUN server: {}", stun);
}
for turn in public_ice::turn_servers() {
tracing::info!("Using public TURN server: {:?}", turn.urls);
turn_servers.push(turn);
}
} else {
tracing::info!("No public ICE servers configured, using host candidates only");
}
} else {
// Use custom servers
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);
}
}
}
@@ -326,6 +345,10 @@ async fn main() -> anyhow::Result<()> {
config.audio.device,
config.audio.quality
);
// Start audio streaming so WebRTC can subscribe to Opus frames
if let Err(e) = controller.start_streaming().await {
tracing::warn!("Failed to start audio streaming: {}", e);
}
} else {
tracing::info!("Audio disabled in configuration");
}

23
src/video/stream_manager.rs
View File

@@ -396,6 +396,29 @@ impl VideoStreamManager {
);
self.webrtc_streamer.update_video_config(resolution, format, fps).await;
self.webrtc_streamer.set_video_source(frame_tx).await;
// Get device path for events
let device_path = self.streamer.current_device().await
.map(|d| d.path.to_string_lossy().to_string())
.unwrap_or_default();
// Publish StreamConfigApplied event - clients can now safely connect
self.publish_event(SystemEvent::StreamConfigApplied {
device: device_path,
resolution: (resolution.width, resolution.height),
format: format!("{:?}", format).to_lowercase(),
fps,
})
.await;
// Publish WebRTCReady event - frame source is now connected
let codec = self.webrtc_streamer.current_video_codec().await;
let is_hardware = self.webrtc_streamer.is_hardware_encoding().await;
self.publish_event(SystemEvent::WebRTCReady {
codec: codec_to_string(codec),
hardware: is_hardware,
})
.await;
} else {
warn!("No frame source available for WebRTC - sessions may fail to receive video");
}

11
src/web/handlers/config/types.rs
View File

@@ -73,6 +73,10 @@ pub struct StreamConfigResponse {
pub mode: StreamMode,
pub encoder: EncoderType,
pub bitrate_preset: BitratePreset,
/// 是否有公共 ICE 服务器可用（编译时确定）
pub has_public_ice_servers: bool,
/// 当前是否正在使用公共 ICE 服务器（STUN/TURN 都为空时）
pub using_public_ice_servers: bool,
pub stun_server: Option<String>,
pub turn_server: Option<String>,
pub turn_username: Option<String>,
@@ -82,10 +86,13 @@ pub struct StreamConfigResponse {
impl From<&StreamConfig> for StreamConfigResponse {
fn from(config: &StreamConfig) -> Self {
use crate::webrtc::config::public_ice;
Self {
mode: config.mode.clone(),
encoder: config.encoder.clone(),
bitrate_preset: config.bitrate_preset,
has_public_ice_servers: public_ice::is_configured(),
using_public_ice_servers: config.is_using_public_ice_servers(),
stun_server: config.stun_server.clone(),
turn_server: config.turn_server.clone(),
turn_username: config.turn_username.clone(),
@@ -101,8 +108,10 @@ pub struct StreamConfigUpdate {
pub encoder: Option<EncoderType>,
pub bitrate_preset: Option<BitratePreset>,
/// STUN server URL (e.g., "stun:stun.l.google.com:19302")
/// Leave empty to use public ICE servers
pub stun_server: Option<String>,
/// TURN server URL (e.g., "turn:turn.example.com:3478")
/// Leave empty to use public ICE servers
pub turn_server: Option<String>,
/// TURN username
pub turn_username: Option<String>,
@@ -142,7 +151,7 @@ impl StreamConfigUpdate {
if let Some(preset) = self.bitrate_preset {
config.bitrate_preset = preset;
}
// STUN/TURN settings - empty string means clear, Some("value") means set
// STUN/TURN settings - empty string means clear (use public servers), Some("value") means set custom
if let Some(ref stun) = self.stun_server {
config.stun_server = if stun.is_empty() { None } else { Some(stun.clone()) };
}

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

@@ -648,6 +648,24 @@ pub async fn setup_init(
}
}
// Start audio streaming if audio device was selected during setup
if new_config.audio.enabled {
let audio_config = crate::audio::AudioControllerConfig {
enabled: true,
device: new_config.audio.device.clone(),
quality: crate::audio::AudioQuality::from_str(&new_config.audio.quality),
};
if let Err(e) = state.audio.update_config(audio_config).await {
tracing::warn!("Failed to start audio during setup: {}", e);
} else {
tracing::info!("Audio started during setup: device={}", new_config.audio.device);
}
// Also enable WebRTC audio
if let Err(e) = state.stream_manager.set_webrtc_audio_enabled(true).await {
tracing::warn!("Failed to enable WebRTC audio during setup: {}", e);
}
}
tracing::info!("System initialized successfully with admin user: {}", req.username);
Ok(Json(LoginResponse {

73
src/webrtc/config.rs
View File

@@ -2,6 +2,63 @@
use serde::{Deserialize, Serialize};
use crate::secrets;
/// Public ICE server utilities
pub mod public_ice {
use super::*;
/// Check if public ICE servers are configured (at compile time)
pub fn is_configured() -> bool {
secrets::ice::is_configured()
}
/// Check if public TURN servers are configured (requires credentials)
pub fn has_turn() -> bool {
secrets::ice::has_turn()
}
/// Get the public STUN server URL
pub fn stun_server() -> Option<String> {
let server = secrets::ice::STUN_SERVER;
if server.is_empty() {
None
} else {
Some(server.to_string())
}
}
/// Get public TURN servers as TurnServer structs
pub fn turn_servers() -> Vec<TurnServer> {
if !secrets::ice::has_turn() {
return vec![];
}
let urls: Vec<String> = secrets::ice::TURN_URLS
.split(',')
.map(|s| s.trim().to_string())
.filter(|s| !s.is_empty())
.collect();
if urls.is_empty() {
return vec![];
}
vec![TurnServer {
urls,
username: secrets::ice::TURN_USERNAME.to_string(),
credential: secrets::ice::TURN_PASSWORD.to_string(),
}]
}
/// Get all public ICE servers (STUN + TURN) for WebRTC configuration
pub fn get_all_servers() -> (Vec<String>, Vec<TurnServer>) {
let stun_servers = stun_server().into_iter().collect();
let turn_servers = turn_servers();
(stun_servers, turn_servers)
}
}
/// WebRTC configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WebRtcConfig {
@@ -46,14 +103,26 @@ impl Default for WebRtcConfig {
/// TURN server configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TurnServer {
/// TURN server URL (e.g., "turn:turn.example.com:3478")
pub url: String,
/// TURN server URLs (e.g., ["turn:turn.example.com:3478?transport=udp", "turn:turn.example.com:3478?transport=tcp"])
/// Multiple URLs allow fallback between UDP and TCP transports
pub urls: Vec<String>,
/// Username for TURN authentication
pub username: String,
/// Credential for TURN authentication
pub credential: String,
}
impl TurnServer {
/// Create a TurnServer with a single URL (for backwards compatibility)
pub fn new(url: String, username: String, credential: String) -> Self {
Self {
urls: vec![url],
username,
credential,
}
}
}
/// Video codec preference
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]

17
src/webrtc/peer.rs
View File

@@ -78,7 +78,7 @@ impl PeerConnection {
for turn in &config.turn_servers {
ice_servers.push(RTCIceServer {
urls: vec![turn.url.clone()],
urls: turn.urls.clone(),
username: turn.username.clone(),
credential: turn.credential.clone(),
..Default::default()
@@ -207,10 +207,11 @@ impl PeerConnection {
*data_channel.write().await = Some(dc.clone());
// Set up message handler with HID processing
// Immediately spawn task in tokio runtime for low latency
dc.on_message(Box::new(move |msg: DataChannelMessage| {
let hid = hid.clone();
Box::pin(async move {
tokio::spawn(async move {
debug!("DataChannel HID message: {} bytes", msg.data.len());
// Parse and process HID message
@@ -233,7 +234,10 @@ impl PeerConnection {
}
}
}
})
});
// Return empty future (actual work is spawned above)
Box::pin(async {})
}));
})
}));
@@ -432,11 +436,10 @@ impl PeerConnectionManager {
// Add video track
peer.add_video_track(VideoTrackConfig::default()).await?;
// Create data channel and set HID controller
// Set HID controller if available
// Note: We DON'T create a data channel here - the frontend creates it.
// The server receives it via on_data_channel callback set in set_hid_controller().
if self.config.enable_datachannel {
peer.create_data_channel("hid").await?;
// Set HID controller if available
if let Some(ref hid) = self.hid_controller {
peer.set_hid_controller(hid.clone());
}

21
src/webrtc/universal_session.rs
View File

@@ -250,13 +250,13 @@ impl UniversalSession {
// Skip TURN servers without credentials (webrtc-rs requires them)
if turn.username.is_empty() || turn.credential.is_empty() {
warn!(
"Skipping TURN server {} - credentials required but missing",
turn.url
"Skipping TURN server {:?} - credentials required but missing",
turn.urls
);
continue;
}
ice_servers.push(RTCIceServer {
urls: vec![turn.url.clone()],
urls: turn.urls.clone(),
username: turn.username.clone(),
credential: turn.credential.clone(),
..Default::default()
@@ -424,7 +424,9 @@ impl UniversalSession {
dc.on_message(Box::new(move |msg: DataChannelMessage| {
let hid = hid.clone();
Box::pin(async move {
// Immediately spawn task in tokio runtime for low latency
// Don't rely on webrtc-rs to poll the returned Future
tokio::spawn(async move {
if let Some(event) = parse_hid_message(&msg.data) {
match event {
HidChannelEvent::Keyboard(kb_event) => {
@@ -444,7 +446,10 @@ impl UniversalSession {
}
}
}
})
});
// Return empty future (actual work is spawned above)
Box::pin(async {})
}));
})
}));
@@ -654,12 +659,6 @@ impl UniversalSession {
}
} else {
packets_sent += 1;
trace!(
"Session {} sent audio packet {}: {} bytes",
session_id,
packets_sent,
opus_frame.data.len()
);
}
}
Err(broadcast::error::RecvError::Lagged(n)) => {

121
src/webrtc/webrtc_streamer.rs
View File

@@ -35,7 +35,7 @@
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::{broadcast, RwLock};
use tracing::{debug, error, info, warn};
use tracing::{debug, error, info, trace, warn};
use crate::audio::shared_pipeline::{SharedAudioPipeline, SharedAudioPipelineConfig};
use crate::audio::{AudioController, OpusFrame};
@@ -315,9 +315,11 @@ impl WebRtcStreamer {
}
drop(pipeline_guard);
// Clear video frame source to signal upstream to stop
*streamer.video_frame_tx.write().await = None;
info!("Cleared video frame source");
// NOTE: Don't clear video_frame_tx here!
// The frame source is managed by stream_manager and should
// remain available for new sessions. Only stream_manager
// should clear it during mode switches.
info!("Video pipeline stopped, but keeping frame source for new sessions");
}
break;
}
@@ -512,13 +514,37 @@ impl WebRtcStreamer {
/// Update video configuration
///
/// This will restart the encoding pipeline and close all sessions.
/// Only restarts the encoding pipeline if configuration actually changed.
/// This allows multiple consumers (WebRTC, RustDesk) to share the same pipeline
/// without interrupting each other when they call this method with the same config.
pub async fn update_video_config(
&self,
resolution: Resolution,
format: PixelFormat,
fps: u32,
) {
// Check if configuration actually changed
let config = self.config.read().await;
let config_changed = config.resolution != resolution
|| config.input_format != format
|| config.fps != fps;
drop(config);
if !config_changed {
// Configuration unchanged, no need to restart pipeline
trace!(
"Video config unchanged: {}x{} {:?} @ {} fps",
resolution.width, resolution.height, format, fps
);
return;
}
// Configuration changed, restart pipeline
info!(
"Video config changed, restarting pipeline: {}x{} {:?} @ {} fps",
resolution.width, resolution.height, format, fps
);
// Stop existing pipeline
if let Some(ref pipeline) = *self.video_pipeline.read().await {
pipeline.stop();
@@ -598,6 +624,8 @@ impl WebRtcStreamer {
///
/// Note: Changes take effect for new sessions only.
/// Existing sessions need to be reconnected to use the new ICE config.
///
/// If both stun_server and turn_server are empty/None, uses public ICE servers.
pub async fn update_ice_config(
&self,
stun_server: Option<String>,
@@ -607,32 +635,49 @@ impl WebRtcStreamer {
) {
let mut config = self.config.write().await;
// Update STUN servers
// Clear existing servers
config.webrtc.stun_servers.clear();
if let Some(ref stun) = stun_server {
if !stun.is_empty() {
config.webrtc.stun_servers.push(stun.clone());
info!("WebRTC STUN server updated: {}", stun);
}
}
// Update TURN servers
config.webrtc.turn_servers.clear();
if let Some(ref turn) = turn_server {
if !turn.is_empty() {
let username = turn_username.unwrap_or_default();
let credential = turn_password.unwrap_or_default();
config.webrtc.turn_servers.push(TurnServer {
url: turn.clone(),
username: username.clone(),
credential,
});
info!("WebRTC TURN server updated: {} (user: {})", turn, username);
}
}
if config.webrtc.stun_servers.is_empty() && config.webrtc.turn_servers.is_empty() {
info!("WebRTC ICE config cleared - only host candidates will be used");
// Check if user configured custom servers
let has_custom_stun = stun_server.as_ref().map(|s| !s.is_empty()).unwrap_or(false);
let has_custom_turn = turn_server.as_ref().map(|s| !s.is_empty()).unwrap_or(false);
// If no custom servers, use public ICE servers (like RustDesk)
if !has_custom_stun && !has_custom_turn {
use crate::webrtc::config::public_ice;
if public_ice::is_configured() {
if let Some(stun) = public_ice::stun_server() {
config.webrtc.stun_servers.push(stun.clone());
info!("Using public STUN server: {}", stun);
}
for turn in public_ice::turn_servers() {
info!("Using public TURN server: {:?}", turn.urls);
config.webrtc.turn_servers.push(turn);
}
} else {
info!("No public ICE servers configured, using host candidates only");
}
} else {
// Use custom servers
if let Some(ref stun) = stun_server {
if !stun.is_empty() {
config.webrtc.stun_servers.push(stun.clone());
info!("Using custom STUN server: {}", stun);
}
}
if let Some(ref turn) = turn_server {
if !turn.is_empty() {
let username = turn_username.unwrap_or_default();
let credential = turn_password.unwrap_or_default();
config.webrtc.turn_servers.push(TurnServer::new(
turn.clone(),
username.clone(),
credential,
));
info!("Using custom TURN server: {} (user: {})", turn, username);
}
}
}
}
@@ -670,15 +715,14 @@ impl WebRtcStreamer {
let mut session = UniversalSession::new(session_config.clone(), session_id.clone()).await?;
// Set HID controller if available
// Note: We DON'T create a data channel here - the frontend creates it.
// The server only receives it via on_data_channel callback set in set_hid_controller().
// If server also created a channel, frontend's ondatachannel would overwrite its
// own channel with server's, but server's channel has no message handler!
if let Some(ref hid) = *self.hid_controller.read().await {
session.set_hid_controller(hid.clone());
}
// Create data channel
if self.config.read().await.webrtc.enable_datachannel {
session.create_data_channel("hid").await?;
}
let session = Arc::new(session);
// Subscribe to video pipeline frames
@@ -901,9 +945,16 @@ impl WebRtcStreamer {
/// Set bitrate using preset
///
/// Note: Hardware encoders (VAAPI, NVENC, etc.) don't support dynamic bitrate changes.
/// This method restarts the pipeline to apply the new bitrate.
/// This method restarts the pipeline to apply the new bitrate only if the preset actually changed.
pub async fn set_bitrate_preset(self: &Arc<Self>, preset: BitratePreset) -> Result<()> {
// Update config first
// Check if preset actually changed
let current_preset = self.config.read().await.bitrate_preset;
if current_preset == preset {
trace!("Bitrate preset unchanged: {}", preset);
return Ok(());
}
// Update config
self.config.write().await.bitrate_preset = preset;
// Check if pipeline exists and is running