make commit filtering an async job (#1842)

Author: extrawurst

asyncgit/Cargo.toml

@@ -25,7 +25,7 @@ openssl-sys = { version = '0.9', features = ["vendored"], optional = true }
 rayon-core = "1.11"
 scopetime = { path = "../scopetime", version = "0.1" }
 serde = { version = "1.0", features = ["derive"] }
-shellexpand = "3.1" 
+shellexpand = "3.1"
 thiserror = "1.0"
 unicode-truncate = "0.2.0"
 url = "2.4"

asyncgit/src/filter_commits.rs

@@ -0,0 +1,149 @@
+use crate::{
+	asyncjob::{AsyncJob, RunParams},
+	error::Result,
+	sync::{self, CommitId, LogWalkerFilter, RepoPath},
+	AsyncGitNotification, ProgressPercent,
+};
+use std::{
+	sync::{Arc, Mutex},
+	time::{Duration, Instant},
+};
+
+///
+pub struct CommitFilterResult {
+	///
+	pub result: Vec<CommitId>,
+	///
+	pub duration: Duration,
+}
+
+enum JobState {
+	Request {
+		commits: Vec<CommitId>,
+		repo_path: RepoPath,
+	},
+	Response(Result<CommitFilterResult>),
+}
+
+///
+#[derive(Clone)]
+pub struct AsyncCommitFilterJob {
+	state: Arc<Mutex<Option<JobState>>>,
+	filter: LogWalkerFilter,
+}
+
+///
+impl AsyncCommitFilterJob {
+	///
+	pub fn new(
+		repo_path: RepoPath,
+		commits: Vec<CommitId>,
+		filter: LogWalkerFilter,
+	) -> Self {
+		Self {
+			state: Arc::new(Mutex::new(Some(JobState::Request {
+				repo_path,
+				commits,
+			}))),
+			filter,
+		}
+	}
+
+	///
+	pub fn result(&self) -> Option<Result<CommitFilterResult>> {
+		if let Ok(mut state) = self.state.lock() {
+			if let Some(state) = state.take() {
+				return match state {
+					JobState::Request { .. } => None,
+					JobState::Response(result) => Some(result),
+				};
+			}
+		}
+
+		None
+	}
+
+	fn run_request(
+		&self,
+		repo_path: &RepoPath,
+		commits: Vec<CommitId>,
+		params: &RunParams<AsyncGitNotification, ProgressPercent>,
+	) -> JobState {
+		let response = sync::repo(repo_path)
+			.map(|repo| self.filter_commits(&repo, commits, params))
+			.map(|(start, result)| CommitFilterResult {
+				result,
+				duration: start.elapsed(),
+			});
+
+		JobState::Response(response)
+	}
+
+	fn filter_commits(
+		&self,
+		repo: &git2::Repository,
+		commits: Vec<CommitId>,
+		params: &RunParams<AsyncGitNotification, ProgressPercent>,
+	) -> (Instant, Vec<CommitId>) {
+		let total_amount = commits.len();
+		let start = Instant::now();
+
+		let mut progress = ProgressPercent::new(0, total_amount);
+
+		let result = commits
+			.into_iter()
+			.enumerate()
+			.filter_map(|(idx, c)| {
+				let new_progress =
+					ProgressPercent::new(idx, total_amount);
+
+				if new_progress != progress {
+					Self::update_progress(params, new_progress);
+					progress = new_progress;
+				}
+
+				(*self.filter)(repo, &c)
+					.ok()
+					.and_then(|res| res.then_some(c))
+			})
+			.collect::<Vec<_>>();
+
+		(start, result)
+	}
+
+	fn update_progress(
+		params: &RunParams<AsyncGitNotification, ProgressPercent>,
+		new_progress: ProgressPercent,
+	) {
+		if let Err(e) = params.set_progress(new_progress) {
+			log::error!("progress error: {e}");
+		} else if let Err(e) =
+			params.send(AsyncGitNotification::CommitFilter)
+		{
+			log::error!("send error: {e}");
+		}
+	}
+}
+
+impl AsyncJob for AsyncCommitFilterJob {
+	type Notification = AsyncGitNotification;
+	type Progress = ProgressPercent;
+
+	fn run(
+		&mut self,
+		params: RunParams<Self::Notification, Self::Progress>,
+	) -> Result<Self::Notification> {
+		if let Ok(mut state) = self.state.lock() {
+			*state = state.take().map(|state| match state {
+				JobState::Request { commits, repo_path } => {
+					self.run_request(&repo_path, commits, &params)
+				}
+				JobState::Response(result) => {
+					JobState::Response(result)
+				}
+			});
+		}
+
+		Ok(AsyncGitNotification::CommitFilter)
+	}
+}

asyncgit/src/lib.rs

@@ -38,6 +38,7 @@ mod commit_files;
 mod diff;
 mod error;
 mod fetch_job;
+mod filter_commits;
 mod progress;
 mod pull;
 mod push;
@@ -57,6 +58,7 @@ pub use crate::{
 	diff::{AsyncDiff, DiffParams, DiffType},
 	error::{Error, Result},
 	fetch_job::AsyncFetchJob,
+	filter_commits::{AsyncCommitFilterJob, CommitFilterResult},
 	progress::ProgressPercent,
 	pull::{AsyncPull, FetchRequest},
 	push::{AsyncPush, PushRequest},
@@ -111,6 +113,8 @@ pub enum AsyncGitNotification {
 	Branches,
 	///
 	TreeFiles,
+	///
+	CommitFilter,
 }
 
 /// helper function to calculate the hash of an arbitrary type that implements the `Hash` trait

asyncgit/src/revlog.rs

@@ -81,6 +81,10 @@ impl AsyncLog {
 		start_index: usize,
 		amount: usize,
 	) -> Result<Vec<CommitId>> {
+		if self.partial_extract.load(Ordering::Relaxed) {
+			return Err(Error::Generic(String::from("Faulty usage of AsyncLog: Cannot partially extract items and rely on get_items slice to still work!")));
+		}
+
 		let list = &self.current.lock()?.commits;
 		let list_len = list.len();
 		let min = start_index.min(list_len);

asyncgit/src/sync/logwalker.rs

@@ -32,6 +32,7 @@ impl<'a> Ord for TimeOrderedCommit<'a> {
 	}
 }
 
+//TODO: since its used in more than just the log walker now, we should rename and put in its own file
 ///
 pub type LogWalkerFilter = Arc<
 	Box<dyn Fn(&Repository, &CommitId) -> Result<bool> + Send + Sync>,

src/app.rs

@@ -976,7 +976,7 @@ impl App {
 				self.reset_popup.open(id)?;
 			}
 			InternalEvent::CommitSearch(options) => {
-				self.revlog.search(options)?;
+				self.revlog.search(options);
 			}
 		};
 

src/components/commitlist.rs

@@ -97,6 +97,11 @@ impl CommitList {
 		self.items.clear();
 	}
 
+	///
+	pub fn copy_items(&self) -> Vec<CommitId> {
+		self.commits.clone()
+	}
+
 	///
 	pub fn set_tags(&mut self, tags: Tags) {
 		self.tags = Some(tags);