diff --git a/src/desktop_control_tests.rs b/src/desktop_control_tests.rs index 7f92760..0bb5761 100644 --- a/src/desktop_control_tests.rs +++ b/src/desktop_control_tests.rs @@ -215,3 +215,47 @@ 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(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. + 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(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(); + }).await.expect("capture_desktop promotion exceeded deadline"); +} diff --git a/src/main.rs b/src/main.rs index be53393..f062df9 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" { @@ -960,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; @@ -981,7 +986,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 +1036,50 @@ 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; + } + let controller_mac = peer.controller_mac; + 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; }, + Err(_) => { log::warn!("Desktop initialization worker failed"); continue; }, + } + } else { None }; + video::warm_auto_probe(); video::check_ffmpeg_usable(); + 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 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(); 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..96ea45f 100644 --- a/src/webrtc.rs +++ b/src/webrtc.rs @@ -42,12 +42,170 @@ 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::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. + 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(), 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() } + 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(); + 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. +#[derive(Clone)] +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), 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; + 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; } + } + }); +} + +type EchoReplay = (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) -> 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())); + 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 { + 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); + } + } + } => {}, + } }).abort_handle()) + } } impl WebRtcSession { @@ -106,7 +264,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 { *role.input.lock().unwrap() = Some(input.desktop_session()); } let context = Arc::new(Mutex::new(SessionContext::default())); crate::network_diagnostics::start(Arc::downgrade(&pc)); @@ -164,13 +323,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 +441,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(), &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 = (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 { let _ = stream.spawn(None, role.active.subscribe()); } + 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 +516,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 +524,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 +554,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 +612,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 +630,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) { @@ -563,10 +676,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() { @@ -675,7 +788,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,82 +805,91 @@ 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; } - 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 @@ -853,58 +976,75 @@ 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, 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(); + // 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 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); } - 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); + screen = role.screen_or(&screen); + replay_clipboard_echo(channel.clone(), sequence.clone(), replay.clone(), role.clone()); + } + 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. - 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; - 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()); @@ -913,12 +1053,18 @@ 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); + replay_clipboard_echo(channel.clone(), sequence.clone(), replay.clone(), role.clone()); + } }); }) })); @@ -927,14 +1073,15 @@ 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) }); 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 +1315,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 +1446,50 @@ 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. + /// `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(); } + *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 { + 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"); + } + + /// 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) { + 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"); + } + /// 应用远端 Offer，生成并返回 Answer pub async fn handle_offer(&self, soac: &SoacMessage) -> Result { // 同步当前的客户端与房间会话上下文