make execution saner

1 files changed, 35 insertions, 25 deletions

diff --git a/src/run/mod.rs b/src/run/mod.rs
index 994e222..f767434 100644
--- a/src/run/mod.rs
+++ b/src/run/mod.rs
@@ -1,4 +1,5 @@
 use std::collections::HashMap;
+use std::io::{PipeReader, PipeWriter};
 use std::path::PathBuf;
 use std::sync::{Arc, Mutex};
 
@@ -12,6 +13,7 @@ enum ExecError {
     ExecError(i32),
 }
 
+#[derive(Clone)]
 struct Executor {
     se: Arc<Mutex<Session>>,
 }
@@ -20,7 +22,8 @@ impl Executor {
     fn execute_pipeline(
         &mut self,
         pipes: parse::Pipes<parse::PostExpansion>,
-        capture: Option<&mut Vec<u8>>,
+        mut stdin: Option<PipeReader>,
+        mut stdout: Option<PipeWriter>,
     ) -> Result<(), ExecError> {
         let mut children = Vec::new();
         let mut threads = Vec::new();
@@ -29,7 +32,7 @@ impl Executor {
 
         let pipelen = pipes.cmds.len();
         for (i, cmd) in pipes.cmds.into_iter().enumerate() {
-            let (reader, writer) = if i < pipelen - 1 || capture.is_some() {
+            let (reader, writer) = if i < pipelen - 1 {
                 let (r, w) = io::pipe().unwrap();
                 (Some(r), Some(w))
             } else {
@@ -47,14 +50,14 @@ impl Executor {
 
                     if let Some(r) = prev_reader.take() {
                         command.stdin(Stdio::from(r));
-                    } else if capture.is_some() {
-                        command.stdin(Stdio::null());
+                    } else if let Some(stdin) = stdin.take() {
+                        command.stdin(Stdio::from(stdin));
                     }
 
                     if let Some(w) = writer {
                         command.stdout(Stdio::from(w));
-                    } else if capture.is_some() {
-                        command.stdout(Stdio::piped());
+                    } else if let Some(stdout) = stdout.take() {
+                        command.stdout(Stdio::from(stdout));
                     }
 
                     let child = match command.spawn() {
@@ -91,15 +94,20 @@ impl Executor {
                 }
 
                 CommandKind::Builtin(builtin) => {
-                    let mut input: Box<dyn io::Read + Send> = match prev_reader.take() {
-                        Some(r) => Box::new(r),
-                        None if capture.is_some() => Box::new(io::empty()),
-                        None => Box::new(io::stdin()),
+                    let mut input: Box<dyn io::Read + Send> = if let Some(r) = prev_reader.take() {
+                        Box::new(r)
+                    } else if let Some(stdin) = stdin.take() {
+                        Box::new(stdin)
+                    } else {
+                        Box::new(io::stdin())
                     };
 
-                    let mut output: Box<dyn io::Write + Send> = match writer {
-                        Some(w) => Box::new(w),
-                        None => Box::new(io::stdout()),
+                    let mut output: Box<dyn io::Write + Send> = if let Some(w) = writer {
+                        Box::new(w)
+                    } else if let Some(w) = stdout.take() {
+                        Box::new(w)
+                    } else {
+                        Box::new(io::stdout())
                     };
 
                     // currently only required for `re`, cannot happen in background thread
@@ -130,10 +138,6 @@ impl Executor {
             return Err(ExecError::ExecError(127));
         }
 
-        if let (Some(cap), Some(mut reader)) = (capture, prev_reader) {
-            reader.read_to_end(cap).unwrap();
-        }
-
         let mut code = 0;
         for jh in threads {
             match jh.join() {
@@ -178,12 +182,13 @@ impl Executor {
     fn execute(
         &mut self,
         ast: Ast<parse::PostExpansion>,
-        capture: Option<&mut Vec<u8>>,
+        stdin: Option<PipeReader>,
+        stdout: Option<PipeWriter>,
     ) -> Result<(), ExecError> {
         match ast {
             Ast::FunDecl(fd) => self.execute_fun_decl(fd),
             Ast::VarAssign(va) => self.execute_var_assign(va),
-            Ast::Pipes(pipes) => self.execute_pipeline(pipes, capture),
+            Ast::Pipes(pipes) => self.execute_pipeline(pipes, stdin, stdout),
         }
     }
 }
@@ -219,19 +224,24 @@ impl parse::Expander for Executor {
     }
 
     fn expand_cmd(&mut self, ast: Ast<parse::PostExpansion>) -> Result<BString, Self::Error> {
-        let mut out = Vec::new();
-        self.execute(ast, Some(&mut out))?;
-        if out.last() == Some(&b'\n') {
-            out.pop();
+        let (stdin, _) = io::pipe().unwrap();
+        let (mut expansion, stdout) = io::pipe().unwrap();
+        let mut this = self.clone();
+        let t = std::thread::spawn(move || this.execute(ast, Some(stdin), Some(stdout)));
+        let mut buf = Vec::new();
+        expansion.read_to_end(&mut buf).unwrap();
+        t.join().unwrap()?;
+        if buf.last() == Some(&b'\n') {
+            buf.pop();
         }
-        Ok(out)
+        Ok(buf)
     }
 }
 
 fn exec(se: Arc<Mutex<Session>>, ast: Ast<PreExpansion>) -> Result<(), ExecError> {
     let mut exec = Executor { se };
     let ast = ast.expand(&mut exec)?;
-    exec.execute(ast, None)
+    exec.execute(ast, None, None)
 }
 
 pub fn run(se: Arc<Mutex<Session>>, cmd: Vec<u8>) {