add: implement last jobs[2/2]

2024-11-25 05:09:04 +08:00 · 2022-07-11 18:23:58 +08:00 · 2022-07-11 18:23:58 +08:00 · 9094999a8a
commit 9094999a8a
parent 5aded67597
6 changed files with 210 additions and 17 deletions
--- a/flutter/lib/desktop/pages/file_manager_page.dart
+++ b/flutter/lib/desktop/pages/file_manager_page.dart
@ -413,6 +413,14 @@ class _FileManagerPageState extends State<FileManagerPage>
                        Row(
                          mainAxisAlignment: MainAxisAlignment.end,
                          children: [
+                            Offstage(
+                              offstage: item.state != JobState.paused,
+                              child: IconButton(
+                                  onPressed: () {
+                                    model.resumeJob(item.id);
+                                  },
+                                  icon: Icon(Icons.restart_alt_rounded)),
+                            ),
                            IconButton(
                              icon: Icon(Icons.delete),
                              onPressed: () {
@ -597,6 +605,7 @@ class _FileManagerPageState extends State<FileManagerPage>
                  onPressed: () {
                    final items = getSelectedItem(isLocal);
                    model.sendFiles(items, isRemote: !isLocal);
+                    items.clear();
                  },
                  icon: Transform.rotate(
                    angle: isLocal ? 0 : pi,
--- a/flutter/lib/models/file_model.dart
+++ b/flutter/lib/models/file_model.dart
@ -309,6 +309,8 @@ class FileModel extends ChangeNotifier {
    if (_currentRemoteDir.path.isEmpty) {
      openDirectory(_remoteOption.home, isLocal: false);
    }
+    // load last transfer jobs
+    await _ffi.target?.bind.sessionLoadLastTransferJobs(id: '${_ffi.target?.id}');
  }

  onClose() {
@ -390,11 +392,11 @@ class FileModel extends ChangeNotifier {
    if (isDesktop) {
      // desktop sendFiles
      final toPath =
-      isRemote ? currentRemoteDir.path : currentLocalDir.path;
+      isRemote ? currentLocalDir.path : currentRemoteDir.path;
      final isWindows =
-      isRemote ?  _localOption.isWindows : _remoteOption.isWindows;
+      isRemote ?  _remoteOption.isWindows : _localOption.isWindows;
      final showHidden =
-      isRemote ? _localOption.showHidden : _remoteOption.showHidden ;
+      isRemote ? _remoteOption.showHidden : _localOption.showHidden;
      items.items.forEach((from) async {
        final jobId = ++_jobId;
        _jobTable.add(JobProgress()
@ -406,6 +408,7 @@ class FileModel extends ChangeNotifier {
        );
        _ffi.target?.bind.sessionSendFiles(id: '${_ffi.target?.id}', actId: _jobId, path: from.path, to: PathUtil.join(toPath, from.name, isWindows)
            ,fileNum: 0, includeHidden: showHidden, isRemote: isRemote);
+        print("path:${from.path}, toPath:${toPath}, to:${PathUtil.join(toPath, from.name, isWindows)}");
      });
    } else {
      if (items.isLocal == null) {
@ -672,6 +675,50 @@ class FileModel extends ChangeNotifier {
  }

  bool get remoteSortAscending => _remoteSortAscending;
+
+  void loadLastJob(Map<String, dynamic> evt) {
+    debugPrint("load last job: ${evt}");
+    Map<String,dynamic> jobDetail = json.decode(evt['value']);
+    // int id = int.parse(jobDetail['id']);
+    String remote = jobDetail['remote'];
+    String to = jobDetail['to'];
+    bool showHidden = jobDetail['show_hidden'];
+    int fileNum = jobDetail['file_num'];
+    bool isRemote = jobDetail['is_remote'];
+    final currJobId = _jobId++;
+    var jobProgress = JobProgress()
+      ..jobName = isRemote ? remote : to
+      ..id = currJobId
+      ..isRemote = isRemote
+      ..fileNum = fileNum
+      ..remote = remote
+      ..to = to
+      ..showHidden = showHidden
+      ..state = JobState.paused;
+    jobTable.add(jobProgress);
+    _ffi.target?.bind.sessionAddJob(id: '${_ffi.target?.id}',
+        isRemote: isRemote,
+        includeHidden: showHidden,
+        actId: currJobId,
+        path: isRemote ? remote : to,
+        to: isRemote ? to: remote,
+        fileNum: fileNum,
+    );
+  }
+
+  resumeJob(int jobId) {
+    final jobIndex = getJob(jobId);
+    if (jobIndex != -1) {
+      final job = jobTable[jobIndex];
+      _ffi.target?.bind.sessionResumeJob(id: '${_ffi.target?.id}',
+          actId: job.id,
+          isRemote: job.isRemote);
+      job.state = JobState.inProgress;
+    } else {
+      debugPrint("jobId ${jobId} is not exists");
+    }
+    notifyListeners();
+  }
 }

 class JobResultListener<T> {
@ -877,7 +924,7 @@ class Entry {
  }
 }

-enum JobState { none, inProgress, done, error }
+enum JobState { none, inProgress, done, error, paused }

 extension JobStateDisplay on JobState {
  String display() {
@ -906,6 +953,9 @@ class JobProgress {
  var fileCount = 0;
  var isRemote = false;
  var jobName = "";
+  var remote = "";
+  var to = "";
+  var showHidden = false;

  clear() {
    state = JobState.none;
@ -915,6 +965,8 @@ class JobProgress {
    finishedSize = 0;
    jobName = "";
    fileCount = 0;
+    remote = "";
+    to = "";
  }
 }

--- a/flutter/lib/models/model.dart
+++ b/flutter/lib/models/model.dart
@ -168,6 +168,8 @@ class FfiModel with ChangeNotifier {
        parent.target?.fileModel.jobError(evt);
      } else if (name == 'override_file_confirm') {
        parent.target?.fileModel.overrideFileConfirm(evt);
+      } else if (name == 'load_last_job') {
+        parent.target?.fileModel.loadLastJob(evt);
      } else if (name == 'update_folder_files') {
        parent.target?.fileModel.updateFolderFiles(evt);
      } else if (name == 'try_start_without_auth') {
@ -219,6 +221,8 @@ class FfiModel with ChangeNotifier {
        parent.target?.fileModel.jobError(evt);
      } else if (name == 'override_file_confirm') {
        parent.target?.fileModel.overrideFileConfirm(evt);
+      } else if (name == 'load_last_job') {
+        parent.target?.fileModel.loadLastJob(evt);
      } else if (name == 'update_folder_files') {
        parent.target?.fileModel.updateFolderFiles(evt);
      } else if (name == 'try_start_without_auth') {
--- a/src/client/file_trait.rs
+++ b/src/client/file_trait.rs
@ -93,7 +93,7 @@ pub trait FileManager: Interface {
    }

    fn add_job(
-        &mut self,
+        &self,
        id: i32,
        path: String,
        to: String,
@ -111,7 +111,7 @@ pub trait FileManager: Interface {
        )));
    }

-    fn resume_job(&mut self, id: i32, is_remote: bool) {
+    fn resume_job(&self, id: i32, is_remote: bool) {
        self.send(Data::ResumeJob((id, is_remote)));
    }
 }
--- a/src/flutter.rs
+++ b/src/flutter.rs
@ -5,8 +5,8 @@ use std::{

 use flutter_rust_bridge::{StreamSink, ZeroCopyBuffer};

-use hbb_common::config::PeerConfig;
-use hbb_common::fs::TransferJobMeta;
+use hbb_common::config::{PeerConfig, TransferSerde};
+use hbb_common::fs::{get_job, TransferJobMeta};
 use hbb_common::{
    allow_err,
    compress::decompress,
@ -471,6 +471,10 @@ impl Session {
        load_config(&self.id)
    }

+    pub fn save_config(&self, config: &PeerConfig) {
+        config.store(&self.id);
+    }
+
    pub fn get_platform(&self, is_remote: bool) -> String {
        if is_remote {
            self.lc.read().unwrap().info.platform.clone()
@ -488,16 +492,16 @@ impl Session {
        let mut cnt = 1;
        for job_str in pc.transfer.read_jobs.iter() {
            if !job_str.is_empty() {
-                self.push_event("addJob", vec![("value", job_str)]);
+                self.push_event("load_last_job", vec![("value", job_str)]);
                cnt += 1;
-                println!("restore read_job: {:?}", job);
+                println!("restore read_job: {:?}", job_str);
            }
        }
        for job_str in pc.transfer.write_jobs.iter() {
            if !job_str.is_empty() {
-                self.push_event("addJob", vec![("value", job_str)]);
+                self.push_event("load_last_job", vec![("value", job_str)]);
                cnt += 1;
-                println!("restore write_job: {:?}", job);
+                println!("restore write_job: {:?}", job_str);
            }
        }
    }
@ -978,6 +982,7 @@ impl Connection {
    async fn handle_msg_from_ui(&mut self, data: Data, peer: &mut Stream) -> bool {
        match data {
            Data::Close => {
+                self.sync_jobs_status_to_local().await;
                return false;
            }
            Data::Login((password, remember)) => {
@ -989,8 +994,7 @@ impl Connection {
                allow_err!(peer.send(&msg).await);
            }
            Data::SendFiles((id, path, to, file_num, include_hidden, is_remote)) => {
-                // in mobile, can_enable_override_detection is always true
-                let od = true;
+                let od = can_enable_overwrite_detection(self.session.lc.read().unwrap().version);
                if is_remote {
                    log::debug!("New job {}, write to {} from remote {}", id, to, path);
                    self.write_jobs.push(fs::TransferJob::new_write(
@ -1001,7 +1005,7 @@ impl Connection {
                        include_hidden,
                        is_remote,
                        Vec::new(),
-                        true,
+                        od,
                    ));
                    allow_err!(
                        peer.send(&fs::new_send(id, path, file_num, include_hidden))
@ -1015,7 +1019,7 @@ impl Connection {
                        file_num,
                        include_hidden,
                        is_remote,
-                        true,
+                        od,
                    ) {
                        Err(err) => {
                            self.handle_job_status(id, -1, Some(err.to_string()));
@ -1180,6 +1184,87 @@ impl Connection {
                    }
                }
            }
+            Data::AddJob((id, path, to, file_num, include_hidden, is_remote)) => {
+                let od = can_enable_overwrite_detection(self.session.lc.read().unwrap().version);
+                if is_remote {
+                    log::debug!(
+                        "new write waiting job {}, write to {} from remote {}",
+                        id,
+                        to,
+                        path
+                    );
+                    let mut job = fs::TransferJob::new_write(
+                        id,
+                        path.clone(),
+                        to,
+                        file_num,
+                        include_hidden,
+                        is_remote,
+                        Vec::new(),
+                        od,
+                    );
+                    job.is_last_job = true;
+                    self.write_jobs.push(job);
+                } else {
+                    match fs::TransferJob::new_read(
+                        id,
+                        to.clone(),
+                        path.clone(),
+                        file_num,
+                        include_hidden,
+                        is_remote,
+                        od,
+                    ) {
+                        Err(err) => {
+                            self.handle_job_status(id, -1, Some(err.to_string()));
+                        }
+                        Ok(mut job) => {
+                            log::debug!(
+                                "new read waiting job {}, read {} to remote {}, {} files",
+                                id,
+                                path,
+                                to,
+                                job.files().len()
+                            );
+                            let m = make_fd_flutter(job.id(), job.files(), true);
+                            self.session
+                                .push_event("update_folder_files", vec![("info", &m)]);
+                            job.is_last_job = true;
+                            self.read_jobs.push(job);
+                            self.timer = time::interval(MILLI1);
+                        }
+                    }
+                }
+            }
+            Data::ResumeJob((id, is_remote)) => {
+                if is_remote {
+                    if let Some(job) = get_job(id, &mut self.write_jobs) {
+                        job.is_last_job = false;
+                        allow_err!(
+                            peer.send(&fs::new_send(
+                                id,
+                                job.remote.clone(),
+                                job.file_num,
+                                job.show_hidden
+                            ))
+                            .await
+                        );
+                    }
+                } else {
+                    if let Some(job) = get_job(id, &mut self.read_jobs) {
+                        job.is_last_job = false;
+                        allow_err!(
+                            peer.send(&fs::new_receive(
+                                id,
+                                job.path.to_string_lossy().to_string(),
+                                job.file_num,
+                                job.files.clone()
+                            ))
+                            .await
+                        );
+                    }
+                }
+            }
            _ => {}
        }
        true
@ -1269,6 +1354,24 @@ impl Connection {
            ],
        );
    }
+
+    async fn sync_jobs_status_to_local(&mut self) -> bool {
+        log::info!("sync transfer job status");
+        let mut config: PeerConfig = self.session.load_config();
+        let mut transfer_metas = TransferSerde::default();
+        for job in self.read_jobs.iter() {
+            let json_str = serde_json::to_string(&job.gen_meta()).unwrap();
+            transfer_metas.read_jobs.push(json_str);
+        }
+        for job in self.write_jobs.iter() {
+            let json_str = serde_json::to_string(&job.gen_meta()).unwrap();
+            transfer_metas.write_jobs.push(json_str);
+        }
+        log::info!("meta: {:?}", transfer_metas);
+        config.transfer = transfer_metas;
+        self.session.save_config(&config);
+        true
+    }
 }

 // Server Side
@ -1510,7 +1613,6 @@ pub mod connection_manager {
                mut files,
            } => {
                // in mobile, can_enable_override_detection is always true
-                let od = true;
                WRITE_JOBS.lock().unwrap().push(fs::TransferJob::new_write(
                    id,
                    "".to_string(),
--- a/src/flutter_ffi.rs
+++ b/src/flutter_ffi.rs
@ -349,6 +349,32 @@ pub fn session_get_platform(id: String, is_remote: bool) -> String {
 pub fn session_load_last_transfer_jobs(id: String) {
    if let Some(session) = SESSIONS.read().unwrap().get(&id) {
        return session.load_last_jobs();
+    } else {
+        // a tip for flutter dev
+        eprintln!(
+            "cannot load last transfer job from non-existed session. Please ensure session \
+        is connected before calling load last transfer jobs."
+        );
+    }
+}
+
+pub fn session_add_job(
+    id: String,
+    act_id: i32,
+    path: String,
+    to: String,
+    file_num: i32,
+    include_hidden: bool,
+    is_remote: bool,
+) {
+    if let Some(session) = SESSIONS.read().unwrap().get(&id) {
+        session.add_job(act_id, path, to, file_num, include_hidden, is_remote);
+    }
+}
+
+pub fn session_resume_job(id: String, act_id: i32, is_remote: bool) {
+    if let Some(session) = SESSIONS.read().unwrap().get(&id) {
+        session.resume_job(act_id, is_remote);
    }
 }