1 files changed, 64 insertions, 31 deletions
diff --git a/src/repl.rs b/src/repl.rs
index c44174a..af62f95 100644
--- a/src/repl.rs
+++ b/src/repl.rs
@@ -1,53 +1,86 @@
-use futures::future::Future;
+use futures::future::{Future, IntoFuture};
+use futures::stream::Stream;
 use std::io::Write;
 
 #[derive(Debug)]
 enum Error {
     ReadError(crate::readline::Error),
-    // EvalError(std::io::Error),
+    EvalError(crate::process::Error),
     PrintError(std::io::Error),
-    // LoopError,
 }
 
 pub fn repl() {
-    tokio::run(tokio::prelude::future::lazy(|| {
-        let mut done = false;
-        while !done {
-            let res = read()
-                .and_then(move |line| eval(&line))
-                .and_then(move |out| print(&out))
-                .wait();
-            match res {
-                Ok(_) => {}
-                Err(Error::ReadError(crate::readline::Error::EOF)) => {
-                    done = true;
-                }
-                Err(e) => {
-                    let stderr = std::io::stderr();
-                    let mut stderr = stderr.lock();
-                    write!(stderr, "error: {:?}", e).unwrap();
-                    stderr.flush().unwrap();
-                    done = true;
-                }
-            }
+    let loop_stream = futures::stream::unfold(false, |done| {
+        if done {
+            return None;
         }
-        futures::future::ok(())
-    }));
+
+        let repl = read().and_then(|line| {
+            eval(&line).and_then(|(out, status)| {
+                out
+                    // print the results as they come in
+                    .and_then(|out| print(&out))
+                    // wait for all output to be finished
+                    .collect()
+                    // ignore io errors since we just keep reading even after
+                    // the process exits and the other end of the pty is
+                    // closed
+                    .or_else(|_| futures::future::ok(vec![]))
+                    // once the output is all processed, then wait on the
+                    // process to exit
+                    .and_then(|_| status)
+            })
+        });
+
+        Some(repl.then(move |res| match res {
+            Ok(status) => {
+                eprint!("process exited with status {}\r\n", status);
+                return Ok((done, false));
+            }
+            Err(Error::ReadError(crate::readline::Error::EOF)) => {
+                return Ok((done, true));
+            }
+            Err(e) => {
+                let stderr = std::io::stderr();
+                let mut stderr = stderr.lock();
+                write!(stderr, "error: {:?}\r\n", e).unwrap();
+                stderr.flush().unwrap();
+                return Err(());
+            }
+        }))
+    });
+    tokio::run(loop_stream.collect().map(|_| ()));
 }
 
 fn read() -> impl futures::future::Future<Item = String, Error = Error> {
     crate::readline::readline("$ ", true).map_err(|e| Error::ReadError(e))
 }
 
-fn eval(line: &str) -> Result<String, Error> {
-    Ok(format!("got line '{}'\r\n", line))
+fn eval(
+    line: &str,
+) -> impl futures::future::Future<
+    Item = (
+        impl futures::stream::Stream<Item = Vec<u8>, Error = Error>,
+        impl futures::future::Future<
+            Item = std::process::ExitStatus,
+            Error = Error,
+        >,
+    ),
+    Error = Error,
+> {
+    match crate::process::spawn(line) {
+        Ok((out, status)) => Ok((
+            out.map_err(|e| Error::EvalError(e)),
+            status.map_err(|e| Error::EvalError(e)),
+        )),
+        Err(e) => Err(e).map_err(|e| Error::EvalError(e)),
+    }
+    .into_future()
 }
 
-fn print(out: &str) -> Result<(), Error> {
+fn print(out: &[u8]) -> Result<(), Error> {
     let stdout = std::io::stdout();
     let mut stdout = stdout.lock();
-    stdout
-        .write(out.as_bytes())
-        .map_err(|e| Error::PrintError(e))?;
+    stdout.write(out).map_err(|e| Error::PrintError(e))?;
     stdout.flush().map_err(|e| Error::PrintError(e))
 }