Merge pull request #1181 from Heap-Hop/flutter_desktop_merge_master_mobile

Flutter desktop merge master mobile

src/flutter.rs

@@ -1,6 +1,9 @@
 use std::{
     collections::{HashMap, VecDeque},
-    sync::{Arc, Mutex, RwLock},
+    sync::{
+        atomic::{AtomicUsize, Ordering},
+        Arc, Mutex, RwLock,
+    },
 };
 
 use flutter_rust_bridge::{StreamSink, ZeroCopyBuffer};
@@ -118,6 +121,9 @@ impl Session {
         }
         lc.set_option(name, value);
     }
+    // TODO
+    // input_os_password
+    // restart_remote_device
 
     /// Input the OS password.
     pub fn input_os_password(&self, pass: String, activate: bool) {
@@ -509,6 +515,26 @@ impl Session {
             }
         }
     }
+
+    fn update_quality_status(&self, status: QualityStatus) {
+        const NULL: String = String::new();
+        self.push_event(
+            "update_quality_status",
+            vec![
+                ("speed", &status.speed.map_or(NULL, |it| it)),
+                ("fps", &status.fps.map_or(NULL, |it| it.to_string())),
+                ("delay", &status.delay.map_or(NULL, |it| it.to_string())),
+                (
+                    "target_bitrate",
+                    &status.target_bitrate.map_or(NULL, |it| it.to_string()),
+                ),
+                (
+                    "codec_format",
+                    &status.codec_format.map_or(NULL, |it| it.to_string()),
+                ),
+            ],
+        );
+    }
 }
 
 impl FileManager for Session {}
@@ -603,7 +629,14 @@ impl Interface for Session {
     }
 
     async fn handle_test_delay(&mut self, t: TestDelay, peer: &mut Stream) {
-        handle_test_delay(t, peer).await;
+        if !t.from_client {
+            self.update_quality_status(QualityStatus {
+                delay: Some(t.last_delay as _),
+                target_bitrate: Some(t.target_bitrate as _),
+                ..Default::default()
+            });
+            handle_test_delay(t, peer).await;
+        }
     }
 }
 
@@ -618,6 +651,9 @@ struct Connection {
     write_jobs: Vec<fs::TransferJob>,
     timer: Interval,
     last_update_jobs_status: (Instant, HashMap<i32, u64>),
+    data_count: Arc<AtomicUsize>,
+    frame_count: Arc<AtomicUsize>,
+    video_format: CodecFormat,
 }
 
 impl Connection {
@@ -650,6 +686,9 @@ impl Connection {
             write_jobs: Vec::new(),
             timer: time::interval(SEC30),
             last_update_jobs_status: (Instant::now(), Default::default()),
+            data_count: Arc::new(AtomicUsize::new(0)),
+            frame_count: Arc::new(AtomicUsize::new(0)),
+            video_format: CodecFormat::Unknown,
         };
         let key = Config::get_option("key");
         let token = Config::get_option("access_token");
@@ -663,6 +702,9 @@ impl Connection {
                         ("direct", &direct.to_string()),
                     ],
                 );
+
+                let mut status_timer = time::interval(Duration::new(1, 0));
+
                 loop {
                     tokio::select! {
                         res = peer.next() => {
@@ -675,14 +717,20 @@ impl Connection {
                                     }
                                     Ok(ref bytes) => {
                                         last_recv_time = Instant::now();
+                                        conn.data_count.fetch_add(bytes.len(), Ordering::Relaxed);
                                         if !conn.handle_msg_from_peer(bytes, &mut peer).await {
                                             break
                                         }
                                     }
                                 }
                             } else {
-                                log::info!("Reset by the peer");
-                                session.msgbox("error", "Connection Error", "Reset by the peer");
+                                if session.lc.read().unwrap().restarting_remote_device {
+                                    log::info!("Restart remote device");
+                                    session.msgbox("restarting", "Restarting Remote Device", "remote_restarting_tip");
+                                } else {
+                                    log::info!("Reset by the peer");
+                                    session.msgbox("error", "Connection Error", "Reset by the peer");
+                                }
                                 break;
                             }
                         }
@@ -708,6 +756,16 @@ impl Connection {
                                 conn.timer = time::interval_at(Instant::now() + SEC30, SEC30);
                             }
                         }
+                        _ = status_timer.tick() => {
+                            let speed = conn.data_count.swap(0, Ordering::Relaxed);
+                            let speed = format!("{:.2}kB/s", speed as f32 / 1024 as f32);
+                            let fps = conn.frame_count.swap(0, Ordering::Relaxed) as _;
+                            conn.session.update_quality_status(QualityStatus {
+                                speed:Some(speed),
+                                fps:Some(fps),
+                                ..Default::default()
+                            });
+                        }
                     }
                 }
                 log::debug!("Exit io_loop of id={}", session.id);
@@ -729,6 +787,14 @@ impl Connection {
                     if !self.first_frame {
                         self.first_frame = true;
                     }
+                    let incomming_format = CodecFormat::from(&vf);
+                    if self.video_format != incomming_format {
+                        self.video_format = incomming_format.clone();
+                        self.session.update_quality_status(QualityStatus {
+                            codec_format: Some(incomming_format),
+                            ..Default::default()
+                        })
+                    };
                     if let Ok(true) = self.video_handler.handle_frame(vf) {
                         let stream = self.session.events2ui.read().unwrap();
                         stream.add(EventToUI::Rgba(ZeroCopyBuffer(
@@ -937,6 +1003,7 @@ impl Connection {
                                     Permission::Keyboard => "keyboard",
                                     Permission::Clipboard => "clipboard",
                                     Permission::Audio => "audio",
+                                    Permission::Restart => "restart",
                                     _ => "",
                                 },
                                 &p.enabled.to_string(),
@@ -1618,8 +1685,8 @@ pub mod connection_manager {
                 id,
                 file_num,
                 mut files,
+                overwrite_detection,
             } => {
-                // in mobile, can_enable_override_detection is always true
                 WRITE_JOBS.lock().unwrap().push(fs::TransferJob::new_write(
                     id,
                     "".to_string(),
@@ -1635,7 +1702,7 @@ pub mod connection_manager {
                             ..Default::default()
                         })
                         .collect(),
-                    true,
+                    overwrite_detection,
                 ));
             }
             ipc::FS::CancelWrite { id } => {