From cd48df752c29c05fe8e7acd624a078413fa9cf78 Mon Sep 17 00:00:00 2001 From: mofeng-git Date: Wed, 28 Oct 2026 21:39:27 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20=E6=B8=85=E7=90=86=E6=9C=AA?= =?UTF-8?q?=E4=BD=BF=E7=94=A8=E4=BB=A3=E7=A0=81=E5=B9=B6=E6=95=B4=E5=90=88?= =?UTF-8?q?=E8=BF=90=E8=A1=8C=E6=97=B6=E9=85=8D=E7=BD=AE=E9=80=BB=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/atx/disabled_led.rs | 18 +- src/atx/led.rs | 6 - src/audio/controller.rs | 16 -- src/auth/session.rs | 7 - src/redfish/schema.rs | 26 -- src/runtime/config_apply.rs | 156 +++++++++- src/runtime/mod.rs | 3 + src/runtime/remote_access.rs | 12 +- src/rustdesk/bytes_codec.rs | 422 ++++++++++------------------ src/rustdesk/config.rs | 6 - src/rustdesk/crypto.rs | 66 ----- src/rustdesk/frame_adapters.rs | 37 +-- src/rustdesk/hid_adapter.rs | 4 - src/rustdesk/mod.rs | 5 - src/stream/mjpeg.rs | 18 +- src/stream/mod.rs | 6 +- src/stream/ws_hid.rs | 237 ---------------- src/video/capture/linux.rs | 1 - src/video/codec_constraints.rs | 22 +- src/video/stream_manager.rs | 28 -- src/video/streamer.rs | 10 - src/web/handlers/config/apply.rs | 170 ----------- src/web/handlers/config/atx.rs | 4 +- src/web/handlers/config/audio.rs | 5 +- src/web/handlers/config/auth.rs | 1 - src/web/handlers/config/mod.rs | 1 - src/web/handlers/config/stream.rs | 2 +- src/web/handlers/config/types.rs | 4 - src/web/handlers/config/video.rs | 11 +- src/web/handlers/config/watchdog.rs | 2 +- src/web/handlers/msd_api.rs | 2 +- src/webrtc/mod.rs | 2 +- src/webrtc/signaling.rs | 61 ---- src/webrtc/webrtc_streamer.rs | 18 -- 34 files changed, 321 insertions(+), 1068 deletions(-) delete mode 100644 src/stream/ws_hid.rs delete mode 100644 src/web/handlers/config/apply.rs diff --git a/src/atx/disabled_led.rs b/src/atx/disabled_led.rs index 47708e12..7a1ed5c2 100644 --- a/src/atx/disabled_led.rs +++ b/src/atx/disabled_led.rs @@ -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<()> { diff --git a/src/atx/led.rs b/src/atx/led.rs index 2d0c3c47..c3ec84eb 100644 --- a/src/atx/led.rs +++ b/src/atx/led.rs @@ -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::*; diff --git a/src/audio/controller.rs b/src/audio/controller.rs index 1e2f1675..b68f7438 100644 --- a/src/audio/controller.rs +++ b/src/audio/controller.rs @@ -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(); diff --git a/src/auth/session.rs b/src/auth/session.rs index 4d7df287..6a0043c0 100644 --- a/src/auth/session.rs +++ b/src/auth/session.rs @@ -92,13 +92,6 @@ impl SessionStore { Ok(()) } - pub async fn cleanup_expired(&self) -> Result { - 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 { let mut guard = self.inner.write().await; let n = guard.len() as u64; diff --git a/src/redfish/schema.rs b/src/redfish/schema.rs index fb8d1840..c92426a5 100644 --- a/src/redfish/schema.rs +++ b/src/redfish/schema.rs @@ -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 { diff --git a/src/runtime/config_apply.rs b/src/runtime/config_apply.rs index 1be6a335..e110bf3f 100644 --- a/src/runtime/config_apply.rs +++ b/src/runtime/config_apply.rs @@ -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>, domain: &str) -> Result 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::(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::()?, + }; + + 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(()) +} diff --git a/src/runtime/mod.rs b/src/runtime/mod.rs index 6f12c72b..8535794a 100644 --- a/src/runtime/mod.rs +++ b/src/runtime/mod.rs @@ -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; diff --git a/src/runtime/remote_access.rs b/src/runtime/remote_access.rs index 91a598c8..9ad568f8 100644 --- a/src/runtime/remote_access.rs +++ b/src/runtime/remote_access.rs @@ -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; diff --git a/src/rustdesk/bytes_codec.rs b/src/rustdesk/bytes_codec.rs index a45dc2bd..92c006d3 100644 --- a/src/rustdesk/bytes_codec.rs +++ b/src/rustdesk/bytes_codec.rs @@ -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> { - 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 { - 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(reader: &mut R) -> io::Result { - read_frame_with_limit(reader, MAX_PACKET_LENGTH).await +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", + )); + } + }; + 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( reader: &mut R, max_packet_length: usize, ) -> io::Result { - 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( } pub async fn write_frame(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( 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( Ok(()) } -pub async fn write_frame_buffered( - 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> { - 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> { - 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> { - 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, + 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> { + 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> { + 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> { + Poll::Ready(Ok(())) + } + + fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + 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]; diff --git a/src/rustdesk/config.rs b/src/rustdesk/config.rs index 12dd09d5..bf5e1d62 100644 --- a/src/rustdesk/config.rs +++ b/src/rustdesk/config.rs @@ -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() { diff --git a/src/rustdesk/crypto.rs b/src/rustdesk/crypto.rs index a4d07c8c..547ee660 100644 --- a/src/rustdesk/crypto.rs +++ b/src/rustdesk/crypto.rs @@ -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) { - 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, 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::from_slice(key).ok_or(CryptoError::InvalidKeyLength) -} - -pub fn nonce_from_slice(bytes: &[u8]) -> Result { - Nonce::from_slice(bytes).ok_or(CryptoError::InvalidNonce) -} - -pub fn public_key_from_slice(bytes: &[u8]) -> Result { - PublicKey::from_slice(bytes).ok_or(CryptoError::InvalidKeyLength) -} - pub fn hash_password(password: &str, salt: &str) -> Vec { 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 { - 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, 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, CryptoError> { - sign::verify(signed_message, public_key).map_err(|_| CryptoError::SignatureVerificationFailed) -} - #[cfg(test)] mod tests { use super::*; diff --git a/src/rustdesk/frame_adapters.rs b/src/rustdesk/frame_adapters.rs index 7a3ed733..18602a98 100644 --- a/src/rustdesk/frame_adapters.rs +++ b/src/rustdesk/frame_adapters.rs @@ -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)] diff --git a/src/rustdesk/hid_adapter.rs b/src/rustdesk/hid_adapter.rs index 07d2cfaa..e76ac6da 100644 --- a/src/rustdesk/hid_adapter.rs +++ b/src/rustdesk/hid_adapter.rs @@ -220,10 +220,6 @@ pub(super) fn convert_key_events_in_code_space( } } -pub fn convert_key_event(event: &KeyEvent) -> Option { - convert_key_events(event).into_iter().next() -} - #[derive(Debug, Clone, Copy)] struct KeyMapping { key: CanonicalKey, diff --git a/src/rustdesk/mod.rs b/src/rustdesk/mod.rs index 035df233..e3a1760b 100644 --- a/src/rustdesk/mod.rs +++ b/src/rustdesk/mod.rs @@ -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) { diff --git a/src/stream/mjpeg.rs b/src/stream/mjpeg.rs index a016f752..8cf80e90 100644 --- a/src/stream/mjpeg.rs +++ b/src/stream/mjpeg.rs @@ -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 { (**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(); diff --git a/src/stream/mod.rs b/src/stream/mod.rs index 495c49f6..c63a760d 100644 --- a/src/stream/mod.rs +++ b/src/stream/mod.rs @@ -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; diff --git a/src/stream/ws_hid.rs b/src/stream/ws_hid.rs deleted file mode 100644 index 3df95d76..00000000 --- a/src/stream/ws_hid.rs +++ /dev/null @@ -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>>, - clients: RwLock>>, - running: AtomicBool, - total_events: AtomicU64, -} - -impl WsHidHandler { - pub fn new() -> Arc { - 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) { - *self.hid_controller.write() = Some(hid); - info!("WsHidHandler: HID controller set"); - } - - pub fn hid_controller(&self) -> Option> { - 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, 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, - 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()); - } -} diff --git a/src/video/capture/linux.rs b/src/video/capture/linux.rs index d00cd5dd..127ba795 100644 --- a/src/video/capture/linux.rs +++ b/src/video/capture/linux.rs @@ -940,7 +940,6 @@ impl Drop for CaptureStream { struct DvTimingsMode { width: u32, height: u32, - #[allow(dead_code)] fps: Option, signature: Option, } diff --git a/src/video/codec_constraints.rs b/src/video/codec_constraints.rs index 0cde662f..379a8c3b 100644 --- a/src/video/codec_constraints.rs +++ b/src/video/codec_constraints.rs @@ -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 { - match encoding { - VncEncoding::TightJpeg => None, - VncEncoding::H264 => Some(VideoCodecType::H264), - } -} - fn rustdesk_lock(config: &AppConfig) -> Option { 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, diff --git a/src/video/stream_manager.rs b/src/video/stream_manager.rs index 761074e1..2e41ce07 100644 --- a/src/video/stream_manager.rs +++ b/src/video/stream_manager.rs @@ -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, 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) { 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) { diff --git a/src/video/streamer.rs b/src/video/streamer.rs index d90c8e9d..b80bf4db 100644 --- a/src/video/streamer.rs +++ b/src/video/streamer.rs @@ -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) { *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, diff --git a/src/web/handlers/config/apply.rs b/src/web/handlers/config/apply.rs deleted file mode 100644 index bed1afb1..00000000 --- a/src/web/handlers/config/apply.rs +++ /dev/null @@ -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, - 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::(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, - 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, - _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, - _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::()?, - }; - - 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(()) -} diff --git a/src/web/handlers/config/atx.rs b/src/web/handlers/config/atx.rs index b3cc221a..9a60dd1a 100644 --- a/src/web/handlers/config/atx.rs +++ b/src/web/handlers/config/atx.rs @@ -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>) -> Json { @@ -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)) } diff --git a/src/web/handlers/config/audio.rs b/src/web/handlers/config/audio.rs index 5e92d340..3e318487 100644 --- a/src/web/handlers/config/audio.rs +++ b/src/web/handlers/config/audio.rs @@ -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>) -> Json { @@ -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)) } diff --git a/src/web/handlers/config/auth.rs b/src/web/handlers/config/auth.rs index 5e358307..647a0621 100644 --- a/src/web/handlers/config/auth.rs +++ b/src/web/handlers/config/auth.rs @@ -15,7 +15,6 @@ pub async fn update_auth_config( State(state): State>, Json(update): Json, ) -> Result> { - update.validate()?; state .config .update(|config| { diff --git a/src/web/handlers/config/mod.rs b/src/web/handlers/config/mod.rs index cbb8af38..247d9487 100644 --- a/src/web/handlers/config/mod.rs +++ b/src/web/handlers/config/mod.rs @@ -1,4 +1,3 @@ -pub(crate) mod apply; mod types; mod atx; diff --git a/src/web/handlers/config/stream.rs b/src/web/handlers/config/stream.rs index d6448b04..4717f076 100644 --- a/src/web/handlers/config/stream.rs +++ b/src/web/handlers/config/stream.rs @@ -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>) -> Json { diff --git a/src/web/handlers/config/types.rs b/src/web/handlers/config/types.rs index 68f60f84..5fe78514 100644 --- a/src/web/handlers/config/types.rs +++ b/src/web/handlers/config/types.rs @@ -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; diff --git a/src/web/handlers/config/video.rs b/src/web/handlers/config/video.rs index d1e3b039..a649fac9 100644 --- a/src/web/handlers/config/video.rs +++ b/src/web/handlers/config/video.rs @@ -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>) -> Json { @@ -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)) } diff --git a/src/web/handlers/config/watchdog.rs b/src/web/handlers/config/watchdog.rs index 0976ceca..45ba927d 100644 --- a/src/web/handlers/config/watchdog.rs +++ b/src/web/handlers/config/watchdog.rs @@ -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; diff --git a/src/web/handlers/msd_api.rs b/src/web/handlers/msd_api.rs index b7261079..176d61ad 100644 --- a/src/web/handlers/msd_api.rs +++ b/src/web/handlers/msd_api.rs @@ -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, diff --git a/src/webrtc/mod.rs b/src/webrtc/mod.rs index 2c1ae15b..2efc6bd5 100644 --- a/src/webrtc/mod.rs +++ b/src/webrtc/mod.rs @@ -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}; diff --git a/src/webrtc/signaling.rs b/src/webrtc/signaling.rs index df490bfc..69173383 100644 --- a/src/webrtc/signaling.rs +++ b/src/webrtc/signaling.rs @@ -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) -> Self { - Self { - sdp: sdp.into(), - ice_candidates: None, - } - } - pub fn with_candidates(sdp: impl Into, candidates: Vec) -> Self { Self { sdp: sdp.into(), @@ -61,50 +44,6 @@ pub struct IceCandidate { pub username_fragment: Option, } -impl IceCandidate { - pub fn new(candidate: impl Into) -> Self { - Self { - candidate: candidate.into(), - sdp_mid: None, - sdp_mline_index: None, - username_fragment: None, - } - } - - pub fn with_mid(mut self, mid: impl Into, 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) -> 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, diff --git a/src/webrtc/webrtc_streamer.rs b/src/webrtc/webrtc_streamer.rs index 852c403d..bccbb3fa 100644 --- a/src/webrtc/webrtc_streamer.rs +++ b/src/webrtc/webrtc_streamer.rs @@ -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