Async fetch (#552)

* async fetch
* reuse remote progress for fetch
* prvent push/fetch popup from closing too soon

Author: Stephan Dilly

.clippy.toml

@@ -1,2 +1,3 @@
 msrv = "1.50.0"
-cognitive-complexity-threshold = 18
+cognitive-complexity-threshold = 18
+too-many-lines-threshold = 105

asyncgit/src/fetch.rs

@@ -0,0 +1,158 @@
+use crate::{
+    error::{Error, Result},
+    sync::{
+        cred::BasicAuthCredential,
+        remotes::{fetch_origin, push::ProgressNotification},
+    },
+    AsyncNotification, RemoteProgress, CWD,
+};
+use crossbeam_channel::{unbounded, Sender};
+use std::{
+    sync::{Arc, Mutex},
+    thread,
+};
+
+///
+#[derive(Default, Clone, Debug)]
+pub struct FetchRequest {
+    ///
+    pub remote: String,
+    ///
+    pub branch: String,
+    ///
+    pub basic_credential: Option<BasicAuthCredential>,
+}
+
+#[derive(Default, Clone, Debug)]
+struct FetchState {
+    request: FetchRequest,
+}
+
+///
+pub struct AsyncFetch {
+    state: Arc<Mutex<Option<FetchState>>>,
+    last_result: Arc<Mutex<Option<(usize, String)>>>,
+    progress: Arc<Mutex<Option<ProgressNotification>>>,
+    sender: Sender<AsyncNotification>,
+}
+
+impl AsyncFetch {
+    ///
+    pub fn new(sender: &Sender<AsyncNotification>) -> Self {
+        Self {
+            state: Arc::new(Mutex::new(None)),
+            last_result: Arc::new(Mutex::new(None)),
+            progress: Arc::new(Mutex::new(None)),
+            sender: sender.clone(),
+        }
+    }
+
+    ///
+    pub fn is_pending(&self) -> Result<bool> {
+        let state = self.state.lock()?;
+        Ok(state.is_some())
+    }
+
+    ///
+    pub fn last_result(&self) -> Result<Option<(usize, String)>> {
+        let res = self.last_result.lock()?;
+        Ok(res.clone())
+    }
+
+    ///
+    pub fn progress(&self) -> Result<Option<RemoteProgress>> {
+        let res = self.progress.lock()?;
+        Ok(res.as_ref().map(|progress| progress.clone().into()))
+    }
+
+    ///
+    pub fn request(&mut self, params: FetchRequest) -> Result<()> {
+        log::trace!("request");
+
+        if self.is_pending()? {
+            return Ok(());
+        }
+
+        self.set_request(&params)?;
+        RemoteProgress::set_progress(self.progress.clone(), None)?;
+
+        let arc_state = Arc::clone(&self.state);
+        let arc_res = Arc::clone(&self.last_result);
+        let arc_progress = Arc::clone(&self.progress);
+        let sender = self.sender.clone();
+
+        thread::spawn(move || {
+            let (progress_sender, receiver) = unbounded();
+
+            let handle = RemoteProgress::spawn_receiver_thread(
+                sender.clone(),
+                receiver,
+                arc_progress,
+            );
+
+            let res = fetch_origin(
+                CWD,
+                &params.branch,
+                params.basic_credential,
+                Some(progress_sender.clone()),
+            );
+
+            progress_sender
+                .send(ProgressNotification::Done)
+                .expect("closing send failed");
+
+            handle.join().expect("joining thread failed");
+
+            Self::set_result(arc_res, res).expect("result error");
+
+            Self::clear_request(arc_state).expect("clear error");
+
+            sender
+                .send(AsyncNotification::Fetch)
+                .expect("AsyncNotification error");
+        });
+
+        Ok(())
+    }
+
+    fn set_request(&self, params: &FetchRequest) -> Result<()> {
+        let mut state = self.state.lock()?;
+
+        if state.is_some() {
+            return Err(Error::Generic("pending request".into()));
+        }
+
+        *state = Some(FetchState {
+            request: params.clone(),
+        });
+
+        Ok(())
+    }
+
+    fn clear_request(
+        state: Arc<Mutex<Option<FetchState>>>,
+    ) -> Result<()> {
+        let mut state = state.lock()?;
+
+        *state = None;
+
+        Ok(())
+    }
+
+    fn set_result(
+        arc_result: Arc<Mutex<Option<(usize, String)>>>,
+        res: Result<usize>,
+    ) -> Result<()> {
+        let mut last_res = arc_result.lock()?;
+
+        *last_res = match res {
+            Ok(bytes) => Some((bytes, String::new())),
+            Err(e) => {
+                log::error!("fetch error: {}", e);
+                Some((0, e.to_string()))
+            }
+        };
+
+        Ok(())
+    }
+}

asyncgit/src/lib.rs

@@ -14,7 +14,9 @@ pub mod cached;
 mod commit_files;
 mod diff;
 mod error;
+mod fetch;
 mod push;
+pub mod remote_progress;
 mod revlog;
 mod status;
 pub mod sync;
@@ -23,7 +25,9 @@ mod tags;
 pub use crate::{
     commit_files::AsyncCommitFiles,
     diff::{AsyncDiff, DiffParams, DiffType},
-    push::{AsyncPush, PushProgress, PushProgressState, PushRequest},
+    fetch::{AsyncFetch, FetchRequest},
+    push::{AsyncPush, PushRequest},
+    remote_progress::{RemoteProgress, RemoteProgressState},
     revlog::{AsyncLog, FetchStatus},
     status::{AsyncStatus, StatusParams},
     sync::{
@@ -54,6 +58,8 @@ pub enum AsyncNotification {
     Tags,
     ///
     Push,
+    ///
+    Fetch,
 }
 
 /// current working director `./`