TheLanc3/rustdesk
1
0
Fork 0
mirror of https://github.com/rustdesk/rustdesk.git synced 2026-09-05 23:51:04 +03:00
Code Issues Packages Projects Releases Wiki Activity

port forward: apply the whole-branch review

Browse Source 
Correctness:
- listen(): the Legacy arm is merged with the Claimed arm. On its own it
  ignored outcome.local_eof, so a client that hung up during login still
  got a target connect, an audit record and a CM row on the controlled
  side, and ignored outcome.mux, so a peer upgraded while a legacy window
  stayed open answered as a tunnel while the controller went raw.
- Refusal dialogs are deduplicated per quiet spell (10 s) rather than per
  tunnel lifetime; the lifetime set went silent for the rest of a
  long-lived window after the first burst.
- Android's CM listener handles UpdatePortForward; it fell into `_ => {}`.
- relay_socket_to_tunnel reads into one scratch buffer per channel and
  sends an exact-size copy. A frame owning its 64 KiB read allocation
  pinned it until sent, once per byte on interactive traffic.

Consistency and cleanups:
- The controlled side's refusal text is the raw pipe's wording, RDP
  substitution included.
- connection.rs: the PortForwardChannel arm is a one-line hook, the CM
  label is pushed from the 1 s tick alone, and the unreachable inner.tx
  fall-through is gone.
- The Ready enum is removed; wait_ready() returns Option<Claim>.
- SendCredit::add wakes with notify_one alone.
- on_ui_command() replaces the two ui_receiver handlers in listen().
- TunnelHandle is no longer re-exported (unused-import warning).

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01EZ49AbZJYfm8NTp5yDPMab
This commit is contained in:
rustdesk
2026-09-04 09:13:16 +08:00
parent e21d875ae7
commit f69e706cfa
5 changed files with 175 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files

63
src/port_forward.rs
View File

@@ -97,24 +97,15 @@ pub async fn listen(
// never shadow it.
Ok((forward, peer_addr)) = listener.accept() => {
log::debug!("new connection from {:?}", peer_addr);
let claim = tunnel.try_claim();
let claim = match claim {
let claim = match tunnel.try_claim() {
Claim::Wait => {
// Keep servicing the UI while the establishing accept holds the
// prompt. `None` means establishment failed: this accept is dropped.
let resolved: Option<Claim> = loop {
let resolved = loop {
tokio::select! {
// `tokio::select!` brings its own `Poll::{Ready, Pending}` into
// scope, so `Ready` must be qualified here or it resolves to that.
r = tunnel.wait_ready() => break match r {
crate::port_forward_mux::Ready::Muxed(h) => Some(Claim::Muxed(h)),
crate::port_forward_mux::Ready::Legacy => Some(Claim::Legacy),
crate::port_forward_mux::Ready::Failed => None,
},
d = ui_receiver.recv() => match d {
Some(Data::Close) => return Ok(()),
Some(Data::NewRDP) => run_rdp(addr.port(), &rdp_display_name(&lc, &id)),
_ => {}
r = tunnel.wait_ready() => break r,
d = ui_receiver.recv() => if on_ui_command(d, addr.port(), &lc, &id) {
return Ok(());
},
}
};
@@ -134,18 +125,10 @@ pub async fn listen(
log::debug!("cannot open channel for {:?}: {}", peer_addr, e);
}
}
Claim::Legacy => {
lc.write().unwrap().port_forward = (remote_host.clone(), remote_port);
let mut forward = Framed::new(forward, BytesCodec::new());
let mut close_port_forward = false;
match connect_and_login(&id, &password, &mut ui_receiver, interface.clone(), &mut forward, key, token, is_rdp, &mut close_port_forward).await {
Ok(Some(outcome)) => run_legacy(outcome, forward, peer_addr, interface.clone()),
_ if close_port_forward => break,
Err(err) => interface.on_establish_connection_error(err.to_string()),
_ => {}
}
}
Claim::Claimed => {
// A `Legacy` window logs in for every accept, as it always did,
// and publishes the outcome the same way the claiming accept
// does: a peer upgraded since the window opened becomes a tunnel.
Claim::Claimed | Claim::Legacy => {
lc.write().unwrap().port_forward = (remote_host.clone(), remote_port);
let mut forward = Framed::new(forward, BytesCodec::new());
let mut close_port_forward = false;
@@ -182,23 +165,27 @@ pub async fn listen(
Claim::Wait => continue,
}
}
d = ui_receiver.recv() => {
match d {
Some(Data::Close) => {
break;
}
Some(Data::NewRDP) => {
println!("receive run_rdp from ui_receiver");
run_rdp(addr.port(), &rdp_display_name(&lc, &id));
}
_ => {}
}
}
d = ui_receiver.recv() => if on_ui_command(d, addr.port(), &lc, &id) {
break;
},
}
}
Ok(())
}
/// Commands the window sends its listener. `true` means stop listening: the
/// window is closing, or its sender is gone.
fn on_ui_command(d: Option<Data>, port: u16, lc: &Arc<RwLock<LoginConfigHandler>>, id: &str) -> bool {
match d {
Some(Data::Close) | None => true,
Some(Data::NewRDP) => {
run_rdp(port, &rdp_display_name(lc, id));
false
}
_ => false,
}
}
/// Today's raw pipe, for peers without multiplexing.
fn run_legacy(
outcome: LoginOutcome,

116
src/port_forward_mux.rs
View File

@@ -1,5 +1,5 @@
use hbb_common::{
bytes::{BufMut, Bytes, BytesMut},
bytes::Bytes,
log,
message_proto::*,
tokio::{
@@ -118,7 +118,6 @@ impl SendCredit {
let mut credit = self.credit.lock().unwrap();
*credit = credit.saturating_add(n).min(MAX_SEND_CREDIT);
}
self.notify.notify_waiters();
self.notify.notify_one();
}
@@ -260,18 +259,17 @@ async fn relay_socket_to_tunnel<R: AsyncRead + Unpin>(
mut cancel: watch::Receiver<bool>,
) -> RelayEnd {
let mut reader = std::io::Cursor::new(prebuf).chain(reader);
// One scratch buffer per channel and an exact-size copy per frame: a frame
// that owned its read allocation would pin up to MAX_FRAME until sent,
// whatever its length, and interactive traffic is mostly tiny frames.
let mut scratch = vec![0u8; MAX_FRAME];
loop {
let allow = tokio::select! {
n = credit.take(MAX_FRAME) => n,
_ = cancel.changed() => return RelayEnd::Cancelled,
};
let mut buf = BytesMut::with_capacity(allow);
let mut limited = (&mut buf).limit(allow);
let got = tokio::select! {
r = reader.read_buf(&mut limited) => match r {
Ok(n) => n,
Err(_) => 0,
},
r = reader.read(&mut scratch[..allow]) => r.unwrap_or(0),
_ = cancel.changed() => {
credit.add(allow as u32);
return RelayEnd::Cancelled;
@@ -284,7 +282,8 @@ async fn relay_socket_to_tunnel<R: AsyncRead + Unpin>(
if got == 0 {
return RelayEnd::LocalEof;
}
if sink.send_ordered(data_msg(id, buf.freeze())).await.is_err() {
let frame = data_msg(id, Bytes::copy_from_slice(&scratch[..got]));
if sink.send_ordered(frame).await.is_err() {
return RelayEnd::TunnelGone;
}
}
@@ -381,7 +380,7 @@ pub async fn run_channel<R, W>(
}
#[cfg(not(any(target_os = "android", target_os = "ios")))]
pub use tunnel::{Claim, Ready, Tunnel, TunnelHandle};
pub use tunnel::{Claim, Tunnel};
#[cfg(not(any(target_os = "android", target_os = "ios")))]
mod tunnel {
@@ -394,15 +393,20 @@ mod tunnel {
Stream,
};
use std::{
collections::{HashMap, HashSet},
collections::HashMap,
sync::atomic::{AtomicI32, Ordering},
time::Duration,
};
/// One dialog per distinct reason, and never an unbounded number: the
/// message text comes from the peer.
const MAX_REPORTED_OPEN_ERRORS: usize = 8;
/// A refused reason is shown once, then not again until it has been quiet
/// for this long: a page load's dozen refusals make one dialog, and a
/// target that breaks again hours later is reported again.
const REPORT_AGAIN_AFTER: Duration = Duration::from_secs(10);
/// Distinct reasons remembered at once, so the dialog count stays bounded
/// however the peer varies the text it sends.
pub(super) const MAX_REPORTED_OPEN_ERRORS: usize = 8;
// Internal state only; `Claim` and `Ready` are the API listeners see.
// Internal state only; `Claim` is the API listeners see.
enum TunnelState {
Unset,
Establishing,
@@ -418,12 +422,6 @@ mod tunnel {
Legacy,
}
pub enum Ready {
Muxed(Arc<TunnelHandle>),
Legacy,
Failed,
}
/// One per port-forward window. `watch::Sender::send_if_modified` is the
/// atomic claim; nothing is ever awaited while it runs.
pub struct Tunnel {
@@ -457,22 +455,20 @@ mod tunnel {
outcome
}
pub async fn wait_ready(&self) -> Ready {
/// What the establishing accept ended up with; `None` when it failed.
pub async fn wait_ready(&self) -> Option<Claim> {
let mut rx = self.state.subscribe();
loop {
let ready = match &*rx.borrow_and_update() {
TunnelState::Muxed(h) => Some(Ready::Muxed(h.clone())),
TunnelState::Legacy => Some(Ready::Legacy),
match &*rx.borrow_and_update() {
TunnelState::Muxed(h) => return Some(Claim::Muxed(h.clone())),
TunnelState::Legacy => return Some(Claim::Legacy),
// `Unset` here means the tunnel died between the claim
// and this wait; the waiter treats it as a failure.
TunnelState::Failed | TunnelState::Unset => Some(Ready::Failed),
TunnelState::Establishing => None,
};
if let Some(r) = ready {
return r;
TunnelState::Failed | TunnelState::Unset => return None,
TunnelState::Establishing => {}
}
if rx.changed().await.is_err() {
return Ready::Failed;
return None;
}
}
}
@@ -515,7 +511,7 @@ mod tunnel {
sink: FrameSink,
channels: Mutex<HashMap<i32, ChannelEntry>>,
next_id: AtomicI32,
reported: Mutex<HashSet<String>>,
reported: Mutex<HashMap<String, Instant>>,
}
impl TunnelHandle {
@@ -626,17 +622,28 @@ mod tunnel {
}
}
/// The peer's reason for refusing a channel, the first time we see it.
/// One page load can have a dozen connections refused for the same
/// reason, and the user needs one dialog, not a dozen.
/// The peer's reason for refusing a channel, unless it was reported
/// within `REPORT_AGAIN_AFTER`. One page load can have a dozen
/// connections refused for the same reason, and the user needs one
/// dialog, not a dozen; a burst that keeps going keeps it quiet.
fn first_report(&self, message: String) -> Option<String> {
self.first_report_at(message, Instant::now())
}
pub(super) fn first_report_at(&self, message: String, now: Instant) -> Option<String> {
if message.is_empty() {
return None;
}
let mut reported = self.reported.lock().unwrap();
if reported.len() >= MAX_REPORTED_OPEN_ERRORS || !reported.insert(message.clone()) {
reported.retain(|_, last| now.duration_since(*last) < REPORT_AGAIN_AFTER);
if let Some(last) = reported.get_mut(&message) {
*last = now;
return None;
}
if reported.len() >= MAX_REPORTED_OPEN_ERRORS {
return None;
}
reported.insert(message.clone(), now);
Some(message)
}
@@ -1039,7 +1046,7 @@ mod tests {
#[cfg(not(any(target_os = "android", target_os = "ios")))]
mod tunnel {
use super::*;
use crate::port_forward_mux::{Claim, Ready, Tunnel, TunnelHandle};
use crate::port_forward_mux::{tunnel::{TunnelHandle, MAX_REPORTED_OPEN_ERRORS}, Claim, Tunnel};
use hbb_common::{
protobuf::Message as _,
tcp::FramedStream,
@@ -1085,11 +1092,11 @@ mod tests {
assert!(matches!(t.try_claim(), Claim::Wait));
let w = { let t = t.clone(); tokio::spawn(async move { t.wait_ready().await }) };
t.set_failed();
assert!(matches!(w.await.unwrap(), Ready::Failed));
assert!(w.await.unwrap().is_none());
assert!(matches!(t.try_claim(), Claim::Claimed));
t.set_legacy();
assert!(matches!(t.try_claim(), Claim::Legacy));
assert!(matches!(t.wait_ready().await, Ready::Legacy));
assert!(matches!(t.wait_ready().await, Some(Claim::Legacy)));
});
}
@@ -1212,6 +1219,35 @@ mod tests {
});
}
#[test]
fn a_refused_reason_is_reported_again_after_a_quiet_spell() {
rt().block_on(async {
let (ours, _peer) = stream_pair().await;
let t = Tunnel::new();
t.try_claim();
let h = t.set_muxed(ours, NoUi::default());
let t0 = Instant::now();
let at = |secs: u64| t0 + std::time::Duration::from_secs(secs);
let down = || "down".to_owned();
assert_eq!(h.first_report_at(down(), at(0)), Some(down()));
// A burst that keeps going keeps the dialog quiet ...
assert_eq!(h.first_report_at(down(), at(8)), None);
assert_eq!(h.first_report_at(down(), at(16)), None);
// ... and one that stopped is reported afresh.
assert_eq!(h.first_report_at(down(), at(27)), Some(down()));
// The cap counts reasons still live, so it cannot silence the
// window for good.
for i in 0..MAX_REPORTED_OPEN_ERRORS {
h.first_report_at(format!("reason {}", i), at(27));
}
assert_eq!(h.first_report_at("one more".to_owned(), at(27)), None);
assert_eq!(
h.first_report_at("one more".to_owned(), at(40)),
Some("one more".to_owned())
);
});
}
#[test]
fn tunnel_death_closes_channels_and_resets_state() {
rt().block_on(async {
@@ -1293,7 +1329,7 @@ mod tests {
Some(Ok(bytes)) => {
let Ok(m) = Message::parse_from_bytes(&bytes) else { continue };
if let Some(message::Union::PortForwardChannel(ch)) = m.union {
mux.handle(ch, true);
mux.handle(ch, || true);
mux.sweep();
}
}

48
src/server/connection.rs
View File

@@ -1669,16 +1669,12 @@ impl Connection {
return true;
};
if pf.multiplex {
if let Some(tx) = self.inner.tx.clone() {
self.port_forward_mux = Some(super::port_forward_mux::PortForwardMux::new(
tx,
self.port_forward_address.clone(),
));
return true;
}
// Without a sender there is no way to answer channel frames: fall
// through to the raw pipe below rather than return with neither
// flag set, which would make the connection look like remote desktop.
// `inner.tx` is set for the connection's whole life; `None` here is
// unreachable, and refusing the login is the only honest answer.
self.port_forward_mux = self.inner.tx.clone().map(|tx| {
super::port_forward_mux::PortForwardMux::new(tx, self.port_forward_address.clone())
});
return self.port_forward_mux.is_some();
}
let mut pf = pf.clone();
let (mut addr, is_rdp) = Self::normalize_port_forward_target(&mut pf);
@@ -2245,6 +2241,16 @@ impl Connection {
self.send_to_cm(ipc::Data::UpdatePortForward(label));
}
fn handle_port_forward_channel(&mut self, ch: PortForwardChannel) {
let Some(mux) = self.port_forward_mux.as_mut() else {
log::debug!("port forward channel frame on a non-multiplexed connection");
return;
};
mux.handle(ch, || {
Self::permission(keys::OPTION_ENABLE_TUNNEL, &self.control_permissions)
});
}
#[inline]
fn send_fs(&mut self, data: ipc::FS) {
self.send_to_cm(ipc::Data::FS(data));
@@ -3901,27 +3907,7 @@ impl Connection {
self.refresh_video_display(Some(request.display as usize));
}
}
Some(message::Union::PortForwardChannel(ch)) => {
// Only open/close can change the target set; sweeping after every
// data frame would walk the whole table per 64 KiB.
let may_change_targets = matches!(
ch.union,
Some(port_forward_channel::Union::Open(_))
| Some(port_forward_channel::Union::Close(_))
);
// Only `open` consults permissions; short-circuit so the lookup
// isn't made per 64 KiB of data.
let permitted = matches!(ch.union, Some(port_forward_channel::Union::Open(_)))
&& Self::permission(keys::OPTION_ENABLE_TUNNEL, &self.control_permissions);
if let Some(mux) = self.port_forward_mux.as_mut() {
mux.handle(ch, permitted);
if may_change_targets {
self.push_port_forward_label();
}
} else {
log::debug!("port forward channel frame on a non-multiplexed connection");
}
}
Some(message::Union::PortForwardChannel(ch)) => self.handle_port_forward_channel(ch),
Some(message::Union::TerminalAction(action)) => {
#[cfg(not(any(target_os = "android", target_os = "ios")))]
allow_err!(self.handle_terminal_action(action).await);

79
src/server/port_forward_mux.rs
View File

@@ -50,9 +50,14 @@ impl PortForwardMux {
}
}
pub fn handle(&mut self, frame: PortForwardChannel, permitted: bool) {
/// `tunnel_permitted` is consulted for `open` alone, so the lookup is not
/// made per 64 KiB of data.
pub fn handle(&mut self, frame: PortForwardChannel, tunnel_permitted: impl FnOnce() -> bool) {
match frame.union {
Some(port_forward_channel::Union::Open(open)) => self.on_open(open, permitted),
Some(port_forward_channel::Union::Open(open)) => {
let permitted = tunnel_permitted();
self.on_open(open, permitted)
}
Some(port_forward_channel::Union::Data(d)) => {
let len = d.data.len();
let Some(entry) = self.channels.get(&d.channel_id) else {
@@ -113,7 +118,7 @@ impl PortForwardMux {
port: open.port,
..Default::default()
};
let (addr, _is_rdp) = Connection::normalize_port_forward_target(&mut pf);
let (addr, is_rdp) = Connection::normalize_port_forward_target(&mut pf);
let (inbound_tx, inbound_rx) = mpsc::unbounded_channel();
let credit = Arc::new(SendCredit::new(effective_window(open.window)));
let window = Arc::new(Mutex::new(RecvWindow::new(CHANNEL_WINDOW)));
@@ -129,6 +134,7 @@ impl PortForwardMux {
tokio::spawn(run_controlled_channel(
id,
addr,
is_rdp,
credit,
window,
inbound_rx,
@@ -184,6 +190,7 @@ impl PortForwardMux {
async fn run_controlled_channel(
id: i32,
addr: String,
is_rdp: bool,
credit: Arc<SendCredit>,
window: Arc<Mutex<RecvWindow>>,
mut inbound: mpsc::UnboundedReceiver<Inbound>,
@@ -215,19 +222,16 @@ async fn run_controlled_channel(
return;
}
},
res = &mut connect => match res {
Ok(Ok(s)) => break s,
Ok(Err(e)) => {
log::debug!("port forward channel {} connect {} failed: {}", id, addr, e);
sink.send_ordered(opened_msg(id, false, &format!("Failed to access remote {}", addr), 0)).await.ok();
return;
}
Err(_) => {
log::debug!("port forward channel {} connect {} timed out", id, addr);
sink.send_ordered(opened_msg(id, false, &format!("Failed to access remote {}", addr), 0)).await.ok();
return;
}
},
res = &mut connect => {
let err = match res {
Ok(Ok(s)) => break s,
Ok(Err(e)) => e.to_string(),
Err(e) => e.to_string(),
};
log::debug!("port forward channel {} connect {} failed: {}", id, addr, err);
sink.send_ordered(opened_msg(id, false, &unreachable_message(&addr, is_rdp), 0)).await.ok();
return;
}
}
};
if sink
@@ -241,6 +245,15 @@ async fn run_controlled_channel(
run_channel(id, reader, writer, Vec::new(), pending, credit, window, inbound, sink).await;
}
/// The same words the raw pipe puts in its login error, so one problem reads
/// the same whichever path the peer takes.
fn unreachable_message(addr: &str, is_rdp: bool) -> String {
format!(
"Failed to access remote {}. Please make sure it is reachable/open.",
if is_rdp { "RDP" } else { addr }
)
}
#[cfg(test)]
mod tests {
use super::*;
@@ -340,11 +353,11 @@ mod tests {
let port = echo_target().await;
let (tx, mut rx) = mpsc::unbounded_channel();
let mut mux = PortForwardMux::new(tx, "localhost:1".to_owned());
mux.handle(open(1, port), true);
mux.handle(data(1, b"ping"), true);
mux.handle(open(1, port), || true);
mux.handle(data(1, b"ping"), || true);
assert_eq!(opened(&next_frame(&mut rx).await), (1, true));
assert_eq!(data_of(&next_frame(&mut rx).await), (1, b"ping".to_vec()));
mux.handle(close(1), true);
mux.handle(close(1), || true);
});
}
@@ -356,8 +369,8 @@ mod tests {
drop(l);
let (tx, mut rx) = mpsc::unbounded_channel();
let mut mux = PortForwardMux::new(tx, "localhost:1".to_owned());
mux.handle(open(1, port), true);
mux.handle(data(1, b"lost"), true);
mux.handle(open(1, port), || true);
mux.handle(data(1, b"lost"), || true);
assert_eq!(opened(&next_frame(&mut rx).await), (1, false));
assert!(tokio::time::timeout(std::time::Duration::from_millis(50), rx.recv()).await.is_err());
});
@@ -369,7 +382,7 @@ mod tests {
let port = echo_target().await;
let (tx, mut rx) = mpsc::unbounded_channel();
let mut mux = PortForwardMux::new(tx, "localhost:1".to_owned());
mux.handle(open(1, port), false);
mux.handle(open(1, port), || false);
assert_eq!(opened(&next_frame(&mut rx).await), (1, false));
assert_eq!(mux.live_channels(), 0);
});
@@ -384,8 +397,8 @@ mod tests {
let port = echo_target().await;
let (tx, mut rx) = mpsc::unbounded_channel();
let mut mux = PortForwardMux::new(tx, "localhost:1".to_owned());
mux.handle(open(1, port), true);
mux.handle(close(1), true);
mux.handle(open(1, port), || true);
mux.handle(close(1), || true);
assert!(tokio::time::timeout(std::time::Duration::from_millis(200), rx.recv()).await.is_err());
assert_eq!(mux.live_channels(), 0);
});
@@ -397,21 +410,21 @@ mod tests {
let port = echo_target().await;
let (tx, mut rx) = mpsc::unbounded_channel();
let mut mux = PortForwardMux::new(tx, "localhost:1".to_owned());
mux.handle(open(1, port), true);
mux.handle(open(2, port), true);
mux.handle(open(1, port), || true);
mux.handle(open(2, port), || true);
let mut seen = 0;
while seen < 2 {
opened(&next_frame(&mut rx).await);
seen += 1;
}
let too_much = vec![0u8; CHANNEL_WINDOW as usize + 1];
mux.handle(data(1, &too_much), true);
mux.handle(data(1, &too_much), || true);
let ch = next_frame(&mut rx).await;
match &ch.union {
Some(port_forward_channel::Union::Close(c)) => assert_eq!(c.channel_id, 1),
other => panic!("expected close, got {:?}", other),
}
mux.handle(data(2, b"still fine"), true);
mux.handle(data(2, b"still fine"), || true);
assert_eq!(data_of(&next_frame(&mut rx).await), (2, b"still fine".to_vec()));
});
}
@@ -454,7 +467,7 @@ mod tests {
let (tx, mut rx) = mpsc::unbounded_channel();
let mut mux = PortForwardMux::new(tx, "localhost:1".to_owned());
for id in 1..=(MAX_CHANNELS as i32 * 2) {
mux.handle(open(id, port), true);
mux.handle(open(id, port), || true);
assert_eq!(opened(&next_frame(&mut rx).await), (id, true));
// The task sends `close` on the target's EOF and exits; the
// entry is dead until the next `open` sweeps it.
@@ -480,15 +493,15 @@ mod tests {
let (tx, mut rx) = mpsc::unbounded_channel();
let mut mux = PortForwardMux::new(tx, format!("127.0.0.1:{}", a));
assert_eq!(mux.sweep(), None);
mux.handle(open(1, a), true);
mux.handle(open(2, a), true);
mux.handle(open(1, a), || true);
mux.handle(open(2, a), || true);
opened(&next_frame(&mut rx).await);
opened(&next_frame(&mut rx).await);
assert_eq!(mux.sweep(), None);
mux.handle(open(3, b), true);
mux.handle(open(3, b), || true);
opened(&next_frame(&mut rx).await);
assert_eq!(mux.sweep(), Some(format!("127.0.0.1:{} +1", a)));
mux.handle(close(3), true);
mux.handle(close(3), || true);
tokio::task::yield_now().await;
assert_eq!(mux.sweep(), Some(format!("127.0.0.1:{}", a)));
});

11
src/ui_cm_interface.rs
View File

@@ -963,6 +963,17 @@ pub async fn start_listen<T: InvokeUiCM>(
Some(Data::Close) => {
break;
}
Some(Data::UpdatePortForward(port_forward)) => {
let updated = {
let mut clients = CLIENTS.write().unwrap();
clients.get_mut(&current_id).map(|c| {
c.port_forward = port_forward.clone();
})
};
if updated.is_some() {
cm.ui_handler.update_port_forward(current_id, port_forward);
}
}
Some(Data::StartVoiceCall) => {
cm.voice_call_started(current_id);
}