Perform git pulls in parallel (fix #48)

src/steps/git.rs

@@ -1,12 +1,16 @@
-use crate::error::Error;
+use crate::error::{Error, ErrorKind};
 use crate::executor::{CommandExt, RunType};
 use crate::terminal::print_separator;
 use crate::utils::{which, HumanizedPath};
+use console::style;
+use futures::future::{join_all, Future};
 use log::{debug, error};
 use std::collections::HashSet;
 use std::io;
 use std::path::{Path, PathBuf};
 use std::process::Command;
+use tokio::runtime::Runtime;
+use tokio_process::CommandExt as TokioCommandExt;
 
 #[derive(Debug)]
 pub struct Git {
@@ -55,18 +59,63 @@ impl Git {
         None
     }
 
-    pub fn pull<P: AsRef<Path>>(&self, path: P, run_type: RunType) -> Result<(), Error> {
-        let path = path.as_ref();
-
-        print_separator(format!("Pulling {}", HumanizedPath::from(path)));
+    pub fn multi_pull(&self, repositories: &Repositories, run_type: RunType) -> Result<(), Error> {
+        if repositories.repositories.is_empty() {
+            return Ok(());
+        }
 
         let git = self.git.as_ref().unwrap();
 
-        run_type
-            .execute(git)
-            .args(&["pull", "--rebase", "--autostash"])
-            .current_dir(&path)
-            .check_run()
+        print_separator("Git repositories");
+
+        if let RunType::Dry = run_type {
+            repositories
+                .repositories
+                .iter()
+                .for_each(|repo| println!("Would pull {}", HumanizedPath::from(std::path::Path::new(&repo))));
+
+            return Ok(());
+        }
+
+        let futures: Vec<_> = repositories
+            .repositories
+            .iter()
+            .map(|repo| {
+                let repo = repo.clone();
+                let path = format!("{}", HumanizedPath::from(std::path::Path::new(&repo)));
+                println!("{} {}", style("Pulling").cyan(), path);
+                Command::new(git)
+                    .args(&["pull", "--rebase", "--autostash"])
+                    .current_dir(&repo)
+                    .output_async()
+                    .then(move |result| match result {
+                        Ok(output) => {
+                            if output.status.success() {
+                                println!("{} {}", style("Pulled").green(), path);
+                                Ok(true) as Result<bool, Error>
+                            } else {
+                                println!("{} pulling {}", style("Failed").red(), path);
+                                if let Ok(text) = std::str::from_utf8(&output.stderr) {
+                                    print!("{}", text);
+                                }
+                                Ok(false)
+                            }
+                        }
+                        Err(e) => {
+                            println!("{} pulling {}: {}", style("Failed").red(), path, e);
+                            Ok(false)
+                        }
+                    })
+            })
+            .collect();
+
+        let mut runtime = Runtime::new().unwrap();
+        let results: Vec<bool> = runtime.block_on(join_all(futures))?;
+        if results.into_iter().any(|success| !success) {
+            Err(ErrorKind::StepFailed.into())
+        } else {
+            Ok(())
+        }
     }
 }
 
@@ -83,8 +132,4 @@ impl<'a> Repositories<'a> {
             self.repositories.insert(repo);
         }
     }
-
-    pub fn repositories(&self) -> &HashSet<String> {
-        &self.repositories
-    }
 }