Ensure that stdout and stderr are output together from the sh step

Fixes #43

Cargo.lock

@@ -2121,6 +2121,7 @@ dependencies = [
 name = "sh-step"
 version = "0.1.0"
 dependencies = [
+ "os_pipe",
  "otto-agent",
  "serde 1.0.117",
  "tempfile",

cli/agent/src/main.rs

@@ -63,7 +63,12 @@ async fn main() -> std::io::Result<()> {
     mkdir_if_not_exists(&work_dir)?;
     mkdir_if_not_exists(&cache_dir)?;
 
-    std::env::set_var("CACHES_DIR", cache_dir.canonicalize().expect("Failed to canonicalize cache directory"));
+    std::env::set_var(
+        "CACHES_DIR",
+        cache_dir
+            .canonicalize()
+            .expect("Failed to canonicalize cache directory"),
+    );
     std::env::set_current_dir(work_dir)?;
 
     let (sender, receiver) = channel(MAX_CONTROL_MSGS);

services/parser/src/lib.rs

@@ -81,11 +81,11 @@ fn parse_kwarg(parser: &mut Pairs<Rule>) -> Option<(String, Value)> {
         match parsed.as_rule() {
             Rule::IDENT => {
                 key = Some(parsed.as_str().to_string());
-            },
+            }
             Rule::STR => {
                 value = Some(Value::String(parse_str(&mut parsed)));
-            },
-            _ => {},
+            }
+            _ => {}
         }
     }
 
@@ -116,12 +116,12 @@ fn parse_steps(parser: &mut Pairs<Rule>, uuid: Uuid) -> Vec<Step> {
                 match part.as_rule() {
                     Rule::IDENT => {
                         symbol = Some(part.as_str().to_string());
-                    },
+                    }
                     Rule::kwarg => {
                         if let Some((key, value)) = parse_kwarg(&mut part.into_inner()) {
                             kwargs.insert(key, value);
                         }
-                    },
+                    }
                     Rule::args => {
                         let mut pairs: Vec<Pair<Rule>> = part.into_inner().collect();
                         for mut pair in pairs.iter_mut() {
@@ -129,21 +129,22 @@ fn parse_steps(parser: &mut Pairs<Rule>, uuid: Uuid) -> Vec<Step> {
                                 args.push(Value::String(parse_str(&mut pair)));
                             }
                         }
-                    },
-                    _ => {},
+                    }
+                    _ => {}
                 }
             }
 
             if let Some(symbol) = symbol {
                 if kwargs.len() > 0 {
                     if args.len() > 0 {
-                        error!("Parsed keyword and positional arguments out, discarding positionals");
+                        error!(
+                            "Parsed keyword and positional arguments out, discarding positionals"
+                        );
                     }
                     let parameters = StepParameters::Keyword(kwargs);
                     let step = Step::new(uuid, symbol, parameters);
                     steps.push(step);
-                }
-                else {
+                } else {
                     let parameters = StepParameters::Positional(args);
                     let step = Step::new(uuid, symbol, parameters);
                     steps.push(step);
@@ -353,7 +354,9 @@ mod tests {
         assert_eq!(step.symbol, "git");
 
         match &step.parameters {
-            StepParameters::Positional(_args) => assert!(false, "Shouldn't have positional arguments"),
+            StepParameters::Positional(_args) => {
+                assert!(false, "Shouldn't have positional arguments")
+            }
             StepParameters::Keyword(kwargs) => {
                 assert_eq!(kwargs.get("url").unwrap(), "https://example.com");
                 assert_eq!(kwargs.get("branch").unwrap(), "main");
@@ -382,10 +385,10 @@ mod tests {
                 assert_eq!(args.len(), 2);
                 assert_eq!(args[0], "https://example.com");
                 assert_eq!(args[1], "main");
-            },
+            }
             StepParameters::Keyword(_kwargs) => {
                 assert!(false, "Not expecting keyword arguments for this step");
-            },
+            }
         }
     }
 }

stdlib/sh/Cargo.toml

@@ -5,6 +5,7 @@ authors = ["R. Tyler Croy <rtyler@brokenco.de>"]
 edition = "2018"
 
 [dependencies]
+os_pipe = "~0.9.2"
 otto-agent = { path = "../../crates/agent" }
 serde = {version = "~1.0.117", features = ["derive"]}
 tempfile = "~3.1.0"

stdlib/sh/src/main.rs

@@ -2,8 +2,9 @@
  * A very simple step which just invokes a shell script with some flags
  */
 
+use os_pipe::pipe;
 use serde::Deserialize;
-use std::io::{stderr, stdout, Write};
+use std::io::{BufRead, BufReader, Write};
 use std::process::Command;
 use tempfile::NamedTempFile;
 
@@ -29,19 +30,30 @@ fn main() -> std::io::Result<()> {
     writeln!(file, "{}", invoke.parameters.script)
         .expect("Failed to write temporary file for script");
 
-    let output = Command::new("/bin/sh")
-        .arg("-xe")
-        .arg(file.path())
-        .output()
-        .expect("Failed to invoke the script");
+    let mut cmd = Command::new("/bin/sh");
+    cmd.arg("-xe");
+    cmd.arg(file.path());
 
-    stdout().write_all(&output.stdout).unwrap();
-    stderr().write_all(&output.stderr).unwrap();
+    let (reader, writer) = pipe().unwrap();
+    let writer_clone = writer.try_clone().unwrap();
+    cmd.stdout(writer);
+    cmd.stderr(writer_clone);
+
+    let mut handle = cmd.spawn()?;
+    drop(cmd);
+
+    let bufr = BufReader::new(reader);
+    for line in bufr.lines() {
+        if let Ok(buffer) = line {
+            println!("{}", buffer);
+        }
+    }
+
+    let status = handle.wait()?;
 
     std::process::exit(
-        output
-            .status
+        status
             .code()
-            .expect("Failed to get status code of script"),
+            .expect("Could not get exit code from subprocess"),
     );
 }