make cast run a command

doesn't do anything with it yet

5 files changed, 449 insertions, 16 deletions

diff --git a/Cargo.lock b/Cargo.lock
index b7cf23b..79b2d6d 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -9,6 +9,11 @@ dependencies = [
 ]
 
 [[package]]
+name = "arc-swap"
+version = "0.4.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+
+[[package]]
 name = "arrayvec"
 version = "0.4.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -141,6 +146,90 @@ dependencies = [
 ]
 
 [[package]]
+name = "crossterm"
+version = "0.10.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "crossterm_cursor 0.2.6 (registry+https://github.com/rust-lang/crates.io-index)",
+ "crossterm_input 0.3.9 (registry+https://github.com/rust-lang/crates.io-index)",
+ "crossterm_screen 0.2.5 (registry+https://github.com/rust-lang/crates.io-index)",
+ "crossterm_style 0.4.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "crossterm_terminal 0.2.6 (registry+https://github.com/rust-lang/crates.io-index)",
+ "crossterm_utils 0.2.4 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
+[[package]]
+name = "crossterm_cursor"
+version = "0.2.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "crossterm_utils 0.2.4 (registry+https://github.com/rust-lang/crates.io-index)",
+ "crossterm_winapi 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)",
+ "winapi 0.3.8 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
+[[package]]
+name = "crossterm_input"
+version = "0.3.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "crossterm_screen 0.2.5 (registry+https://github.com/rust-lang/crates.io-index)",
+ "crossterm_utils 0.2.4 (registry+https://github.com/rust-lang/crates.io-index)",
+ "crossterm_winapi 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)",
+ "libc 0.2.62 (registry+https://github.com/rust-lang/crates.io-index)",
+ "winapi 0.3.8 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
+[[package]]
+name = "crossterm_screen"
+version = "0.2.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "crossterm_utils 0.2.4 (registry+https://github.com/rust-lang/crates.io-index)",
+ "crossterm_winapi 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)",
+ "winapi 0.3.8 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
+[[package]]
+name = "crossterm_style"
+version = "0.4.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "crossterm_utils 0.2.4 (registry+https://github.com/rust-lang/crates.io-index)",
+ "crossterm_winapi 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)",
+ "winapi 0.3.8 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
+[[package]]
+name = "crossterm_terminal"
+version = "0.2.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "crossterm_cursor 0.2.6 (registry+https://github.com/rust-lang/crates.io-index)",
+ "crossterm_utils 0.2.4 (registry+https://github.com/rust-lang/crates.io-index)",
+ "crossterm_winapi 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)",
+ "libc 0.2.62 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
+[[package]]
+name = "crossterm_utils"
+version = "0.2.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "crossterm_winapi 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)",
+ "libc 0.2.62 (registry+https://github.com/rust-lang/crates.io-index)",
+ "winapi 0.3.8 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
+[[package]]
+name = "crossterm_winapi"
+version = "0.1.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "winapi 0.3.8 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
+[[package]]
 name = "doc-comment"
 version = "0.3.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -482,6 +571,24 @@ version = "0.7.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 
 [[package]]
+name = "signal-hook"
+version = "0.1.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "libc 0.2.62 (registry+https://github.com/rust-lang/crates.io-index)",
+ "signal-hook-registry 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
+[[package]]
+name = "signal-hook-registry"
+version = "1.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "arc-swap 0.4.2 (registry+https://github.com/rust-lang/crates.io-index)",
+ "libc 0.2.62 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
+[[package]]
 name = "slab"
 version = "0.4.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -537,10 +644,12 @@ name = "termcast"
 version = "0.1.0"
 dependencies = [
  "clap 2.33.0 (registry+https://github.com/rust-lang/crates.io-index)",
+ "crossterm 0.10.2 (registry+https://github.com/rust-lang/crates.io-index)",
  "futures 0.1.28 (registry+https://github.com/rust-lang/crates.io-index)",
  "mio 0.6.19 (registry+https://github.com/rust-lang/crates.io-index)",
  "snafu 0.4.4 (registry+https://github.com/rust-lang/crates.io-index)",
  "tokio 0.1.22 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tokio-pty-process 0.4.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "uuid 0.7.4 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
@@ -624,6 +733,20 @@ dependencies = [
 ]
 
 [[package]]
+name = "tokio-pty-process"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "bytes 0.4.12 (registry+https://github.com/rust-lang/crates.io-index)",
+ "futures 0.1.28 (registry+https://github.com/rust-lang/crates.io-index)",
+ "libc 0.2.62 (registry+https://github.com/rust-lang/crates.io-index)",
+ "mio 0.6.19 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tokio 0.1.22 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tokio-io 0.1.12 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tokio-signal 0.2.7 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
+[[package]]
 name = "tokio-reactor"
 version = "0.1.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -642,6 +765,22 @@ dependencies = [
 ]
 
 [[package]]
+name = "tokio-signal"
+version = "0.2.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "futures 0.1.28 (registry+https://github.com/rust-lang/crates.io-index)",
+ "libc 0.2.62 (registry+https://github.com/rust-lang/crates.io-index)",
+ "mio 0.6.19 (registry+https://github.com/rust-lang/crates.io-index)",
+ "mio-uds 0.6.7 (registry+https://github.com/rust-lang/crates.io-index)",
+ "signal-hook 0.1.10 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tokio-executor 0.1.8 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tokio-io 0.1.12 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tokio-reactor 0.1.9 (registry+https://github.com/rust-lang/crates.io-index)",
+ "winapi 0.3.8 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
+[[package]]
 name = "tokio-sync"
 version = "0.1.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -784,6 +923,7 @@ dependencies = [
 
 [metadata]
 "checksum ansi_term 0.11.0 (registry+https://github.com/rust-lang/crates.io-index)" = "ee49baf6cb617b853aa8d93bf420db2383fab46d314482ca2803b40d5fde979b"
+"checksum arc-swap 0.4.2 (registry+https://github.com/rust-lang/crates.io-index)" = "854ede29f7a0ce90519fb2439d030320c6201119b87dab0ee96044603e1130b9"
 "checksum arrayvec 0.4.11 (registry+https://github.com/rust-lang/crates.io-index)" = "b8d73f9beda665eaa98ab9e4f7442bd4e7de6652587de55b2525e52e29c1b0ba"
 "checksum atty 0.2.13 (registry+https://github.com/rust-lang/crates.io-index)" = "1803c647a3ec87095e7ae7acfca019e98de5ec9a7d01343f611cf3152ed71a90"
 "checksum autocfg 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)" = "b671c8fb71b457dd4ae18c4ba1e59aa81793daacc361d82fcd410cef0d491875"
@@ -800,6 +940,14 @@ dependencies = [
 "checksum crossbeam-epoch 0.7.2 (registry+https://github.com/rust-lang/crates.io-index)" = "fedcd6772e37f3da2a9af9bf12ebe046c0dfe657992377b4df982a2b54cd37a9"
 "checksum crossbeam-queue 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)" = "7c979cd6cfe72335896575c6b5688da489e420d36a27a0b9eb0c73db574b4a4b"
 "checksum crossbeam-utils 0.6.6 (registry+https://github.com/rust-lang/crates.io-index)" = "04973fa96e96579258a5091af6003abde64af786b860f18622b82e026cca60e6"
+"checksum crossterm 0.10.2 (registry+https://github.com/rust-lang/crates.io-index)" = "9abce7d7c50e9823ea0c0dbeb8f16d7e247af06d75b4c6244ea0a0998b3a6f35"
+"checksum crossterm_cursor 0.2.6 (registry+https://github.com/rust-lang/crates.io-index)" = "fb4bfd085f17d83e6cd2943f0150d3b4331e465de8dba1750d1966192faf63dc"
+"checksum crossterm_input 0.3.9 (registry+https://github.com/rust-lang/crates.io-index)" = "c6dd255ca05a596bae31ec392fdb67a829509bb767213f00f37c6b62814db663"
+"checksum crossterm_screen 0.2.5 (registry+https://github.com/rust-lang/crates.io-index)" = "0bf294484fc34c22d514c41afc0b97ce74e10ea54d6eb5fe4806d1e1ac0f7b76"
+"checksum crossterm_style 0.4.1 (registry+https://github.com/rust-lang/crates.io-index)" = "8b950f8262e29a446a8a976e0290b67a9067ddc9620f9fb37961d2377f0d8c09"
+"checksum crossterm_terminal 0.2.6 (registry+https://github.com/rust-lang/crates.io-index)" = "db8546b519e0c26aa1f43a4a4ea45ccb41eaca74b9a753ea1788f9ad90212636"
+"checksum crossterm_utils 0.2.4 (registry+https://github.com/rust-lang/crates.io-index)" = "f874a71b2040c730669ddff805c9bc2a1a2f6de9d7f6aab2ae8d29ccbf8a0617"
+"checksum crossterm_winapi 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)" = "b055e7cc627c452e6a9b977022f48a2db6f0ff73df446ca970f95eef9c381d45"
 "checksum doc-comment 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "923dea538cea0aa3025e8685b20d6ee21ef99c4f77e954a30febbaac5ec73a97"
 "checksum fnv 1.0.6 (registry+https://github.com/rust-lang/crates.io-index)" = "2fad85553e09a6f881f739c29f0b00b0f01357c743266d478b68951ce23285f3"
 "checksum fuchsia-cprng 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)" = "a06f77d526c1a601b7c4cdd98f54b5eaabffc14d5f2f0296febdc7f357c6d3ba"
@@ -842,6 +990,8 @@ dependencies = [
 "checksum scopeguard 1.0.0 (registry+https://github.com/rust-lang/crates.io-index)" = "b42e15e59b18a828bbf5c58ea01debb36b9b096346de35d941dcb89009f24a0d"
 "checksum semver 0.9.0 (registry+https://github.com/rust-lang/crates.io-index)" = "1d7eb9ef2c18661902cc47e535f9bc51b78acd254da71d375c2f6720d9a40403"
 "checksum semver-parser 0.7.0 (registry+https://github.com/rust-lang/crates.io-index)" = "388a1df253eca08550bef6c72392cfe7c30914bf41df5269b68cbd6ff8f570a3"
+"checksum signal-hook 0.1.10 (registry+https://github.com/rust-lang/crates.io-index)" = "4f61c4d59f3aaa9f61bba6450a9b80ba48362fd7d651689e7a10c453b1f6dc68"
+"checksum signal-hook-registry 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)" = "1797d48f38f91643908bb14e35e79928f9f4b3cefb2420a564dde0991b4358dc"
 "checksum slab 0.4.2 (registry+https://github.com/rust-lang/crates.io-index)" = "c111b5bd5695e56cffe5129854aa230b39c93a305372fdbb2668ca2394eea9f8"
 "checksum smallvec 0.6.10 (registry+https://github.com/rust-lang/crates.io-index)" = "ab606a9c5e214920bb66c458cd7be8ef094f813f20fe77a54cc7dbfff220d4b7"
 "checksum snafu 0.4.4 (registry+https://github.com/rust-lang/crates.io-index)" = "b028158eb06caa8345bee10cccfb25fa632beccf0ef5308832b4fd4b78a7db48"
@@ -856,7 +1006,9 @@ dependencies = [
 "checksum tokio-executor 0.1.8 (registry+https://github.com/rust-lang/crates.io-index)" = "0f27ee0e6db01c5f0b2973824547ce7e637b2ed79b891a9677b0de9bd532b6ac"
 "checksum tokio-fs 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)" = "3fe6dc22b08d6993916647d108a1a7d15b9cd29c4f4496c62b92c45b5041b7af"
 "checksum tokio-io 0.1.12 (registry+https://github.com/rust-lang/crates.io-index)" = "5090db468dad16e1a7a54c8c67280c5e4b544f3d3e018f0b913b400261f85926"
+"checksum tokio-pty-process 0.4.0 (registry+https://github.com/rust-lang/crates.io-index)" = "b9e3193b62e8c2277534e195d8f8ec4cb43d28a92f89494dd755686026795175"
 "checksum tokio-reactor 0.1.9 (registry+https://github.com/rust-lang/crates.io-index)" = "6af16bfac7e112bea8b0442542161bfc41cbfa4466b580bdda7d18cb88b911ce"
+"checksum tokio-signal 0.2.7 (registry+https://github.com/rust-lang/crates.io-index)" = "dd6dc5276ea05ce379a16de90083ec80836440d5ef8a6a39545a3207373b8296"
 "checksum tokio-sync 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)" = "2162248ff317e2bc713b261f242b69dbb838b85248ed20bb21df56d60ea4cae7"
 "checksum tokio-tcp 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)" = "1d14b10654be682ac43efee27401d792507e30fd8d26389e1da3b185de2e4119"
 "checksum tokio-threadpool 0.1.15 (registry+https://github.com/rust-lang/crates.io-index)" = "90ca01319dea1e376a001e8dc192d42ebde6dd532532a5bad988ac37db365b19"
diff --git a/Cargo.toml b/Cargo.toml
index d6777eb..987df46 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -6,8 +6,10 @@ edition = "2018"
 
 [dependencies]
 clap = "2"
+crossterm = "0.10"
 futures = "0.1"
 mio = "0.6"
 snafu = { version = "0.4", features = ["futures-01"] }
 tokio = "0.1"
+tokio-pty-process = "0.4"
 uuid = { version = "0.7", features = ["v4"] }
diff --git a/src/cmd/cast.rs b/src/cmd/cast.rs
index c4ed386..9260d8c 100644
--- a/src/cmd/cast.rs
+++ b/src/cmd/cast.rs
@@ -1,16 +1,25 @@
+use futures::future::Future as _;
+use futures::stream::Stream as _;
 use snafu::ResultExt as _;
+use std::io::Write as _;
 
 #[derive(Debug, snafu::Snafu)]
 pub enum Error {
     #[snafu(display("failed to connect: {}", source))]
     Connect { source: std::io::Error },
 
+    #[snafu(display("failed to run process: {}", source))]
+    Spawn { source: crate::process::Error },
+
     #[snafu(display("failed to read message: {}", source))]
     Read { source: crate::protocol::Error },
 
     #[snafu(display("failed to write message: {}", source))]
     Write { source: crate::protocol::Error },
 
+    #[snafu(display("failed to write to stdout: {}", source))]
+    Print { source: std::io::Error },
+
     #[snafu(display("failed to read message: unexpected message received"))]
     UnexpectedMessage,
 }
@@ -28,21 +37,31 @@ pub fn run<'a>(_matches: &clap::ArgMatches<'a>) -> super::Result<()> {
 fn run_impl() -> Result<()> {
     let sock =
         std::net::TcpStream::connect("127.0.0.1:8000").context(Connect)?;
-    let msg = crate::protocol::Message::start_casting("doy");
-    msg.write(&sock).context(Write)?;
-    loop {
-        std::thread::sleep(std::time::Duration::from_secs(5));
-        crate::protocol::Message::heartbeat()
-            .write(&sock)
-            .context(Write)?;
-        let res = crate::protocol::Message::read(&sock).context(Read)?;
-        match res {
-            crate::protocol::Message::Heartbeat => {
-                println!("received heartbeat response");
-            }
-            _ => {
-                return Err(Error::UnexpectedMessage);
+
+    crate::protocol::Message::start_casting("doy")
+        .write(&sock)
+        .context(Write)?;
+
+    let future = crate::process::Process::new("zsh", &[])
+        .context(Spawn)?
+        .map_err(|e| Error::Spawn { source: e })
+        .for_each(|e| {
+            match e {
+                crate::process::CommandEvent::CommandStart(..) => {}
+                crate::process::CommandEvent::CommandExit(..) => {}
+                crate::process::CommandEvent::Output(output) => {
+                    let stdout = std::io::stdout();
+                    let mut stdout = stdout.lock();
+                    stdout.write(&output).context(Print)?;
+                    stdout.flush().context(Print)?;
+                }
             }
-        }
-    }
+            Ok(())
+        })
+        .map_err(|e| {
+            eprintln!("cast: {}", e);
+        });
+
+    tokio::run(future);
+    Ok(())
 }
diff --git a/src/main.rs b/src/main.rs
index 6a15d83..399438f 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -1,4 +1,5 @@
 mod cmd;
+mod process;
 mod protocol;
 mod util;
 
diff --git a/src/process.rs b/src/process.rs
new file mode 100644
index 0000000..62b469d
--- /dev/null
+++ b/src/process.rs
@@ -0,0 +1,259 @@
+use futures::future::Future as _;
+use snafu::ResultExt as _;
+use std::io::{Read as _, Write as _};
+use tokio::io::AsyncRead as _;
+use tokio_pty_process::CommandExt as _;
+
+#[derive(Debug, snafu::Snafu)]
+pub enum Error {
+    #[snafu(display("failed to open a pty: {}", source))]
+    OpenPty { source: std::io::Error },
+
+    #[snafu(display("failed to spawn process for `{}`: {}", cmd, source))]
+    SpawnProcess { cmd: String, source: std::io::Error },
+
+    #[snafu(display("failed to resize pty: {}", source))]
+    ResizePty { source: std::io::Error },
+
+    #[snafu(display("failed to write to pty: {}", source))]
+    WriteToPty { source: std::io::Error },
+
+    #[snafu(display("failed to read from terminal: {}", source))]
+    ReadFromTerminal { source: std::io::Error },
+
+    #[snafu(display(
+        "failed to clear ready state on pty for reading: {}",
+        source
+    ))]
+    PtyClearReadReady { source: std::io::Error },
+
+    #[snafu(display("failed to poll for process exit: {}", source))]
+    ProcessExitPoll { source: std::io::Error },
+
+    #[snafu(display(
+        "failed to put the terminal into raw mode: {}",
+        source
+    ))]
+    IntoRawMode { source: std::io::Error },
+}
+
+pub type Result<T> = std::result::Result<T, Error>;
+
+pub enum CommandEvent {
+    CommandStart(String, Vec<String>),
+    Output(Vec<u8>),
+    CommandExit(std::process::ExitStatus),
+}
+
+pub struct Process {
+    pty: tokio_pty_process::AsyncPtyMaster,
+    process: tokio_pty_process::Child,
+    // TODO: tokio::io::Stdin is broken
+    // input: tokio::io::Stdin,
+    input: tokio::reactor::PollEvented2<EventedStdin>,
+    cmd: String,
+    args: Vec<String>,
+    buf: Vec<u8>,
+    started: bool,
+    output_done: bool,
+    exit_done: bool,
+    manage_screen: bool,
+    raw_screen: Option<crossterm::RawScreen>,
+}
+
+struct Resizer<'a, T> {
+    rows: u16,
+    cols: u16,
+    pty: &'a T,
+}
+
+impl<'a, T: tokio_pty_process::PtyMaster> futures::future::Future
+    for Resizer<'a, T>
+{
+    type Item = ();
+    type Error = std::io::Error;
+
+    fn poll(&mut self) -> futures::Poll<Self::Item, Self::Error> {
+        self.pty.resize(self.rows, self.cols)
+    }
+}
+
+impl Process {
+    pub fn new(cmd: &str, args: &[String]) -> Result<Self> {
+        let pty =
+            tokio_pty_process::AsyncPtyMaster::open().context(OpenPty)?;
+
+        let process = std::process::Command::new(cmd)
+            .args(args)
+            .spawn_pty_async(&pty)
+            .context(SpawnProcess { cmd })?;
+
+        let (cols, rows) = crossterm::terminal().terminal_size();
+        Resizer {
+            rows,
+            cols,
+            pty: &pty,
+        }
+        .wait()
+        .context(ResizePty)?;
+
+        // TODO: tokio::io::stdin is broken (it's blocking)
+        // let input = tokio::io::stdin();
+        let input = tokio::reactor::PollEvented2::new(EventedStdin);
+
+        Ok(Self {
+            pty,
+            process,
+            input,
+            cmd: cmd.to_string(),
+            args: args.to_vec(),
+            buf: Vec::with_capacity(4096),
+            started: false,
+            output_done: false,
+            exit_done: false,
+            manage_screen: true,
+            raw_screen: None,
+        })
+    }
+
+    #[allow(dead_code)]
+    pub fn set_raw(mut self, raw: bool) -> Self {
+        self.manage_screen = raw;
+        self
+    }
+}
+
+#[must_use = "streams do nothing unless polled"]
+impl futures::stream::Stream for Process {
+    type Item = CommandEvent;
+    type Error = Error;
+
+    fn poll(&mut self) -> futures::Poll<Option<Self::Item>, Self::Error> {
+        if self.manage_screen && self.raw_screen.is_none() {
+            self.raw_screen = Some(
+                crossterm::RawScreen::into_raw_mode().context(IntoRawMode)?,
+            );
+        }
+
+        if !self.started {
+            self.started = true;
+            return Ok(futures::Async::Ready(Some(
+                CommandEvent::CommandStart(
+                    self.cmd.clone(),
+                    self.args.clone(),
+                ),
+            )));
+        }
+
+        let ready = mio::Ready::readable();
+        let input_poll = self.input.poll_read_ready(ready);
+        if let Ok(futures::Async::Ready(_)) = input_poll {
+            let stdin = std::io::stdin();
+            let mut stdin = stdin.lock();
+            let mut buf = vec![0; 4096];
+            // TODO: async
+            let n = stdin.read(&mut buf).context(ReadFromTerminal)?;
+            if n > 0 {
+                let bytes = buf[..n].to_vec();
+
+                // TODO: async
+                self.pty.write_all(&bytes).context(WriteToPty)?;
+            }
+        }
+        // TODO: this could lose pending bytes if there is stuff to read in
+        // the buffer but we don't read it all in the previous read call,
+        // since i think we won't get another notification until new bytes
+        // actually arrive even if there are bytes in the buffer
+        self.input
+            .clear_read_ready(ready)
+            .context(PtyClearReadReady)?;
+
+        if !self.output_done {
+            self.buf.clear();
+            let output_poll = self.pty.read_buf(&mut self.buf);
+            match output_poll {
+                Ok(futures::Async::Ready(n)) => {
+                    let bytes = self.buf[..n].to_vec();
+                    let bytes: Vec<_> = bytes
+                        .iter()
+                        // replace \n with \r\n
+                        .fold(vec![], |mut acc, &c| {
+                            if c == b'\n' {
+                                acc.push(b'\r');
+                                acc.push(b'\n');
+                            } else {
+                                acc.push(c);
+                            }
+                            acc
+                        });
+                    return Ok(futures::Async::Ready(Some(
+                        CommandEvent::Output(bytes),
+                    )));
+                }
+                Ok(futures::Async::NotReady) => {
+                    return Ok(futures::Async::NotReady);
+                }
+                Err(_) => {
+                    // explicitly ignoring errors (for now?) because we
+                    // always read off the end of the pty after the process
+                    // is done
+                    self.output_done = true;
+                }
+            }
+        }
+
+        if !self.exit_done {
+            let exit_poll = self.process.poll().context(ProcessExitPoll);
+            match exit_poll {
+                Ok(futures::Async::Ready(status)) => {
+                    self.exit_done = true;
+                    return Ok(futures::Async::Ready(Some(
+                        CommandEvent::CommandExit(status),
+                    )));
+                }
+                Ok(futures::Async::NotReady) => {
+                    return Ok(futures::Async::NotReady);
+                }
+                Err(e) => {
+                    return Err(e);
+                }
+            }
+        }
+
+        Ok(futures::Async::Ready(None))
+    }
+}
+
+struct EventedStdin;
+
+impl mio::Evented for EventedStdin {
+    fn register(
+        &self,
+        poll: &mio::Poll,
+        token: mio::Token,
+        interest: mio::Ready,
+        opts: mio::PollOpt,
+    ) -> std::io::Result<()> {
+        let fd = 0 as std::os::unix::io::RawFd;
+        let eventedfd = mio::unix::EventedFd(&fd);
+        eventedfd.register(poll, token, interest, opts)
+    }
+
+    fn reregister(
+        &self,
+        poll: &mio::Poll,
+        token: mio::Token,
+        interest: mio::Ready,
+        opts: mio::PollOpt,
+    ) -> std::io::Result<()> {
+        let fd = 0 as std::os::unix::io::RawFd;
+        let eventedfd = mio::unix::EventedFd(&fd);
+        eventedfd.reregister(poll, token, interest, opts)
+    }
+
+    fn deregister(&self, poll: &mio::Poll) -> std::io::Result<()> {
+        let fd = 0 as std::os::unix::io::RawFd;
+        let eventedfd = mio::unix::EventedFd(&fd);
+        eventedfd.deregister(poll)
+    }
+}