https://github.com/rustdesk/rustdesk/issues/327

libs/enigo/examples/key.rs

@@ -6,7 +6,7 @@ fn main() {
     thread::sleep(Duration::from_secs(2));
     let mut enigo = Enigo::new();
 
-    enigo.key_down(Key::Layout('a'));
+    enigo.key_down(Key::Layout('a')).ok();
     thread::sleep(Duration::from_secs(1));
     enigo.key_up(Key::Layout('a'));
 }

libs/enigo/examples/keyboard.rs

@@ -10,7 +10,7 @@ fn main() {
     enigo.key_sequence("Hello World! here is a lot of text  ❤️");
 
     // select all
-    enigo.key_down(Key::Control);
+    enigo.key_down(Key::Control).ok();
     enigo.key_click(Key::Layout('a'));
     enigo.key_up(Key::Control);
 }

libs/enigo/examples/mouse.rs

@@ -11,7 +11,7 @@ fn main() {
     enigo.mouse_move_to(500, 200);
     thread::sleep(wait_time);
 
-    enigo.mouse_down(MouseButton::Left);
+    enigo.mouse_down(MouseButton::Left).ok();
     thread::sleep(wait_time);
 
     enigo.mouse_move_relative(100, 100);

libs/enigo/examples/timer.rs

@@ -16,7 +16,7 @@ fn main() {
     println!("{:?}", time);
 
     // select all
-    enigo.key_down(Key::Control);
+    enigo.key_down(Key::Control).ok();
     enigo.key_click(Key::Layout('a'));
     enigo.key_up(Key::Control);
 }

libs/hbb_common/src/tcp.rs

@@ -10,7 +10,7 @@ use std::{
 use tokio::net::{lookup_host, TcpListener, TcpSocket, TcpStream, ToSocketAddrs};
 use tokio_util::codec::Framed;
 
-pub struct FramedStream(Framed<TcpStream, BytesCodec>, Option<(Key, u64, u64)>);
+pub struct FramedStream(Framed<TcpStream, BytesCodec>, Option<(Key, u64, u64)>, u64);
 
 impl Deref for FramedStream {
     type Target = Framed<TcpStream, BytesCodec>;
@@ -56,14 +56,18 @@ impl FramedStream {
                     new_socket(local_addr, true)?.connect(remote_addr),
                 )
                 .await??;
-                return Ok(Self(Framed::new(stream, BytesCodec::new()), None));
+                return Ok(Self(Framed::new(stream, BytesCodec::new()), None, 0));
             }
         }
         bail!("could not resolve to any address");
     }
 
+    pub fn set_send_timeout(&mut self, ms: u64) {
+        self.2 = ms;
+    }
+
     pub fn from(stream: TcpStream) -> Self {
-        Self(Framed::new(stream, BytesCodec::new()), None)
+        Self(Framed::new(stream, BytesCodec::new()), None, 0)
     }
 
     pub fn set_raw(&mut self) {
@@ -88,12 +92,17 @@ impl FramedStream {
             let nonce = Self::get_nonce(key.1);
             msg = secretbox::seal(&msg, &nonce, &key.0);
         }
-        self.0.send(bytes::Bytes::from(msg)).await?;
+        self.send_bytes(bytes::Bytes::from(msg)).await?;
         Ok(())
     }
 
+    #[inline]
     pub async fn send_bytes(&mut self, bytes: Bytes) -> ResultType<()> {
+        if self.2 > 0 {
+            super::timeout(self.2, self.0.send(bytes)).await??;
+        } else {
             self.0.send(bytes).await?;
+        }
         Ok(())
     }
 

libs/scrap/examples/record-screen.rs

@@ -43,7 +43,6 @@ struct Args {
     flag_time: Option<u64>,
     flag_fps: u64,
     flag_bv: u32,
-    flag_ba: u32,
 }
 
 #[derive(Debug, serde::Deserialize)]

src/server/connection.rs

@@ -81,6 +81,8 @@ const TEST_DELAY_TIMEOUT: Duration = Duration::from_secs(3);
 const SEC30: Duration = Duration::from_secs(30);
 const H1: Duration = Duration::from_secs(3600);
 const MILLI1: Duration = Duration::from_millis(1);
+const SEND_TIMEOUT_VIDEO: u64 = 12_000;
+const SEND_TIMEOUT_OTHER: u64 = SEND_TIMEOUT_VIDEO * 10;
 
 impl Connection {
     pub async fn start(
@@ -147,6 +149,14 @@ impl Connection {
             time::interval_at(Instant::now() + TEST_DELAY_TIMEOUT, TEST_DELAY_TIMEOUT);
         let mut last_recv_time = Instant::now();
 
+        conn.stream.set_send_timeout(
+            if conn.file_transfer.is_some() || conn.port_forward_socket.is_some() {
+                SEND_TIMEOUT_OTHER
+            } else {
+                SEND_TIMEOUT_VIDEO
+            },
+        );
+
         loop {
             tokio::select! {
                 biased; // video has higher priority
@@ -286,73 +296,58 @@ impl Connection {
                 }
             }
         }
-        video_service::notify_video_frame_feched(id, None);
 
+        video_service::notify_video_frame_feched(id, None);
         super::video_service::update_test_latency(id, 0);
         super::video_service::update_image_quality(id, None);
-        if let Some(forward) = conn.port_forward_socket.as_mut() {
+        if let Err(err) = conn.try_port_forward_loop(&mut rx_from_cm).await {
+            conn.on_close(&err.to_string(), false);
+        }
+    }
+
+    async fn try_port_forward_loop(
+        &mut self,
+        rx_from_cm: &mut mpsc::UnboundedReceiver<Data>,
+    ) -> ResultType<()> {
+        let mut last_recv_time = Instant::now();
+        if let Some(forward) = self.port_forward_socket.as_mut() {
             log::info!("Running port forwarding loop");
-            conn.stream.set_raw();
+            self.stream.set_raw();
             loop {
                 tokio::select! {
                     Some(data) = rx_from_cm.recv() => {
                         match data {
                             ipc::Data::Close => {
-                                conn.on_close("Close requested from connection manager", false);
+                                bail!("Close requested from selfection manager");
-                                break;
                             }
                             _ => {}
                         }
                     }
                     res = forward.next() => {
                         if let Some(res) = res {
-                            match res {
-                                Err(err) => {
-                                    conn.on_close(&err.to_string(), false);
-                                    break;
-                                },
-                                Ok(bytes) => {
                             last_recv_time = Instant::now();
-                                    if let Err(err) = conn.stream.send_bytes(bytes.into()).await {
+                            self.stream.send_bytes(res?.into()).await?;
-                                        conn.on_close(&err.to_string(), false);
-                                        break;
-                                    }
-                                }
-                            }
                         } else {
-                            conn.on_close("Forward reset by the peer", false);
+                            bail!("Forward reset by the peer");
-                            break;
                         }
                     },
-                    res = conn.stream.next() => {
+                    res = self.stream.next() => {
                         if let Some(res) = res {
-                            match res {
-                                Err(err) => {
-                                    conn.on_close(&err.to_string(), false);
-                                    break;
-                                },
-                                Ok(bytes) => {
                             last_recv_time = Instant::now();
-                                    if let Err(err) = forward.send(bytes.into()).await {
+                            timeout(SEND_TIMEOUT_OTHER, forward.send(res?.into())).await??;
-                                        conn.on_close(&err.to_string(), false);
-                                        break;
-                                    }
-                                }
-                            }
                         } else {
-                            conn.on_close("Stream reset by the peer", false);
+                            bail!("Stream reset by the peer");
-                            break;
                         }
                     },
-                    _ = conn.timer.tick() => {
+                    _ = self.timer.tick() => {
                         if last_recv_time.elapsed() >= H1 {
-                            conn.on_close("Timeout", false);
+                            bail!("Timeout");
-                            break;
                         }
                     }
                 }
             }
         }
+        Ok(())
     }
 
     async fn send_permission(&mut self, permission: Permission, enabled: bool) {