GithubFollow/One-KVM
1
0
Fork 0
mirror of https://github.com/mofeng-git/One-KVM.git synced 2026-10-31 01:40:49 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

refactor: 清理未使用代码并整合运行时配置逻辑

Browse Source
This commit is contained in:
mofeng-git
2026-10-28 21:39:27 +08:00
parent 63ba4c3980
commit cd48df752c
34 changed files with 321 additions and 1068 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
src/atx/disabled_led.rs
View File

@@ -1,23 +1,11 @@
#![allow(dead_code)]
use super::types::{AtxInputBinding, PowerStatus};
use crate::error::Result;
pub struct LedSensor {
config: AtxInputBinding,
}
pub struct LedSensor;
impl LedSensor {
pub fn new(config: AtxInputBinding) -> Self {
Self { config }
}
pub fn is_configured(&self) -> bool {
self.config.is_configured()
}
pub fn is_initialized(&self) -> bool {
false
pub fn new(_config: AtxInputBinding) -> Self {
Self
}
pub async fn init(&mut self) -> Result<()> {

6
src/atx/led.rs
View File

@@ -101,12 +101,6 @@ impl LedSensor {
}
}
impl Drop for LedSensor {
fn drop(&mut self) {
*self.handle.lock().unwrap() = None;
}
}
#[cfg(test)]
mod tests {
use super::*;

16
src/audio/controller.rs
View File

@@ -269,22 +269,6 @@ impl AudioController {
.map(|s| s.subscribe_opus())
}
pub async fn set_enabled(&self, enabled: bool) -> Result<()> {
let _operation = self.operation.lock().await;
self.recovery.cancel();
{
let mut config = self.config.write().await;
config.enabled = enabled;
}
if !enabled && self.is_streaming().await {
self.stop_streaming_inner().await?;
}
info!("Audio enabled: {}", enabled);
Ok(())
}
pub async fn update_config(&self, new_config: AudioControllerConfig) -> Result<()> {
let _operation = self.operation.lock().await;
self.recovery.cancel();

7
src/auth/session.rs
View File

@@ -92,13 +92,6 @@ impl SessionStore {
Ok(())
}
pub async fn cleanup_expired(&self) -> Result<u64> {
let mut guard = self.inner.write().await;
let before = guard.len();
guard.retain(|_, s| !s.is_expired());
Ok((before - guard.len()) as u64)
}
pub async fn delete_all(&self) -> Result<u64> {
let mut guard = self.inner.write().await;
let n = guard.len() as u64;

26
src/redfish/schema.rs
View File

@@ -33,14 +33,6 @@ impl Status {
}
}
pub fn enabled_health(health: &str) -> Self {
Self {
state: "Enabled".to_string(),
health: health.to_string(),
health_rollup: None,
}
}
pub fn disabled_ok() -> Self {
Self {
state: "Disabled".to_string(),
@@ -48,14 +40,6 @@ impl Status {
health_rollup: None,
}
}
pub fn offline_ok() -> Self {
Self {
state: "Offline".to_string(),
health: "OK".to_string(),
health_rollup: None,
}
}
}
#[derive(Debug, Clone, Serialize)]
@@ -626,16 +610,6 @@ impl RedfishError {
}
}
pub fn property_missing(property: &str) -> Self {
Self {
error: RedfishErrorBody {
code: "Base.1.18.PropertyMissing".to_string(),
message: format!("Property '{}' is required", property),
extended_info: vec![],
},
}
}
pub fn service_unavailable(msg: &str) -> Self {
Self {
error: RedfishErrorBody {

156
src/runtime/config_apply.rs
View File

@@ -2,7 +2,10 @@ use std::sync::Arc;
use tokio::sync::{Mutex, OwnedMutexGuard};
use crate::config::{AtxConfig, AudioConfig, StreamConfig, VideoConfig};
use crate::error::{AppError, Result};
use crate::state::AppState;
use crate::stream_encoder::encoder_type_to_backend;
#[derive(Debug, Clone, Copy, Default)]
pub struct ConfigApplyOptions {
@@ -12,14 +15,6 @@ pub struct ConfigApplyOptions {
}
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,
@@ -42,3 +37,148 @@ pub fn try_apply_lock(lock: &Arc<Mutex<()>>, domain: &str) -> Result<OwnedMutexG
AppError::ServiceUnavailable(format!("{domain} configuration is already applying"))
})
}
pub(crate) async fn apply_video_config(state: &AppState, new_config: &VideoConfig) -> Result<()> {
tracing::info!("Applying video config changes...");
let device = new_config
.device
.as_deref()
.ok_or_else(|| AppError::BadRequest("video_device is required".to_string()))?;
let format = new_config
.format
.as_ref()
.and_then(|f| {
serde_json::from_value::<crate::video::format::PixelFormat>(serde_json::Value::String(
f.clone(),
))
.ok()
})
.unwrap_or(crate::video::format::PixelFormat::Mjpeg);
let resolution = crate::video::format::Resolution::new(new_config.width, new_config.height);
state
.stream_manager
.apply_video_config(device, format, resolution, new_config.fps)
.await
.map_err(|e| AppError::VideoError(format!("Failed to apply video config: {}", e)))?;
tracing::info!("Video config applied successfully");
Ok(())
}
pub(crate) async fn apply_stream_config(
state: &AppState,
old_config: &StreamConfig,
new_config: &StreamConfig,
options: ConfigApplyOptions,
) -> Result<()> {
tracing::info!("Applying stream config changes...");
if options.force || old_config.encoder != new_config.encoder {
let encoder_backend = encoder_type_to_backend(new_config.encoder.clone());
tracing::info!(
"Updating encoder backend to: {:?} (from config: {:?})",
encoder_backend,
new_config.encoder
);
state.webrtc.update_encoder_backend(encoder_backend).await;
}
if options.force || old_config.bitrate_preset != new_config.bitrate_preset {
state
.stream_manager
.set_bitrate_preset(new_config.bitrate_preset)
.await?;
}
let ice_changed = old_config.stun_server != new_config.stun_server
|| old_config.turn_server != new_config.turn_server
|| old_config.turn_username != new_config.turn_username
|| old_config.turn_password != new_config.turn_password;
if options.force || ice_changed {
tracing::info!(
"Updating ICE config: STUN={:?}, TURN={:?}",
new_config.stun_server,
new_config.turn_server
);
state
.webrtc
.update_ice_config(
new_config.stun_server.clone(),
new_config.turn_server.clone(),
new_config.turn_username.clone(),
new_config.turn_password.clone(),
)
.await;
}
tracing::info!(
"Stream config applied: encoder={:?}, bitrate={}",
new_config.encoder,
new_config.bitrate_preset
);
Ok(())
}
pub(crate) async fn apply_atx_config(state: &AppState, new_config: &AtxConfig) -> Result<()> {
tracing::info!("Applying ATX config changes...");
let controller_config = new_config.to_controller_config();
let atx_guard = state.atx.read().await;
if let Some(atx) = atx_guard.as_ref() {
if let Err(e) = atx.reload(controller_config).await {
tracing::error!("ATX reload failed: {}", e);
return Err(AppError::Config(format!("ATX reload failed: {}", e)));
}
tracing::info!("ATX controller reloaded successfully");
} else {
drop(atx_guard);
if new_config.enabled {
tracing::info!("ATX enabled in config, initializing...");
let atx = crate::atx::AtxController::new(controller_config);
atx.init()
.await
.map_err(|e| AppError::Config(format!("ATX initialization failed: {}", e)))?;
*state.atx.write().await = Some(atx);
tracing::info!("ATX controller initialized successfully");
}
}
Ok(())
}
pub(crate) async fn apply_audio_config(state: &AppState, new_config: &AudioConfig) -> Result<()> {
tracing::info!("Applying audio config changes...");
let audio_config = crate::audio::AudioControllerConfig {
enabled: new_config.enabled,
device: new_config.device.clone(),
quality: new_config.quality.parse::<crate::audio::AudioQuality>()?,
};
state.audio.update_config(audio_config).await?;
tracing::info!(
"Audio config applied: enabled={}, device={}",
new_config.enabled,
new_config.device
);
state
.stream_manager
.set_webrtc_audio_enabled(new_config.enabled)
.await?;
tracing::debug!("WebRTC audio enabled: {}", new_config.enabled);
if new_config.enabled {
state.stream_manager.reconnect_webrtc_audio_sources().await;
}
Ok(())
}

3
src/runtime/mod.rs
View File

@@ -5,6 +5,9 @@ mod supervisor;
mod usb;
pub use builder::{ApplicationRuntime, RuntimeBuilder, WebConfigOverrides};
pub(crate) use config_apply::{
apply_atx_config, apply_audio_config, apply_stream_config, apply_video_config,
};
pub use config_apply::{try_apply_lock, ConfigApplyOptions};
pub use remote_access::{RemoteAccessCoordinator, RustDeskRuntimeStatus};
pub use usb::UsbCoordinator;

12
src/runtime/remote_access.rs
View File

@@ -219,7 +219,7 @@ impl RemoteAccessCoordinator {
tracing::info!("RustDesk service stopped");
}
*self.rustdesk.write().await = None;
} else if !options.preserve_service_state && new_config.enabled {
} else if !options.preserve_service_state {
match current {
None => {
tracing::info!("Initializing RustDesk service...");
@@ -258,7 +258,7 @@ impl RemoteAccessCoordinator {
credentials_to_save = service.save_credentials();
}
}
} else if options.preserve_service_state && need_restart {
} else if need_restart {
if let Some(service) = current {
let mut runtime_config = new_config.clone();
runtime_config.enabled = true;
@@ -336,7 +336,7 @@ impl RemoteAccessCoordinator {
service.stop().await?;
}
*self.vnc.write().await = None;
} else if !options.preserve_service_state && new_config.enabled {
} else if !options.preserve_service_state {
match current {
None => {
let service = Arc::new(VncService::new(
@@ -360,7 +360,7 @@ impl RemoteAccessCoordinator {
}
}
}
} else if options.preserve_service_state && need_restart {
} else if need_restart {
if let Some(service) = current {
let mut runtime_config = new_config.clone();
runtime_config.enabled = true;
@@ -399,7 +399,7 @@ impl RemoteAccessCoordinator {
})?;
}
*self.rtsp.write().await = None;
} else if !options.preserve_service_state && new_config.enabled {
} else if !options.preserve_service_state {
match current {
None => {
let service = Arc::new(RtspService::new(
@@ -422,7 +422,7 @@ impl RemoteAccessCoordinator {
}
}
}
} else if options.preserve_service_state && need_restart {
} else if need_restart {
if let Some(service) = current {
let mut runtime_config = new_config.clone();
runtime_config.enabled = true;

412
src/rustdesk/bytes_codec.rs
View File

@@ -1,59 +1,26 @@
//! Variable-length TCP framing (RustDesk wire format).
use bytes::{Buf, BufMut, Bytes, BytesMut};
use bytes::BytesMut;
use std::io::{self, IoSlice};
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
const MAX_PACKET_LENGTH: usize = 0x3FFFFFFF;
pub fn encode_frame(data: &[u8]) -> io::Result<Vec<u8>> {
let len = data.len();
let mut buf = Vec::with_capacity(len + 4);
if len <= 0x3F {
buf.push((len << 2) as u8);
} else if len <= 0x3FFF {
let h = ((len << 2) as u16) | 0x1;
buf.extend_from_slice(&h.to_le_bytes());
} else if len <= 0x3FFFFF {
let h = ((len << 2) as u32) | 0x2;
buf.push((h & 0xFF) as u8);
buf.push(((h >> 8) & 0xFF) as u8);
buf.push(((h >> 16) & 0xFF) as u8);
} else if len <= MAX_PACKET_LENGTH {
let h = ((len << 2) as u32) | 0x3;
buf.extend_from_slice(&h.to_le_bytes());
} else {
fn encode_header(len: usize) -> io::Result<([u8; 4], usize)> {
let header_len = match len {
0..=0x3F => 1,
0x40..=0x3FFF => 2,
0x4000..=0x3FFFFF => 3,
0x400000..=MAX_PACKET_LENGTH => 4,
_ => {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
"Message too large",
));
}
buf.extend_from_slice(data);
Ok(buf)
}
fn decode_header(first_byte: u8, header_bytes: &[u8]) -> (usize, usize) {
let head_len = ((first_byte & 0x3) + 1) as usize;
let mut n = first_byte as usize;
if head_len > 1 && !header_bytes.is_empty() {
n |= (header_bytes[0] as usize) << 8;
}
if head_len > 2 && header_bytes.len() >= 2 {
n |= (header_bytes[1] as usize) << 16;
}
if head_len > 3 && header_bytes.len() >= 3 {
n |= (header_bytes[2] as usize) << 24;
}
let msg_len = n >> 2;
(head_len, msg_len)
}
pub async fn read_frame<R: AsyncRead + Unpin>(reader: &mut R) -> io::Result<BytesMut> {
read_frame_with_limit(reader, MAX_PACKET_LENGTH).await
};
let header = (((len as u32) << 2) | (header_len as u32 - 1)).to_le_bytes();
Ok((header, header_len))
}
/// Read one framed message while enforcing a caller-selected allocation limit.
@@ -65,17 +32,11 @@ pub async fn read_frame_with_limit<R: AsyncRead + Unpin>(
reader: &mut R,
max_packet_length: usize,
) -> io::Result<BytesMut> {
let mut first_byte = [0u8; 1];
reader.read_exact(&mut first_byte).await?;
let head_len = ((first_byte[0] & 0x3) + 1) as usize;
let mut header_rest = [0u8; 3];
if head_len > 1 {
reader.read_exact(&mut header_rest[..head_len - 1]).await?;
}
let (_, msg_len) = decode_header(first_byte[0], &header_rest);
let mut header = [0u8; 4];
reader.read_exact(&mut header[..1]).await?;
let header_len = ((header[0] & 0x3) + 1) as usize;
reader.read_exact(&mut header[1..header_len]).await?;
let msg_len = (u32::from_le_bytes(header) >> 2) as usize;
if msg_len > max_packet_length {
return Err(io::Error::new(
@@ -92,8 +53,7 @@ pub async fn read_frame_with_limit<R: AsyncRead + Unpin>(
}
pub async fn write_frame<W: AsyncWrite + Unpin>(writer: &mut W, data: &[u8]) -> io::Result<()> {
let frame = encode_frame(data)?;
writer.write_all(&frame).await?;
write_frame_vectored(writer, data).await?;
writer.flush().await?;
Ok(())
}
@@ -104,29 +64,7 @@ pub async fn write_frame_vectored<W: AsyncWrite + Unpin>(
writer: &mut W,
data: &[u8],
) -> io::Result<()> {
let len = data.len();
let mut header = [0u8; 4];
let header_len = if len <= 0x3F {
header[0] = (len << 2) as u8;
1
} else if len <= 0x3FFF {
header[..2].copy_from_slice(&(((len << 2) as u16) | 0x1).to_le_bytes());
2
} else if len <= 0x3FFFFF {
let value = ((len << 2) as u32) | 0x2;
header[0] = (value & 0xFF) as u8;
header[1] = ((value >> 8) & 0xFF) as u8;
header[2] = ((value >> 16) & 0xFF) as u8;
3
} else if len <= MAX_PACKET_LENGTH {
header.copy_from_slice(&(((len << 2) as u32) | 0x3).to_le_bytes());
4
} else {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
"Message too large",
));
};
let (header, header_len) = encode_header(data.len())?;
let slices = [IoSlice::new(&header[..header_len]), IoSlice::new(data)];
let written = writer.write_vectored(&slices).await?;
@@ -145,214 +83,148 @@ pub async fn write_frame_vectored<W: AsyncWrite + Unpin>(
Ok(())
}
pub async fn write_frame_buffered<W: AsyncWrite + Unpin>(
writer: &mut W,
data: &[u8],
buf: &mut BytesMut,
) -> io::Result<()> {
buf.clear();
encode_frame_into(data, buf)?;
writer.write_all(buf).await?;
writer.flush().await?;
Ok(())
}
pub fn encode_frame_into(data: &[u8], buf: &mut BytesMut) -> io::Result<()> {
let len = data.len();
buf.reserve(4 + len);
if len <= 0x3F {
buf.put_u8((len << 2) as u8);
} else if len <= 0x3FFF {
buf.put_u16_le(((len << 2) as u16) | 0x1);
} else if len <= 0x3FFFFF {
let h = ((len << 2) as u32) | 0x2;
buf.put_u8((h & 0xFF) as u8);
buf.put_u8(((h >> 8) & 0xFF) as u8);
buf.put_u8(((h >> 16) & 0xFF) as u8);
} else if len <= MAX_PACKET_LENGTH {
buf.put_u32_le(((len << 2) as u32) | 0x3);
} else {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
"Message too large",
));
}
buf.extend_from_slice(data);
Ok(())
}
/// Stateful decoder for `Framed`.
#[derive(Debug, Clone, Copy)]
pub struct BytesCodec {
state: DecodeState,
max_packet_length: usize,
}
#[derive(Debug, Clone, Copy)]
enum DecodeState {
Head,
Data(usize),
}
impl Default for BytesCodec {
fn default() -> Self {
Self::new()
}
}
impl BytesCodec {
pub fn new() -> Self {
Self {
state: DecodeState::Head,
max_packet_length: MAX_PACKET_LENGTH,
}
}
pub fn set_max_packet_length(&mut self, n: usize) {
self.max_packet_length = n;
}
pub fn decode(&mut self, src: &mut BytesMut) -> io::Result<Option<BytesMut>> {
let n = match self.state {
DecodeState::Head => match self.decode_head(src)? {
Some(n) => {
self.state = DecodeState::Data(n);
n
}
None => return Ok(None),
},
DecodeState::Data(n) => n,
};
match self.decode_data(n, src)? {
Some(data) => {
self.state = DecodeState::Head;
Ok(Some(data))
}
None => Ok(None),
}
}
fn decode_head(&mut self, src: &mut BytesMut) -> io::Result<Option<usize>> {
if src.is_empty() {
return Ok(None);
}
let head_len = ((src[0] & 0x3) + 1) as usize;
if src.len() < head_len {
return Ok(None);
}
let mut n = src[0] as usize;
if head_len > 1 {
n |= (src[1] as usize) << 8;
}
if head_len > 2 {
n |= (src[2] as usize) << 16;
}
if head_len > 3 {
n |= (src[3] as usize) << 24;
}
n >>= 2;
if n > self.max_packet_length {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"Message too large",
));
}
src.advance(head_len);
Ok(Some(n))
}
fn decode_data(&self, n: usize, src: &mut BytesMut) -> io::Result<Option<BytesMut>> {
if src.len() < n {
return Ok(None);
}
Ok(Some(src.split_to(n)))
}
pub fn encode(&mut self, data: Bytes, buf: &mut BytesMut) -> io::Result<()> {
let len = data.len();
if len <= 0x3F {
buf.put_u8((len << 2) as u8);
} else if len <= 0x3FFF {
buf.put_u16_le(((len << 2) as u16) | 0x1);
} else if len <= 0x3FFFFF {
let h = ((len << 2) as u32) | 0x2;
buf.put_u16_le((h & 0xFFFF) as u16);
buf.put_u8((h >> 16) as u8);
} else if len <= MAX_PACKET_LENGTH {
buf.put_u32_le(((len << 2) as u32) | 0x3);
} else {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
"Message too large",
));
}
buf.extend(data);
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::pin::Pin;
use std::task::{Context, Poll};
#[test]
fn test_encode_decode_small() {
let data = vec![1u8; 63];
let encoded = encode_frame(&data).unwrap();
assert_eq!(encoded.len(), 63 + 1);
struct ShortWriter {
bytes: Vec<u8>,
max_write: usize,
}
let mut codec = BytesCodec::new();
let mut buf = BytesMut::from(&encoded[..]);
let decoded = codec.decode(&mut buf).unwrap().unwrap();
assert_eq!(decoded.len(), 63);
impl AsyncWrite for ShortWriter {
fn poll_write(
mut self: Pin<&mut Self>,
_cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<io::Result<usize>> {
let len = buf.len().min(self.max_write);
self.bytes.extend_from_slice(&buf[..len]);
Poll::Ready(Ok(len))
}
fn poll_write_vectored(
mut self: Pin<&mut Self>,
_cx: &mut Context<'_>,
bufs: &[IoSlice<'_>],
) -> Poll<io::Result<usize>> {
let mut written = 0;
for buf in bufs {
let len = buf.len().min(self.max_write - written);
self.bytes.extend_from_slice(&buf[..len]);
written += len;
}
Poll::Ready(Ok(written))
}
fn is_write_vectored(&self) -> bool {
true
}
fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
Poll::Ready(Ok(()))
}
fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
Poll::Ready(Ok(()))
}
}
#[test]
fn test_encode_decode_medium() {
let data = vec![2u8; 1000];
let encoded = encode_frame(&data).unwrap();
assert_eq!(encoded.len(), 1000 + 2);
let mut codec = BytesCodec::new();
let mut buf = BytesMut::from(&encoded[..]);
let decoded = codec.decode(&mut buf).unwrap().unwrap();
assert_eq!(decoded.len(), 1000);
fn header_boundaries_match_wire_format() {
for (len, expected) in [
(0, &[0x00][..]),
(63, &[0xfc][..]),
(64, &[0x01, 0x01][..]),
(16383, &[0xfd, 0xff][..]),
(16384, &[0x02, 0x00, 0x01][..]),
(4194303, &[0xfe, 0xff, 0xff][..]),
(4194304, &[0x03, 0x00, 0x00, 0x01][..]),
(MAX_PACKET_LENGTH, &[0xff, 0xff, 0xff, 0xff][..]),
] {
let (header, header_len) = encode_header(len).unwrap();
assert_eq!(&header[..header_len], expected, "payload length {len}");
}
assert_eq!(
encode_header(MAX_PACKET_LENGTH + 1).unwrap_err().kind(),
io::ErrorKind::InvalidInput
);
}
#[test]
fn test_encode_decode_large() {
let data = vec![3u8; 100000];
let encoded = encode_frame(&data).unwrap();
assert_eq!(encoded.len(), 100000 + 3);
let mut codec = BytesCodec::new();
let mut buf = BytesMut::from(&encoded[..]);
let decoded = codec.decode(&mut buf).unwrap().unwrap();
assert_eq!(decoded.len(), 100000);
#[tokio::test]
async fn frame_round_trips_at_header_boundaries() {
for len in [0, 63, 64, 16383, 16384, 4194303, 4194304] {
let payload = vec![0x5a; len];
let mut encoded = Vec::new();
write_frame(&mut encoded, &payload).await.unwrap();
let mut reader = encoded.as_slice();
let decoded = read_frame_with_limit(&mut reader, len).await.unwrap();
assert_eq!(decoded.as_ref(), payload.as_slice());
assert!(reader.is_empty());
}
}
#[tokio::test]
async fn read_limit_rejects_length_before_allocating_payload() {
let encoded = encode_frame(&vec![0u8; 1024]).unwrap();
let (mut writer, mut reader) = tokio::io::duplex(encoded.len());
tokio::spawn(async move {
writer.write_all(&encoded).await.unwrap();
});
// A 1024-byte length header with no payload must fail the limit check,
// rather than attempt to read the missing payload and return EOF.
let mut reader = &[0x01, 0x10][..];
let error = read_frame_with_limit(&mut reader, 128).await.unwrap_err();
assert_eq!(error.kind(), io::ErrorKind::InvalidData);
}
#[tokio::test]
async fn truncated_frames_return_unexpected_eof() {
for bytes in [
&[0x03][..],
&[0x03, 0x00][..],
&[0x03, 0x00, 0x00][..],
&[0x08, 0x5a][..],
] {
let mut reader = bytes;
let error = read_frame_with_limit(&mut reader, 1024).await.unwrap_err();
assert_eq!(error.kind(), io::ErrorKind::UnexpectedEof);
}
}
#[tokio::test]
async fn consecutive_frames_preserve_message_boundaries() {
let mut reader = &[0x08, 0x11, 0x22, 0x00, 0x04, 0x33][..];
for expected in [&[0x11, 0x22][..], &[][..], &[0x33][..]] {
let frame = read_frame_with_limit(&mut reader, 2).await.unwrap();
assert_eq!(frame.as_ref(), expected);
}
assert!(reader.is_empty());
}
#[tokio::test]
async fn vectored_writer_completes_partial_header_and_payload_writes() {
let payload = vec![0x5a; 16384];
for max_write in [1, 2, 3, 4, 1024] {
let mut writer = ShortWriter {
bytes: Vec::new(),
max_write,
};
write_frame_vectored(&mut writer, &payload).await.unwrap();
assert_eq!(&writer.bytes[..3], &[0x02, 0x00, 0x01]);
assert_eq!(&writer.bytes[3..], payload.as_slice());
}
}
#[tokio::test]
async fn vectored_writer_reports_zero_writes() {
let mut writer = ShortWriter {
bytes: Vec::new(),
max_write: 0,
};
let error = write_frame_vectored(&mut writer, b"payload")
.await
.unwrap_err();
assert_eq!(error.kind(), io::ErrorKind::WriteZero);
}
#[tokio::test]
async fn vectored_writer_round_trips() {
let payload = vec![0x5a; 100_000];

6
src/rustdesk/config.rs
View File

@@ -103,12 +103,6 @@ impl RustDeskConfig {
(*new_uuid.as_bytes(), true)
}
pub fn get_uuid_bytes(&self) -> Option<[u8; 16]> {
self.uuid
.as_ref()
.and_then(|s| uuid::Uuid::parse_str(s).ok().map(|u| *u.as_bytes()))
}
pub fn rendezvous_addr(&self) -> String {
let server = &self.rendezvous_server;
if server.is_empty() {

66
src/rustdesk/crypto.rs
View File

@@ -10,16 +10,10 @@ use thiserror::Error;
pub enum CryptoError {
#[error("Failed to initialize sodiumoxide")]
InitError,
#[error("Encryption failed")]
EncryptionFailed,
#[error("Decryption failed")]
DecryptionFailed,
#[error("Invalid key length")]
InvalidKeyLength,
#[error("Invalid nonce")]
InvalidNonce,
#[error("Signature verification failed")]
SignatureVerificationFailed,
#[error("Key conversion failed")]
KeyConversionFailed,
}
@@ -103,39 +97,6 @@ pub fn decrypt_box(
.map_err(|_| CryptoError::DecryptionFailed)
}
pub fn encrypt_with_key(data: &[u8], key: &secretbox::Key) -> (secretbox::Nonce, Vec<u8>) {
let nonce = secretbox::gen_nonce();
let ciphertext = secretbox::seal(data, &nonce, key);
(nonce, ciphertext)
}
pub fn decrypt_with_key(
ciphertext: &[u8],
nonce: &secretbox::Nonce,
key: &secretbox::Key,
) -> Result<Vec<u8>, CryptoError> {
secretbox::open(ciphertext, nonce, key).map_err(|_| CryptoError::DecryptionFailed)
}
pub fn precompute_key(
their_public_key: &PublicKey,
our_secret_key: &SecretKey,
) -> box_::PrecomputedKey {
box_::precompute(their_public_key, our_secret_key)
}
pub fn symmetric_key_from_slice(key: &[u8]) -> Result<secretbox::Key, CryptoError> {
secretbox::Key::from_slice(key).ok_or(CryptoError::InvalidKeyLength)
}
pub fn nonce_from_slice(bytes: &[u8]) -> Result<Nonce, CryptoError> {
Nonce::from_slice(bytes).ok_or(CryptoError::InvalidNonce)
}
pub fn public_key_from_slice(bytes: &[u8]) -> Result<PublicKey, CryptoError> {
PublicKey::from_slice(bytes).ok_or(CryptoError::InvalidKeyLength)
}
pub fn hash_password(password: &str, salt: &str) -> Vec<u8> {
use sha2::{Digest, Sha256};
let mut hasher = Sha256::new();
@@ -182,26 +143,6 @@ pub fn decrypt_symmetric_key(
secretbox::Key::from_slice(&key_bytes).ok_or(CryptoError::InvalidKeyLength)
}
pub fn encrypt_message(data: &[u8], key: &secretbox::Key, nonce_counter: u64) -> Vec<u8> {
let mut nonce_bytes = [0u8; secretbox::NONCEBYTES];
nonce_bytes[..8].copy_from_slice(&nonce_counter.to_le_bytes());
let nonce = secretbox::Nonce(nonce_bytes);
secretbox::seal(data, &nonce, key)
}
pub fn decrypt_message(
ciphertext: &[u8],
key: &secretbox::Key,
nonce_counter: u64,
) -> Result<Vec<u8>, CryptoError> {
let mut nonce_bytes = [0u8; secretbox::NONCEBYTES];
nonce_bytes[..8].copy_from_slice(&nonce_counter.to_le_bytes());
let nonce = secretbox::Nonce(nonce_bytes);
secretbox::open(ciphertext, &nonce, key).map_err(|_| CryptoError::DecryptionFailed)
}
#[derive(Clone)]
pub struct SigningKeyPair {
pub public_key: sign::PublicKey,
@@ -273,13 +214,6 @@ impl SigningKeyPair {
}
}
pub fn verify_signed(
signed_message: &[u8],
public_key: &sign::PublicKey,
) -> Result<Vec<u8>, CryptoError> {
sign::verify(signed_message, public_key).map_err(|_| CryptoError::SignatureVerificationFailed)
}
#[cfg(test)]
mod tests {
use super::*;

37
src/rustdesk/frame_adapters.rs
View File

@@ -3,7 +3,7 @@ use protobuf::Message as ProtobufMessage;
use super::protocol::hbb::message::{
message as msg_union, misc as misc_union, video_frame as vf_union, AudioFormat, AudioFrame,
CursorData, CursorPosition, EncodedVideoFrame, EncodedVideoFrames, Message, Misc, VideoFrame,
CursorData, EncodedVideoFrame, EncodedVideoFrames, Message, Misc, VideoFrame,
};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -15,18 +15,6 @@ pub enum VideoCodec {
AV1,
}
impl VideoCodec {
pub fn to_codec_id(self) -> i32 {
match self {
VideoCodec::H264 => 0,
VideoCodec::H265 => 1,
VideoCodec::VP8 => 2,
VideoCodec::VP9 => 3,
VideoCodec::AV1 => 4,
}
}
}
pub struct VideoFrameAdapter {
codec: VideoCodec,
seq: u32,
@@ -42,10 +30,6 @@ impl VideoFrameAdapter {
}
}
pub fn set_codec(&mut self, codec: VideoCodec) {
self.codec = codec;
}
pub fn encode_frame_from_bytes(
&mut self,
data: Bytes,
@@ -95,15 +79,6 @@ impl VideoFrameAdapter {
Bytes::from(msg.write_to_bytes().unwrap_or_default())
}
pub fn encode_frame_bytes(
&mut self,
data: &[u8],
is_keyframe: bool,
timestamp_ms: u64,
) -> Bytes {
self.encode_frame_bytes_zero_copy(Bytes::copy_from_slice(data), is_keyframe, timestamp_ms)
}
pub fn seq(&self) -> u32 {
self.seq
}
@@ -185,16 +160,6 @@ impl CursorAdapter {
msg.union = Some(msg_union::Union::CursorData(cursor));
msg
}
pub fn encode_position(x: i32, y: i32) -> Message {
let mut pos = CursorPosition::new();
pos.x = x;
pos.y = y;
let mut msg = Message::new();
msg.union = Some(msg_union::Union::CursorPosition(pos));
msg
}
}
#[cfg(test)]

4
src/rustdesk/hid_adapter.rs
View File

@@ -220,10 +220,6 @@ pub(super) fn convert_key_events_in_code_space(
}
}
pub fn convert_key_event(event: &KeyEvent) -> Option<KeyboardEvent> {
convert_key_events(event).into_iter().next()
}
#[derive(Debug, Clone, Copy)]
struct KeyMapping {
key: CanonicalKey,

5
src/rustdesk/mod.rs
View File

@@ -520,11 +520,6 @@ impl RustDeskService {
}
None
}
#[deprecated(note = "Use save_credentials instead")]
pub fn save_keypair(&self) {
let _ = self.save_credentials();
}
}
async fn wait_for_service_task(mut handle: JoinHandle<()>, task_name: &str) {

18
src/stream/mjpeg.rs
View File

@@ -414,14 +414,6 @@ impl MjpegStreamHandler {
self.auto_pause_config.read().clone()
}
pub fn set_auto_pause_config(&self, config: AutoPauseConfig) {
info!(
"Auto-pause config updated: enabled={}, delay={}s, timeout={}s",
config.enabled, config.shutdown_delay_secs, config.client_timeout_secs
);
*self.auto_pause_config.write() = config;
}
pub fn current_frame(&self) -> Option<VideoFrame> {
(**self.current_frame.load()).clone()
}
@@ -430,19 +422,11 @@ impl MjpegStreamHandler {
self.frame_notify.subscribe()
}
pub fn disconnect_all_clients(&self) {
self.disconnect_clients_matching(|_| true);
}
pub fn disconnect_non_vnc_clients(&self) {
self.disconnect_clients_matching(|id| !id.starts_with("vnc-"));
}
fn disconnect_clients_matching(&self, should_disconnect: impl Fn(&str) -> bool) {
let count = {
let mut clients = self.clients.write();
let before = clients.len();
clients.retain(|id, _| !should_disconnect(id));
clients.retain(|id, _| id.starts_with("vnc-"));
before - clients.len()
};
let remaining = self.client_count();

6
src/stream/mod.rs
View File

@@ -1,10 +1,6 @@
//! MJPEG multipart streaming and WebSocket HID (for MJPEG mode).
//! MJPEG multipart streaming.
pub mod mjpeg;
pub(crate) mod mjpeg_http;
#[cfg(feature = "desktop")]
pub mod ws_hid;
pub use mjpeg::{ClientGuard, MjpegStreamHandler};
#[cfg(feature = "desktop")]
pub use ws_hid::WsHidHandler;

237
src/stream/ws_hid.rs
View File

@@ -1,237 +0,0 @@
//! WebSocket HID for MJPEG mode; binary messages per `crate::hid::datachannel`.
use axum::extract::ws::{Message, WebSocket};
use futures::{SinkExt, StreamExt};
use parking_lot::RwLock;
use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use std::time::Instant;
use tokio::sync::mpsc;
use tracing::{debug, error, info, warn};
use crate::hid::datachannel::{parse_hid_message, HidChannelEvent};
use crate::hid::HidController;
pub type ClientId = String;
#[derive(Debug)]
pub struct WsHidClient {
pub id: ClientId,
pub connected_at: Instant,
pub events_processed: AtomicU64,
shutdown_tx: mpsc::Sender<()>,
}
impl WsHidClient {
pub fn events_count(&self) -> u64 {
self.events_processed.load(Ordering::Relaxed)
}
pub fn connected_secs(&self) -> u64 {
self.connected_at.elapsed().as_secs()
}
}
pub struct WsHidHandler {
hid_controller: RwLock<Option<Arc<HidController>>>,
clients: RwLock<HashMap<ClientId, Arc<WsHidClient>>>,
running: AtomicBool,
total_events: AtomicU64,
}
impl WsHidHandler {
pub fn new() -> Arc<Self> {
Arc::new(Self {
hid_controller: RwLock::new(None),
clients: RwLock::new(HashMap::new()),
running: AtomicBool::new(true),
total_events: AtomicU64::new(0),
})
}
pub fn set_hid_controller(&self, hid: Arc<HidController>) {
*self.hid_controller.write() = Some(hid);
info!("WsHidHandler: HID controller set");
}
pub fn hid_controller(&self) -> Option<Arc<HidController>> {
self.hid_controller.read().clone()
}
pub fn is_hid_available(&self) -> bool {
self.hid_controller.read().is_some()
}
pub fn client_count(&self) -> usize {
self.clients.read().len()
}
pub fn is_running(&self) -> bool {
self.running.load(Ordering::SeqCst)
}
pub fn stop(&self) {
self.running.store(false, Ordering::SeqCst);
let clients = self.clients.read();
for client in clients.values() {
let _ = client.shutdown_tx.try_send(());
}
}
pub fn total_events(&self) -> u64 {
self.total_events.load(Ordering::Relaxed)
}
pub async fn add_client(self: &Arc<Self>, client_id: ClientId, socket: WebSocket) {
let (shutdown_tx, shutdown_rx) = mpsc::channel(1);
let client = Arc::new(WsHidClient {
id: client_id.clone(),
connected_at: Instant::now(),
events_processed: AtomicU64::new(0),
shutdown_tx,
});
self.clients
.write()
.insert(client_id.clone(), client.clone());
info!(
"WsHidHandler: Client {} connected (total: {})",
client_id,
self.client_count()
);
let handler = self.clone();
tokio::spawn(async move {
handler
.handle_client(client_id.clone(), socket, client, shutdown_rx)
.await;
handler.remove_client(&client_id);
});
}
pub fn remove_client(&self, client_id: &str) {
if let Some(client) = self.clients.write().remove(client_id) {
info!(
"WsHidHandler: Client {} disconnected after {}s ({} events)",
client_id,
client.connected_secs(),
client.events_count()
);
}
}
async fn handle_client(
&self,
client_id: ClientId,
socket: WebSocket,
client: Arc<WsHidClient>,
mut shutdown_rx: mpsc::Receiver<()>,
) {
let (mut sender, mut receiver) = socket.split();
let status_byte = if self.is_hid_available() {
0x00u8
} else {
0x01u8
};
let _ = sender.send(Message::Binary(vec![status_byte].into())).await;
loop {
tokio::select! {
biased;
_ = shutdown_rx.recv() => {
debug!("WsHidHandler: Client {} received shutdown signal", client_id);
break;
}
msg = receiver.next() => {
match msg {
Some(Ok(Message::Binary(data))) => {
if let Err(e) = self.handle_binary_message(&data, &client).await {
warn!("WsHidHandler: Failed to handle binary message: {}", e);
}
}
Some(Ok(Message::Ping(data))) => {
let _ = sender.send(Message::Pong(data)).await;
}
Some(Ok(Message::Close(_))) => {
debug!("WsHidHandler: Client {} closed connection", client_id);
break;
}
Some(Err(e)) => {
error!("WsHidHandler: WebSocket error for client {}: {}", client_id, e);
break;
}
None => {
debug!("WsHidHandler: Client {} stream ended", client_id);
break;
}
Some(Ok(Message::Text(_))) => {
warn!("WsHidHandler: Ignoring text message from client {} (binary protocol only)", client_id);
}
_ => {}
}
}
}
}
let hid = self.hid_controller.read().clone();
if let Some(hid) = hid {
if let Err(e) = hid.reset().await {
warn!(
"WsHidHandler: Failed to reset HID on client {} disconnect: {}",
client_id, e
);
} else {
debug!("WsHidHandler: HID reset on client {} disconnect", client_id);
}
}
}
async fn handle_binary_message(&self, data: &[u8], client: &WsHidClient) -> Result<(), String> {
let hid = self
.hid_controller
.read()
.clone()
.ok_or("HID controller not available")?;
let event = parse_hid_message(data).ok_or("Invalid binary HID message")?;
match event {
HidChannelEvent::Keyboard(kb_event) => {
hid.send_keyboard(kb_event)
.await
.map_err(|e| e.to_string())?;
}
HidChannelEvent::Mouse(ms_event) => {
hid.send_mouse(ms_event).await.map_err(|e| e.to_string())?;
}
HidChannelEvent::Consumer(consumer_event) => {
hid.send_consumer(consumer_event)
.await
.map_err(|e| e.to_string())?;
}
}
client.events_processed.fetch_add(1, Ordering::Relaxed);
self.total_events.fetch_add(1, Ordering::Relaxed);
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_ws_hid_handler_creation() {
let handler = WsHidHandler::new();
assert!(handler.is_running());
assert_eq!(handler.client_count(), 0);
assert!(!handler.is_hid_available());
}
}

1
src/video/capture/linux.rs
View File

@@ -940,7 +940,6 @@ impl Drop for CaptureStream {
struct DvTimingsMode {
width: u32,
height: u32,
#[allow(dead_code)]
fps: Option<f64>,
signature: Option<DvTimingsSignature>,
}

22
src/video/codec_constraints.rs
View File

@@ -36,10 +36,6 @@ impl ThirdPartyCodecLock {
ThirdPartyCodecLock::Mjpeg => "mjpeg",
}
}
fn compatible_with(self, other: Self) -> bool {
self == other
}
}
#[derive(Debug, Clone, Copy)]
@@ -152,13 +148,6 @@ pub fn rtsp_codec_to_video_codec(codec: RtspCodec) -> VideoCodecType {
}
}
pub fn vnc_encoding_to_video_codec(encoding: VncEncoding) -> Option<VideoCodecType> {
match encoding {
VncEncoding::TightJpeg => None,
VncEncoding::H264 => Some(VideoCodecType::H264),
}
}
fn rustdesk_lock(config: &AppConfig) -> Option<ThirdPartySourceLock> {
if config.rustdesk.enabled {
return Some(ThirdPartySourceLock {
@@ -204,7 +193,7 @@ pub fn validate_third_party_codec_compatibility(config: &AppConfig) -> Result<()
let locks = third_party_locks(config);
if let Some(first) = locks.first() {
for item in locks.iter().skip(1) {
if !first.lock.compatible_with(item.lock) {
if first.lock != item.lock {
return Err(AppError::BadRequest(format!(
"{} codec {} conflicts with {} codec {}; choose a compatible codec or stop the running service first",
item.source,
@@ -290,15 +279,6 @@ pub fn encoder_codec_to_id(codec: VideoEncoderType) -> &'static str {
}
}
pub fn video_codec_to_encoder_codec(codec: VideoCodecType) -> VideoEncoderType {
match codec {
VideoCodecType::H264 => VideoEncoderType::H264,
VideoCodecType::H265 => VideoEncoderType::H265,
VideoCodecType::VP8 => VideoEncoderType::VP8,
VideoCodecType::VP9 => VideoEncoderType::VP9,
}
}
pub fn encoder_codec_to_video_codec(codec: VideoEncoderType) -> VideoCodecType {
match codec {
VideoEncoderType::H264 => VideoCodecType::H264,

28
src/video/stream_manager.rs
View File

@@ -211,19 +211,6 @@ impl VideoStreamManager {
Ok(())
}
/// Switch streaming mode
///
/// This will:
/// 1. Acquire switching lock (prevent concurrent switches)
/// 2. Notify clients of the mode change
/// 3. Stop the current mode
/// 4. Start the new mode (ensuring video capture runs for WebRTC)
/// 5. Update configuration
pub async fn switch_mode(self: &Arc<Self>, new_mode: StreamMode) -> Result<()> {
let _ = self.switch_mode_transaction(new_mode).await?;
Ok(())
}
/// Switch streaming mode with a transaction ID for correlating events
///
/// If a switch is already in progress, returns `accepted=false` with the
@@ -614,16 +601,6 @@ impl VideoStreamManager {
}
}
/// Get MJPEG client count
pub fn mjpeg_client_count(&self) -> u64 {
self.streamer.mjpeg_handler().client_count()
}
/// Get WebRTC session count
pub async fn webrtc_session_count(&self) -> usize {
self.webrtc_streamer.session_count().await
}
/// Set HID controller for WebRTC DataChannel
pub async fn set_hid_controller(&self, hid: Arc<HidController>) {
self.webrtc_streamer.set_hid_controller(hid).await;
@@ -634,11 +611,6 @@ impl VideoStreamManager {
self.webrtc_streamer.set_audio_enabled(enabled).await
}
/// Check if WebRTC audio is enabled
pub async fn is_webrtc_audio_enabled(&self) -> bool {
self.webrtc_streamer.is_audio_enabled().await
}
/// Reconnect audio sources for all WebRTC sessions
/// Call this after audio controller restarts (e.g., quality change)
pub async fn reconnect_webrtc_audio_sources(&self) {

10
src/video/streamer.rs
View File

@@ -263,10 +263,6 @@ impl Streamer {
}
}
pub fn set_next_retry_ms(&self, ms: u64) {
self.next_retry_ms.store(ms, Ordering::Relaxed);
}
/// Set event bus for broadcasting state changes
pub async fn set_event_bus(&self, events: Arc<EventBus>) {
*self.events.write().await = Some(events);
@@ -294,12 +290,6 @@ impl Streamer {
self.current_device.read().await.clone()
}
/// Get current video configuration (format, resolution, fps)
pub async fn current_video_config(&self) -> (PixelFormat, Resolution, u32) {
let config = self.config.read().await;
(config.format, config.resolution, config.fps)
}
/// Get current capture configuration for direct pipelines
pub async fn current_capture_config(
&self,

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

@@ -1,170 +0,0 @@
use std::sync::Arc;
use crate::config::*;
use crate::error::{AppError, Result};
pub use crate::runtime::{try_apply_lock, ConfigApplyOptions};
use crate::state::AppState;
use crate::stream_encoder::encoder_type_to_backend;
pub async fn apply_video_config(
state: &Arc<AppState>,
old_config: &VideoConfig,
new_config: &VideoConfig,
options: ConfigApplyOptions,
) -> Result<()> {
if old_config == new_config && !options.force {
tracing::info!("Video config unchanged, skipping reload");
return Ok(());
}
tracing::info!("Applying video config changes...");
let device = new_config
.device
.clone()
.ok_or_else(|| AppError::BadRequest("video_device is required".to_string()))?;
let format = new_config
.format
.as_ref()
.and_then(|f| {
serde_json::from_value::<crate::video::format::PixelFormat>(serde_json::Value::String(
f.clone(),
))
.ok()
})
.unwrap_or(crate::video::format::PixelFormat::Mjpeg);
let resolution = crate::video::format::Resolution::new(new_config.width, new_config.height);
state
.stream_manager
.apply_video_config(&device, format, resolution, new_config.fps)
.await
.map_err(|e| AppError::VideoError(format!("Failed to apply video config: {}", e)))?;
tracing::info!("Video config applied successfully");
Ok(())
}
pub async fn apply_stream_config(
state: &Arc<AppState>,
old_config: &StreamConfig,
new_config: &StreamConfig,
options: ConfigApplyOptions,
) -> Result<()> {
tracing::info!("Applying stream config changes...");
if options.force || old_config.encoder != new_config.encoder {
let encoder_backend = encoder_type_to_backend(new_config.encoder.clone());
tracing::info!(
"Updating encoder backend to: {:?} (from config: {:?})",
encoder_backend,
new_config.encoder
);
state.webrtc.update_encoder_backend(encoder_backend).await;
}
if options.force || old_config.bitrate_preset != new_config.bitrate_preset {
state
.stream_manager
.set_bitrate_preset(new_config.bitrate_preset)
.await?;
}
let ice_changed = old_config.stun_server != new_config.stun_server
|| old_config.turn_server != new_config.turn_server
|| old_config.turn_username != new_config.turn_username
|| old_config.turn_password != new_config.turn_password;
if options.force || ice_changed {
tracing::info!(
"Updating ICE config: STUN={:?}, TURN={:?}",
new_config.stun_server,
new_config.turn_server
);
state
.webrtc
.update_ice_config(
new_config.stun_server.clone(),
new_config.turn_server.clone(),
new_config.turn_username.clone(),
new_config.turn_password.clone(),
)
.await;
}
tracing::info!(
"Stream config applied: encoder={:?}, bitrate={}",
new_config.encoder,
new_config.bitrate_preset
);
Ok(())
}
pub async fn apply_atx_config(
state: &Arc<AppState>,
_old_config: &AtxConfig,
new_config: &AtxConfig,
) -> Result<()> {
tracing::info!("Applying ATX config changes...");
let controller_config = new_config.to_controller_config();
let atx_guard = state.atx.read().await;
if let Some(atx) = atx_guard.as_ref() {
if let Err(e) = atx.reload(controller_config).await {
tracing::error!("ATX reload failed: {}", e);
return Err(AppError::Config(format!("ATX reload failed: {}", e)));
}
tracing::info!("ATX controller reloaded successfully");
} else {
drop(atx_guard);
if new_config.enabled {
tracing::info!("ATX enabled in config, initializing...");
let atx = crate::atx::AtxController::new(controller_config);
atx.init()
.await
.map_err(|e| AppError::Config(format!("ATX initialization failed: {}", e)))?;
*state.atx.write().await = Some(atx);
tracing::info!("ATX controller initialized successfully");
}
}
Ok(())
}
pub async fn apply_audio_config(
state: &Arc<AppState>,
_old_config: &AudioConfig,
new_config: &AudioConfig,
) -> Result<()> {
tracing::info!("Applying audio config changes...");
let audio_config = crate::audio::AudioControllerConfig {
enabled: new_config.enabled,
device: new_config.device.clone(),
quality: new_config.quality.parse::<crate::audio::AudioQuality>()?,
};
state.audio.update_config(audio_config).await?;
tracing::info!(
"Audio config applied: enabled={}, device={}",
new_config.enabled,
new_config.device
);
state
.stream_manager
.set_webrtc_audio_enabled(new_config.enabled)
.await?;
tracing::debug!("WebRTC audio enabled: {}", new_config.enabled);
if new_config.enabled {
state.stream_manager.reconnect_webrtc_audio_sources().await;
}
Ok(())
}

4
src/web/handlers/config/atx.rs
View File

@@ -4,9 +4,9 @@ use std::sync::Arc;
use crate::atx::AtxDriverType;
use crate::config::{AtxConfig, HidConfig};
use crate::error::{AppError, Result};
use crate::runtime::{apply_atx_config, try_apply_lock};
use crate::state::AppState;
use super::apply::{apply_atx_config, try_apply_lock};
use super::types::AtxConfigUpdate;
pub async fn get_atx_config(State(state): State<Arc<AppState>>) -> Json<AtxConfig> {
@@ -38,7 +38,7 @@ pub async fn update_atx_config(
let new_atx_config = state.config.get().atx.clone();
apply_atx_config(&state, &old_atx_config, &new_atx_config).await?;
apply_atx_config(&state, &new_atx_config).await?;
Ok(Json(new_atx_config))
}

5
src/web/handlers/config/audio.rs
View File

@@ -3,9 +3,9 @@ use std::sync::Arc;
use crate::config::AudioConfig;
use crate::error::Result;
use crate::runtime::{apply_audio_config, try_apply_lock};
use crate::state::AppState;
use super::apply::{apply_audio_config, try_apply_lock};
use super::types::AudioConfigUpdate;
pub async fn get_audio_config(State(state): State<Arc<AppState>>) -> Json<AudioConfig> {
@@ -19,7 +19,6 @@ pub async fn update_audio_config(
req.validate()?;
let _apply_guard = try_apply_lock(&state.config_apply_locks.audio, "audio")?;
let old_audio_config = state.config.get().audio.clone();
state
.config
@@ -30,7 +29,7 @@ pub async fn update_audio_config(
let new_audio_config = state.config.get().audio.clone();
apply_audio_config(&state, &old_audio_config, &new_audio_config).await?;
apply_audio_config(&state, &new_audio_config).await?;
Ok(Json(new_audio_config))
}

1
src/web/handlers/config/auth.rs
View File

@@ -15,7 +15,6 @@ pub async fn update_auth_config(
State(state): State<Arc<AppState>>,
Json(update): Json<AuthConfigUpdate>,
) -> Result<Json<AuthConfig>> {
update.validate()?;
state
.config
.update(|config| {

1
src/web/handlers/config/mod.rs
View File

@@ -1,4 +1,3 @@
pub(crate) mod apply;
mod types;
mod atx;

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

@@ -2,9 +2,9 @@ use axum::{extract::State, Json};
use std::sync::Arc;
use crate::error::Result;
use crate::runtime::{apply_stream_config, try_apply_lock, ConfigApplyOptions};
use crate::state::AppState;
use super::apply::{apply_stream_config, try_apply_lock, ConfigApplyOptions};
use super::types::{StreamConfigResponse, StreamConfigUpdate};
pub async fn get_stream_config(State(state): State<Arc<AppState>>) -> Json<StreamConfigResponse> {

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

@@ -31,10 +31,6 @@ pub struct AuthConfigUpdate {
}
impl AuthConfigUpdate {
pub fn validate(&self) -> crate::error::Result<()> {
Ok(())
}
pub fn apply_to(&self, config: &mut AuthConfig) {
if let Some(allow_multiple) = self.single_user_allow_multiple_sessions {
config.single_user_allow_multiple_sessions = allow_multiple;

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

@@ -3,9 +3,9 @@ use std::sync::Arc;
use crate::config::VideoConfig;
use crate::error::Result;
use crate::runtime::{apply_video_config, try_apply_lock};
use crate::state::AppState;
use super::apply::{apply_video_config, try_apply_lock, ConfigApplyOptions};
use super::types::VideoConfigUpdate;
pub async fn get_video_config(State(state): State<Arc<AppState>>) -> Json<VideoConfig> {
@@ -41,7 +41,6 @@ pub async fn update_video_config(
req.validate()?;
let _apply_guard = try_apply_lock(&state.config_apply_locks.video, "video")?;
let old_video_config = state.config.get().video.clone();
state
.config
@@ -52,13 +51,7 @@ pub async fn update_video_config(
let new_video_config = state.config.get().video.clone();
apply_video_config(
&state,
&old_video_config,
&new_video_config,
ConfigApplyOptions::forced(),
)
.await?;
apply_video_config(&state, &new_video_config).await?;
Ok(Json(new_video_config))
}

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

@@ -5,8 +5,8 @@ use axum::{extract::State, Json};
use crate::error::{AppError, Result};
use crate::state::AppState;
use super::apply::try_apply_lock;
use super::types::{WatchdogConfigResponse, WatchdogConfigUpdate};
use crate::runtime::try_apply_lock;
async fn response(state: &AppState) -> WatchdogConfigResponse {
let runtime = state.watchdog.status().await;

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

@@ -1,5 +1,5 @@
use super::config::apply::try_apply_lock;
use super::*;
use crate::runtime::try_apply_lock;
use crate::msd::{
DiskModeRequest, DownloadProgress, DriveFile, DriveFileAccess, DriveInfo, DriveInitRequest,

2
src/webrtc/mod.rs
View File

@@ -12,7 +12,7 @@ pub mod webrtc_streamer;
pub use config::WebRtcConfig;
pub use rtp::OpusAudioTrack;
pub use signaling::{ConnectionState, IceCandidate, SdpAnswer, SdpOffer, SignalingMessage};
pub use signaling::{ConnectionState, IceCandidate, SdpAnswer, SdpOffer};
pub use universal_session::{UniversalSession, UniversalSessionConfig, UniversalSessionInfo};
pub use video_track::{UniversalVideoTrack, UniversalVideoTrackConfig, VideoCodec};
pub use webrtc_streamer::{SessionInfo, WebRtcStreamer, WebRtcStreamerConfig, WebRtcStreamerStats};

61
src/webrtc/signaling.rs
View File

@@ -2,16 +2,6 @@
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "lowercase")]
pub enum SignalingMessage {
Offer(SdpOffer),
Answer(SdpAnswer),
Candidate(IceCandidate),
Error(SignalingError),
Close,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SdpOffer {
pub sdp: String,
@@ -31,13 +21,6 @@ pub struct SdpAnswer {
}
impl SdpAnswer {
pub fn new(sdp: impl Into<String>) -> Self {
Self {
sdp: sdp.into(),
ice_candidates: None,
}
}
pub fn with_candidates(sdp: impl Into<String>, candidates: Vec<IceCandidate>) -> Self {
Self {
sdp: sdp.into(),
@@ -61,50 +44,6 @@ pub struct IceCandidate {
pub username_fragment: Option<String>,
}
impl IceCandidate {
pub fn new(candidate: impl Into<String>) -> Self {
Self {
candidate: candidate.into(),
sdp_mid: None,
sdp_mline_index: None,
username_fragment: None,
}
}
pub fn with_mid(mut self, mid: impl Into<String>, index: u16) -> Self {
self.sdp_mid = Some(mid.into());
self.sdp_mline_index = Some(index);
self
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SignalingError {
pub code: u32,
pub message: String,
}
impl SignalingError {
pub fn new(code: u32, message: impl Into<String>) -> Self {
Self {
code,
message: message.into(),
}
}
pub fn invalid_offer() -> Self {
Self::new(400, "Invalid SDP offer")
}
pub fn connection_failed() -> Self {
Self::new(500, "Connection failed")
}
pub fn media_error() -> Self {
Self::new(502, "Media error")
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct OfferRequest {
pub sdp: String,

18
src/webrtc/webrtc_streamer.rs
View File

@@ -811,24 +811,6 @@ impl WebRtcStreamer {
});
}
/// Clear direct capture device configuration
pub async fn clear_capture_device(&self) {
*self.capture_device.write().await = None;
}
/// Prepare for configuration change
///
/// This stops the encoding pipeline and closes all sessions.
pub async fn prepare_for_config_change(&self) {
self.close_all_sessions().await;
if let Err(error) = self.stop_video_pipeline_and_release().await {
warn!(
"Failed to release video pipeline for config change: {}",
error
);
}
}
// === Configuration ===
/// Update video configuration