move more tui behavior into the state object

3 files changed, 111 insertions, 78 deletions

diff --git a/src/lib.rs b/src/lib.rs
index ebe3785..df7b61e 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -8,6 +8,7 @@
 #![allow(clippy::match_same_arms)]
 #![allow(clippy::module_name_repetitions)]
 #![allow(clippy::multiple_crate_versions)]
+#![allow(clippy::similar_names)]
 #![allow(clippy::single_match)]
 #![allow(clippy::write_with_newline)]
 
diff --git a/src/state.rs b/src/state.rs
index 93abb26..652911c 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -1,5 +1,3 @@
-use futures::future::Future as _;
-use futures::sink::Sink as _;
 use futures::stream::Stream as _;
 use snafu::{OptionExt as _, ResultExt as _};
 use std::io::Write as _;
@@ -16,18 +14,18 @@ pub enum Error {
 
     #[snafu(display("error printing output: {}", source))]
     PrintOutput { source: std::io::Error },
+
+    #[snafu(display("error during eval: {}", source))]
+    Eval { source: crate::eval::Error },
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
 
 #[derive(Debug)]
 pub enum StateEvent {
-    Start(usize, String, Vec<String>),
-    Output(usize, Vec<u8>),
-    Exit(usize, std::process::ExitStatus),
+    Line(usize, String, futures::sync::oneshot::Sender<()>),
 }
 
-#[derive(Debug)]
 pub struct State {
     r: futures::sync::mpsc::Receiver<StateEvent>,
     commands: std::collections::HashMap<usize, Command>,
@@ -41,19 +39,36 @@ impl State {
         }
     }
 
-    pub fn command_start(
+    pub fn eval(
         &mut self,
         idx: usize,
-        cmd: &str,
-        args: &[String],
+        line: &str,
+        res: futures::sync::oneshot::Sender<()>,
     ) -> Result<()> {
         snafu::ensure!(
             !self.commands.contains_key(&idx),
             InvalidCommandIndex { idx }
         );
-        let command = Command::new(cmd, args);
-        self.commands.insert(idx, command.clone());
-        eprint!("running '{} {:?}'\r\n", command.cmd, command.args);
+        let eval = crate::eval::eval(line).context(Eval)?;
+        self.commands.insert(idx, Command::new(eval, res));
+        Ok(())
+    }
+
+    pub fn command_start(
+        &mut self,
+        idx: usize,
+        cmd: &str,
+        args: &[String],
+    ) -> Result<()> {
+        let command = self
+            .commands
+            .get_mut(&idx)
+            .context(InvalidCommandIndex { idx })?;
+        let cmd = cmd.to_string();
+        let args = args.to_vec();
+        eprint!("running '{} {:?}'\r\n", cmd, args);
+        command.cmd = Some(cmd);
+        command.args = Some(args);
         Ok(())
     }
 
@@ -97,63 +112,88 @@ impl futures::future::Future for State {
 
     fn poll(&mut self) -> futures::Poll<Self::Item, Self::Error> {
         loop {
-            let event = futures::try_ready!(self
-                .r
-                .poll()
-                .map_err(|_: ()| unreachable!()));
-            match event {
-                Some(StateEvent::Start(idx, cmd, args)) => {
-                    self.command_start(idx, &cmd, &args)?;
+            let mut did_work = false;
+
+            match self.r.poll().map_err(|()| unreachable!())? {
+                futures::Async::Ready(Some(StateEvent::Line(
+                    idx,
+                    line,
+                    res,
+                ))) => {
+                    self.eval(idx, &line, res)?;
+                    did_work = true;
                 }
-                Some(StateEvent::Output(idx, output)) => {
-                    self.command_output(idx, &output)?;
+                futures::Async::Ready(None) => {
+                    return Ok(futures::Async::Ready(()));
                 }
-                Some(StateEvent::Exit(idx, status)) => {
-                    self.command_exit(idx, status)?;
+                futures::Async::NotReady => {}
+            }
+
+            for idx in self.commands.keys().cloned().collect::<Vec<usize>>() {
+                match self
+                    .commands
+                    .get_mut(&idx)
+                    .unwrap()
+                    .future
+                    .poll()
+                    .map_err(|e| Error::Eval { source: e })?
+                {
+                    futures::Async::Ready(Some(event)) => match event {
+                        crate::eval::CommandEvent::CommandStart(
+                            cmd,
+                            args,
+                        ) => {
+                            self.command_start(idx, &cmd, &args)?;
+                            did_work = true;
+                        }
+                        crate::eval::CommandEvent::Output(out) => {
+                            self.command_output(idx, &out)?;
+                            did_work = true;
+                        }
+                        crate::eval::CommandEvent::CommandExit(status) => {
+                            self.command_exit(idx, status)?;
+                            did_work = true;
+                        }
+                    },
+                    futures::Async::Ready(None) => {
+                        self.commands
+                            .remove(&idx)
+                            .context(InvalidCommandIndex { idx })?
+                            .res
+                            .send(())
+                            .map_err(|()| unreachable!())?;
+                        did_work = true;
+                    }
+                    futures::Async::NotReady => {}
                 }
-                None => return Ok(futures::Async::Ready(())),
             }
-        }
-    }
-}
 
-pub fn update(
-    w: futures::sync::mpsc::Sender<StateEvent>,
-    idx: usize,
-    event: &crate::eval::CommandEvent,
-) -> impl futures::future::Future<Item = (), Error = Error> {
-    match event {
-        crate::eval::CommandEvent::CommandStart(cmd, args) => {
-            w.send(crate::state::StateEvent::Start(
-                idx,
-                cmd.to_string(),
-                args.to_vec(),
-            ))
-        }
-        crate::eval::CommandEvent::Output(out) => {
-            w.send(crate::state::StateEvent::Output(idx, out.to_vec()))
-        }
-        crate::eval::CommandEvent::CommandExit(status) => {
-            w.send(crate::state::StateEvent::Exit(idx, *status))
+            if !did_work {
+                return Ok(futures::Async::NotReady);
+            }
         }
     }
-    .map(|_| ())
-    .map_err(|e| Error::Sending { source: e })
 }
 
-#[derive(Debug, Clone)]
 struct Command {
-    cmd: String,
-    args: Vec<String>,
+    future: crate::eval::Eval,
+    res: futures::sync::oneshot::Sender<()>,
+    cmd: Option<String>,
+    args: Option<Vec<String>>,
     output: Vec<u8>,
     status: Option<std::process::ExitStatus>,
 }
 
 impl Command {
-    fn new(cmd: &str, args: &[String]) -> Self {
+    fn new(
+        future: crate::eval::Eval,
+        res: futures::sync::oneshot::Sender<()>,
+    ) -> Self {
         Self {
-            cmd: cmd.to_string(),
-            args: args.to_vec(),
+            future,
+            res,
+            cmd: None,
+            args: None,
             output: vec![],
             status: None,
         }
diff --git a/src/tui.rs b/src/tui.rs
index 8d84a7f..88a169d 100644
--- a/src/tui.rs
+++ b/src/tui.rs
@@ -1,5 +1,5 @@
 use futures::future::{Future as _, IntoFuture as _};
-use futures::stream::Stream as _;
+use futures::sink::Sink as _;
 use std::io::Write as _;
 
 #[derive(Debug, snafu::Snafu)]
@@ -12,6 +12,16 @@ pub enum Error {
 
     #[snafu(display("error during print: {}", source))]
     Print { source: crate::state::Error },
+
+    #[snafu(display("error during sending: {}", source))]
+    Sending {
+        source: futures::sync::mpsc::SendError<crate::state::StateEvent>,
+    },
+
+    #[snafu(display("error during receiving: {}", source))]
+    Receiving {
+        source: futures::sync::oneshot::Canceled,
+    },
 }
 
 pub fn tui() {
@@ -26,11 +36,12 @@ pub fn tui() {
             let w = w.clone();
             read()
                 .and_then(move |line| {
-                    let w = w.clone();
-                    eval(&line).for_each(move |event| {
-                        let w = w.clone();
-                        print(w, idx, &event)
-                    })
+                    let (res, req) = futures::sync::oneshot::channel();
+                    w.send(crate::state::StateEvent::Line(idx, line, res))
+                        .map_err(|e| Error::Sending { source: e })
+                        .and_then(|_| {
+                            req.map_err(|e| Error::Receiving { source: e })
+                        })
                 })
                 .then(move |res| match res {
                     // successful run or empty input means prompt again
@@ -64,25 +75,6 @@ fn read() -> impl futures::future::Future<Item = String, Error = Error> {
         .map_err(|e| Error::Read { source: e })
 }
 
-fn eval(
-    line: &str,
-) -> impl futures::stream::Stream<Item = crate::eval::CommandEvent, Error = Error>
-{
-    crate::eval::eval(line)
-        .into_future()
-        .flatten_stream()
-        .map_err(|e| Error::Eval { source: e })
-}
-
-fn print(
-    w: futures::sync::mpsc::Sender<crate::state::StateEvent>,
-    idx: usize,
-    event: &crate::eval::CommandEvent,
-) -> impl futures::future::Future<Item = (), Error = Error> {
-    crate::state::update(w, idx, event)
-        .map_err(|e| Error::Print { source: e })
-}
-
 fn error(e: &Error) {
     let stderr = std::io::stderr();
     let mut stderr = stderr.lock();