support stream API with wait_with_pipe()

// Continuously process child process' outputs
spawn!(journalctl)?.wait_with_pipe(&mut |pipe| {
    BufReader::new(pipe)
        .lines()
        .filter_map(|line| line.ok())
        .filter(|line| line.find("usb").is_some())
        .take(10)
        .for_each(|line| println!("{}", line));
    Ok(())
})?;

Author: tao-guo

examples/rust_cookbook.rs

@@ -2,7 +2,8 @@
 // Rewrite examples with rust_cmd_lib from
 // https://rust-lang-nursery.github.io/rust-cookbook/os/external.html
 //
-use cmd_lib::{init_builtin_logger, run_cmd, run_fun, CmdResult};
+use cmd_lib::*;
+use std::io::{BufRead, BufReader};
 fn main() -> CmdResult {
     init_builtin_logger();
     cmd_lib::set_pipefail(false); // do not fail due to pipe errors
@@ -26,7 +27,15 @@ fn main() -> CmdResult {
     run_cmd!(rm -f out.txt)?;
 
     // Continuously process child process' outputs
-    run_cmd!(ping -c 5 www.google.com | awk r#"/time/ {print $(NF-3) " " $(NF-1) " " $NF}"#)?;
+    spawn!(journalctl)?.wait_with_pipe(&mut |pipe| {
+        BufReader::new(pipe)
+            .lines()
+            .filter_map(|line| line.ok())
+            .filter(|line| line.find("usb").is_some())
+            .take(10)
+            .for_each(|line| println!("{}", line));
+        Ok(())
+    })?;
 
     Ok(())
 }

src/child.rs

@@ -6,9 +6,13 @@ use std::process::{Child, ExitStatus};
 use std::thread::JoinHandle;
 use CmdChild::{ProcChild, SyncChild, ThreadChild};
 
+/// Representation of running or exited children processes, connected with pipes
+/// optionally.
+///
+/// Calling `spawn!` macro will return `Result<CmdChildren>`
 pub struct CmdChildren(Vec<CmdChild>);
 impl CmdChildren {
-    pub fn from(children: Vec<CmdChild>) -> Self {
+    pub(crate) fn from(children: Vec<CmdChild>) -> Self {
         Self(children)
     }
 
@@ -39,33 +43,6 @@ impl CmdChildren {
         Ok(())
     }
 
-    pub fn wait_raw_result(&mut self) -> Result<Vec<u8>> {
-        let ret = self.wait_raw_result_nolog();
-        if let Err(ref err) = ret {
-            error!(
-                "Running {} failed, Error: {}",
-                CmdChild::get_full_cmd(&self.0),
-                err
-            );
-        }
-        ret
-    }
-
-    fn wait_raw_result_nolog(&mut self) -> Result<Vec<u8>> {
-        let handle = self.0.pop().unwrap();
-        let wait_last = handle.wait_with_output();
-        match wait_last {
-            Err(e) => {
-                let _ = Self::wait_children(&mut self.0);
-                Err(e)
-            }
-            Ok(output) => {
-                Self::wait_children(&mut self.0)?;
-                Ok(output)
-            }
-        }
-    }
-
     pub fn wait_fun_result(&mut self) -> FunResult {
         let ret = self.wait_fun_result_nolog();
         if let Err(ref err) = ret {
@@ -97,23 +74,55 @@ impl CmdChildren {
             }
         }
     }
+
+    pub fn wait_with_pipe(&mut self, f: &mut dyn FnMut(PipeReader) -> CmdResult) -> CmdResult {
+        let handle = self.0.pop().unwrap();
+        let mut ret = Ok(());
+        match handle {
+            ProcChild {
+                mut child,
+                stderr,
+                stdout,
+                ..
+            } => {
+                if let Some(stdout) = stdout {
+                    ret = f(stdout);
+                    let _ = child.kill();
+                }
+                CmdChild::log_stderr_output(stderr);
+            }
+            ThreadChild { .. } => {
+                panic!("should not wait pipe on thread");
+            }
+            SyncChild { stderr, stdout, .. } => {
+                CmdChild::log_stderr_output(stderr);
+                if let Some(stdout) = stdout {
+                    ret = f(stdout);
+                }
+            }
+        };
+        let _ = Self::wait_children(&mut self.0);
+        ret
+    }
 }
 
 #[derive(Debug)]
 pub enum CmdChild {
     ProcChild {
         child: Child,
         cmd: String,
+        stdout: Option<PipeReader>,
         stderr: Option<PipeReader>,
     },
     ThreadChild {
         child: JoinHandle<CmdResult>,
         cmd: String,
+        stdout: Option<PipeReader>,
         stderr: Option<PipeReader>,
     },
     SyncChild {
-        output: Option<PipeReader>,
         cmd: String,
+        stdout: Option<PipeReader>,
         stderr: Option<PipeReader>,
     },
 }
@@ -134,6 +143,7 @@ impl CmdChild {
                 mut child,
                 stderr,
                 cmd,
+                ..
             } => {
                 Self::log_stderr_output(stderr);
                 if let Some(stdout) = child.stdout.take() {
@@ -150,7 +160,9 @@ impl CmdChild {
                     ));
                 }
             }
-            ThreadChild { child, cmd, stderr } => {
+            ThreadChild {
+                child, cmd, stderr, ..
+            } => {
                 let status = child.join();
                 Self::log_stderr_output(stderr);
                 match status {
@@ -167,12 +179,12 @@ impl CmdChild {
                     }
                 }
             }
-            SyncChild { output, stderr, .. } => {
+            SyncChild { stdout, stderr, .. } => {
                 Self::log_stderr_output(stderr);
-                if let Some(mut out) = output {
+                if let Some(mut out) = stdout {
                     let mut buf = vec![];
                     check_result(out.read_to_end(&mut buf).map(|_| ()))?;
-                    println!("{}", String::from_utf8_lossy(&buf));
+                    print!("{}", String::from_utf8_lossy(&buf));
                 }
             }
         }
@@ -181,24 +193,32 @@ impl CmdChild {
 
     pub fn wait_with_output(self) -> Result<Vec<u8>> {
         match self {
-            ProcChild { child, cmd, stderr } => {
+            ProcChild {
+                mut child,
+                cmd,
+                stdout,
+                stderr,
+            } => {
+                let mut buf = vec![];
+                if let Some(mut stdout) = stdout {
+                    stdout.read_to_end(&mut buf)?;
+                }
                 Self::log_stderr_output(stderr);
-                let output = child.wait_with_output()?;
-                if !output.status.success() {
+                let status = child.wait()?;
+                if !status.success() {
                     return Err(Self::status_to_io_error(
-                        output.status,
+                        status,
                         &format!("{} exited with error", cmd),
                     ));
-                } else {
-                    Ok(output.stdout)
                 }
+                Ok(buf)
             }
             ThreadChild { cmd, .. } => {
                 panic!("{} thread should not be waited for output", cmd);
             }
-            SyncChild { output, stderr, .. } => {
+            SyncChild { stdout, stderr, .. } => {
                 Self::log_stderr_output(stderr);
-                if let Some(mut out) = output {
+                if let Some(mut out) = stdout {
                     let mut buf = vec![];
                     out.read_to_end(&mut buf)?;
                     return Ok(buf);

src/lib.rs

@@ -331,6 +331,7 @@ pub use builtins::{
     builtin_cat, builtin_debug, builtin_die, builtin_echo, builtin_error, builtin_info,
     builtin_trace, builtin_true, builtin_warn,
 };
+pub use child::CmdChildren;
 #[doc(hidden)]
 pub use log;
 pub use logger::init_builtin_logger;

src/process.rs

@@ -10,7 +10,7 @@ use std::fmt;
 use std::fs::{File, OpenOptions};
 use std::io::{Error, ErrorKind, Read, Result, Write};
 use std::path::Path;
-use std::process::{Command, Stdio};
+use std::process::Command;
 use std::sync::Mutex;
 
 /// Environment for builtin or custom commands
@@ -249,7 +249,8 @@ pub struct Cmd {
     stdin_redirect: Option<CmdIn>,
     stdout_redirect: Option<CmdOut>,
     stderr_redirect: Option<CmdOut>,
-    stderr_logging: Option<PipeReader>, // for builtin/custom commands
+    stdout_logging: Option<PipeReader>,
+    stderr_logging: Option<PipeReader>,
 }
 
 impl Default for Cmd {
@@ -263,6 +264,7 @@ impl Default for Cmd {
             stdin_redirect: None,
             stdout_redirect: None,
             stderr_redirect: None,
+            stdout_logging: None,
             stderr_logging: None,
         }
     }
@@ -342,12 +344,11 @@ impl Cmd {
             self.run_cd_cmd(current_dir)?;
             Ok(CmdChild::SyncChild {
                 cmd: full_cmd,
-                output: None,
+                stdout: None,
                 stderr: None,
             })
         } else if self.in_cmd_map {
-            let pipe_out = matches!(self.stdout_redirect, Some(CmdOut::CmdPipe(_)));
-            let mut new_pipe_out = None;
+            let pipe_out = self.stdout_logging.is_none();
             let mut env = CmdEnv {
                 args: self.args,
                 vars: self.vars,
@@ -360,9 +361,7 @@ impl Cmd {
                 stdout: if let Some(redirect_out) = self.stdout_redirect.take() {
                     redirect_out
                 } else {
-                    let (pipe_reader, pipe_writer) = os_pipe::pipe()?;
-                    new_pipe_out = Some(pipe_reader);
-                    CmdOut::CmdPipe(pipe_writer)
+                    CmdOut::CmdPipe(os_pipe::dup_stdout()?)
                 },
                 stderr: if let Some(redirect_err) = self.stderr_redirect.take() {
                     redirect_err
@@ -376,6 +375,7 @@ impl Cmd {
                 let handle = std::thread::spawn(move || internal_cmd(&mut env));
                 Ok(CmdChild::ThreadChild {
                     child: handle,
+                    stdout: self.stdout_logging,
                     stderr: self.stderr_logging,
                     cmd: full_cmd,
                 })
@@ -384,7 +384,7 @@ impl Cmd {
                 drop(env);
                 Ok(CmdChild::SyncChild {
                     cmd: full_cmd,
-                    output: new_pipe_out,
+                    stdout: self.stdout_logging,
                     stderr: self.stderr_logging,
                 })
             }
@@ -404,8 +404,6 @@ impl Cmd {
             // update stdout
             if let Some(redirect_out) = self.stdout_redirect.take() {
                 cmd.stdout(redirect_out);
-            } else {
-                cmd.stdout(Stdio::piped());
             }
 
             // update stderr
@@ -417,6 +415,7 @@ impl Cmd {
             let child = cmd.spawn()?;
             Ok(CmdChild::ProcChild {
                 cmd: full_cmd,
+                stdout: self.stdout_logging,
                 stderr: self.stderr_logging,
                 child,
             })
@@ -470,11 +469,17 @@ impl Cmd {
         self.stderr_redirect = Some(CmdOut::CmdPipe(pipe_writer));
         self.stderr_logging = Some(pipe_reader);
 
-        if let Some(pipe) = pipe_in.take() {
-            self.stdin_redirect = Some(CmdIn::CmdPipe(pipe));
-        }
         if let Some(pipe) = pipe_out {
             self.stdout_redirect = Some(CmdOut::CmdPipe(pipe));
+        } else {
+            // set up stdout pipe
+            let (pipe_reader, pipe_writer) = os_pipe::pipe()?;
+            self.stdout_redirect = Some(CmdOut::CmdPipe(pipe_writer));
+            self.stdout_logging = Some(pipe_reader);
+        }
+
+        if let Some(pipe) = pipe_in.take() {
+            self.stdin_redirect = Some(CmdIn::CmdPipe(pipe));
         }
 
         for redirect in self.redirects.iter() {