Test push progress (#272)

Author: Stephan Dilly

Cargo.lock

@@ -13,7 +13,8 @@ keywords = ["git"]
 
 [dependencies]
 scopetime = { path = "../scopetime", version = "0.1" }
-git2 = { version = "0.13", features = ["vendored-openssl"] }
+# git2 = { version = "0.13", features = ["vendored-openssl"] }
+git2 = { git="https://github.com/rust-lang/git2-rs.git", rev="5fddf7e04dc76e70873569ca9f1de3287ec3edda", features = ["vendored-openssl"] }
 rayon-core = "1.8"
 crossbeam-channel = "0.4"
 log = "0.4"

asyncgit/Cargo.toml

@@ -52,11 +52,11 @@ pub struct AsyncDiff {
 
 impl AsyncDiff {
     ///
-    pub fn new(sender: Sender<AsyncNotification>) -> Self {
+    pub fn new(sender: &Sender<AsyncNotification>) -> Self {
         Self {
             current: Arc::new(Mutex::new(Request(0, None))),
             last: Arc::new(Mutex::new(None)),
-            sender,
+            sender: sender.clone(),
             pending: Arc::new(AtomicUsize::new(0)),
         }
     }

asyncgit/src/diff.rs

@@ -11,6 +11,7 @@ pub mod cached;
 mod commit_files;
 mod diff;
 mod error;
+mod push;
 mod revlog;
 mod status;
 pub mod sync;
@@ -19,6 +20,7 @@ mod tags;
 pub use crate::{
     commit_files::AsyncCommitFiles,
     diff::{AsyncDiff, DiffParams, DiffType},
+    push::{AsyncPush, PushRequest},
     revlog::{AsyncLog, FetchStatus},
     status::{AsyncStatus, StatusParams},
     sync::{
@@ -47,6 +49,8 @@ pub enum AsyncNotification {
     CommitFiles,
     ///
     Tags,
+    ///
+    Push,
 }
 
 /// current working director `./`

asyncgit/src/lib.rs

@@ -0,0 +1,140 @@
+use crate::{
+    error::{Error, Result},
+    sync, AsyncNotification, CWD,
+};
+use crossbeam_channel::Sender;
+use std::sync::{Arc, Mutex};
+
+#[derive(Clone, Debug)]
+enum PushStates {
+    None,
+    // Packing,
+    // Pushing(usize, usize),
+}
+
+impl Default for PushStates {
+    fn default() -> Self {
+        PushStates::None
+    }
+}
+
+///
+#[derive(Default, Clone, Debug)]
+pub struct PushRequest {
+    ///
+    pub remote: String,
+    ///
+    pub branch: String,
+}
+
+#[derive(Default, Clone, Debug)]
+struct PushState {
+    request: PushRequest,
+    state: PushStates,
+}
+
+///
+pub struct AsyncPush {
+    state: Arc<Mutex<Option<PushState>>>,
+    last_result: Arc<Mutex<Option<String>>>,
+    sender: Sender<AsyncNotification>,
+}
+
+impl AsyncPush {
+    ///
+    pub fn new(sender: &Sender<AsyncNotification>) -> Self {
+        Self {
+            state: Arc::new(Mutex::new(None)),
+            last_result: 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<String>> {
+        let res = self.last_result.lock()?;
+        Ok(res.clone())
+    }
+
+    ///
+    pub fn request(&mut self, params: PushRequest) -> Result<()> {
+        log::trace!("request");
+
+        if self.is_pending()? {
+            return Ok(());
+        }
+
+        self.set_request(&params)?;
+
+        let arc_state = Arc::clone(&self.state);
+        let arc_res = Arc::clone(&self.last_result);
+        let sender = self.sender.clone();
+
+        rayon_core::spawn(move || {
+            //TODO: use channels to communicate progress
+            let res = sync::push_origin(
+                CWD,
+                params.remote.as_str(),
+                params.branch.as_str(),
+            );
+
+            Self::set_result(arc_res, res).expect("result error");
+
+            Self::clear_request(arc_state).expect("clear error");
+
+            sender
+                .send(AsyncNotification::Push)
+                .expect("error sending push");
+        });
+
+        Ok(())
+    }
+
+    fn set_request(&self, params: &PushRequest) -> Result<()> {
+        let mut state = self.state.lock()?;
+
+        if state.is_some() {
+            return Err(Error::Generic("pending request".into()));
+        }
+
+        *state = Some(PushState {
+            request: params.clone(),
+            ..PushState::default()
+        });
+
+        Ok(())
+    }
+
+    fn clear_request(
+        state: Arc<Mutex<Option<PushState>>>,
+    ) -> Result<()> {
+        let mut state = state.lock()?;
+
+        *state = None;
+
+        Ok(())
+    }
+
+    fn set_result(
+        arc_result: Arc<Mutex<Option<String>>>,
+        res: Result<()>,
+    ) -> Result<()> {
+        let mut last_res = arc_result.lock()?;
+
+        *last_res = match res {
+            Ok(_) => None,
+            Err(e) => {
+                log::error!("push error: {}", e);
+                Some(e.to_string())
+            }
+        };
+
+        Ok(())
+    }
+}

asyncgit/src/push.rs

@@ -30,9 +30,7 @@ pub use hooks::{hooks_commit_msg, hooks_post_commit, HookResult};
 pub use hunks::{reset_hunk, stage_hunk, unstage_hunk};
 pub use ignore::add_to_ignore;
 pub use logwalker::LogWalker;
-pub use remotes::{
-    fetch_origin, get_remotes, push_origin, remote_push_master,
-};
+pub use remotes::{fetch_origin, get_remotes, push_origin};
 pub use reset::{reset_stage, reset_workdir};
 pub use stash::{get_stashes, stash_apply, stash_drop, stash_save};
 pub use tags::{get_tags, CommitTags, Tags};

asyncgit/src/sync/mod.rs

@@ -16,18 +16,6 @@ pub fn get_remotes(repo_path: &str) -> Result<Vec<String>> {
     Ok(remotes)
 }
 
-///
-pub fn remote_push_master(repo_path: &str) -> Result<()> {
-    scope_time!("remote_push_master");
-
-    let repo = utils::repo(repo_path)?;
-    let mut remote = repo.find_remote("origin")?;
-
-    remote.push(&["refs/heads/master"], None)?;
-
-    Ok(())
-}
-
 ///
 pub fn fetch_origin(repo_path: &str, branch: &str) -> Result<usize> {
     scope_time!("remote_fetch_master");
@@ -44,14 +32,19 @@ pub fn fetch_origin(repo_path: &str, branch: &str) -> Result<usize> {
 }
 
 ///
-pub fn push_origin(repo_path: &str, branch: &str) -> Result<()> {
+pub fn push_origin(
+    repo_path: &str,
+    remote: &str,
+    branch: &str,
+) -> Result<()> {
     scope_time!("push_origin");
 
     let repo = utils::repo(repo_path)?;
-    let mut remote = repo.find_remote("origin")?;
+    let mut remote = repo.find_remote(remote)?;
 
     let mut options = PushOptions::new();
     options.remote_callbacks(remote_callbacks());
+    options.packbuilder_parallelism(0);
 
     remote.push(&[branch], Some(&mut options))?;
 
@@ -60,6 +53,17 @@ pub fn push_origin(repo_path: &str, branch: &str) -> Result<()> {
 
 fn remote_callbacks<'a>() -> RemoteCallbacks<'a> {
     let mut callbacks = RemoteCallbacks::new();
+    callbacks.push_transfer_progress(|progress, total, bytes| {
+        log::debug!(
+            "progress: {}/{} ({} B)",
+            progress,
+            total,
+            bytes,
+        );
+    });
+    callbacks.pack_progress(|stage, current, total| {
+        log::debug!("packing: {:?} - {}/{}", stage, current, total);
+    });
     callbacks.credentials(|url, username_from_url, allowed_types| {
         log::debug!(
             "creds: '{}' {:?} ({:?})",