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

feat(rustdesk): 优化视频编码协商和添加公共服务器支持

Browse Source 
- 调整视频编码优先级为 H264 > H265 > VP8 > VP9，优先使用硬件编码
- 对接 RustDesk 客户端质量预设 (Low/Balanced/Best) 到 BitratePreset
- 添加 secrets.toml 编译时读取机制，支持配置公共服务器
- 默认公共服务器: rustdesk.mofeng.run:21116
- 前端 ID 服务器输入框添加问号提示，显示公共服务器信息
- 用户留空时自动使用公共服务器
This commit is contained in:
mofeng-git
2026-01-02 17:22:34 +08:00
parent be4de59f3b
commit 28ecf951df
29 changed files with 776 additions and 316 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
src/config/schema.rs
View File

@@ -1,5 +1,6 @@
use serde::{Deserialize, Serialize};
use typeshare::typeshare;
use crate::video::encoder::BitratePreset;
// Re-export ExtensionsConfig from extensions module
pub use crate::extensions::ExtensionsConfig;
@@ -347,10 +348,8 @@ pub struct StreamConfig {
pub mode: StreamMode,
/// Encoder type for H264/H265
pub encoder: EncoderType,
/// Target bitrate in kbps (for H264/H265)
pub bitrate_kbps: u32,
/// GOP size
pub gop_size: u32,
/// Bitrate preset (Speed/Balanced/Quality)
pub bitrate_preset: BitratePreset,
/// Custom STUN server (e.g., "stun:stun.l.google.com:19302")
pub stun_server: Option<String>,
/// Custom TURN server (e.g., "turn:turn.example.com:3478")
@@ -375,8 +374,7 @@ impl Default for StreamConfig {
Self {
mode: StreamMode::Mjpeg,
encoder: EncoderType::Auto,
bitrate_kbps: 1000,
gop_size: 30,
bitrate_preset: BitratePreset::Balanced,
stun_server: Some("stun:stun.l.google.com:19302".to_string()),
turn_server: None,
turn_username: None,

5
src/lib.rs
View File

@@ -22,4 +22,9 @@ pub mod video;
pub mod web;
pub mod webrtc;
/// Auto-generated secrets module (from secrets.toml at compile time)
pub mod secrets {
include!(concat!(env!("OUT_DIR"), "/secrets_generated.rs"));
}
pub use error::{AppError, Result};

3
src/main.rs
View File

@@ -170,8 +170,7 @@ async fn main() -> anyhow::Result<()> {
resolution: video_resolution,
input_format: video_format,
fps: config.video.fps,
bitrate_kbps: config.stream.bitrate_kbps,
gop_size: config.stream.gop_size,
bitrate_preset: config.stream.bitrate_preset,
encoder_backend: config.stream.encoder.to_backend(),
webrtc: {
let mut stun_servers = vec![];

75
src/rustdesk/config.rs
View File

@@ -5,6 +5,8 @@
use serde::{Deserialize, Serialize};
use typeshare::typeshare;
use crate::secrets;
/// RustDesk configuration
#[typeshare]
#[derive(Debug, Clone, Serialize, Deserialize)]
@@ -15,6 +17,7 @@ pub struct RustDeskConfig {
/// Rendezvous server address (hbbs), e.g., "rs.example.com" or "192.168.1.100"
/// Port defaults to 21116 if not specified
/// If empty, uses the public server from secrets.toml
pub rendezvous_server: String,
/// Relay server address (hbbr), if different from rendezvous server
@@ -70,13 +73,41 @@ impl Default for RustDeskConfig {
impl RustDeskConfig {
/// Check if the configuration is valid for starting the service
/// Returns true if enabled and has a valid server (user-configured or public)
pub fn is_valid(&self) -> bool {
self.enabled
&& !self.rendezvous_server.is_empty()
&& !self.effective_rendezvous_server().is_empty()
&& !self.device_id.is_empty()
&& !self.device_password.is_empty()
}
/// Check if using the public server (user left rendezvous_server empty)
pub fn is_using_public_server(&self) -> bool {
self.rendezvous_server.is_empty() && secrets::rustdesk::has_public_server()
}
/// Get the effective rendezvous server (user-configured or public fallback)
pub fn effective_rendezvous_server(&self) -> &str {
if self.rendezvous_server.is_empty() {
secrets::rustdesk::PUBLIC_SERVER
} else {
&self.rendezvous_server
}
}
/// Get public server info for display (server address and public key)
/// Returns None if no public server is configured
pub fn public_server_info() -> Option<PublicServerInfo> {
if secrets::rustdesk::has_public_server() {
Some(PublicServerInfo {
server: secrets::rustdesk::PUBLIC_SERVER.to_string(),
public_key: secrets::rustdesk::PUBLIC_KEY.to_string(),
})
} else {
None
}
}
/// Generate a new random device ID
pub fn generate_device_id() -> String {
generate_device_id()
@@ -111,10 +142,11 @@ impl RustDeskConfig {
/// Get the rendezvous server address with default port
pub fn rendezvous_addr(&self) -> String {
if self.rendezvous_server.contains(':') {
self.rendezvous_server.clone()
let server = self.effective_rendezvous_server();
if server.contains(':') {
server.to_string()
} else {
format!("{}:21116", self.rendezvous_server)
format!("{}:21116", server)
}
}
@@ -127,9 +159,10 @@ impl RustDeskConfig {
format!("{}:21117", s)
}
}).or_else(|| {
// Default: same host as rendezvous server
if !self.rendezvous_server.is_empty() {
let host = self.rendezvous_server.split(':').next().unwrap_or("");
// Default: same host as effective rendezvous server
let server = self.effective_rendezvous_server();
if !server.is_empty() {
let host = server.split(':').next().unwrap_or("");
if !host.is_empty() {
Some(format!("{}:21117", host))
} else {
@@ -142,6 +175,16 @@ impl RustDeskConfig {
}
}
/// Public server information for display to users
#[derive(Debug, Clone, Serialize, Deserialize)]
#[typeshare]
pub struct PublicServerInfo {
/// Public server address
pub server: String,
/// Public key for client connection
pub public_key: String,
}
/// Generate a random 9-digit device ID
pub fn generate_device_id() -> String {
use rand::Rng;
@@ -196,9 +239,6 @@ mod tests {
fn test_relay_addr() {
let mut config = RustDeskConfig::default();
// No server configured
assert!(config.relay_addr().is_none());
// Rendezvous server configured, relay defaults to same host
config.rendezvous_server = "example.com".to_string();
assert_eq!(config.relay_addr(), Some("example.com:21117".to_string()));
@@ -207,4 +247,19 @@ mod tests {
config.relay_server = Some("relay.example.com".to_string());
assert_eq!(config.relay_addr(), Some("relay.example.com:21117".to_string()));
}
#[test]
fn test_effective_rendezvous_server() {
let mut config = RustDeskConfig::default();
// When user sets a server, use it
config.rendezvous_server = "custom.example.com".to_string();
assert_eq!(config.effective_rendezvous_server(), "custom.example.com");
// When empty, falls back to public server (if configured)
config.rendezvous_server = String::new();
// This will return PUBLIC_SERVER from secrets
let effective = config.effective_rendezvous_server();
assert!(!effective.is_empty() || !secrets::rustdesk::has_public_server());
}
}

60
src/rustdesk/connection.rs
View File

@@ -22,6 +22,7 @@ use tracing::{debug, error, info, warn};
use crate::hid::HidController;
use crate::video::encoder::registry::{EncoderRegistry, VideoEncoderType};
use crate::video::encoder::BitratePreset;
use crate::video::stream_manager::VideoStreamManager;
use super::bytes_codec::{read_frame, write_frame};
@@ -507,7 +508,7 @@ impl Connection {
*self.state.write() = ConnectionState::Active;
// Select the best available video codec
// Priority: VP8 > VP9 > H264 > H265 (VP8/VP9 are more widely supported by software decoders)
// Priority: H264 > H265 > VP8 > VP9 (H264/H265 leverage hardware encoding)
let negotiated = self.negotiate_video_codec();
self.negotiated_codec = Some(negotiated);
info!("Negotiated video codec: {:?}", negotiated);
@@ -519,28 +520,29 @@ impl Connection {
}
/// Negotiate video codec - select the best available encoder
/// Priority: VP8 > VP9 > H264 > H265 (VP8/VP9 have better software decoder support)
/// Priority: H264 > H265 > VP8 > VP9 (H264/H265 leverage hardware encoding on embedded devices)
fn negotiate_video_codec(&self) -> VideoEncoderType {
let registry = EncoderRegistry::global();
// Check availability in priority order
// VP8 is preferred because it has the best compatibility with software decoders
if registry.is_format_available(VideoEncoderType::VP8, false) {
return VideoEncoderType::VP8;
}
if registry.is_format_available(VideoEncoderType::VP9, false) {
return VideoEncoderType::VP9;
}
// H264 is preferred because it has the best hardware encoder support (RKMPP, VAAPI, etc.)
// and most RustDesk clients support H264 hardware decoding
if registry.is_format_available(VideoEncoderType::H264, false) {
return VideoEncoderType::H264;
}
if registry.is_format_available(VideoEncoderType::H265, false) {
return VideoEncoderType::H265;
}
if registry.is_format_available(VideoEncoderType::VP8, false) {
return VideoEncoderType::VP8;
}
if registry.is_format_available(VideoEncoderType::VP9, false) {
return VideoEncoderType::VP9;
}
// Fallback to VP8 (should always be available via libvpx)
warn!("No video encoder available, defaulting to VP8");
VideoEncoderType::VP8
// Fallback to H264 (should be available via hardware or software encoder)
warn!("No video encoder available, defaulting to H264");
VideoEncoderType::H264
}
/// Handle misc message with Arc writer
@@ -575,8 +577,30 @@ impl Connection {
Ok(())
}
/// Handle Option message from client (includes codec preference)
/// Handle Option message from client (includes codec and quality preferences)
async fn handle_option_message(&mut self, opt: &hbb::OptionMessage) -> anyhow::Result<()> {
// Handle image quality preset
// RustDesk ImageQuality: NotSet=0, Low=2, Balanced=3, Best=4
// Map to One-KVM BitratePreset: Low->Speed, Balanced->Balanced, Best->Quality
let image_quality = opt.image_quality;
if image_quality != 0 {
let preset = match image_quality {
2 => Some(BitratePreset::Speed), // Low -> Speed (1 Mbps)
3 => Some(BitratePreset::Balanced), // Balanced -> Balanced (4 Mbps)
4 => Some(BitratePreset::Quality), // Best -> Quality (8 Mbps)
_ => None,
};
if let Some(preset) = preset {
info!("Client requested quality preset: {:?} (image_quality={})", preset, image_quality);
if let Some(ref video_manager) = self.video_manager {
if let Err(e) = video_manager.set_bitrate_preset(preset).await {
warn!("Failed to set bitrate preset: {}", e);
}
}
}
}
// Check if client sent supported_decoding with a codec preference
if let Some(ref supported_decoding) = opt.supported_decoding {
let prefer = supported_decoding.prefer;
@@ -616,9 +640,9 @@ impl Connection {
}
}
// Log other options for debugging
// Log custom_image_quality (accept but don't process)
if opt.custom_image_quality > 0 {
debug!("Client requested image quality: {}", opt.custom_image_quality);
debug!("Client sent custom_image_quality: {} (ignored)", opt.custom_image_quality);
}
if opt.custom_fps > 0 {
debug!("Client requested FPS: {}", opt.custom_fps);
@@ -665,7 +689,7 @@ impl Connection {
let state = self.state.clone();
let conn_id = self.id;
let shutdown_tx = self.shutdown_tx.clone();
let negotiated_codec = self.negotiated_codec.unwrap_or(VideoEncoderType::VP8);
let negotiated_codec = self.negotiated_codec.unwrap_or(VideoEncoderType::H264);
let task = tokio::spawn(async move {
info!("Starting video streaming for connection {} with codec {:?}", conn_id, negotiated_codec);
@@ -1298,12 +1322,12 @@ async fn run_video_streaming(
// Get encoding config for logging
if let Some(config) = video_manager.get_encoding_config().await {
info!(
"RustDesk connection {} using shared video pipeline: {:?} {}x{} @ {} kbps",
"RustDesk connection {} using shared video pipeline: {:?} {}x{} @ {}",
conn_id,
config.output_codec,
config.resolution.width,
config.resolution.height,
config.bitrate_kbps
config.bitrate_preset
);
}

2
src/video/encoder/mod.rs
View File

@@ -19,7 +19,7 @@ pub mod vp8;
pub mod vp9;
// Core traits and types
pub use traits::{EncodedFormat, EncodedFrame, Encoder, EncoderConfig, EncoderFactory};
pub use traits::{BitratePreset, EncodedFormat, EncodedFrame, Encoder, EncoderConfig, EncoderFactory};
// WebRTC codec abstraction
pub use codec::{CodecFrame, VideoCodec, VideoCodecConfig, VideoCodecFactory, VideoCodecType};

85
src/video/encoder/traits.rs
View File

@@ -1,11 +1,96 @@
//! Encoder traits and common types
use bytes::Bytes;
use serde::{Deserialize, Serialize};
use std::time::Instant;
use typeshare::typeshare;
use crate::video::format::{PixelFormat, Resolution};
use crate::error::Result;
/// Bitrate preset for video encoding
///
/// Simplifies bitrate configuration by providing three intuitive presets
/// plus a custom option for advanced users.
#[typeshare]
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "type", content = "value")]
pub enum BitratePreset {
/// Speed priority: 1 Mbps, lowest latency, smaller GOP
/// Best for: slow networks, remote management, low-bandwidth scenarios
Speed,
/// Balanced: 4 Mbps, good quality/latency tradeoff
/// Best for: typical usage, recommended default
Balanced,
/// Quality priority: 8 Mbps, best visual quality
/// Best for: local network, high-bandwidth scenarios, detailed work
Quality,
/// Custom bitrate in kbps (for advanced users)
Custom(u32),
}
impl BitratePreset {
/// Get bitrate value in kbps
pub fn bitrate_kbps(&self) -> u32 {
match self {
Self::Speed => 1000,
Self::Balanced => 4000,
Self::Quality => 8000,
Self::Custom(kbps) => *kbps,
}
}
/// Get recommended GOP size based on preset
///
/// Speed preset uses shorter GOP for faster recovery from packet loss.
/// Quality preset uses longer GOP for better compression efficiency.
pub fn gop_size(&self, fps: u32) -> u32 {
match self {
Self::Speed => (fps / 2).max(15), // 0.5 second, minimum 15 frames
Self::Balanced => fps, // 1 second
Self::Quality => fps * 2, // 2 seconds
Self::Custom(_) => fps, // Default 1 second for custom
}
}
/// Get quality preset name for encoder configuration
pub fn quality_level(&self) -> &'static str {
match self {
Self::Speed => "low", // ultrafast/veryfast preset
Self::Balanced => "medium", // medium preset
Self::Quality => "high", // slower preset, better quality
Self::Custom(_) => "medium",
}
}
/// Create from kbps value, mapping to nearest preset or Custom
pub fn from_kbps(kbps: u32) -> Self {
match kbps {
0..=1500 => Self::Speed,
1501..=6000 => Self::Balanced,
6001..=10000 => Self::Quality,
_ => Self::Custom(kbps),
}
}
}
impl Default for BitratePreset {
fn default() -> Self {
Self::Balanced
}
}
impl std::fmt::Display for BitratePreset {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Speed => write!(f, "Speed (1 Mbps)"),
Self::Balanced => write!(f, "Balanced (4 Mbps)"),
Self::Quality => write!(f, "Quality (8 Mbps)"),
Self::Custom(kbps) => write!(f, "Custom ({} kbps)", kbps),
}
}
}
/// Encoder configuration
#[derive(Debug, Clone)]
pub struct EncoderConfig {

2
src/video/mod.rs
View File

@@ -10,6 +10,7 @@ pub mod encoder;
pub mod format;
pub mod frame;
pub mod h264_pipeline;
pub mod pacer;
pub mod shared_video_pipeline;
pub mod stream_manager;
pub mod streamer;
@@ -18,6 +19,7 @@ pub mod video_session;
pub use capture::VideoCapturer;
pub use convert::{MjpegDecoder, MjpegToYuv420Converter, PixelConverter, Yuv420pBuffer};
pub use decoder::{MjpegVaapiDecoder, MjpegVaapiDecoderConfig};
pub use pacer::{EncoderPacer, PacerStats};
pub use device::{VideoDevice, VideoDeviceInfo};
pub use encoder::{JpegEncoder, H264Encoder, H264EncoderType};
pub use format::PixelFormat;

72
src/video/pacer.rs Normal file
View File

@@ -0,0 +1,72 @@
//! Encoder Pacer - Placeholder for future backpressure control
//!
//! Currently a pass-through that allows all frames.
//! TODO: Implement effective backpressure control.
use std::sync::atomic::{AtomicU64, Ordering};
use tracing::debug;
/// Encoder pacing statistics
#[derive(Debug, Clone, Default)]
pub struct PacerStats {
/// Total frames processed
pub frames_processed: u64,
/// Frames skipped (currently always 0)
pub frames_skipped: u64,
/// Keyframes processed
pub keyframes_processed: u64,
}
/// Encoder pacer (currently pass-through)
///
/// This is a placeholder for future backpressure control.
/// Currently allows all frames through without throttling.
pub struct EncoderPacer {
frames_processed: AtomicU64,
keyframes_processed: AtomicU64,
}
impl EncoderPacer {
/// Create a new encoder pacer
pub fn new(_max_in_flight: usize) -> Self {
debug!("Creating encoder pacer (pass-through mode)");
Self {
frames_processed: AtomicU64::new(0),
keyframes_processed: AtomicU64::new(0),
}
}
/// Check if encoding should proceed (always returns true)
pub async fn should_encode(&self, is_keyframe: bool) -> bool {
self.frames_processed.fetch_add(1, Ordering::Relaxed);
if is_keyframe {
self.keyframes_processed.fetch_add(1, Ordering::Relaxed);
}
true // Always allow encoding
}
/// Report lag from receiver (currently no-op)
pub async fn report_lag(&self, _frames_lagged: u64) {
// TODO: Implement effective backpressure control
// Currently this is a no-op
}
/// Check if throttling (always false)
pub fn is_throttling(&self) -> bool {
false
}
/// Get pacer statistics
pub fn stats(&self) -> PacerStats {
PacerStats {
frames_processed: self.frames_processed.load(Ordering::Relaxed),
frames_skipped: 0,
keyframes_processed: self.keyframes_processed.load(Ordering::Relaxed),
}
}
/// Get in-flight count (always 0)
pub fn in_flight(&self) -> usize {
0
}
}

138
src/video/shared_video_pipeline.rs
View File

@@ -37,6 +37,7 @@ use crate::video::encoder::vp8::{VP8Config, VP8Encoder};
use crate::video::encoder::vp9::{VP9Config, VP9Encoder};
use crate::video::format::{PixelFormat, Resolution};
use crate::video::frame::VideoFrame;
use crate::video::pacer::EncoderPacer;
/// Encoded video frame for distribution
#[derive(Debug, Clone)]
@@ -64,14 +65,14 @@ pub struct SharedVideoPipelineConfig {
pub input_format: PixelFormat,
/// Output codec type
pub output_codec: VideoEncoderType,
/// Target bitrate in kbps
pub bitrate_kbps: u32,
/// Bitrate preset (replaces raw bitrate_kbps)
pub bitrate_preset: crate::video::encoder::BitratePreset,
/// Target FPS
pub fps: u32,
/// GOP size
pub gop_size: u32,
/// Encoder backend (None = auto select best available)
pub encoder_backend: Option<EncoderBackend>,
/// Maximum in-flight frames for backpressure control
pub max_in_flight_frames: usize,
}
impl Default for SharedVideoPipelineConfig {
@@ -80,54 +81,70 @@ impl Default for SharedVideoPipelineConfig {
resolution: Resolution::HD720,
input_format: PixelFormat::Yuyv,
output_codec: VideoEncoderType::H264,
bitrate_kbps: 1000,
bitrate_preset: crate::video::encoder::BitratePreset::Balanced,
fps: 30,
gop_size: 30,
encoder_backend: None,
max_in_flight_frames: 8, // Default: allow 8 frames in flight
}
}
}
impl SharedVideoPipelineConfig {
/// Create H264 config
pub fn h264(resolution: Resolution, bitrate_kbps: u32) -> Self {
/// Get effective bitrate in kbps
pub fn bitrate_kbps(&self) -> u32 {
self.bitrate_preset.bitrate_kbps()
}
/// Get effective GOP size
pub fn gop_size(&self) -> u32 {
self.bitrate_preset.gop_size(self.fps)
}
/// Create H264 config with bitrate preset
pub fn h264(resolution: Resolution, preset: crate::video::encoder::BitratePreset) -> Self {
Self {
resolution,
output_codec: VideoEncoderType::H264,
bitrate_kbps,
bitrate_preset: preset,
..Default::default()
}
}
/// Create H265 config
pub fn h265(resolution: Resolution, bitrate_kbps: u32) -> Self {
/// Create H265 config with bitrate preset
pub fn h265(resolution: Resolution, preset: crate::video::encoder::BitratePreset) -> Self {
Self {
resolution,
output_codec: VideoEncoderType::H265,
bitrate_kbps,
bitrate_preset: preset,
..Default::default()
}
}
/// Create VP8 config
pub fn vp8(resolution: Resolution, bitrate_kbps: u32) -> Self {
/// Create VP8 config with bitrate preset
pub fn vp8(resolution: Resolution, preset: crate::video::encoder::BitratePreset) -> Self {
Self {
resolution,
output_codec: VideoEncoderType::VP8,
bitrate_kbps,
bitrate_preset: preset,
..Default::default()
}
}
/// Create VP9 config
pub fn vp9(resolution: Resolution, bitrate_kbps: u32) -> Self {
/// Create VP9 config with bitrate preset
pub fn vp9(resolution: Resolution, preset: crate::video::encoder::BitratePreset) -> Self {
Self {
resolution,
output_codec: VideoEncoderType::VP9,
bitrate_kbps,
bitrate_preset: preset,
..Default::default()
}
}
/// Create config with legacy bitrate_kbps (for compatibility during migration)
pub fn with_bitrate_kbps(mut self, bitrate_kbps: u32) -> Self {
self.bitrate_preset = crate::video::encoder::BitratePreset::from_kbps(bitrate_kbps);
self
}
}
/// Pipeline statistics
@@ -136,12 +153,16 @@ pub struct SharedVideoPipelineStats {
pub frames_captured: u64,
pub frames_encoded: u64,
pub frames_dropped: u64,
/// Frames skipped due to backpressure (pacer)
pub frames_skipped: u64,
pub bytes_encoded: u64,
pub keyframes_encoded: u64,
pub avg_encode_time_ms: f32,
pub current_fps: f32,
pub errors: u64,
pub subscribers: u64,
/// Current number of frames in-flight (waiting to be sent)
pub pending_frames: usize,
}
@@ -305,18 +326,21 @@ pub struct SharedVideoPipeline {
/// Pipeline start time for PTS calculation (epoch millis, 0 = not set)
/// Uses AtomicI64 instead of Mutex for lock-free access
pipeline_start_time_ms: AtomicI64,
/// Encoder pacer for backpressure control
pacer: EncoderPacer,
}
impl SharedVideoPipeline {
/// Create a new shared video pipeline
pub fn new(config: SharedVideoPipelineConfig) -> Result<Arc<Self>> {
info!(
"Creating shared video pipeline: {} {}x{} @ {} kbps (input: {})",
"Creating shared video pipeline: {} {}x{} @ {} (input: {}, max_in_flight: {})",
config.output_codec,
config.resolution.width,
config.resolution.height,
config.bitrate_kbps,
config.input_format
config.bitrate_preset,
config.input_format,
config.max_in_flight_frames
);
let (frame_tx, _) = broadcast::channel(16); // Reduced from 64 for lower latency
@@ -324,6 +348,9 @@ impl SharedVideoPipeline {
let nv12_size = (config.resolution.width * config.resolution.height * 3 / 2) as usize;
let yuv420p_size = nv12_size; // Same size as NV12
// Create pacer for backpressure control
let pacer = EncoderPacer::new(config.max_in_flight_frames);
let pipeline = Arc::new(Self {
config: RwLock::new(config),
encoder: Mutex::new(None),
@@ -342,6 +369,7 @@ impl SharedVideoPipeline {
sequence: AtomicU64::new(0),
keyframe_requested: AtomicBool::new(false),
pipeline_start_time_ms: AtomicI64::new(0),
pacer,
});
Ok(pipeline)
@@ -379,9 +407,9 @@ impl SharedVideoPipeline {
};
let encoder_config = H264Config {
base: EncoderConfig::h264(config.resolution, config.bitrate_kbps),
bitrate_kbps: config.bitrate_kbps,
gop_size: config.gop_size,
base: EncoderConfig::h264(config.resolution, config.bitrate_kbps()),
bitrate_kbps: config.bitrate_kbps(),
gop_size: config.gop_size(),
fps: config.fps,
input_format: h264_input_format,
};
@@ -413,9 +441,9 @@ impl SharedVideoPipeline {
VideoEncoderType::H265 => {
// Determine H265 input format based on backend and input format
let encoder_config = if use_yuyv_direct {
H265Config::low_latency_yuyv422(config.resolution, config.bitrate_kbps)
H265Config::low_latency_yuyv422(config.resolution, config.bitrate_kbps())
} else {
H265Config::low_latency(config.resolution, config.bitrate_kbps)
H265Config::low_latency(config.resolution, config.bitrate_kbps())
};
let encoder = if use_yuyv_direct {
@@ -441,7 +469,7 @@ impl SharedVideoPipeline {
Box::new(H265EncoderWrapper(encoder))
}
VideoEncoderType::VP8 => {
let encoder_config = VP8Config::low_latency(config.resolution, config.bitrate_kbps);
let encoder_config = VP8Config::low_latency(config.resolution, config.bitrate_kbps());
let encoder = if let Some(ref backend) = config.encoder_backend {
let codec_name = get_codec_name(VideoEncoderType::VP8, Some(*backend))
@@ -458,7 +486,7 @@ impl SharedVideoPipeline {
Box::new(VP8EncoderWrapper(encoder))
}
VideoEncoderType::VP9 => {
let encoder_config = VP9Config::low_latency(config.resolution, config.bitrate_kbps);
let encoder_config = VP9Config::low_latency(config.resolution, config.bitrate_kbps());
let encoder = if let Some(ref backend) = config.encoder_backend {
let codec_name = get_codec_name(VideoEncoderType::VP9, Some(*backend))
@@ -589,6 +617,19 @@ impl SharedVideoPipeline {
self.frame_tx.receiver_count()
}
/// Report that a receiver has lagged behind
///
/// Call this when a broadcast receiver detects it has fallen behind
/// (e.g., when RecvError::Lagged is received). This triggers throttle
/// mode in the encoder to reduce encoding rate.
///
/// # Arguments
///
/// * `frames_lagged` - Number of frames the receiver has lagged
pub async fn report_lag(&self, frames_lagged: u64) {
self.pacer.report_lag(frames_lagged).await;
}
/// Request encoder to produce a keyframe on next encode
///
/// This is useful when a new client connects and needs an immediate
@@ -604,9 +645,15 @@ impl SharedVideoPipeline {
pub async fn stats(&self) -> SharedVideoPipelineStats {
let mut stats = self.stats.lock().await.clone();
stats.subscribers = self.frame_tx.receiver_count() as u64;
stats.pending_frames = if self.pacer.is_throttling() { 1 } else { 0 };
stats
}
/// Get pacer statistics for debugging
pub fn pacer_stats(&self) -> crate::video::pacer::PacerStats {
self.pacer.stats()
}
/// Check if running
pub fn is_running(&self) -> bool {
*self.running_rx.borrow()
@@ -662,7 +709,8 @@ impl SharedVideoPipeline {
let _ = self.running.send(true);
let config = self.config.read().await.clone();
info!("Starting {} pipeline", config.output_codec);
let gop_size = config.gop_size();
info!("Starting {} pipeline (GOP={})", config.output_codec, gop_size);
let pipeline = self.clone();
@@ -678,6 +726,7 @@ impl SharedVideoPipeline {
let mut local_keyframes: u64 = 0;
let mut local_errors: u64 = 0;
let mut local_dropped: u64 = 0;
let mut local_skipped: u64 = 0;
// Track when we last had subscribers for auto-stop feature
let mut no_subscribers_since: Option<Instant> = None;
@@ -728,8 +777,18 @@ impl SharedVideoPipeline {
}
}
// === Lag-feedback based flow control ===
// Check if this is a keyframe interval
let is_keyframe_interval = frame_count % gop_size as u64 == 0;
// Note: pacer.should_encode() currently always returns true
// TODO: Implement effective backpressure control
let _ = pipeline.pacer.should_encode(is_keyframe_interval).await;
match pipeline.encode_frame(&video_frame, frame_count).await {
Ok(Some(encoded_frame)) => {
// Send frame to all subscribers
// Note: broadcast::send is non-blocking
let _ = pipeline.frame_tx.send(encoded_frame.clone());
// Update local counters (no lock)
@@ -762,6 +821,8 @@ impl SharedVideoPipeline {
s.keyframes_encoded += local_keyframes;
s.errors += local_errors;
s.frames_dropped += local_dropped;
s.frames_skipped += local_skipped;
s.pending_frames = if pipeline.pacer.is_throttling() { 1 } else { 0 };
s.current_fps = current_fps;
// Reset local counters
@@ -770,6 +831,7 @@ impl SharedVideoPipeline {
local_keyframes = 0;
local_errors = 0;
local_dropped = 0;
local_skipped = 0;
}
}
Err(broadcast::error::RecvError::Lagged(n)) => {
@@ -958,15 +1020,22 @@ impl SharedVideoPipeline {
}
}
/// Set bitrate
pub async fn set_bitrate(&self, bitrate_kbps: u32) -> Result<()> {
/// Set bitrate using preset
pub async fn set_bitrate_preset(&self, preset: crate::video::encoder::BitratePreset) -> Result<()> {
let bitrate_kbps = preset.bitrate_kbps();
if let Some(ref mut encoder) = *self.encoder.lock().await {
encoder.set_bitrate(bitrate_kbps)?;
self.config.write().await.bitrate_kbps = bitrate_kbps;
self.config.write().await.bitrate_preset = preset;
}
Ok(())
}
/// Set bitrate using raw kbps value (converts to appropriate preset)
pub async fn set_bitrate(&self, bitrate_kbps: u32) -> Result<()> {
let preset = crate::video::encoder::BitratePreset::from_kbps(bitrate_kbps);
self.set_bitrate_preset(preset).await
}
/// Get current config
pub async fn config(&self) -> SharedVideoPipelineConfig {
self.config.read().await.clone()
@@ -1038,13 +1107,14 @@ fn parse_h265_nal_types(data: &[u8]) -> Vec<(u8, usize)> {
#[cfg(test)]
mod tests {
use super::*;
use crate::video::encoder::BitratePreset;
#[test]
fn test_pipeline_config() {
let h264 = SharedVideoPipelineConfig::h264(Resolution::HD1080, 4000);
let h264 = SharedVideoPipelineConfig::h264(Resolution::HD1080, BitratePreset::Balanced);
assert_eq!(h264.output_codec, VideoEncoderType::H264);
let h265 = SharedVideoPipelineConfig::h265(Resolution::HD720, 2000);
let h265 = SharedVideoPipelineConfig::h265(Resolution::HD720, BitratePreset::Speed);
assert_eq!(h265.output_codec, VideoEncoderType::H265);
}
}

8
src/video/stream_manager.rs
View File

@@ -613,6 +613,14 @@ impl VideoStreamManager {
self.webrtc_streamer.set_video_codec(codec).await
}
/// Set bitrate preset for the shared video pipeline
///
/// This allows external consumers (like RustDesk) to adjust the video quality
/// based on client preferences.
pub async fn set_bitrate_preset(&self, preset: crate::video::encoder::BitratePreset) -> crate::error::Result<()> {
self.webrtc_streamer.set_bitrate_preset(preset).await
}
/// Publish event to event bus
async fn publish_event(&self, event: SystemEvent) {
if let Some(ref events) = *self.events.read().await {

11
src/video/video_session.rs
View File

@@ -13,6 +13,7 @@ use tokio::sync::{broadcast, RwLock};
use tracing::{debug, info, warn};
use super::encoder::registry::{EncoderBackend, EncoderRegistry, VideoEncoderType};
use super::encoder::BitratePreset;
use super::format::Resolution;
use super::frame::VideoFrame;
use super::shared_video_pipeline::{
@@ -123,8 +124,8 @@ pub struct VideoSessionManagerConfig {
pub default_codec: VideoEncoderType,
/// Default resolution
pub resolution: Resolution,
/// Default bitrate (kbps)
pub bitrate_kbps: u32,
/// Bitrate preset
pub bitrate_preset: BitratePreset,
/// Default FPS
pub fps: u32,
/// Session timeout (seconds)
@@ -138,7 +139,7 @@ impl Default for VideoSessionManagerConfig {
Self {
default_codec: VideoEncoderType::H264,
resolution: Resolution::HD720,
bitrate_kbps: 8000,
bitrate_preset: BitratePreset::Balanced,
fps: 30,
session_timeout_secs: 300,
encoder_backend: None,
@@ -325,10 +326,10 @@ impl VideoSessionManager {
resolution: self.config.resolution,
input_format: crate::video::format::PixelFormat::Mjpeg, // Common input
output_codec: codec,
bitrate_kbps: self.config.bitrate_kbps,
bitrate_preset: self.config.bitrate_preset,
fps: self.config.fps,
gop_size: 30,
encoder_backend: self.config.encoder_backend,
..Default::default()
};
// Create new pipeline

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

@@ -109,11 +109,11 @@ pub async fn apply_stream_config(
}
// 更新码率
if old_config.bitrate_kbps != new_config.bitrate_kbps {
if old_config.bitrate_preset != new_config.bitrate_preset {
state
.stream_manager
.webrtc_streamer()
.set_bitrate(new_config.bitrate_kbps)
.set_bitrate_preset(new_config.bitrate_preset)
.await
.ok(); // Ignore error if no active stream
}
@@ -143,9 +143,9 @@ pub async fn apply_stream_config(
}
tracing::info!(
"Stream config applied: encoder={:?}, bitrate={} kbps",
"Stream config applied: encoder={:?}, bitrate={}",
new_config.encoder,
new_config.bitrate_kbps
new_config.bitrate_preset
);
Ok(())
}

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

@@ -4,7 +4,7 @@ use axum::{extract::State, Json};
use std::sync::Arc;
use crate::error::Result;
use crate::rustdesk::config::RustDeskConfig;
use crate::rustdesk::config::{PublicServerInfo, RustDeskConfig};
use crate::state::AppState;
use super::apply::apply_rustdesk_config;
@@ -21,6 +21,8 @@ pub struct RustDeskConfigResponse {
pub has_password: bool,
/// 是否已设置密钥对
pub has_keypair: bool,
/// 是否使用公共服务器（用户留空时）
pub using_public_server: bool,
}
impl From<&RustDeskConfig> for RustDeskConfigResponse {
@@ -32,6 +34,7 @@ impl From<&RustDeskConfig> for RustDeskConfigResponse {
device_id: config.device_id.clone(),
has_password: !config.device_password.is_empty(),
has_keypair: config.public_key.is_some() && config.private_key.is_some(),
using_public_server: config.is_using_public_server(),
}
}
}
@@ -42,6 +45,8 @@ pub struct RustDeskStatusResponse {
pub config: RustDeskConfigResponse,
pub service_status: String,
pub rendezvous_status: Option<String>,
/// 公共服务器信息（仅当有公共服务器配置时返回）
pub public_server: Option<PublicServerInfo>,
}
/// 获取 RustDesk 配置
@@ -65,10 +70,14 @@ pub async fn get_rustdesk_status(State(state): State<Arc<AppState>>) -> Json<Rus
}
};
// 获取公共服务器信息
let public_server = RustDeskConfig::public_server_info();
Json(RustDeskStatusResponse {
config: RustDeskConfigResponse::from(&config),
service_status,
rendezvous_status,
public_server,
})
}

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

@@ -3,6 +3,7 @@ use typeshare::typeshare;
use crate::config::*;
use crate::error::AppError;
use crate::rustdesk::config::RustDeskConfig;
use crate::video::encoder::BitratePreset;
// ===== Video Config =====
#[typeshare]
@@ -71,8 +72,7 @@ impl VideoConfigUpdate {
pub struct StreamConfigResponse {
pub mode: StreamMode,
pub encoder: EncoderType,
pub bitrate_kbps: u32,
pub gop_size: u32,
pub bitrate_preset: BitratePreset,
pub stun_server: Option<String>,
pub turn_server: Option<String>,
pub turn_username: Option<String>,
@@ -85,8 +85,7 @@ impl From<&StreamConfig> for StreamConfigResponse {
Self {
mode: config.mode.clone(),
encoder: config.encoder.clone(),
bitrate_kbps: config.bitrate_kbps,
gop_size: config.gop_size,
bitrate_preset: config.bitrate_preset,
stun_server: config.stun_server.clone(),
turn_server: config.turn_server.clone(),
turn_username: config.turn_username.clone(),
@@ -100,8 +99,7 @@ impl From<&StreamConfig> for StreamConfigResponse {
pub struct StreamConfigUpdate {
pub mode: Option<StreamMode>,
pub encoder: Option<EncoderType>,
pub bitrate_kbps: Option<u32>,
pub gop_size: Option<u32>,
pub bitrate_preset: Option<BitratePreset>,
/// STUN server URL (e.g., "stun:stun.l.google.com:19302")
pub stun_server: Option<String>,
/// TURN server URL (e.g., "turn:turn.example.com:3478")
@@ -114,16 +112,7 @@ pub struct StreamConfigUpdate {
impl StreamConfigUpdate {
pub fn validate(&self) -> crate::error::Result<()> {
if let Some(bitrate) = self.bitrate_kbps {
if !(1000..=15000).contains(&bitrate) {
return Err(AppError::BadRequest("Bitrate must be 1000-15000 kbps".into()));
}
}
if let Some(gop) = self.gop_size {
if !(10..=300).contains(&gop) {
return Err(AppError::BadRequest("GOP size must be 10-300".into()));
}
}
// BitratePreset is always valid (enum)
// Validate STUN server format
if let Some(ref stun) = self.stun_server {
if !stun.is_empty() && !stun.starts_with("stun:") {
@@ -150,11 +139,8 @@ impl StreamConfigUpdate {
if let Some(encoder) = self.encoder.clone() {
config.encoder = encoder;
}
if let Some(bitrate) = self.bitrate_kbps {
config.bitrate_kbps = bitrate;
}
if let Some(gop) = self.gop_size {
config.gop_size = gop;
if let Some(preset) = self.bitrate_preset {
config.bitrate_preset = preset;
}
// STUN/TURN settings - empty string means clear, Some("value") means set
if let Some(ref stun) = self.stun_server {

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

@@ -13,6 +13,7 @@ use crate::auth::{Session, SESSION_COOKIE};
use crate::config::{AppConfig, StreamMode};
use crate::error::{AppError, Result};
use crate::state::AppState;
use crate::video::encoder::BitratePreset;
// ============================================================================
// Health & Info
@@ -742,12 +743,12 @@ pub async fn update_config(
state
.stream_manager
.webrtc_streamer()
.set_bitrate(new_config.stream.bitrate_kbps)
.set_bitrate_preset(new_config.stream.bitrate_preset)
.await
.ok(); // Ignore error if no active stream
tracing::info!("Stream config applied: encoder={:?}, bitrate={} kbps",
new_config.stream.encoder, new_config.stream.bitrate_kbps);
tracing::info!("Stream config applied: encoder={:?}, bitrate={}",
new_config.stream.encoder, new_config.stream.bitrate_preset);
}
// HID config processing - always reload if section was sent
@@ -1191,7 +1192,7 @@ pub struct AvailableCodecsResponse {
/// Set bitrate request
#[derive(Deserialize)]
pub struct SetBitrateRequest {
pub bitrate_kbps: u32,
pub bitrate_preset: BitratePreset,
}
/// Set stream bitrate (real-time adjustment)
@@ -1199,19 +1200,11 @@ pub async fn stream_set_bitrate(
State(state): State<Arc<AppState>>,
Json(req): Json<SetBitrateRequest>,
) -> Result<Json<LoginResponse>> {
// Validate bitrate range (1000-15000 kbps)
if req.bitrate_kbps < 1000 || req.bitrate_kbps > 15000 {
return Err(AppError::BadRequest(format!(
"Bitrate must be between 1000 and 15000 kbps, got {}",
req.bitrate_kbps
)));
}
// Update config
state
.config
.update(|config| {
config.stream.bitrate_kbps = req.bitrate_kbps;
config.stream.bitrate_preset = req.bitrate_preset;
})
.await?;
@@ -1219,18 +1212,18 @@ pub async fn stream_set_bitrate(
if let Err(e) = state
.stream_manager
.webrtc_streamer()
.set_bitrate(req.bitrate_kbps)
.set_bitrate_preset(req.bitrate_preset)
.await
{
warn!("Failed to set bitrate dynamically: {}", e);
// Don't fail the request - config is saved, will apply on next connection
} else {
info!("Bitrate updated to {} kbps", req.bitrate_kbps);
info!("Bitrate updated to {}", req.bitrate_preset);
}
Ok(Json(LoginResponse {
success: true,
message: Some(format!("Bitrate set to {} kbps", req.bitrate_kbps)),
message: Some(format!("Bitrate set to {}", req.bitrate_preset)),
}))
}

12
src/webrtc/universal_session.rs
View File

@@ -30,6 +30,7 @@ use crate::error::{AppError, Result};
use crate::hid::datachannel::{parse_hid_message, HidChannelEvent};
use crate::hid::HidController;
use crate::video::encoder::registry::VideoEncoderType;
use crate::video::encoder::BitratePreset;
use crate::video::format::{PixelFormat, Resolution};
use crate::video::shared_video_pipeline::EncodedVideoFrame;
@@ -47,12 +48,10 @@ pub struct UniversalSessionConfig {
pub resolution: Resolution,
/// Input pixel format
pub input_format: PixelFormat,
/// Target bitrate in kbps
pub bitrate_kbps: u32,
/// Bitrate preset
pub bitrate_preset: BitratePreset,
/// Target FPS
pub fps: u32,
/// GOP size
pub gop_size: u32,
/// Enable audio track
pub audio_enabled: bool,
}
@@ -64,9 +63,8 @@ impl Default for UniversalSessionConfig {
codec: VideoEncoderType::H264,
resolution: Resolution::HD720,
input_format: PixelFormat::Mjpeg,
bitrate_kbps: 1000,
bitrate_preset: BitratePreset::Balanced,
fps: 30,
gop_size: 30,
audio_enabled: false,
}
}
@@ -144,7 +142,7 @@ impl UniversalSession {
stream_id: "one-kvm-stream".to_string(),
codec: video_codec,
resolution: config.resolution,
bitrate_kbps: config.bitrate_kbps,
bitrate_kbps: config.bitrate_preset.bitrate_kbps(),
fps: config.fps,
};
let video_track = Arc::new(UniversalVideoTrack::new(track_config));

151
src/webrtc/video_track.rs
View File

@@ -17,12 +17,10 @@
//! ```
use bytes::Bytes;
use std::io::Cursor;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Mutex;
use tracing::{debug, trace, warn};
use webrtc::media::io::h264_reader::H264Reader;
use webrtc::media::Sample;
use webrtc::rtp_transceiver::rtp_codec::RTCRtpCodecCapability;
use webrtc::track::track_local::track_local_static_rtp::TrackLocalStaticRTP;
@@ -201,18 +199,6 @@ pub struct VideoTrackStats {
pub errors: u64,
}
/// Cached codec parameters for H264/H265
#[derive(Debug, Default)]
struct CachedParams {
/// H264: SPS, H265: VPS
#[allow(dead_code)]
vps: Option<Bytes>,
/// SPS (both H264 and H265)
sps: Option<Bytes>,
/// PPS (both H264 and H265)
pps: Option<Bytes>,
}
/// Track type wrapper to support different underlying track implementations
enum TrackType {
/// Sample-based track with built-in payloader (H264, VP8, VP9)
@@ -243,8 +229,6 @@ pub struct UniversalVideoTrack {
config: UniversalVideoTrackConfig,
/// Statistics
stats: Mutex<VideoTrackStats>,
/// Cached parameters for H264/H265
cached_params: Mutex<CachedParams>,
/// H265 RTP state (only used for H265)
h265_state: Option<Mutex<H265RtpState>>,
}
@@ -294,7 +278,6 @@ impl UniversalVideoTrack {
codec: config.codec,
config,
stats: Mutex::new(VideoTrackStats::default()),
cached_params: Mutex::new(CachedParams::default()),
h265_state,
}
}
@@ -341,71 +324,43 @@ impl UniversalVideoTrack {
}
/// Write H264 frame (Annex B format)
///
/// Sends the entire Annex B frame as a single Sample to allow the
/// H264Payloader to aggregate SPS+PPS into STAP-A packets.
async fn write_h264_frame(&self, data: &[u8], is_keyframe: bool) -> Result<()> {
let cursor = Cursor::new(data);
let mut h264_reader = H264Reader::new(cursor, 1024 * 1024);
// Send entire Annex B frame as one Sample
// The H264Payloader in rtp crate will:
// 1. Parse NAL units from Annex B format
// 2. Cache SPS and PPS
// 3. Aggregate SPS+PPS+IDR into STAP-A when possible
// 4. Fragment large NALs using FU-A
let frame_duration = Duration::from_micros(1_000_000 / self.config.fps.max(1) as u64);
let sample = Sample {
data: Bytes::copy_from_slice(data),
duration: frame_duration,
..Default::default()
};
let mut nals: Vec<Bytes> = Vec::new();
let mut has_sps = false;
let mut has_pps = false;
let mut has_idr = false;
// Parse NAL units
while let Ok(nal) = h264_reader.next_nal() {
if nal.data.is_empty() {
continue;
}
let nal_type = nal.data[0] & 0x1F;
// Skip AUD (9) and filler (12)
if nal_type == 9 || nal_type == 12 {
continue;
}
match nal_type {
5 => has_idr = true,
7 => {
has_sps = true;
self.cached_params.lock().await.sps = Some(nal.data.clone().freeze());
}
8 => {
has_pps = true;
self.cached_params.lock().await.pps = Some(nal.data.clone().freeze());
}
_ => {}
}
nals.push(nal.data.freeze());
}
// Inject cached SPS/PPS before IDR if missing
if has_idr && (!has_sps || !has_pps) {
let mut injected: Vec<Bytes> = Vec::new();
let params = self.cached_params.lock().await;
if !has_sps {
if let Some(ref sps) = params.sps {
debug!("Injecting cached H264 SPS");
injected.push(sps.clone());
match &self.track {
TrackType::Sample(track) => {
if let Err(e) = track.write_sample(&sample).await {
debug!("H264 write_sample failed: {}", e);
}
}
if !has_pps {
if let Some(ref pps) = params.pps {
debug!("Injecting cached H264 PPS");
injected.push(pps.clone());
}
}
drop(params);
if !injected.is_empty() {
injected.extend(nals);
nals = injected;
TrackType::Rtp(_) => {
warn!("H264 should not use RTP track");
}
}
// Send NAL units
self.send_nals(nals, is_keyframe).await
// Update stats
let mut stats = self.stats.lock().await;
stats.frames_sent += 1;
stats.bytes_sent += data.len() as u64;
if is_keyframe {
stats.keyframes_sent += 1;
}
Ok(())
}
/// Write H265 frame (Annex B format)
@@ -483,52 +438,6 @@ impl UniversalVideoTrack {
Ok(())
}
/// Send NAL units as samples (H264 only)
///
/// Important: Only the last NAL unit should have the frame duration set.
/// All NAL units in a frame share the same RTP timestamp, so only the last
/// one should increment the timestamp by the frame duration.
async fn send_nals(&self, nals: Vec<Bytes>, is_keyframe: bool) -> Result<()> {
let mut total_bytes = 0u64;
// Calculate frame duration based on configured FPS
let frame_duration = Duration::from_micros(1_000_000 / self.config.fps.max(1) as u64);
let nal_count = nals.len();
match &self.track {
TrackType::Sample(track) => {
for (i, nal_data) in nals.into_iter().enumerate() {
let is_last = i == nal_count - 1;
// Only the last NAL should have duration set
// This ensures all NALs in a frame share the same RTP timestamp
let sample = Sample {
data: nal_data.clone(),
duration: if is_last { frame_duration } else { Duration::ZERO },
..Default::default()
};
if let Err(e) = track.write_sample(&sample).await {
debug!("NAL write_sample failed: {}", e);
}
total_bytes += nal_data.len() as u64;
}
}
TrackType::Rtp(_) => {
warn!("send_nals should not be called for RTP track (H265)");
}
}
// Update stats
let mut stats = self.stats.lock().await;
stats.frames_sent += 1;
stats.bytes_sent += total_bytes;
if is_keyframe {
stats.keyframes_sent += 1;
}
Ok(())
}
/// Send H265 NAL units via custom H265Payloader
async fn send_h265_rtp(&self, data: &[u8], is_keyframe: bool) -> Result<()> {
let rtp_track = match &self.track {

41
src/webrtc/webrtc_streamer.rs
View File

@@ -51,6 +51,7 @@ use crate::video::shared_video_pipeline::{SharedVideoPipeline, SharedVideoPipeli
use super::config::{TurnServer, WebRtcConfig};
use super::signaling::{ConnectionState, IceCandidate, SdpAnswer, SdpOffer};
use super::universal_session::{UniversalSession, UniversalSessionConfig};
use crate::video::encoder::BitratePreset;
/// WebRTC streamer configuration
#[derive(Debug, Clone)]
@@ -63,12 +64,10 @@ pub struct WebRtcStreamerConfig {
pub resolution: Resolution,
/// Input pixel format
pub input_format: PixelFormat,
/// Target bitrate in kbps
pub bitrate_kbps: u32,
/// Bitrate preset
pub bitrate_preset: BitratePreset,
/// Target FPS
pub fps: u32,
/// GOP size (keyframe interval)
pub gop_size: u32,
/// Enable audio (reserved)
pub audio_enabled: bool,
/// Encoder backend (None = auto select best available)
@@ -82,9 +81,8 @@ impl Default for WebRtcStreamerConfig {
video_codec: VideoCodecType::H264,
resolution: Resolution::HD720,
input_format: PixelFormat::Mjpeg,
bitrate_kbps: 8000,
bitrate_preset: BitratePreset::Balanced,
fps: 30,
gop_size: 30,
audio_enabled: false,
encoder_backend: None,
}
@@ -282,10 +280,10 @@ impl WebRtcStreamer {
resolution: config.resolution,
input_format: config.input_format,
output_codec: Self::codec_type_to_encoder_type(codec),
bitrate_kbps: config.bitrate_kbps,
bitrate_preset: config.bitrate_preset,
fps: config.fps,
gop_size: config.gop_size,
encoder_backend: config.encoder_backend,
..Default::default()
};
info!("Creating shared video pipeline for {:?}", codec);
@@ -541,8 +539,8 @@ impl WebRtcStreamer {
// Note: bitrate is NOT auto-scaled here - use set_bitrate() or config to change it
info!(
"WebRTC config updated: {}x{} {:?} @ {} fps, {} kbps",
resolution.width, resolution.height, format, fps, config.bitrate_kbps
"WebRTC config updated: {}x{} {:?} @ {} fps, {}",
resolution.width, resolution.height, format, fps, config.bitrate_preset
);
}
@@ -636,9 +634,8 @@ impl WebRtcStreamer {
codec: Self::codec_type_to_encoder_type(codec),
resolution: config.resolution,
input_format: config.input_format,
bitrate_kbps: config.bitrate_kbps,
bitrate_preset: config.bitrate_preset,
fps: config.fps,
gop_size: config.gop_size,
audio_enabled: *self.audio_enabled.read().await,
};
drop(config);
@@ -875,13 +872,13 @@ impl WebRtcStreamer {
}
}
/// Set bitrate
/// Set bitrate using preset
///
/// Note: Hardware encoders (VAAPI, NVENC, etc.) don't support dynamic bitrate changes.
/// This method restarts the pipeline to apply the new bitrate.
pub async fn set_bitrate(self: &Arc<Self>, bitrate_kbps: u32) -> Result<()> {
pub async fn set_bitrate_preset(self: &Arc<Self>, preset: BitratePreset) -> Result<()> {
// Update config first
self.config.write().await.bitrate_kbps = bitrate_kbps;
self.config.write().await.bitrate_preset = preset;
// Check if pipeline exists and is running
let pipeline_running = {
@@ -894,8 +891,8 @@ impl WebRtcStreamer {
if pipeline_running {
info!(
"Restarting video pipeline to apply new bitrate: {} kbps",
bitrate_kbps
"Restarting video pipeline to apply new bitrate: {}",
preset
);
// Stop existing pipeline
@@ -936,16 +933,16 @@ impl WebRtcStreamer {
}
info!(
"Video pipeline restarted with {} kbps, reconnected {} sessions",
bitrate_kbps,
"Video pipeline restarted with {}, reconnected {} sessions",
preset,
session_ids.len()
);
}
}
} else {
debug!(
"Pipeline not running, bitrate {} kbps will apply on next start",
bitrate_kbps
"Pipeline not running, bitrate {} will apply on next start",
preset
);
}
@@ -978,7 +975,7 @@ mod tests {
let config = WebRtcStreamerConfig::default();
assert_eq!(config.video_codec, VideoCodecType::H264);
assert_eq!(config.resolution, Resolution::HD720);
assert_eq!(config.bitrate_kbps, 8000);
assert_eq!(config.bitrate_preset, BitratePreset::Quality);
assert_eq!(config.fps, 30);
assert!(!config.audio_enabled);
}