GithubFollow/One-KVM
1
0
Fork 0
mirror of https://github.com/mofeng-git/One-KVM.git synced 2026-04-29 17:36:35 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix: 优化视频切换流畅性；修复 OTG HID 功能无法一次保存成功和页面未即刻生效问题

Browse Source
This commit is contained in:
mofeng-git
2026-04-11 21:20:54 +08:00
parent eecbc0fc13
commit 099f0b1ca2
8 changed files with 219 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
src/config/store.rs
View File

@@ -4,6 +4,7 @@ use std::path::Path;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::broadcast;
use tokio::sync::Mutex;
use super::AppConfig;
use crate::error::{AppError, Result};
@@ -18,6 +19,8 @@ pub struct ConfigStore {
/// Lock-free cache using ArcSwap for zero-cost reads
cache: Arc<ArcSwap<AppConfig>>,
change_tx: broadcast::Sender<ConfigChange>,
/// Serializes `set` / `update` so concurrent PATCH handlers cannot clobber each other
write_lock: Arc<Mutex<()>>,
}
/// Configuration change event
@@ -59,6 +62,7 @@ impl ConfigStore {
pool,
cache,
change_tx,
write_lock: Arc::new(Mutex::new(())),
})
}
@@ -191,6 +195,7 @@ impl ConfigStore {
/// Set entire configuration
pub async fn set(&self, config: AppConfig) -> Result<()> {
let _guard = self.write_lock.lock().await;
Self::save_config_to_db(&self.pool, &config).await?;
self.cache.store(Arc::new(config));
@@ -204,13 +209,13 @@ impl ConfigStore {
/// Update configuration with a closure
///
/// Note: This uses a read-modify-write pattern. For concurrent updates,
/// the last write wins. This is acceptable for configuration changes
/// which are infrequent and typically user-initiated.
/// Uses read-modify-write under a mutex so concurrent `update` / `set` calls are serialized
/// and merged correctly (each closure sees the latest stored config).
pub async fn update<F>(&self, f: F) -> Result<()>
where
F: FnOnce(&mut AppConfig),
{
let _guard = self.write_lock.lock().await;
// Load current config, clone it for modification
let current = self.cache.load();
let mut config = (**current).clone();

35
src/video/shared_video_pipeline.rs
View File

@@ -844,7 +844,7 @@ impl SharedVideoPipeline {
}
}
/// Stop the pipeline
/// Stop the pipeline (non-blocking, does not wait for capture thread to exit)
pub fn stop(&self) {
if *self.running_rx.borrow() {
let _ = self.running.send(false);
@@ -854,6 +854,39 @@ impl SharedVideoPipeline {
}
}
/// Stop the pipeline and wait for the capture thread to fully exit.
///
/// This ensures the V4L2 device is released before returning, which is
/// necessary when another consumer (e.g. MJPEG streamer) needs to open
/// the same device immediately after.
pub async fn stop_and_wait(&self, timeout: std::time::Duration) {
self.stop();
let mut rx = self.running_watch();
if !*rx.borrow() {
// Capture thread may still be running from a previous `stop()` call.
// Wait for the "Video pipeline stopped" log (thread sets running=false
// at exit), unless it already happened.
}
let deadline = tokio::time::Instant::now() + timeout;
loop {
if !self.running_flag.load(Ordering::Acquire) {
// Flag is cleared, but the capture thread may still be unwinding
// (dropping the V4L2 stream). Give it a brief moment.
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
break;
}
let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
if remaining.is_zero() {
warn!(
"Timed out waiting for video pipeline to stop after {:?}",
timeout
);
break;
}
let _ = tokio::time::timeout(remaining, rx.changed()).await;
}
}
/// Set bitrate using preset
pub async fn set_bitrate_preset(
&self,

62
src/video/stream_manager.rs
View File

@@ -404,8 +404,11 @@ impl VideoStreamManager {
}
}
StreamMode::WebRTC => {
info!("Closing all WebRTC sessions");
let closed = self.webrtc_streamer.close_all_sessions().await;
info!("Closing all WebRTC sessions and releasing capture device");
let closed = self
.webrtc_streamer
.close_all_sessions_and_release_device()
.await;
if closed > 0 {
info!("Closed {} WebRTC sessions", closed);
}
@@ -781,6 +784,61 @@ impl VideoStreamManager {
self.webrtc_streamer.request_keyframe().await
}
/// Notify frontend about a codec-only switch (WebRTC mode unchanged, codec changed).
///
/// `set_video_codec` already rebuilt the pipeline synchronously, so we just
/// emit the events the frontend waits on: `StreamModeChanged`, `WebRTCReady`,
/// and `StreamModeReady`.
///
/// Events are spawned asynchronously so the HTTP response (carrying the
/// `transition_id`) reaches the client before the WebSocket events, giving
/// the frontend time to call `registerTransition()` first.
pub async fn notify_codec_switch(
self: &Arc<Self>,
transition_id: &str,
new_codec_str: &str,
previous_codec_str: &str,
) {
let manager = Arc::clone(self);
let transition_id = transition_id.to_string();
let new_codec = new_codec_str.to_string();
let prev_codec = previous_codec_str.to_string();
tokio::spawn(async move {
// Small yield to ensure the HTTP response is flushed first.
tokio::task::yield_now().await;
manager
.publish_event(SystemEvent::StreamModeChanged {
transition_id: Some(transition_id.clone()),
mode: new_codec.clone(),
previous_mode: prev_codec.clone(),
})
.await;
let is_hardware = manager.webrtc_streamer.is_hardware_encoding().await;
manager
.publish_event(SystemEvent::WebRTCReady {
transition_id: Some(transition_id.clone()),
codec: new_codec.clone(),
hardware: is_hardware,
})
.await;
manager
.publish_event(SystemEvent::StreamModeReady {
transition_id: transition_id.clone(),
mode: new_codec.clone(),
})
.await;
info!(
"Codec switch notified: {} -> {} (transition: {})",
prev_codec, new_codec, transition_id
);
});
}
/// Publish event to event bus
async fn publish_event(&self, event: SystemEvent) {
if let Some(ref events) = *self.events.read().await {

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

@@ -1282,7 +1282,26 @@ pub async fn stream_mode_set(
}
}
// Set video codec if switching to WebRTC mode with specific codec
let requested_mode_str = match (&new_mode, &video_codec) {
(StreamMode::Mjpeg, _) => "mjpeg",
(StreamMode::WebRTC, Some(VideoCodecType::H264)) => "h264",
(StreamMode::WebRTC, Some(VideoCodecType::H265)) => "h265",
(StreamMode::WebRTC, Some(VideoCodecType::VP8)) => "vp8",
(StreamMode::WebRTC, Some(VideoCodecType::VP9)) => "vp9",
(StreamMode::WebRTC, None) => "webrtc",
};
// Detect codec-only switch: already in WebRTC mode, just changing codec.
// switch_mode_transaction treats this as "no switch needed" since StreamMode
// is still WebRTC, so we handle codec change + event emission here.
let current_mode = state.stream_manager.current_mode().await;
let prev_codec = state.stream_manager.webrtc_streamer().current_video_codec().await;
let codec_changed = video_codec.is_some_and(|c| c != prev_codec);
let is_codec_only_switch = current_mode == StreamMode::WebRTC
&& new_mode == StreamMode::WebRTC
&& codec_changed;
if let Some(codec) = video_codec {
info!("Setting WebRTC video codec to {:?}", codec);
if let Err(e) = state
@@ -1295,22 +1314,34 @@ pub async fn stream_mode_set(
}
}
// For codec-only switch, emit events directly instead of going through
// switch_mode_transaction (which short-circuits when mode is unchanged).
if is_codec_only_switch {
let transition_id = uuid::Uuid::new_v4().to_string();
state
.stream_manager
.notify_codec_switch(
&transition_id,
requested_mode_str,
&codec_to_id(prev_codec),
)
.await;
return Ok(Json(StreamModeResponse {
success: true,
mode: requested_mode_str.to_string(),
transition_id: Some(transition_id),
switching: false,
message: Some(format!("Codec switched to {}", requested_mode_str)),
}));
}
let tx = state
.stream_manager
.switch_mode_transaction(new_mode.clone())
.await?;
// Return the requested codec identifier (for UI display). The actual active mode
// may differ if the request was rejected due to an in-progress switch.
let requested_mode_str = match (&new_mode, &video_codec) {
(StreamMode::Mjpeg, _) => "mjpeg",
(StreamMode::WebRTC, Some(VideoCodecType::H264)) => "h264",
(StreamMode::WebRTC, Some(VideoCodecType::H265)) => "h265",
(StreamMode::WebRTC, Some(VideoCodecType::VP8)) => "vp8",
(StreamMode::WebRTC, Some(VideoCodecType::VP9)) => "vp9",
(StreamMode::WebRTC, None) => "webrtc",
};
let active_mode_str = match state.stream_manager.current_mode().await {
StreamMode::Mjpeg => "mjpeg".to_string(),
StreamMode::WebRTC => {

16
src/webrtc/webrtc_streamer.rs
View File

@@ -787,6 +787,22 @@ impl WebRtcStreamer {
count
}
/// Close all sessions and wait for the video pipeline to fully release the
/// capture device. Use this when the caller needs the V4L2 device immediately
/// afterwards (e.g. switching to MJPEG mode).
pub async fn close_all_sessions_and_release_device(&self) -> usize {
let count = self.close_all_sessions().await;
if let Some(ref pipeline) = *self.video_pipeline.read().await {
pipeline
.stop_and_wait(std::time::Duration::from_secs(3))
.await;
}
*self.video_pipeline.write().await = None;
count
}
/// Get session count
pub async fn session_count(&self) -> usize {
self.sessions.read().await.len()

16
web/src/composables/useWebRTC.ts
View File

@@ -334,15 +334,14 @@ async function addRemoteIceCandidate(candidate: IceCandidate) {
async function flushPendingRemoteIce() {
if (!peerConnection || !sessionId || !peerConnection.remoteDescription) return
const remaining: WebRTCIceCandidateEvent[] = []
for (const event of pendingRemoteCandidates) {
const queued = pendingRemoteCandidates
pendingRemoteCandidates = []
for (const event of queued) {
if (event.session_id === sessionId) {
await addRemoteIceCandidate(event.candidate)
} else {
// Drop candidates for old sessions
}
}
pendingRemoteCandidates = remaining
if (pendingRemoteIceComplete.has(sessionId)) {
pendingRemoteIceComplete.delete(sessionId)
@@ -546,10 +545,8 @@ async function connect(): Promise<boolean> {
}
}
// 等待连接真正建立（最多等待 15 秒）
// 直接检查 peerConnection.connectionState 而不是 reactive state
// 因为 TypeScript 不知道 state 会被 onconnectionstatechange 回调异步修改
const connectionTimeout = 15000
// Wait for connection to establish (5s for LAN, sufficient for most scenarios)
const connectionTimeout = 5000
const pollInterval = 100
let waited = 0
connectStage.value = 'waiting_connection'
@@ -568,7 +565,6 @@ async function connect(): Promise<boolean> {
waited += pollInterval
}
// 超时
throw new Error('Connection timeout waiting for ICE negotiation')
} catch (err) {
state.value = 'failed'

37
web/src/views/ConsoleView.vue
View File

@@ -1312,18 +1312,17 @@ async function switchToWebRTC(codec: VideoMode = 'h264') {
}
}
// Step 3: Connect WebRTC with retry
let retries = 3
// Step 3: Connect WebRTC with retry (backoff between retries)
const MAX_ATTEMPTS = 3
const RETRY_DELAYS = [200, 800]
let success = false
while (retries > 0 && !success) {
success = await connectWebRTCSerial('switchToWebRTC')
if (!success) {
retries--
if (retries > 0) {
console.log(`[WebRTC] Connection failed, retrying (${retries} attempts left)`)
await new Promise(resolve => setTimeout(resolve, 500))
}
for (let attempt = 0; attempt < MAX_ATTEMPTS && !success; attempt++) {
if (attempt > 0) {
const delay = RETRY_DELAYS[attempt - 1] ?? RETRY_DELAYS[RETRY_DELAYS.length - 1]
console.log(`[WebRTC] Connection failed, retrying in ${delay}ms (${MAX_ATTEMPTS - attempt} attempts left)`)
await new Promise(resolve => setTimeout(resolve, delay))
}
success = await connectWebRTCSerial('switchToWebRTC')
}
if (success) {
toast.success(t('console.webrtcConnected'), {
@@ -1526,10 +1525,22 @@ watch(() => webrtc.state.value, (newState, oldState) => {
}, 1000)
}
// Handle direct 'failed' state (ICE or DTLS failure)
// Allow one automatic retry before marking as failed, consistent with
// the disconnected->reconnect path that allows 2 failures.
if (newState === 'failed' && videoMode.value !== 'mjpeg') {
webrtcReconnectFailures += 1
if (webrtcReconnectFailures >= 1) {
if (webrtcReconnectFailures >= 2) {
markWebRTCFailure(t('console.webrtcFailed'))
} else {
webrtcReconnectTimeout = setTimeout(async () => {
if (videoMode.value !== 'mjpeg' && webrtc.state.value !== 'connected') {
const success = await connectWebRTCSerial('auto reconnect after failed')
if (!success) {
markWebRTCFailure(t('console.webrtcFailed'))
}
}
}, 1000)
}
}
})
@@ -2155,6 +2166,10 @@ async function activateConsoleView() {
isConsoleActive.value = true
registerInteractionListeners()
// REST snapshot: returning from Settings (or other routes) may have missed WS device_info
void systemStore.fetchAllStates()
void configStore.refreshHid().then(() => syncMouseModeFromConfig()).catch(() => {})
// Ensure HID WebSocket is connected when console becomes active
if (!hidWs.connected.value) {
hidWs.connect().catch(() => {})

60
web/src/views/SettingsView.vue
View File

@@ -974,33 +974,29 @@ async function saveConfig() {
saved.value = false
try {
// Save only config related to the active section
const savePromises: Promise<unknown>[] = []
// Save only config related to the active section.
// Sequential awaits: backend ConfigStore uses read-modify-write; parallel PATCH
// requests could overwrite each other's section (last writer wins on full JSON).
// Video config (including encoder and WebRTC/STUN/TURN settings)
if (activeSection.value === 'video') {
savePromises.push(
configStore.updateVideo({
device: config.value.video_device || undefined,
format: config.value.video_format || undefined,
width: config.value.video_width,
height: config.value.video_height,
fps: toConfigFps(config.value.video_fps),
})
)
// Save Stream/Encoder and STUN/TURN config together
savePromises.push(
configStore.updateStream({
encoder: config.value.encoder_backend as any,
stun_server: config.value.stun_server || undefined,
turn_server: config.value.turn_server || undefined,
turn_username: config.value.turn_username || undefined,
turn_password: config.value.turn_password || undefined,
})
)
await configStore.updateVideo({
device: config.value.video_device || undefined,
format: config.value.video_format || undefined,
width: config.value.video_width,
height: config.value.video_height,
fps: toConfigFps(config.value.video_fps),
})
await configStore.updateStream({
encoder: config.value.encoder_backend as any,
stun_server: config.value.stun_server || undefined,
turn_server: config.value.turn_server || undefined,
turn_username: config.value.turn_username || undefined,
turn_password: config.value.turn_password || undefined,
})
}
// HID config
// HID config (includes MSD enable — same gadget; must not race with updateHid)
if (activeSection.value === 'hid') {
if (!isHidFunctionSelectionValid.value || !isOtgEndpointBudgetValid.value) {
return
@@ -1024,24 +1020,20 @@ async function saveConfig() {
hidUpdate.otg_functions = { ...config.value.hid_otg_functions }
hidUpdate.otg_keyboard_leds = config.value.hid_otg_keyboard_leds
}
savePromises.push(configStore.updateHid(hidUpdate))
savePromises.push(
configStore.updateMsd({
enabled: config.value.msd_enabled,
})
)
await configStore.updateHid(hidUpdate)
await configStore.updateMsd({
enabled: config.value.msd_enabled,
})
}
// MSD config
if (activeSection.value === 'msd') {
savePromises.push(
configStore.updateMsd({
msd_dir: config.value.msd_dir || undefined,
})
)
await configStore.updateMsd({
msd_dir: config.value.msd_dir || undefined,
})
}
await Promise.all(savePromises)
await loadConfig()
saved.value = true
setTimeout(() => (saved.value = false), 2000)
} catch (e) {