From 5787d4cd20cc42eea414372e82bdbfa4db80923d Mon Sep 17 00:00:00 2001 From: VallierDev <24893073+VallierDev@users.noreply.github.com> Date: Sat, 24 Oct 2026 00:45:52 +0800 Subject: [PATCH 1/5] Start the desktop on a tool connection when the viewer sends capture_desktop (#99) With a port-mapping connection open, the Mac viewer (4.42.0) does not offer a new desktop peer; it sends SimpleAction 4 capture_desktop on the existing capture_type=9 peer and waits for video, which the host never sent. A tool peer now keeps its desktop video unstarted and becomes the desktop on capture_desktop: the main loop checks the backend (headless and Portal are skipped with a log line), applies the peer's controller platform, closes peers that were offered as desktops, then the peer takes the input lease, starts the H.264 stream it negotiated, counts the session and repeats the desktop CONTROL initialization, lock state, clipboard and cursor. Port mapping, files and terminal keep running on the same peer. Co-Authored-By: Claude Opus 5.5 --- src/desktop_control_tests.rs | 34 ++++ src/main.rs | 47 ++++- src/signaling.rs | 2 + src/system_actions.rs | 2 + src/webrtc.rs | 322 +++++++++++++++++++++++------------ 5 files changed, 292 insertions(+), 115 deletions(-) diff --git a/src/desktop_control_tests.rs b/src/desktop_control_tests.rs index 7f92760..e08afad 100644 --- a/src/desktop_control_tests.rs +++ b/src/desktop_control_tests.rs @@ -215,3 +215,37 @@ async fn locked_control_unlock_requests_and_tools_work_without_a_video_frame() { drop(lock); }).await.expect("locked bootstrap/tool regression exceeded deadline"); } + +#[tokio::test] +async fn capture_desktop_on_a_port_mapping_peer_promotes_it_once() { + tokio::time::timeout(Duration::from_secs(25), async { + let input = Arc::new(InputInjector::without_desktop()); + let (host, mut viewer) = connect(input.clone(), false, None, &["CONTROL_DATA_CHANNEL", "FILE_DATA_CHANNEL"]).await; + let (requests, mut requested) = mpsc::unbounded_channel(); + host.on_desktop_request(move || { let _ = requests.send(()); }); + receive(&mut viewer.messages, "CONTROL_DATA_CHANNEL", 13).await; + assert!(!host.role.is_desktop() && host.role.input().is_none()); + // The Mac viewer opens the desktop on its existing port-mapping peer (#99). + send(&viewer, "CONTROL_DATA_CHANNEL", crate::initialization::envelope(5, 0, 3, &[8, 4])).await; + tokio::time::timeout(Duration::from_secs(5), requested.recv()).await.unwrap().unwrap(); + host.promote(); + assert!(host.role.is_desktop()); + assert!(host.role.input().is_some_and(|session| session.is_current())); + // CONTROL repeats the desktop initialization for the same viewer. + receive(&mut viewer.messages, "CONTROL_DATA_CHANNEL", 13).await; + // The port-mapping session on this peer still answers. + send(&viewer, "FILE_DATA_CHANNEL", mapping(2, b"still mapped")).await; + receive(&mut viewer.messages, "FILE_DATA_CHANNEL", 27).await; + // A repeated capture_desktop does not ask again, and a later promote is a no-op. + send(&viewer, "CONTROL_DATA_CHANNEL", crate::initialization::envelope(6, 0, 3, &[8, 4])).await; + send(&viewer, "CONTROL_DATA_CHANNEL", crate::initialization::envelope(7, 0, 3, &[8, 0])).await; + receive(&mut viewer.messages, "CONTROL_DATA_CHANNEL", 3).await; + assert!(requested.try_recv().is_err()); + let lease = host.role.input().unwrap(); + host.promote(); + assert!(lease.is_current()); + host.pc.close().await.unwrap(); + viewer.pc.close().await.unwrap(); + viewer.router.lock().await.stop().await.unwrap(); + }).await.expect("capture_desktop promotion exceeded deadline"); +} diff --git a/src/main.rs b/src/main.rs index be53393..32a9d37 100644 --- a/src/main.rs +++ b/src/main.rs @@ -737,6 +737,8 @@ async fn serve(mut capture: Arc, input: Arc, shutd let mut capture_types: HashMap = HashMap::new(); let mut mac_by_ice: HashMap = HashMap::new(); let mut next_mac = false; + // Tool peers ask here to become the desktop (capture_desktop on an existing connection, #99). + let (desktop_requests, mut desktop_requested) = mpsc::unbounded_channel::(); loop { if *shutdown.borrow() { @@ -821,6 +823,7 @@ async fn serve(mut capture: Arc, input: Arc, shutd _ = shutdown.changed() => Some(SignalingEvent::Stop), command = commands.recv(), if assist.is_some() => Some(SignalingEvent::AssistCommand(command.unwrap_or(assist::Command::Stop))), _ = assist_tick.tick(), if assist.is_some() => Some(SignalingEvent::AssistTick), + Some(ice) = desktop_requested.recv() => Some(SignalingEvent::DesktopRequested(ice)), event = rx_event.recv() => event, }; let Some(evt) = event else { break; }; @@ -855,8 +858,8 @@ async fn serve(mut capture: Arc, input: Arc, shutd ::webrtc::peer_connection::peer_connection_state::RTCPeerConnectionState::Closed | ::webrtc::peer_connection::peer_connection_state::RTCPeerConnectionState::Failed)).map(|(id, _)| id.clone()).collect(); let mut desktop_ended = false; - for id in ended { if let Some(s) = sessions.remove(&id) { desktop_ended |= s.desktop; let _ = s.pc.close().await; } } - if desktop_ended && !sessions.values().any(|s| s.desktop) { if let Some(s) = assist.as_mut() { s.session_ended()?; } } + for id in ended { if let Some(s) = sessions.remove(&id) { desktop_ended |= s.role.is_desktop(); let _ = s.pc.close().await; } } + if desktop_ended && !sessions.values().any(|s| s.role.is_desktop()) { if let Some(s) = assist.as_mut() { s.session_ended()?; } } if let Some(s) = assist.as_mut() { s.tick()?; } } SignalingEvent::Connected => { @@ -943,8 +946,8 @@ async fn serve(mut capture: Arc, input: Arc, shutd ::webrtc::peer_connection::peer_connection_state::RTCPeerConnectionState::Closed | ::webrtc::peer_connection::peer_connection_state::RTCPeerConnectionState::Failed)).map(|(k, _)| k.clone()).collect(); let mut desktop_ended = false; - for k in ended { if let Some(old) = sessions.remove(&k) { desktop_ended |= old.desktop; let _ = old.pc.close().await; } } - if desktop_ended && !sessions.values().any(|s| s.desktop) { if let Some(s) = assist.as_mut() { s.session_ended()?; } } + for k in ended { if let Some(old) = sessions.remove(&k) { desktop_ended |= old.role.is_desktop(); let _ = old.pc.close().await; } } + if desktop_ended && !sessions.values().any(|s| s.role.is_desktop()) { if let Some(s) = assist.as_mut() { s.session_ended()?; } } if let Some(s) = assist.as_mut() { s.tick()?; } if soac.msg_type == "offer" { @@ -981,7 +984,12 @@ async fn serve(mut capture: Arc, input: Arc, shutd if desktop { if let Some(portal) = &portal_desktop { portal.begin(); } } match WebRtcSession::new(session_capture, input.clone(), signaling_client.clone(), desktop, transport, if desktop { portal_desktop.clone() } else { None }).await { - Ok(sess) => { + Ok(mut sess) => { + sess.controller_mac = controller_mac; + if !desktop { + let (requests, peer) = (desktop_requests.clone(), ice.clone()); + sess.on_desktop_request(move || { let _ = requests.send(peer.clone()); }); + } // Register before awaiting SDP: shutdown can always close // a peer whose answer may already be queued to the viewer. sessions.insert(ice.clone(), sess); @@ -1026,13 +1034,38 @@ async fn serve(mut capture: Arc, input: Arc, shutd let mut desktop_ended = false; for k in targets { if let Some(old) = sessions.remove(&k) { - desktop_ended |= old.desktop; + desktop_ended |= old.role.is_desktop(); if let Err(e) = old.pc.close().await { log::warn!("Close released peer failed: {}", e); } } } - if desktop_ended && !sessions.values().any(|s| s.desktop) { if let Some(s) = assist.as_mut() { s.session_ended()?; } } + if desktop_ended && !sessions.values().any(|s| s.role.is_desktop()) { if let Some(s) = assist.as_mut() { s.session_ended()?; } } if let Some(s) = assist.as_mut() { s.tick()?; } } + SignalingEvent::DesktopRequested(ice) => { + let mut sessions = webrtc_sessions.lock().await; + let Some(peer) = sessions.get(&ice).filter(|peer| !peer.role.is_desktop()) else { continue; }; + if desktop_backend::headless() { + log::info!("Desktop request on a tool connection ignored: headless mode"); + continue; + } + if portal_desktop.is_some() { + log::info!("Desktop request on a tool connection not supported with the Portal backend; open the desktop before port mapping"); + continue; + } + if peer.capture.init_error.is_some() { + log::info!("Desktop request on a tool connection ignored: desktop capture unavailable"); + continue; + } + video::warm_auto_probe(); video::check_ffmpeg_usable(); + input.set_mac_controller(peer.controller_mac); + input.bind_portal(peer.capture.portal.clone()); + // One desktop at a time, as for a new desktop offer. Only peers offered as + // desktops close; this peer keeps its port mapping, files or terminal. + let superseded: Vec = sessions.iter().filter(|(id, old)| id.as_str() != ice && old.desktop) + .map(|(id, _)| id.clone()).collect(); + for id in superseded { if let Some(old) = sessions.remove(&id) { let _ = old.pc.close().await; } } + if let Some(peer) = sessions.get(&ice) { peer.promote(); } + } SignalingEvent::Disconnected => { capture_types.clear(); next_capture_type=None; diff --git a/src/signaling.rs b/src/signaling.rs index d65aaba..3448a7a 100644 --- a/src/signaling.rs +++ b/src/signaling.rs @@ -46,6 +46,8 @@ pub enum SignalingEvent { Released(Value), Left(Value), Disconnected, + /// capture_desktop on this tool peer (ice_id); raised by the peer, not by signaling. + DesktopRequested(String), } pub struct SignalingClient { diff --git a/src/system_actions.rs b/src/system_actions.rs index aeb633d..443bb9e 100644 --- a/src/system_actions.rs +++ b/src/system_actions.rs @@ -12,6 +12,8 @@ pub const FF_SIMPLE_ACTION: u8 = 2; pub const FF_SYSTEM_METRICS: u8 = 2; pub const SHOW_DESKTOP: u64 = 2; +/// Opens the desktop on an existing connection (the viewer's port mapping reuses its peer, #99). +pub const CAPTURE_DESKTOP: u64 = 4; pub const REBOOT_SYSTEM: u64 = 9; pub const LOCK_SYSTEM: u64 = 10; pub const SHUTDOWN_SYSTEM: u64 = 12; diff --git a/src/webrtc.rs b/src/webrtc.rs index d1e7613..551bfff 100644 --- a/src/webrtc.rs +++ b/src/webrtc.rs @@ -42,12 +42,121 @@ pub struct SessionContext { pub bundle_media: Option<(String, u16)>, } +/// Desktop role of one peer. A tool peer (file transfer, terminal, port mapping) gains it +/// when the viewer opens the desktop on that same connection with SimpleAction 4 +/// capture_desktop instead of a new desktop offer (#99). +pub struct DesktopRole { + active: tokio::sync::watch::Sender, + input: std::sync::OnceLock>, + stream: std::sync::Mutex>, + request: std::sync::OnceLock>, +} +impl DesktopRole { + fn new(desktop: bool) -> Self { + Self { active: tokio::sync::watch::channel(desktop).0, input: Default::default(), + stream: Default::default(), request: Default::default() } + } + pub fn is_desktop(&self) -> bool { *self.active.borrow() } + fn input(&self) -> Option> { self.input.get().cloned() } + /// Waits until this peer is a desktop; false once the channel closes first. + async fn wait_desktop(&self, channel: &RTCDataChannel) -> bool { + let mut active = self.active.subscribe(); + loop { + if *active.borrow_and_update() { return true; } + tokio::select! { + changed = active.changed() => if changed.is_err() { return false; }, + _ = tokio::time::sleep(std::time::Duration::from_millis(200)) => { + if matches!(channel.ready_state(), webrtc::data_channel::data_channel_state::RTCDataChannelState::Closing + | webrtc::data_channel::data_channel_state::RTCDataChannelState::Closed) { return false; } + } + } + } + } +} + +/// Live desktop video for one peer; a tool peer keeps it unstarted until it becomes a desktop. +struct DesktopStream { + pc: std::sync::Weak, + track: Arc, + portal_desktop: Option>, + capture_size: (u32, u32), + capture_display: String, + portal: Option>, + settings_rx: tokio::sync::watch::Receiver, + settings: Arc>, + idr_request: Arc, + selection: crate::video_select::Selection, + chroma_sync: Option>>, + reload_at_session: u64, +} + pub struct WebRtcSession { pub pc: Arc, pub capture: Arc, pub input: Arc, + /// The role this peer was offered with; `role` also covers a later capture_desktop. pub desktop: bool, + pub role: Arc, + pub controller_mac: bool, pub context: Arc>, + actions: Arc, +} + +/// Counts a desktop session once it is connected (a failed offer is not a session). +fn track_session(end_pc: Arc, actions: Arc) { + tokio::spawn(async move { + use webrtc::peer_connection::peer_connection_state::RTCPeerConnectionState as State; + while end_pc.connection_state() != State::Connected { + if matches!(end_pc.connection_state(), State::Closed | State::Failed) { return; } + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + } + let starting = actions.clone(); + let _ = tokio::task::spawn_blocking(move || starting.session_started()).await; + while !matches!(end_pc.connection_state(), State::Closed | State::Failed) { + tokio::time::sleep(std::time::Duration::from_secs(1)).await; + } + drop(end_pc); + let _ = tokio::task::spawn_blocking(move || actions.session_ended()).await; + }); +} + +impl DesktopStream { + /// Encode current frames only while the peer is connected. + fn spawn(self) { + let Some(streaming_pc) = self.pc.upgrade() else { return; }; + let DesktopStream { track: video_track, portal_desktop, capture_size, capture_display, portal, + settings_rx, settings, idr_request, selection, chroma_sync, reload_at_session, .. } = self; + tokio::spawn(async move { + let screen = if let Some(desktop) = portal_desktop { + if let Err(error) = crate::locked_video::run(&streaming_pc, &video_track, &desktop, selection).await { + log::warn!("Locked desktop video unavailable: {error}"); + } + let mut updates = desktop.subscribe(); + loop { + let state = updates.borrow_and_update().clone(); + if state.ready() { break Some(state.screen); } + if !state.waiting { return; } + if crate::video::transport_closed(&streaming_pc) { return; } + tokio::select! { + changed = updates.changed() => if changed.is_err() { return; }, + _ = tokio::time::sleep(std::time::Duration::from_millis(100)) => {}, + } + } + } else { None }; + let (capture_size, capture_display, portal) = screen.as_ref().map_or( + (capture_size, capture_display, portal), + |screen| ((screen.width, screen.height), screen.display.clone(), screen.portal.clone())); + let state_pc = streaming_pc.clone(); + if let Err(e) = crate::video::stream_desktop_at(streaming_pc, video_track, capture_size, capture_display, portal, settings_rx, settings, idr_request, selection, chroma_sync, reload_at_session).await { + // A send racing the viewer's close fails with a misleading track error. + if matches!(state_pc.connection_state(), RTCPeerConnectionState::Closed | RTCPeerConnectionState::Failed) { + log::info!("Live desktop stream ended: peer connection closed ({})", e); + } else { + log::error!("Live desktop stream stopped: {}", e); + } + } + }); + } } impl WebRtcSession { @@ -106,7 +215,8 @@ impl WebRtcSession { .map_err(|e| format!("Create PeerConnection failed: {}", e))?, ); - let input_session=desktop.then(||input.desktop_session()); + let role = Arc::new(DesktopRole::new(desktop)); + if desktop { let _ = role.input.set(input.desktop_session()); } let context = Arc::new(Mutex::new(SessionContext::default())); crate::network_diagnostics::start(Arc::downgrade(&pc)); @@ -164,13 +274,13 @@ impl WebRtcSession { })); let connected_context = context.clone(); - let connection_input=input_session.clone(); + let connection_role=role.clone(); let connected_signal = signaling_client.clone(); let notified = Arc::new(std::sync::atomic::AtomicBool::new(false)); let connected_selection = video_selection; pc.on_peer_connection_state_change(Box::new(move |state| { let ctx = connected_context.clone(); - let input=connection_input.clone(); + let input=connection_role.input(); let sig = connected_signal.clone(); let notified = notified.clone(); let video_selection = connected_selection; @@ -282,72 +392,24 @@ impl WebRtcSession { let video_settings = Arc::new(video_settings); // Per-connection mute / lock-after-session settings (安全 menu). let peer_actions = Arc::new(crate::system_actions::PeerActions::default()); + let session_actions = peer_actions.clone(); let desktop_sound = desktop && crate::audio::enabled(); - if desktop && !cfg!(test) { - let (end_pc, actions) = (pc.clone(), peer_actions.clone()); - tokio::spawn(async move { - { - use webrtc::peer_connection::peer_connection_state::RTCPeerConnectionState as State; - // Count the session once it is actually connected (a failed offer is not a session). - while end_pc.connection_state() != State::Connected { - if matches!(end_pc.connection_state(), State::Closed | State::Failed) { return; } - tokio::time::sleep(std::time::Duration::from_millis(200)).await; - } - let starting = actions.clone(); - let _ = tokio::task::spawn_blocking(move || starting.session_started()).await; - } - use webrtc::peer_connection::peer_connection_state::RTCPeerConnectionState as State; - while !matches!(end_pc.connection_state(), State::Closed | State::Failed) { - tokio::time::sleep(std::time::Duration::from_secs(1)).await; - } - drop(end_pc); - let _ = tokio::task::spawn_blocking(move || actions.session_ended()).await; - }); - } - let portal=capture.portal.clone(); - let stream_desktop = portal_desktop.clone(); - let stream_selection = video_selection; + if desktop && !cfg!(test) { track_session(pc.clone(), peer_actions.clone()); } // Live encoder switch (SIGUSR2) sends CaptureSettingSync when chroma is downgraded. let (chroma_sync_tx, chroma_sync_rx) = tokio::sync::mpsc::unbounded_channel::>(); - let chroma_sync = (desktop && !cfg!(test)).then_some(chroma_sync_tx); let chroma_sync_rx = Arc::new(std::sync::Mutex::new((!cfg!(test)).then_some(chroma_sync_rx))); - // File/terminal/port-mapping connections carry no desktop video. + // File/terminal/port-mapping connections carry no desktop video until capture_desktop. let control_video_track = video_track.clone(); - let stream_settings = video_settings.clone(); - if desktop && !cfg!(test) { - let state_pc = streaming_pc.clone(); - tokio::spawn(async move { - let screen = if let Some(desktop) = stream_desktop { - if let Err(error) = crate::locked_video::run(&streaming_pc, &video_track, &desktop, stream_selection).await { - log::warn!("Locked desktop video unavailable: {error}"); - } - let mut updates = desktop.subscribe(); - loop { - let state = updates.borrow_and_update().clone(); - if state.ready() { break Some(state.screen); } - if !state.waiting { return; } - if crate::video::transport_closed(&streaming_pc) { return; } - tokio::select! { - changed = updates.changed() => if changed.is_err() { return; }, - _ = tokio::time::sleep(std::time::Duration::from_millis(100)) => {}, - } - } - } else { None }; - let (capture_size, capture_display, portal) = screen.as_ref().map_or( - (capture_size, capture_display, portal), - |screen| ((screen.width, screen.height), screen.display.clone(), screen.portal.clone())); - if let Err(e) = crate::video::stream_desktop_at(streaming_pc, video_track, capture_size, capture_display, portal, video_settings_rx, stream_settings, idr_request, stream_selection, chroma_sync.clone(), reload_at_session).await { - // A send racing the viewer's close fails with a misleading track error. - if matches!(state_pc.connection_state(), RTCPeerConnectionState::Closed | RTCPeerConnectionState::Failed) { - log::info!("Live desktop stream ended: peer connection closed ({})", e); - } else { - log::error!("Live desktop stream stopped: {}", e); - } - } - }); - } else { - drop((streaming_pc, video_track, capture_size, capture_display, video_settings_rx)); - } + let stream = DesktopStream { + pc: Arc::downgrade(&streaming_pc), track: video_track, portal_desktop: portal_desktop.clone(), + capture_size, capture_display, portal: capture.portal.clone(), settings_rx: video_settings_rx, + settings: video_settings.clone(), idr_request, selection: video_selection, + chroma_sync: (!cfg!(test)).then_some(chroma_sync_tx), reload_at_session, + }; + drop(streaming_pc); + if cfg!(test) { drop(stream); } + else if desktop { stream.spawn(); } + else { *role.stream.lock().unwrap() = Some(stream); } let input_clone = input.clone(); let control_capture = capture.clone(); let desktop_pc = Arc::downgrade(&pc); @@ -405,6 +467,7 @@ impl WebRtcSession { // Settings arrive on CONTROL or TEXT; serialize both queues for this peer. let settings_serial=Arc::new(Mutex::new(())); + let data_role = role.clone(); // 监听 DataChannel pc.on_data_channel(Box::new(move |d: Arc| { let label = d.label().to_string(); @@ -412,7 +475,7 @@ impl WebRtcSession { let peer_settings_serial=settings_serial.clone(); let input_inst = input_clone.clone(); - let channel_input_session=input_session.clone(); + let channel_role=data_role.clone(); let screen = control_capture.clone(); let portal_desktop = portal_desktop.clone(); let desktop_pc = desktop_pc.clone(); @@ -442,9 +505,9 @@ impl WebRtcSession { let rich_state=Arc::new(std::sync::Mutex::new(crate::clipboard_rich::State::default())); let clipboard_enabled = shared_clipboard_ready.clone(); if label=="CONTROL_DATA_CHANNEL" { - let closing=channel_input_session.clone(); + let closing=channel_role.clone(); d.on_close(Box::new(move || { - let closing=closing.clone(); + let closing=closing.input(); Box::pin(async move {if let Some(session)=closing {session.disconnected();}}) })); } @@ -500,7 +563,7 @@ impl WebRtcSession { let manager = portal_desktop.clone(); let channel = Arc::downgrade(&d); let initial_screen = screen.clone(); - let switch_owner=channel_input_session.clone(); + let capture_role=channel_role.clone(); let settings_serial=peer_settings_serial.clone(); let capture_selection = peer_selection; tokio::spawn(async move { @@ -518,6 +581,7 @@ impl WebRtcSession { } }; let _serial=settings_serial.lock().await; + let (desktop,switch_owner)=(capture_role.is_desktop(),capture_role.input()); if desktop && switch_owner.as_ref().is_none_or(|owner|!owner.is_current()) {continue;} let reject=|reason:&str| { if let Some(response)=crate::capture_setting::rejected_request(&request,reason) { @@ -675,7 +739,8 @@ impl WebRtcSession { let open_video = channel_video_settings.clone(); let open_track = peer_video_track.clone(); let open_desktop = portal_desktop.clone(); - let open_owner = channel_input_session.clone(); + let open_owner = channel_role.input(); + let open_role = channel_role.clone(); let open_screen = screen.clone(); let open_selection = peer_selection; d.on_open(Box::new(move || { @@ -691,13 +756,14 @@ impl WebRtcSession { let rich_outgoing=rich_outgoing.clone(); let portal_desktop = open_desktop.clone(); let owner = open_owner.clone(); + let role = open_role.clone(); let pc = desktop_pc.upgrade(); Box::pin(async move { // Permission dialogs and first-frame waits must not occupy the // channel-open callback or delay CONTROL message dispatch. tokio::spawn(async move { if channel.label() == "TEXT_DATA_CHANNEL" { - if !desktop { return; } + if !desktop && !role.wait_desktop(&channel).await { return; } let screen = if let Some(desktop) = &portal_desktop { let mut updates = desktop.subscribe(); let Some(_) = crate::desktop_session::wait_ready(&mut updates, &channel).await else { return; }; @@ -853,49 +919,58 @@ impl WebRtcSession { } } else { snapshot.screen } } else { screen }; - let timestamp = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_secs(); let next_seq = || sequence.fetch_add(1, std::sync::atomic::Ordering::Relaxed); // Default file-transfer folder: the desktop user's home directory. let home = std::env::var("HOME").unwrap_or_else(|_| "/".into()); - // Tool peers are permitted without a desktop; the port-mapping viewer aborts on Reject. - let desktop_permission=if screen.init_error.is_none() || !desktop { - crate::system_metrics::Permission::Allow - } else { crate::system_metrics::Permission::Reject }; - // Without a desktop (headless/portal before consent) the port-mapping tool still waits - // for ScreenSources/CaptureChange and aborts otherwise: tool peers get a nominal screen. - let screen_size=crate::wayland::input_size(if screen.init_error.is_none() { (screen.width, screen.height) } else { (1920, 1080) }); - let messages = [ - (8, crate::initialization::encode_capture_change(next_seq(), timestamp, 0, crate::wayland::selected() as i32, "Linux desktop")), - (13, Ok(crate::initialization::envelope(next_seq(), timestamp, 13, &init_actions.metrics_payload(desktop_sound, desktop_permission)))), - (26, Ok(crate::initialization::envelope(next_seq(), timestamp, 26, - &crate::system_metrics::remote_download_path(&home)))), - (18, Ok(crate::initialization::envelope(next_seq(), timestamp, 18, - &crate::capture_setting::codec_negotiation(screen_size, crate::video::display_fps(), init_track.supports(crate::capture_setting::VideoCodec::H265))))), - (7, crate::capture_setting::screen_sources(next_seq(), timestamp, screen_size, init_video.borrow().output, "Linux desktop")), - ]; - if desktop && screen.init_error.is_none() { - let codec = crate::capture_setting::preferred_codec_for(init_track.supports(crate::capture_setting::VideoCodec::H264), - init_track.supports(crate::capture_setting::VideoCodec::H265), open_selection); - init_video.send_if_modified(|current| if current.codec != codec { - log::info!("Desktop codec selected: {:?}", codec); current.codec = codec; true } else { false }); - } - for (field, message) in messages { - if early_desktop_selection && field == 8 { continue; } - if (!desktop || screen.init_error.is_some()) && field == 18 { continue; } - if desktop && screen.init_error.is_some() && matches!(field,7|8) { continue; } - let message = match message { - Ok(message) => message, - Err(e) => { log::error!("CONTROL initialization encode failed: {}", e); return; } - }; - if let Err(e) = channel.send(&bytes::Bytes::from(message)).await { - log::error!("CONTROL initialization field {} failed: {}", field, e); - return; + let (mut desktop, mut promoted) = (desktop, false); + loop { + let timestamp = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_secs(); + // Tool peers are permitted without a desktop; the port-mapping viewer aborts on Reject. + let desktop_permission=if screen.init_error.is_none() || !desktop { + crate::system_metrics::Permission::Allow + } else { crate::system_metrics::Permission::Reject }; + // Without a desktop (headless/portal before consent) the port-mapping tool still waits + // for ScreenSources/CaptureChange and aborts otherwise: tool peers get a nominal screen. + let screen_size=crate::wayland::input_size(if screen.init_error.is_none() { (screen.width, screen.height) } else { (1920, 1080) }); + let messages = [ + (8, crate::initialization::encode_capture_change(next_seq(), timestamp, 0, crate::wayland::selected() as i32, "Linux desktop")), + (13, Ok(crate::initialization::envelope(next_seq(), timestamp, 13, &init_actions.metrics_payload(desktop_sound, desktop_permission)))), + (26, Ok(crate::initialization::envelope(next_seq(), timestamp, 26, + &crate::system_metrics::remote_download_path(&home)))), + (18, Ok(crate::initialization::envelope(next_seq(), timestamp, 18, + &crate::capture_setting::codec_negotiation(screen_size, crate::video::display_fps(), init_track.supports(crate::capture_setting::VideoCodec::H265))))), + (7, crate::capture_setting::screen_sources(next_seq(), timestamp, screen_size, init_video.borrow().output, "Linux desktop")), + ]; + if desktop && screen.init_error.is_none() { + let codec = crate::capture_setting::preferred_codec_for(init_track.supports(crate::capture_setting::VideoCodec::H264), + init_track.supports(crate::capture_setting::VideoCodec::H265), open_selection); + init_video.send_if_modified(|current| if current.codec != codec { + log::info!("Desktop codec selected: {:?}", codec); current.codec = codec; true } else { false }); } - log::info!("CONTROL initialization field {} sent", field); + for (field, message) in messages { + if early_desktop_selection && field == 8 { continue; } + if promoted && field == 26 { continue; } + if (!desktop || screen.init_error.is_some()) && field == 18 { continue; } + if desktop && screen.init_error.is_some() && matches!(field,7|8) { continue; } + let message = match message { + Ok(message) => message, + Err(e) => { log::error!("CONTROL initialization encode failed: {}", e); return; } + }; + if let Err(e) = channel.send(&bytes::Bytes::from(message)).await { + log::error!("CONTROL initialization field {} failed: {}", field, e); + return; + } + log::info!("CONTROL initialization field {} sent", field); + } + if desktop { break; } + // The viewer opened the desktop on this tool connection (#99): repeat the + // desktop initialization once the host has started its video and input. + if !role.wait_desktop(&channel).await { return; } + (desktop, promoted) = (true, true); + tokio::spawn(crate::session_lock::report_states(channel.clone(), sequence.clone(), crate::session_lock::watch_locked())); } // Mirror the Linux pointer shape; the video itself carries no pointer. - if !desktop { return; } let mut shapes = crate::cursor::monitor(&screen.display); let mut health = tokio::time::interval(std::time::Duration::from_millis(200)); let mut sent = 0usize; @@ -934,7 +1009,8 @@ impl WebRtcSession { }); let channel_name = lbl.clone(); let input_worker = inp.clone(); - let message_input_session=channel_input_session.clone(); + let message_input_session=channel_role.input(); + let message_role=channel_role.clone(); let terminal_worker = terminal_conn.clone(); let rpc_out = outbound.clone(); let reply_chan = d_reply.clone(); @@ -1168,6 +1244,15 @@ impl WebRtcSession { if let Err(e) = chan.send(&bytes::Bytes::from(message)).await { log::warn!("Settings report failed: {}", e); } } }); + } else if action == sa::CAPTURE_DESKTOP && channel_name == "CONTROL_DATA_CHANNEL" { + if message_role.is_desktop() { + log::info!("SimpleAction 4 (capture_desktop) received on a desktop connection"); + } else if let Some(request) = message_role.request.get() { + log::info!("SimpleAction 4 (capture_desktop) received on a tool connection; starting the desktop"); + request(); + } else { + log::info!("SimpleAction 4 (capture_desktop) not handled: tool connection without a desktop route"); + } } else if action > 1 && action != 15 { let args = action_fields.iter().find_map(|f| match (&f.value, f.number) { (WireValue::Bytes(v), 2) => Some(*v), _ => None }).unwrap_or(&[]); log::info!("SimpleAction {} ({}) not handled: args={}", action, crate::control_protocol::simple_action_name(action).unwrap_or("?"), sa::args_shape(args)); @@ -1290,10 +1375,31 @@ impl WebRtcSession { capture, input, desktop, + role, + controller_mac: false, context, + actions: session_actions, }) } + /// Route for capture_desktop on this tool peer; the caller decides whether it may become the desktop. + pub fn on_desktop_request(&self, request: impl Fn() + Send + Sync + 'static) { + let _ = self.role.request.set(Box::new(request)); + } + + /// Makes this tool peer the desktop: it takes the input lease, starts the video it negotiated + /// (H.264; audio was never negotiated) and repeats the desktop CONTROL initialization. + pub fn promote(&self) { + if self.role.is_desktop() { return; } + let session = self.input.desktop_session(); + if self.pc.connection_state() == RTCPeerConnectionState::Connected { session.connected(); } + let _ = self.role.input.set(session); + if let Some(stream) = self.role.stream.lock().unwrap().take() { stream.spawn(); } + if !cfg!(test) { track_session(self.pc.clone(), self.actions.clone()); } + self.role.active.send_replace(true); + log::info!("Desktop started on an existing tool connection"); + } + /// 应用远端 Offer，生成并返回 Answer pub async fn handle_offer(&self, soac: &SoacMessage) -> Result { // 同步当前的客户端与房间会话上下文 From f5fd9a6bdba51515b32c361ec31ca62995e386fe Mon Sep 17 00:00:00 2001 From: VallierDev <24893073+VallierDev@users.noreply.github.com> Date: Sat, 24 Oct 2026 02:01:34 +0800 Subject: [PATCH 2/5] Demote promoted tool peers and retry capture on capture_desktop (#99) A newer desktop (offer or another capture_desktop) now stops a promoted tool peer's video, input lease, session count and cursor sync while its port mapping, files and terminal keep running; capture_desktop on it promotes it again. Peers offered as desktops still close. capture_desktop on a tool peer whose capture was unavailable when offered retries ScreenCapture::new like a desktop offer, and the peer streams, maps input and reports sizes from the new capture. Co-Authored-By: Claude Opus 5.5 --- src/desktop_control_tests.rs | 14 +++++- src/main.rs | 32 +++++++++----- src/webrtc.rs | 84 +++++++++++++++++++++++++----------- 3 files changed, 93 insertions(+), 37 deletions(-) diff --git a/src/desktop_control_tests.rs b/src/desktop_control_tests.rs index e08afad..0bb5761 100644 --- a/src/desktop_control_tests.rs +++ b/src/desktop_control_tests.rs @@ -228,7 +228,7 @@ async fn capture_desktop_on_a_port_mapping_peer_promotes_it_once() { // The Mac viewer opens the desktop on its existing port-mapping peer (#99). send(&viewer, "CONTROL_DATA_CHANNEL", crate::initialization::envelope(5, 0, 3, &[8, 4])).await; tokio::time::timeout(Duration::from_secs(5), requested.recv()).await.unwrap().unwrap(); - host.promote(); + host.promote(None); assert!(host.role.is_desktop()); assert!(host.role.input().is_some_and(|session| session.is_current())); // CONTROL repeats the desktop initialization for the same viewer. @@ -242,8 +242,18 @@ async fn capture_desktop_on_a_port_mapping_peer_promotes_it_once() { receive(&mut viewer.messages, "CONTROL_DATA_CHANNEL", 3).await; assert!(requested.try_recv().is_err()); let lease = host.role.input().unwrap(); - host.promote(); + host.promote(None); assert!(lease.is_current()); + // A newer desktop demotes it; the port mapping stays, and capture_desktop promotes it again. + host.demote(); + assert!(!host.role.is_desktop() && host.role.input().is_none()); + send(&viewer, "FILE_DATA_CHANNEL", mapping(2, b"after demotion")).await; + receive(&mut viewer.messages, "FILE_DATA_CHANNEL", 27).await; + send(&viewer, "CONTROL_DATA_CHANNEL", crate::initialization::envelope(8, 0, 3, &[8, 4])).await; + tokio::time::timeout(Duration::from_secs(5), requested.recv()).await.unwrap().unwrap(); + host.promote(None); + assert!(host.role.is_desktop() && host.role.input().is_some_and(|session| session.is_current())); + receive(&mut viewer.messages, "CONTROL_DATA_CHANNEL", 13).await; host.pc.close().await.unwrap(); viewer.pc.close().await.unwrap(); viewer.router.lock().await.stop().await.unwrap(); diff --git a/src/main.rs b/src/main.rs index 32a9d37..fdf3026 100644 --- a/src/main.rs +++ b/src/main.rs @@ -963,6 +963,8 @@ async fn serve(mut capture: Arc, input: Arc, shutd for id in superseded_peer_ids(&sessions, &ice, desktop) { if let Some(old) = sessions.remove(&id) { let _ = old.pc.close().await; } } + // A tool peer promoted by capture_desktop stops its desktop but keeps its tools. + if desktop { for old in sessions.values() { old.demote(); } } if desktop && desktop_backend::headless() { log::info!("Desktop connection ignored: headless mode (terminal, file transfer and port mapping remain available)"); continue; @@ -1052,19 +1054,27 @@ async fn serve(mut capture: Arc, input: Arc, shutd log::info!("Desktop request on a tool connection not supported with the Portal backend; open the desktop before port mapping"); continue; } - if peer.capture.init_error.is_some() { - log::info!("Desktop request on a tool connection ignored: desktop capture unavailable"); - continue; - } + let controller_mac = peer.controller_mac; + // Same recovery as a desktop offer: the display may have come up after this peer. + let screen = if peer.capture.init_error.is_some() { + match tokio::task::spawn_blocking(ScreenCapture::new).await { + Ok(Ok(screen)) => { capture = Arc::new(screen); Some(capture.clone()) }, + Ok(Err(error)) => { log::info!("Desktop request on a tool connection ignored: {error}"); continue; }, + Err(_) => { log::warn!("Desktop initialization worker failed"); continue; }, + } + } else { None }; video::warm_auto_probe(); video::check_ffmpeg_usable(); - input.set_mac_controller(peer.controller_mac); - input.bind_portal(peer.capture.portal.clone()); - // One desktop at a time, as for a new desktop offer. Only peers offered as - // desktops close; this peer keeps its port mapping, files or terminal. - let superseded: Vec = sessions.iter().filter(|(id, old)| id.as_str() != ice && old.desktop) + input.set_mac_controller(controller_mac); + input.bind_portal(screen.as_ref().map_or_else(|| sessions[&ice].capture.portal.clone(), |screen| screen.portal.clone())); + // One desktop at a time, as for a new desktop offer. Peers offered as desktops + // close; a promoted tool peer only stops its desktop and keeps its tools. + let superseded: Vec = sessions.iter().filter(|(id, old)| id.as_str() != ice && old.role.is_desktop()) .map(|(id, _)| id.clone()).collect(); - for id in superseded { if let Some(old) = sessions.remove(&id) { let _ = old.pc.close().await; } } - if let Some(peer) = sessions.get(&ice) { peer.promote(); } + for id in superseded { + if sessions[&id].desktop { if let Some(old) = sessions.remove(&id) { let _ = old.pc.close().await; } } + else { sessions[&id].demote(); } + } + if let Some(peer) = sessions.get(&ice) { peer.promote(screen); } } SignalingEvent::Disconnected => { capture_types.clear(); diff --git a/src/webrtc.rs b/src/webrtc.rs index 551bfff..150db91 100644 --- a/src/webrtc.rs +++ b/src/webrtc.rs @@ -47,17 +47,22 @@ pub struct SessionContext { /// capture_desktop instead of a new desktop offer (#99). pub struct DesktopRole { active: tokio::sync::watch::Sender, - input: std::sync::OnceLock>, + input: std::sync::Mutex>>, stream: std::sync::Mutex>, + /// Capture reinitialized at promotion; the peer's own capture was unavailable when offered. + screen: std::sync::Mutex>>, request: std::sync::OnceLock>, } impl DesktopRole { fn new(desktop: bool) -> Self { Self { active: tokio::sync::watch::channel(desktop).0, input: Default::default(), - stream: Default::default(), request: Default::default() } + stream: Default::default(), screen: Default::default(), request: Default::default() } } pub fn is_desktop(&self) -> bool { *self.active.borrow() } - fn input(&self) -> Option> { self.input.get().cloned() } + fn input(&self) -> Option> { self.input.lock().unwrap().clone() } + fn screen_or(&self, screen: &Arc) -> Arc { + self.screen.lock().unwrap().clone().unwrap_or_else(|| screen.clone()) + } /// Waits until this peer is a desktop; false once the channel closes first. async fn wait_desktop(&self, channel: &RTCDataChannel) -> bool { let mut active = self.active.subscribe(); @@ -75,6 +80,7 @@ impl DesktopRole { } /// Live desktop video for one peer; a tool peer keeps it unstarted until it becomes a desktop. +#[derive(Clone)] struct DesktopStream { pc: std::sync::Weak, track: Arc, @@ -102,17 +108,18 @@ pub struct WebRtcSession { actions: Arc, } -/// Counts a desktop session once it is connected (a failed offer is not a session). -fn track_session(end_pc: Arc, actions: Arc) { +/// Counts a desktop session once it is connected (a failed offer is not a session), +/// until the peer closes or a newer desktop demotes it. +fn track_session(end_pc: Arc, actions: Arc, active: tokio::sync::watch::Receiver) { tokio::spawn(async move { use webrtc::peer_connection::peer_connection_state::RTCPeerConnectionState as State; while end_pc.connection_state() != State::Connected { - if matches!(end_pc.connection_state(), State::Closed | State::Failed) { return; } + if matches!(end_pc.connection_state(), State::Closed | State::Failed) || !*active.borrow() { return; } tokio::time::sleep(std::time::Duration::from_millis(200)).await; } let starting = actions.clone(); let _ = tokio::task::spawn_blocking(move || starting.session_started()).await; - while !matches!(end_pc.connection_state(), State::Closed | State::Failed) { + while !matches!(end_pc.connection_state(), State::Closed | State::Failed) && *active.borrow() { tokio::time::sleep(std::time::Duration::from_secs(1)).await; } drop(end_pc); @@ -121,12 +128,16 @@ fn track_session(end_pc: Arc, actions: Arc>, mut active: tokio::sync::watch::Receiver) { let Some(streaming_pc) = self.pc.upgrade() else { return; }; let DesktopStream { track: video_track, portal_desktop, capture_size, capture_display, portal, settings_rx, settings, idr_request, selection, chroma_sync, reload_at_session, .. } = self; - tokio::spawn(async move { + let (capture_size, capture_display, portal) = screen.map_or((capture_size, capture_display, portal), + |screen| ((screen.width, screen.height), screen.display.clone(), screen.portal.clone())); + tokio::spawn(async move { tokio::select! { + _ = active.wait_for(|active| !*active) => log::info!("Live desktop stream stopped: a newer desktop took over"), + _ = async move { let screen = if let Some(desktop) = portal_desktop { if let Err(error) = crate::locked_video::run(&streaming_pc, &video_track, &desktop, selection).await { log::warn!("Locked desktop video unavailable: {error}"); @@ -155,7 +166,8 @@ impl DesktopStream { log::error!("Live desktop stream stopped: {}", e); } } - }); + } => {}, + } }); } } @@ -216,7 +228,7 @@ impl WebRtcSession { ); let role = Arc::new(DesktopRole::new(desktop)); - if desktop { let _ = role.input.set(input.desktop_session()); } + if desktop { *role.input.lock().unwrap() = Some(input.desktop_session()); } let context = Arc::new(Mutex::new(SessionContext::default())); crate::network_diagnostics::start(Arc::downgrade(&pc)); @@ -394,7 +406,7 @@ impl WebRtcSession { let peer_actions = Arc::new(crate::system_actions::PeerActions::default()); let session_actions = peer_actions.clone(); let desktop_sound = desktop && crate::audio::enabled(); - if desktop && !cfg!(test) { track_session(pc.clone(), peer_actions.clone()); } + if desktop && !cfg!(test) { track_session(pc.clone(), peer_actions.clone(), role.active.subscribe()); } // Live encoder switch (SIGUSR2) sends CaptureSettingSync when chroma is downgraded. let (chroma_sync_tx, chroma_sync_rx) = tokio::sync::mpsc::unbounded_channel::>(); let chroma_sync_rx = Arc::new(std::sync::Mutex::new((!cfg!(test)).then_some(chroma_sync_rx))); @@ -408,7 +420,7 @@ impl WebRtcSession { }; drop(streaming_pc); if cfg!(test) { drop(stream); } - else if desktop { stream.spawn(); } + else if desktop { stream.spawn(None, role.active.subscribe()); } else { *role.stream.lock().unwrap() = Some(stream); } let input_clone = input.clone(); let control_capture = capture.clone(); @@ -627,10 +639,10 @@ impl WebRtcSession { if channel.upgrade().is_none_or(|channel| channel.ready_state() != webrtc::data_channel::data_channel_state::RTCDataChannelState::Open) { return; } } } else { - Some(initial_screen.clone()) + Some(capture_role.screen_or(&initial_screen)) } } else { - Some(initial_screen.clone()) + Some(capture_role.screen_or(&initial_screen)) }; let Some(active_screen) = active_screen else { continue; }; let input_size = if active_screen.init_error.is_none() { @@ -922,7 +934,9 @@ impl WebRtcSession { let next_seq = || sequence.fetch_add(1, std::sync::atomic::Ordering::Relaxed); // Default file-transfer folder: the desktop user's home directory. let home = std::env::var("HOME").unwrap_or_else(|_| "/".into()); - let (mut desktop, mut promoted) = (desktop, false); + let (mut desktop, mut promoted, mut lock_reported) = (desktop, false, desktop); + let mut screen = screen; + loop { loop { let timestamp = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_secs(); @@ -968,18 +982,23 @@ impl WebRtcSession { // desktop initialization once the host has started its video and input. if !role.wait_desktop(&channel).await { return; } (desktop, promoted) = (true, true); + screen = role.screen_or(&screen); + } + if !lock_reported { + lock_reported = true; tokio::spawn(crate::session_lock::report_states(channel.clone(), sequence.clone(), crate::session_lock::watch_locked())); } // Mirror the Linux pointer shape; the video itself carries no pointer. let mut shapes = crate::cursor::monitor(&screen.display); let mut health = tokio::time::interval(std::time::Duration::from_millis(200)); let mut sent = 0usize; - loop { + let mut active = role.active.subscribe(); + 'cursor: loop { let current = shapes.borrow_and_update().clone(); if let Some(cursor) = current { let timestamp = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_secs(); let message = crate::cursor::encode_cursor_shape(next_seq(), timestamp, &cursor); - if channel.send(&bytes::Bytes::from(message)).await.is_err() { break; } + if channel.send(&bytes::Bytes::from(message)).await.is_err() { return; } sent += 1; if sent <= 3 || sent % 100 == 0 { log::info!("Cursor shape sent: count={} size={}x{} hotspot={},{} png_bytes={}", sent, cursor.width, cursor.height, cursor.hot_x, cursor.hot_y, cursor.png.len()); @@ -988,12 +1007,17 @@ impl WebRtcSession { loop { tokio::select! { changed = shapes.changed() => { if changed.is_err() { return; } break; }, + _ = active.wait_for(|active| !*active) => break 'cursor, _ = health.tick() => { if channel.ready_state() != webrtc::data_channel::data_channel_state::RTCDataChannelState::Open { return; } }, } } } + // A newer desktop took over this promoted tool peer; wait for capture_desktop again. + if !role.wait_desktop(&channel).await { return; } + screen = role.screen_or(&screen); + } }); }) })); @@ -1002,7 +1026,7 @@ impl WebRtcSession { d.on_message(Box::new(move |msg: DataChannelMessage| { // Keep the whole state: a cached screen waiting for a closed Portal is not ready for input. - let (screen, desktop_ready) = portal_desktop.as_ref().map_or_else(|| (screen.clone(), true), |desktop| { + let (screen, desktop_ready) = portal_desktop.as_ref().map_or_else(|| (channel_role.screen_or(&screen), true), |desktop| { let state = desktop.snapshot(); let ready = state.ready(); (state.screen, ready) @@ -1389,17 +1413,29 @@ impl WebRtcSession { /// Makes this tool peer the desktop: it takes the input lease, starts the video it negotiated /// (H.264; audio was never negotiated) and repeats the desktop CONTROL initialization. - pub fn promote(&self) { + /// `screen` replaces a capture that was unavailable when the peer was offered. + pub fn promote(&self, screen: Option>) { if self.role.is_desktop() { return; } + if let Some(screen) = screen { *self.role.screen.lock().unwrap() = Some(screen); } let session = self.input.desktop_session(); if self.pc.connection_state() == RTCPeerConnectionState::Connected { session.connected(); } - let _ = self.role.input.set(session); - if let Some(stream) = self.role.stream.lock().unwrap().take() { stream.spawn(); } - if !cfg!(test) { track_session(self.pc.clone(), self.actions.clone()); } + *self.role.input.lock().unwrap() = Some(session); self.role.active.send_replace(true); + let stream = self.role.stream.lock().unwrap().clone(); + if let Some(stream) = stream { stream.spawn(self.role.screen.lock().unwrap().clone(), self.role.active.subscribe()); } + if !cfg!(test) { track_session(self.pc.clone(), self.actions.clone(), self.role.active.subscribe()); } log::info!("Desktop started on an existing tool connection"); } + /// A newer desktop took over a promoted tool peer: stop its video, input and session count, + /// keep the connection's tools. capture_desktop on it promotes it again. + pub fn demote(&self) { + if self.desktop || !self.role.is_desktop() { return; } + self.role.active.send_replace(false); + if let Some(session) = self.role.input.lock().unwrap().take() { session.disconnected(); } + log::info!("Desktop on a tool connection stopped: a newer desktop took over"); + } + /// 应用远端 Offer，生成并返回 Answer pub async fn handle_offer(&self, soac: &SoacMessage) -> Result { // 同步当前的客户端与房间会话上下文 From c1201a1f3f229c6542a61d8a38dbbd05b2d0b1be Mon Sep 17 00:00:00 2001 From: VallierDev <24893073+VallierDev@users.noreply.github.com> Date: Sat, 24 Oct 2026 02:08:27 +0800 Subject: [PATCH 3/5] Follow the desktop role in the TEXT clipboard of promoted tool peers (#99) The TEXT clipboard bridge of a promoted tool peer now stops when a newer desktop demotes it (no host offers, incoming updates are dropped) and starts again on the next capture_desktop. After promotion CONTROL declares ff_clipboard again once the bridge is ready, since the viewer's echo was answered before the clipboard existed. Co-Authored-By: Claude Opus 5.5 --- src/webrtc.rs | 173 ++++++++++++++++++++++++++++++-------------------- 1 file changed, 104 insertions(+), 69 deletions(-) diff --git a/src/webrtc.rs b/src/webrtc.rs index 150db91..236616a 100644 --- a/src/webrtc.rs +++ b/src/webrtc.rs @@ -127,6 +127,31 @@ fn track_session(end_pc: Arc, actions: Arc, Arc, Arc); + +/// A promoted tool peer starts its TEXT clipboard after the viewer's echo was answered without +/// ff_clipboard; declare it again on CONTROL once the bridge is ready, as the Portal path does. +fn replay_clipboard_echo(channel: Arc, sequence: Arc, + (echo_request, echo_seen, clipboard_ready): EchoReplay, role: Arc) { + tokio::spawn(async move { + for _ in 0..100 { + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + if !role.is_desktop() || channel.ready_state() != webrtc::data_channel::data_channel_state::RTCDataChannelState::Open { return; } + if !clipboard_ready.load(std::sync::atomic::Ordering::Acquire) { continue; } + // echo_seen is published after the request seq, so reading it first gives a current seq. + if echo_seen.load(std::sync::atomic::Ordering::Acquire) { + let request_seq = echo_request.load(std::sync::atomic::Ordering::Acquire); + let timestamp = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_secs(); + let response = crate::control_protocol::encode_echo_response_with_clipboard( + sequence.fetch_add(1, std::sync::atomic::Ordering::Relaxed), timestamp, request_seq, true); + let _ = channel.send(&bytes::Bytes::from(response)).await; + log::info!("CONTROL clipboard capability declared after capture_desktop"); + } + return; + } + }); +} + impl DesktopStream { /// Encode current frames only while the peer is connected and still the desktop. fn spawn(self, screen: Option>, mut active: tokio::sync::watch::Receiver) { @@ -775,76 +800,84 @@ impl WebRtcSession { // channel-open callback or delay CONTROL message dispatch. tokio::spawn(async move { if channel.label() == "TEXT_DATA_CHANNEL" { - if !desktop && !role.wait_desktop(&channel).await { return; } - let screen = if let Some(desktop) = &portal_desktop { - let mut updates = desktop.subscribe(); - let Some(_) = crate::desktop_session::wait_ready(&mut updates, &channel).await else { return; }; - let mut locked = desktop.lock_state(); - loop { - let state = desktop.snapshot(); - if !state.waiting && state.screen.init_error.is_some() { return; } - if let Some(session) = state.screen.portal.as_ref() { - if session.is_alive() && *locked.borrow() == Some(false) { - break state.screen; - } - } - tokio::select! { - _ = tokio::time::sleep(std::time::Duration::from_millis(100)) => {}, - changed = locked.changed() => if changed.is_err() { return; }, - _ = async { - loop { - if matches!(channel.ready_state(), webrtc::data_channel::data_channel_state::RTCDataChannelState::Closing - | webrtc::data_channel::data_channel_state::RTCDataChannelState::Closed) { return; } - tokio::time::sleep(std::time::Duration::from_millis(100)).await; - } - } => return, - } - } - } else { screen }; - let bridge = if let Some(session)=screen.portal.clone() { crate::clipboard::start_portal(session) } else { crate::clipboard::start(screen.display.clone()) }; - *clipboard_sink.lock().unwrap() = Some(bridge.incoming); - let clipboard_channel = channel.clone(); - let clipboard_sequence = sequence.clone(); - let clipboard_cleanup = clipboard_sink.clone(); - let ready = bridge.ready.await.unwrap_or(false); - clipboard_ready.store(ready,std::sync::atomic::Ordering::Relaxed); - tokio::spawn(async move { - if !ready { - *clipboard_cleanup.lock().unwrap() = None; - log::warn!("Native text clipboard unavailable"); return; - } - log::info!("Native text clipboard bridge ready"); - let mut updates = bridge.outgoing; - let mut health = tokio::time::interval(std::time::Duration::from_millis(200)); - loop { - tokio::select! { - _ = health.tick() => { - if clipboard_channel.ready_state() != webrtc::data_channel::data_channel_state::RTCDataChannelState::Open { break; } - }, - changed = updates.changed() => { - if changed.is_err() { break; } - let text = updates.borrow_and_update().clone(); - let Some(text) = text else { continue; }; - { - let request_id=clipboard_sequence.fetch_add(1,std::sync::atomic::Ordering::Relaxed); - let payload = match rich_outgoing.lock().unwrap().offer(request_id,text.clone()) { - Ok(payload)=>payload,Err(reason)=>{log::warn!("Clipboard offer rejected: {}",reason);continue;} - }; - let timestamp = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_secs(); - let message = crate::initialization::envelope(clipboard_sequence.fetch_add(1,std::sync::atomic::Ordering::Relaxed),timestamp,21,&payload); - if !matches!(tokio::time::timeout(std::time::Duration::from_secs(2),clipboard_channel.send(&bytes::Bytes::from(message))).await,Ok(Ok(_))) { - log::warn!("Native clipboard RPC send failed"); break; - } - } - log::info!("Native clipboard text format offered: bytes={}",text.len()); - } - } - } - *clipboard_cleanup.lock().unwrap() = None; - clipboard_ready.store(false,std::sync::atomic::Ordering::Relaxed); - }); - return; + loop { + if !role.is_desktop() && !role.wait_desktop(&channel).await { return; } + let screen = if let Some(desktop) = &portal_desktop { + let mut updates = desktop.subscribe(); + let Some(_) = crate::desktop_session::wait_ready(&mut updates, &channel).await else { return; }; + let mut locked = desktop.lock_state(); + loop { + let state = desktop.snapshot(); + if !state.waiting && state.screen.init_error.is_some() { return; } + if let Some(session) = state.screen.portal.as_ref() { + if session.is_alive() && *locked.borrow() == Some(false) { + break state.screen; + } + } + tokio::select! { + _ = tokio::time::sleep(std::time::Duration::from_millis(100)) => {}, + changed = locked.changed() => if changed.is_err() { return; }, + _ = async { + loop { + if matches!(channel.ready_state(), webrtc::data_channel::data_channel_state::RTCDataChannelState::Closing + | webrtc::data_channel::data_channel_state::RTCDataChannelState::Closed) { return; } + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } + } => return, + } + } + } else { role.screen_or(&screen) }; + let bridge = if let Some(session)=screen.portal.clone() { crate::clipboard::start_portal(session) } else { crate::clipboard::start(screen.display.clone()) }; + *clipboard_sink.lock().unwrap() = Some(bridge.incoming); + let clipboard_channel = channel.clone(); + let clipboard_sequence = sequence.clone(); + let clipboard_cleanup = clipboard_sink.clone(); + let rich_outgoing = rich_outgoing.clone(); + let ready = bridge.ready.await.unwrap_or(false); + clipboard_ready.store(ready,std::sync::atomic::Ordering::Relaxed); + { + if !ready { + *clipboard_cleanup.lock().unwrap() = None; + log::warn!("Native text clipboard unavailable"); return; + } + log::info!("Native text clipboard bridge ready"); + let mut updates = bridge.outgoing; + let mut health = tokio::time::interval(std::time::Duration::from_millis(200)); + let mut active = role.active.subscribe(); + loop { + tokio::select! { + _ = health.tick() => { + if clipboard_channel.ready_state() != webrtc::data_channel::data_channel_state::RTCDataChannelState::Open { break; } + }, + _ = async { let _ = active.wait_for(|active| !*active).await; } => break, + changed = updates.changed() => { + if changed.is_err() { break; } + let text = updates.borrow_and_update().clone(); + let Some(text) = text else { continue; }; + { + let request_id=clipboard_sequence.fetch_add(1,std::sync::atomic::Ordering::Relaxed); + let payload = match rich_outgoing.lock().unwrap().offer(request_id,text.clone()) { + Ok(payload)=>payload,Err(reason)=>{log::warn!("Clipboard offer rejected: {}",reason);continue;} + }; + let timestamp = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_secs(); + let message = crate::initialization::envelope(clipboard_sequence.fetch_add(1,std::sync::atomic::Ordering::Relaxed),timestamp,21,&payload); + if !matches!(tokio::time::timeout(std::time::Duration::from_secs(2),clipboard_channel.send(&bytes::Bytes::from(message))).await,Ok(Ok(_))) { + log::warn!("Native clipboard RPC send failed"); break; + } + } + log::info!("Native clipboard text format offered: bytes={}",text.len()); + } + } + } + *clipboard_cleanup.lock().unwrap() = None; + clipboard_ready.store(false,std::sync::atomic::Ordering::Relaxed); + } + if channel.ready_state() != webrtc::data_channel::data_channel_state::RTCDataChannelState::Open { return; } + // A newer desktop took over this promoted tool peer; capture_desktop restarts the clipboard. + log::info!("Native text clipboard stopped: a newer desktop took over"); + } } + let replay = (last_echo_request.clone(), echo_seen.clone(), clipboard_ready.clone()); let early_desktop_selection = desktop && portal_desktop.is_some(); if early_desktop_selection { // Android 4.43.0's unlock presenter requires CaptureChange @@ -983,6 +1016,7 @@ impl WebRtcSession { if !role.wait_desktop(&channel).await { return; } (desktop, promoted) = (true, true); screen = role.screen_or(&screen); + replay_clipboard_echo(channel.clone(), sequence.clone(), replay.clone(), role.clone()); } if !lock_reported { lock_reported = true; @@ -1017,6 +1051,7 @@ impl WebRtcSession { // A newer desktop took over this promoted tool peer; wait for capture_desktop again. if !role.wait_desktop(&channel).await { return; } screen = role.screen_or(&screen); + replay_clipboard_echo(channel.clone(), sequence.clone(), replay.clone(), role.clone()); } }); }) From ce03b4c1c0da6cd6c725c15064e87d2642bb6d6d Mon Sep 17 00:00:00 2001 From: VallierDev <24893073+VallierDev@users.noreply.github.com> Date: Sat, 24 Oct 2026 02:14:34 +0800 Subject: [PATCH 4/5] One session tracker per promoted peer; refresh X11 geometry on capture_desktop (#99) A peer's session tracker now starts once and follows every promotion and demotion, so a quick demote and re-promote cannot end the desktop count of a stream that is still running. capture_desktop rebuilds the capture when the X11 root size changed since the peer opened, as a desktop offer does. Co-Authored-By: Claude Opus 5.5 --- src/main.rs | 8 ++++++-- src/webrtc.rs | 44 ++++++++++++++++++++++++++++---------------- 2 files changed, 34 insertions(+), 18 deletions(-) diff --git a/src/main.rs b/src/main.rs index fdf3026..f062df9 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1055,8 +1055,12 @@ async fn serve(mut capture: Arc, input: Arc, shutd continue; } let controller_mac = peer.controller_mac; - // Same recovery as a desktop offer: the display may have come up after this peer. - let screen = if peer.capture.init_error.is_some() { + let current = peer.desktop_capture(); + // Same recovery as a desktop offer: the display may have come up or been resized after this peer. + let resized = current.init_error.is_none() && current.backend == desktop_backend::DesktopBackend::X11 + && tokio::task::spawn_blocking(crate::capture::x11_size).await.ok().flatten().is_some_and(|size| size != (current.width, current.height)); + if resized { log::info!("Desktop geometry changed since this connection opened; reinitializing capture"); } + let screen = if current.init_error.is_some() || resized { match tokio::task::spawn_blocking(ScreenCapture::new).await { Ok(Ok(screen)) => { capture = Arc::new(screen); Some(capture.clone()) }, Ok(Err(error)) => { log::info!("Desktop request on a tool connection ignored: {error}"); continue; }, diff --git a/src/webrtc.rs b/src/webrtc.rs index 236616a..1c8edf3 100644 --- a/src/webrtc.rs +++ b/src/webrtc.rs @@ -51,12 +51,14 @@ pub struct DesktopRole { stream: std::sync::Mutex>, /// Capture reinitialized at promotion; the peer's own capture was unavailable when offered. screen: std::sync::Mutex>>, + /// One session tracker per peer follows every promotion and demotion. + tracking: std::sync::atomic::AtomicBool, request: std::sync::OnceLock>, } impl DesktopRole { fn new(desktop: bool) -> Self { Self { active: tokio::sync::watch::channel(desktop).0, input: Default::default(), - stream: Default::default(), screen: Default::default(), request: Default::default() } + stream: Default::default(), screen: Default::default(), tracking: Default::default(), request: Default::default() } } pub fn is_desktop(&self) -> bool { *self.active.borrow() } fn input(&self) -> Option> { self.input.lock().unwrap().clone() } @@ -108,22 +110,29 @@ pub struct WebRtcSession { actions: Arc, } -/// Counts a desktop session once it is connected (a failed offer is not a session), -/// until the peer closes or a newer desktop demotes it. -fn track_session(end_pc: Arc, actions: Arc, active: tokio::sync::watch::Receiver) { +/// Counts a desktop session once it is connected (a failed offer is not a session), while the +/// peer is the desktop. Started once per peer, so a quick demote and re-promote cannot end the +/// count of a session that is still streaming. +fn track_session(end_pc: Arc, actions: Arc, role: &DesktopRole) { + if role.tracking.swap(true, std::sync::atomic::Ordering::AcqRel) { return; } + let active = role.active.subscribe(); tokio::spawn(async move { use webrtc::peer_connection::peer_connection_state::RTCPeerConnectionState as State; - while end_pc.connection_state() != State::Connected { - if matches!(end_pc.connection_state(), State::Closed | State::Failed) || !*active.borrow() { return; } - tokio::time::sleep(std::time::Duration::from_millis(200)).await; - } - let starting = actions.clone(); - let _ = tokio::task::spawn_blocking(move || starting.session_started()).await; - while !matches!(end_pc.connection_state(), State::Closed | State::Failed) && *active.borrow() { - tokio::time::sleep(std::time::Duration::from_secs(1)).await; + loop { + while end_pc.connection_state() != State::Connected || !*active.borrow() { + if matches!(end_pc.connection_state(), State::Closed | State::Failed) { return; } + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + } + let starting = actions.clone(); + let _ = tokio::task::spawn_blocking(move || starting.session_started()).await; + while !matches!(end_pc.connection_state(), State::Closed | State::Failed) && *active.borrow() { + tokio::time::sleep(std::time::Duration::from_secs(1)).await; + } + let closed = matches!(end_pc.connection_state(), State::Closed | State::Failed); + let ending = actions.clone(); + let _ = tokio::task::spawn_blocking(move || ending.session_ended()).await; + if closed { return; } } - drop(end_pc); - let _ = tokio::task::spawn_blocking(move || actions.session_ended()).await; }); } @@ -431,7 +440,7 @@ impl WebRtcSession { let peer_actions = Arc::new(crate::system_actions::PeerActions::default()); let session_actions = peer_actions.clone(); let desktop_sound = desktop && crate::audio::enabled(); - if desktop && !cfg!(test) { track_session(pc.clone(), peer_actions.clone(), role.active.subscribe()); } + if desktop && !cfg!(test) { track_session(pc.clone(), peer_actions.clone(), &role); } // Live encoder switch (SIGUSR2) sends CaptureSettingSync when chroma is downgraded. let (chroma_sync_tx, chroma_sync_rx) = tokio::sync::mpsc::unbounded_channel::>(); let chroma_sync_rx = Arc::new(std::sync::Mutex::new((!cfg!(test)).then_some(chroma_sync_rx))); @@ -1458,10 +1467,13 @@ impl WebRtcSession { self.role.active.send_replace(true); let stream = self.role.stream.lock().unwrap().clone(); if let Some(stream) = stream { stream.spawn(self.role.screen.lock().unwrap().clone(), self.role.active.subscribe()); } - if !cfg!(test) { track_session(self.pc.clone(), self.actions.clone(), self.role.active.subscribe()); } + if !cfg!(test) { track_session(self.pc.clone(), self.actions.clone(), &self.role); } log::info!("Desktop started on an existing tool connection"); } + /// The capture this peer's desktop uses: its own, or the one reinitialized at promotion. + pub fn desktop_capture(&self) -> Arc { self.role.screen_or(&self.capture) } + /// A newer desktop took over a promoted tool peer: stop its video, input and session count, /// keep the connection's tools. capture_desktop on it promotes it again. pub fn demote(&self) { From 305d42f1866cdf09430114d8804b7ff233c54481 Mon Sep 17 00:00:00 2001 From: VallierDev <24893073+VallierDev@users.noreply.github.com> Date: Sat, 24 Oct 2026 02:19:14 +0800 Subject: [PATCH 5/5] Abort a promoted peer's stream on demotion and re-promotion (#99) The stream task's abort handle is kept on the peer; demotion aborts it and a new promotion replaces it, so a demote and re-promote faster than the watch can report never runs two encoders on one track. Dropping the stream stops its encoder (kill_on_drop) and capture workers (stop-on-drop). Co-Authored-By: Claude Opus 5.5 --- src/webrtc.rs | 21 ++++++++++++++------- 1 file changed, 14 insertions(+), 7 deletions(-) diff --git a/src/webrtc.rs b/src/webrtc.rs index 1c8edf3..96ea45f 100644 --- a/src/webrtc.rs +++ b/src/webrtc.rs @@ -49,6 +49,9 @@ pub struct DesktopRole { active: tokio::sync::watch::Sender, input: std::sync::Mutex>>, stream: std::sync::Mutex>, + /// The running stream of a promoted peer; aborted on demotion so a quick re-promotion + /// never leaves two encoders on one track. + stream_task: std::sync::Mutex>, /// Capture reinitialized at promotion; the peer's own capture was unavailable when offered. screen: std::sync::Mutex>>, /// One session tracker per peer follows every promotion and demotion. @@ -58,7 +61,7 @@ pub struct DesktopRole { impl DesktopRole { fn new(desktop: bool) -> Self { Self { active: tokio::sync::watch::channel(desktop).0, input: Default::default(), - stream: Default::default(), screen: Default::default(), tracking: Default::default(), request: Default::default() } + stream: Default::default(), stream_task: Default::default(), screen: Default::default(), tracking: Default::default(), request: Default::default() } } pub fn is_desktop(&self) -> bool { *self.active.borrow() } fn input(&self) -> Option> { self.input.lock().unwrap().clone() } @@ -163,13 +166,13 @@ fn replay_clipboard_echo(channel: Arc, sequence: Arc>, mut active: tokio::sync::watch::Receiver) { - let Some(streaming_pc) = self.pc.upgrade() else { return; }; + fn spawn(self, screen: Option>, mut active: tokio::sync::watch::Receiver) -> Option { + let streaming_pc = self.pc.upgrade()?; let DesktopStream { track: video_track, portal_desktop, capture_size, capture_display, portal, settings_rx, settings, idr_request, selection, chroma_sync, reload_at_session, .. } = self; let (capture_size, capture_display, portal) = screen.map_or((capture_size, capture_display, portal), |screen| ((screen.width, screen.height), screen.display.clone(), screen.portal.clone())); - tokio::spawn(async move { tokio::select! { + Some(tokio::spawn(async move { tokio::select! { _ = active.wait_for(|active| !*active) => log::info!("Live desktop stream stopped: a newer desktop took over"), _ = async move { let screen = if let Some(desktop) = portal_desktop { @@ -201,7 +204,7 @@ impl DesktopStream { } } } => {}, - } }); + } }).abort_handle()) } } @@ -454,7 +457,7 @@ impl WebRtcSession { }; drop(streaming_pc); if cfg!(test) { drop(stream); } - else if desktop { stream.spawn(None, role.active.subscribe()); } + else if desktop { let _ = stream.spawn(None, role.active.subscribe()); } else { *role.stream.lock().unwrap() = Some(stream); } let input_clone = input.clone(); let control_capture = capture.clone(); @@ -1466,7 +1469,10 @@ impl WebRtcSession { *self.role.input.lock().unwrap() = Some(session); self.role.active.send_replace(true); let stream = self.role.stream.lock().unwrap().clone(); - if let Some(stream) = stream { stream.spawn(self.role.screen.lock().unwrap().clone(), self.role.active.subscribe()); } + if let Some(stream) = stream { + let task = stream.spawn(self.role.screen.lock().unwrap().clone(), self.role.active.subscribe()); + if let Some(old) = std::mem::replace(&mut *self.role.stream_task.lock().unwrap(), task) { old.abort(); } + } if !cfg!(test) { track_session(self.pc.clone(), self.actions.clone(), &self.role); } log::info!("Desktop started on an existing tool connection"); } @@ -1479,6 +1485,7 @@ impl WebRtcSession { pub fn demote(&self) { if self.desktop || !self.role.is_desktop() { return; } self.role.active.send_replace(false); + if let Some(task) = self.role.stream_task.lock().unwrap().take() { task.abort(); } if let Some(session) = self.role.input.lock().unwrap().take() { session.disconnected(); } log::info!("Desktop on a tool connection stopped: a newer desktop took over"); }