Update output logging to stream in realtime (closes #138)

Author: jbeemster

src/factotum/executor/execution_strategy/mod.rs

@@ -14,8 +14,11 @@
 
 #[cfg(test)]
 mod tests;
-use std::process::Command;
+use std::process::{Command, Stdio};
 use std::time::{Instant, Duration};
+use std::io::{BufRead, BufReader};
+use std::thread;
+use colored::*;
 
 #[derive(Clone, PartialEq, Debug)]
 pub struct RunResult {
@@ -65,34 +68,92 @@ pub fn execute_simulation(name: &str, command: &mut Command) -> RunResult {
 pub fn execute_os(name: &str, command: &mut Command) -> RunResult {
     let run_start = Instant::now();
     info!("Executing sh {:?}", command);
-    match command.output() {
-        Ok(r) => {
-            let run_duration = run_start.elapsed();
-            let return_code = r.status.code().unwrap_or(1); // 1 will be returned if the process was killed by a signal
-
-            let task_stdout: String = String::from_utf8_lossy(&r.stdout).trim_right().into();
-            let task_stderr: String = String::from_utf8_lossy(&r.stderr).trim_right().into();
-
-            info!("task '{}' stdout:\n'{}'", name, task_stdout);
-            info!("task '{}' stderr:\n'{}'", name, task_stderr);
-
-            let task_stdout_opt = if task_stdout.is_empty() {
-                None
-            } else {
-                Some(task_stdout)
-            };
-            let task_stderr_opt = if task_stderr.is_empty() {
-                None
-            } else {
-                Some(task_stderr)
-            };
 
-            RunResult {
-                duration: run_duration,
-                task_execution_error: None,
-                stdout: task_stdout_opt,
-                stderr: task_stderr_opt,
-                return_code: return_code,
+    command.stdout(Stdio::piped());
+    command.stderr(Stdio::piped());
+
+    match command.spawn() {
+        Ok(mut child) => {
+            let stdout = child.stdout.take().expect("Failed to capture stdout");
+            let stderr = child.stderr.take().expect("Failed to capture stderr");
+
+            let name_stdout = name.to_string();
+            let name_stderr = name.to_string();
+
+            let stdout_handle = thread::spawn(move || {
+                let reader = BufReader::new(stdout);
+                let mut lines = Vec::new();
+                for line in reader.lines() {
+                    if let Ok(line) = line {
+                        println!("[{}] {}", name_stdout.cyan(), line);
+                        lines.push(line);
+                    }
+                }
+                lines.join("\n")
+            });
+
+            let stderr_handle = thread::spawn(move || {
+                let reader = BufReader::new(stderr);
+                let mut lines = Vec::new();
+                for line in reader.lines() {
+                    if let Ok(line) = line {
+                        eprintln!("[{}] {}", name_stderr.cyan(), line.red());
+                        lines.push(line);
+                    }
+                }
+                lines.join("\n")
+            });
+
+            match child.wait() {
+                Ok(status) => {
+                    let run_duration = run_start.elapsed();
+                    let return_code = status.code().unwrap_or(1);
+
+                    let task_stdout = match stdout_handle.join() {
+                        Ok(output) => output,
+                        Err(_) => {
+                            warn!("stdout reader thread panicked for task '{}'", name);
+                            String::new()
+                        }
+                    };
+                    let task_stderr = match stderr_handle.join() {
+                        Ok(output) => output,
+                        Err(_) => {
+                            warn!("stderr reader thread panicked for task '{}'", name);
+                            String::new()
+                        }
+                    };
+
+                    info!("task '{}' completed with return code {}", name, return_code);
+
+                    let task_stdout_opt = if task_stdout.is_empty() {
+                        None
+                    } else {
+                        Some(task_stdout)
+                    };
+                    let task_stderr_opt = if task_stderr.is_empty() {
+                        None
+                    } else {
+                        Some(task_stderr)
+                    };
+
+                    RunResult {
+                        duration: run_duration,
+                        task_execution_error: None,
+                        stdout: task_stdout_opt,
+                        stderr: task_stderr_opt,
+                        return_code: return_code,
+                    }
+                }
+                Err(e) => {
+                    RunResult {
+                        duration: run_start.elapsed(),
+                        task_execution_error: Some(format!("Error waiting for process - {}", e)),
+                        stdout: None,
+                        stderr: None,
+                        return_code: -1,
+                    }
+                }
             }
         }
         Err(message) => {

src/main.rs

@@ -245,34 +245,21 @@ fn get_task_result_line_str(task_result: &Task<&FactfileTask>) -> (String, Optio
     return (result, stderr);
 }
 
-fn get_task_results_str(task_results: &Vec<&Task<&FactfileTask>>) -> (String, String) {
-    let mut stderr = String::new();
-    let mut stdout = String::new();
-
+fn get_task_summary_str(task_results: &Vec<&Task<&FactfileTask>>) -> String {
     let mut total_run_time = Duration::new(0, 0);
     let mut executed = 0;
 
     for task in task_results.iter() {
-        let (task_stdout, task_stderr) = get_task_result_line_str(task);
-        stdout.push_str(&task_stdout);
-
-        if let Some(task_stderr_str) = task_stderr {
-            stderr.push_str(&task_stderr_str);
-        }
-
         if let Some(ref run_result) = task.run_result {
             total_run_time = total_run_time + run_result.duration;
             executed += 1;
         }
     }
 
-    let summary = format!("{}/{} tasks run in {}\n",
-                          executed,
-                          task_results.len(),
-                          get_duration_as_string(&total_run_time));
-    stdout.push_str(&summary.green().to_string());
-
-    (stdout, stderr)
+    format!("{}/{} tasks run in {}\n",
+            executed,
+            task_results.len(),
+            get_duration_as_string(&total_run_time)).green().to_string()
 }
 
 fn validate_start_task(job: &Factfile, start_task: &str) -> Result<(), &'static str> {
@@ -422,18 +409,12 @@ fn parse_file_and_execute_with_strategy<F>(factfile: &str,
             let normal_completion = !has_errors && !has_early_finish;
 
             let result = if normal_completion {
-                let (stdout_summary, stderr_summary) = get_task_results_str(&tasks);
-                print!("{}", stdout_summary);
-                if !stderr_summary.trim_right().is_empty() {
-                    print_err!("{}", stderr_summary.trim_right());
-                }
+                let summary = get_task_summary_str(&tasks);
+                print!("{}", summary);
                 PROC_SUCCESS
             } else if has_early_finish && !has_errors {
-                let (stdout_summary, stderr_summary) = get_task_results_str(&tasks);
-                print!("{}", stdout_summary);
-                if !stderr_summary.trim_right().is_empty() {
-                    print_err!("{}", stderr_summary.trim_right());
-                }
+                let summary = get_task_summary_str(&tasks);
+                print!("{}", summary);
                 let incomplete_tasks = tasks.iter()
                     .filter(|r| !r.run_result.is_some())
                     .map(|r| format!("'{}'", r.name.cyan()))
@@ -453,12 +434,8 @@ fn parse_file_and_execute_with_strategy<F>(factfile: &str,
                          incomplete_tasks);
                 PROC_SUCCESS
             } else {
-                let (stdout_summary, stderr_summary) = get_task_results_str(&tasks);
-                print!("{}", stdout_summary);
-
-                if !stderr_summary.trim_right().is_empty() {
-                    print_err!("{}", stderr_summary.trim_right());
-                }
+                let summary = get_task_summary_str(&tasks);
+                print!("{}", summary);
 
                 let incomplete_tasks = tasks.iter()
                     .filter(|r| !r.run_result.is_some())
@@ -1189,102 +1166,6 @@ fn test_get_task_result_line_str() {
 
 }
 
-#[test]
-fn test_get_task_results_str_summary() {
-    use chrono::UTC;
-    use factotum::executor::execution_strategy::RunResult;
-    use factotum::factfile::{Task as FactfileTask, OnResult};
-
-    let dt = UTC::now();
-
-    let task_one_spec = FactfileTask {
-        name: "hello world".to_string(),
-        depends_on: vec![],
-        executor: "".to_string(),
-        command: "".to_string(),
-        arguments: vec![],
-        on_result: OnResult {
-            terminate_job: vec![],
-            continue_job: vec![],
-        },
-    };
-
-    let task_one = Task::<&FactfileTask> {
-        name: String::from("hello world"),
-        // children: vec![],
-        state: State::Success,
-        task_spec: &task_one_spec,
-        run_started: Some(dt),
-        run_result: Some(RunResult {
-            duration: Duration::from_secs(20),
-            task_execution_error: None,
-            stdout: Some(String::from("hello world")),
-            stderr: Some(String::from("Mistake")),
-            return_code: 0,
-        }),
-    };
-
-
-    let task_two_spec = FactfileTask {
-        name: "hello world 2".to_string(),
-        depends_on: vec![],
-        executor: "".to_string(),
-        command: "".to_string(),
-        arguments: vec![],
-        on_result: OnResult {
-            terminate_job: vec![],
-            continue_job: vec![],
-        },
-    };
-
-    let task_two = Task::<&FactfileTask> {
-        name: String::from("hello world 2"),
-        // children: vec![],
-        state: State::Success,
-        task_spec: &task_two_spec,
-        run_started: Some(dt),
-        run_result: Some(RunResult {
-            duration: Duration::from_secs(80),
-            task_execution_error: None,
-            stdout: Some(String::from("hello world")),
-            stderr: Some(String::from("Mistake")),
-            return_code: 0,
-        }),
-    };
-
-    let mut tasks: Vec<&Task<&FactfileTask>> = vec![];
-    let (stdout, stderr) = get_task_results_str(&tasks);
-    let expected: String = format!("{}", "0/0 tasks run in 0.0s\n".green());
-
-    assert_eq!(stdout, expected);
-    assert_eq!(stderr, "");
-
-    tasks.push(&task_one);
-
-    let (one_task_stdout, one_task_stderr) = get_task_results_str(&tasks);
-    let (first_task_stdout, first_task_stderr) = get_task_result_line_str(&tasks[0]);
-    let expected_one_task = format!("{}{}",
-                                    first_task_stdout,
-                                    "1/1 tasks run in 20.0s\n".green());
-
-    assert_eq!(one_task_stdout, expected_one_task);
-    let first_task_stderr_str = first_task_stderr.unwrap();
-    assert_eq!(one_task_stderr, first_task_stderr_str);
-
-    tasks.push(&task_two);
-
-    let (two_task_stdout, two_task_stderr) = get_task_results_str(&tasks);
-    let (task_two_stdout, task_two_stderr) = get_task_result_line_str(&tasks[1]);
-    let expected_two_task = format!("{}{}{}",
-                                    first_task_stdout,
-                                    task_two_stdout,
-                                    "2/2 tasks run in 1m, 40s\n".green());
-    assert_eq!(two_task_stdout, expected_two_task);
-    assert_eq!(two_task_stderr,
-               format!("{}{}", first_task_stderr_str, task_two_stderr.unwrap()));
-
-}
-
 #[test]
 fn test_start_task_validation_not_present() {
     let mut factfile = Factfile::new("N/A", "test");