Merge branch 'tokio-process'

Cargo.lock

@@ -194,6 +194,7 @@ dependencies = [
  "lazy_static 1.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "mktemp 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "notify 4.0.4 (registry+https://github.com/rust-lang/crates.io-index)",
+ "owning_ref 0.3.3 (registry+https://github.com/rust-lang/crates.io-index)",
  "regex 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)",
  "secstr 0.3.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "serde 1.0.71 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -201,6 +202,8 @@ dependencies = [
  "serde_json 1.0.26 (registry+https://github.com/rust-lang/crates.io-index)",
  "sha-1 0.7.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "structopt 0.2.10 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tokio 0.1.7 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tokio-process 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)",
  "toml 0.4.6 (registry+https://github.com/rust-lang/crates.io-index)",
  "walkdir 2.2.0 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
@@ -483,6 +486,27 @@ dependencies = [
  "winapi 0.2.8 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
+[[package]]
+name = "mio-named-pipes"
+version = "0.1.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "log 0.4.3 (registry+https://github.com/rust-lang/crates.io-index)",
+ "mio 0.6.15 (registry+https://github.com/rust-lang/crates.io-index)",
+ "miow 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "winapi 0.3.5 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
+[[package]]
+name = "mio-uds"
+version = "0.6.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "iovec 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)",
+ "libc 0.2.43 (registry+https://github.com/rust-lang/crates.io-index)",
+ "mio 0.6.15 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
 [[package]]
 name = "miow"
 version = "0.1.5"
@@ -505,6 +529,15 @@ dependencies = [
  "ws2_32-sys 0.2.1 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
+[[package]]
+name = "miow"
+version = "0.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "socket2 0.3.7 (registry+https://github.com/rust-lang/crates.io-index)",
+ "winapi 0.3.5 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
 [[package]]
 name = "mktemp"
 version = "0.3.1"
@@ -562,6 +595,14 @@ dependencies = [
  "libc 0.2.43 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
+[[package]]
+name = "owning_ref"
+version = "0.3.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "stable_deref_trait 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
 [[package]]
 name = "proc-macro2"
 version = "0.4.11"
@@ -734,6 +775,22 @@ name = "slab"
 version = "0.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 
+[[package]]
+name = "socket2"
+version = "0.3.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "cfg-if 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)",
+ "libc 0.2.43 (registry+https://github.com/rust-lang/crates.io-index)",
+ "redox_syscall 0.1.40 (registry+https://github.com/rust-lang/crates.io-index)",
+ "winapi 0.3.5 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
+[[package]]
+name = "stable_deref_trait"
+version = "1.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+
 [[package]]
 name = "string"
 version = "0.1.1"
@@ -876,6 +933,21 @@ dependencies = [
  "log 0.4.3 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
+[[package]]
+name = "tokio-process"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "futures 0.1.23 (registry+https://github.com/rust-lang/crates.io-index)",
+ "libc 0.2.43 (registry+https://github.com/rust-lang/crates.io-index)",
+ "mio 0.6.15 (registry+https://github.com/rust-lang/crates.io-index)",
+ "mio-named-pipes 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tokio-io 0.1.7 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tokio-reactor 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tokio-signal 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)",
+ "winapi 0.3.5 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
 [[package]]
 name = "tokio-reactor"
 version = "0.1.3"
@@ -889,6 +961,21 @@ dependencies = [
  "tokio-io 0.1.7 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
+[[package]]
+name = "tokio-signal"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "futures 0.1.23 (registry+https://github.com/rust-lang/crates.io-index)",
+ "libc 0.2.43 (registry+https://github.com/rust-lang/crates.io-index)",
+ "mio 0.6.15 (registry+https://github.com/rust-lang/crates.io-index)",
+ "mio-uds 0.6.6 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tokio-executor 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tokio-io 0.1.7 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tokio-reactor 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)",
+ "winapi 0.3.5 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
 [[package]]
 name = "tokio-tcp"
 version = "0.1.1"
@@ -1121,14 +1208,18 @@ dependencies = [
 "checksum memoffset 0.2.1 (registry+https://github.com/rust-lang/crates.io-index)" = "0f9dc261e2b62d7a622bf416ea3c5245cdd5d9a7fcc428c0d06804dfce1775b3"
 "checksum mio 0.5.1 (registry+https://github.com/rust-lang/crates.io-index)" = "a637d1ca14eacae06296a008fa7ad955347e34efcb5891cfd8ba05491a37907e"
 "checksum mio 0.6.15 (registry+https://github.com/rust-lang/crates.io-index)" = "4fcfcb32d63961fb6f367bfd5d21e4600b92cd310f71f9dca25acae196eb1560"
+"checksum mio-named-pipes 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)" = "f5e374eff525ce1c5b7687c4cef63943e7686524a387933ad27ca7ec43779cb3"
+"checksum mio-uds 0.6.6 (registry+https://github.com/rust-lang/crates.io-index)" = "84c7b5caa3a118a6e34dbac36504503b1e8dc5835e833306b9d6af0e05929f79"
 "checksum miow 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)" = "3e690c5df6b2f60acd45d56378981e827ff8295562fc8d34f573deb267a59cd1"
 "checksum miow 0.2.1 (registry+https://github.com/rust-lang/crates.io-index)" = "8c1f2f3b1cf331de6896aabf6e9d55dca90356cc9960cca7eaaf408a355ae919"
+"checksum miow 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "9224c91f82b3c47cf53dcf78dfaa20d6888fbcc5d272d5f2fcdf8a697f3c987d"
 "checksum mktemp 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "77001ceb9eed65439f3dc2a2543f9ba1417d912686bf224a7738d0966e6dcd69"
 "checksum net2 0.2.33 (registry+https://github.com/rust-lang/crates.io-index)" = "42550d9fb7b6684a6d404d9fa7250c2eb2646df731d1c06afc06dcee9e1bcf88"
 "checksum nix 0.5.1 (registry+https://github.com/rust-lang/crates.io-index)" = "bfb3ddedaa14746434a02041940495bf11325c22f6d36125d3bdd56090d50a79"
 "checksum nodrop 0.1.12 (registry+https://github.com/rust-lang/crates.io-index)" = "9a2228dca57108069a5262f2ed8bd2e82496d2e074a06d1ccc7ce1687b6ae0a2"
 "checksum notify 4.0.4 (registry+https://github.com/rust-lang/crates.io-index)" = "d023ef40ca7680784b07be3f49913e1ea176da1b63949f2eb2fed96438bd7f42"
 "checksum num_cpus 1.8.0 (registry+https://github.com/rust-lang/crates.io-index)" = "c51a3322e4bca9d212ad9a158a02abc6934d005490c054a2778df73a70aa0a30"
+"checksum owning_ref 0.3.3 (registry+https://github.com/rust-lang/crates.io-index)" = "cdf84f41639e037b484f93433aa3897863b561ed65c6e59c7073d7c561710f37"
 "checksum proc-macro2 0.4.11 (registry+https://github.com/rust-lang/crates.io-index)" = "762eea716b821300a86da08870a64b597304866ceb9f54a11d67b4cf56459c6a"
 "checksum quote 0.6.6 (registry+https://github.com/rust-lang/crates.io-index)" = "ed7d650913520df631972f21e104a4fa2f9c82a14afc65d17b388a2e29731e7c"
 "checksum rand 0.3.22 (registry+https://github.com/rust-lang/crates.io-index)" = "15a732abf9d20f0ad8eeb6f909bf6868722d9a06e1e50802b6a70351f40b4eb1"
@@ -1152,6 +1243,8 @@ dependencies = [
 "checksum sha-1 0.7.0 (registry+https://github.com/rust-lang/crates.io-index)" = "51b9d1f3b5de8a167ab06834a7c883bd197f2191e1dda1a22d9ccfeedbf9aded"
 "checksum slab 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)" = "d807fd58c4181bbabed77cb3b891ba9748241a552bcc5be698faaebefc54f46e"
 "checksum slab 0.4.1 (registry+https://github.com/rust-lang/crates.io-index)" = "5f9776d6b986f77b35c6cf846c11ad986ff128fe0b2b63a3628e3755e8d3102d"
+"checksum socket2 0.3.7 (registry+https://github.com/rust-lang/crates.io-index)" = "962a516af4d3a7c272cb3a1d50a8cc4e5b41802e4ad54cfb7bee8ba61d37d703"
+"checksum stable_deref_trait 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)" = "dba1a27d3efae4351c8051072d619e3ade2820635c3958d826bfea39d59b54c8"
 "checksum string 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)" = "00caf261d6f90f588f8450b8e1230fa0d5be49ee6140fdfbcb55335aff350970"
 "checksum strsim 0.7.0 (registry+https://github.com/rust-lang/crates.io-index)" = "bb4f380125926a99e52bc279241539c018323fab05ad6368b56f93d9369ff550"
 "checksum structopt 0.2.10 (registry+https://github.com/rust-lang/crates.io-index)" = "d8e9ad6a11096cbecdcca0cc6aa403fdfdbaeda2fb3323a39c98e6a166a1e45a"
@@ -1167,7 +1260,9 @@ dependencies = [
 "checksum tokio-executor 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)" = "424f0c87ecd66b863045d84e384cb7ce0ae384d8b065b9f0363d29c0d1b30b2f"
 "checksum tokio-fs 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)" = "b5cbe4ca6e71cb0b62a66e4e6f53a8c06a6eefe46cc5f665ad6f274c9906f135"
 "checksum tokio-io 0.1.7 (registry+https://github.com/rust-lang/crates.io-index)" = "a5c9635ee806f26d302b8baa1e145689a280d8f5aa8d0552e7344808da54cc21"
+"checksum tokio-process 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)" = "0832648d1ff7ca42c06ca45dc76797b92c56500de828e33c77276fa1449947b6"
 "checksum tokio-reactor 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)" = "8703a5762ff6913510dc64272c714c4389ffd8c4b3cf602879b8bd14ff06b604"
+"checksum tokio-signal 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)" = "52ba234f769af7a85b0cf579da820a82675940054b99819beba2a01a356f5f8e"
 "checksum tokio-tcp 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)" = "5b4c329b47f071eb8a746040465fa751bd95e4716e98daef6a9b4e434c17d565"
 "checksum tokio-threadpool 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)" = "24ab84f574027b0e875378f31575cf175360891919e93a3490f07e76e00e4efb"
 "checksum tokio-timer 0.2.5 (registry+https://github.com/rust-lang/crates.io-index)" = "1c76b4e97a4f61030edff8bd272364e4f731b9f54c7307eb4eb733c3926eb96a"

Cargo.toml

@@ -19,11 +19,14 @@ hyper = "0.12"
 mktemp = "0.3"
 lazy_static = "1.1"
 notify = "4.0"
+owning_ref = "0.3"
 regex = "1.0"
 serde = "1.0"
 serde_derive = "1.0"
 serde_json = "1.0"
 structopt = "0.2"
+tokio = "0.1"
+tokio-process = "0.2"
 toml = "0.4"
 walkdir = "2.2"
 

examples/github.rs

@@ -17,7 +17,7 @@ use std::iter::FromIterator;
 use std::path::PathBuf;
 use std::process::Command;
 
-use failure::{err_msg, Error};
+use failure::err_msg;
 use generic_array::GenericArray;
 use hmac::{Hmac, Mac};
 use secstr::*;
@@ -60,20 +60,23 @@ fn default_path() -> PathBuf {
     PathBuf::from(".")
 }
 
-fn main() -> Result<(), Error> {
+fn main() {
     let args = Opt::from_args();
-    let config: Config = serde_json::from_str(&args.config)?;
+    let config: Config = serde_json::from_str(&args.config).expect("Could not parse config.");
-    println!("{:?}", config);
 
     let mut payload = String::new();
-    io::stdin().read_to_string(&mut payload)?;
+    io::stdin()
-    let payload: Payload = serde_json::from_str(&payload)?;
+        .read_to_string(&mut payload)
+        .expect("Could not read from stdin");
+    let payload: Payload = serde_json::from_str(&payload)
+        .expect(&format!("Could not parse stdin into json: '{}'", payload));
 
     if !config.disable_hmac_verify {
         let secret = GenericArray::from_iter(config.secret.bytes());
         let mut mac = Hmac::<Sha1>::new(&secret);
         mac.input(payload.body.as_bytes());
-        let signature = mac.result()
+        let signature = mac
+            .result()
             .code()
             .into_iter()
             .map(|b| format!("{:02x}", b))
@@ -83,15 +86,23 @@ fn main() -> Result<(), Error> {
         let auth = payload
             .headers
             .get("x-hub-signature")
-            .ok_or(err_msg("Missing auth header"))?;
+            .ok_or(err_msg("Missing auth header"))
+            .expect("Missing auth header");
 
         let left = SecStr::from(format!("sha1={}", signature));
         let right = SecStr::from(auth.bytes().collect::<Vec<_>>());
-        assert!(left == right, "HMAC signature didn't match",);
+        assert!(
+            left == right,
+            "HMAC signature didn't match: {} vs. {}",
+            signature,
+            auth
+        );
     }
 
-    let payload: GithubPayload = serde_json::from_str(&payload.body)?;
+    let payload: GithubPayload =
-    let mut target_path = PathBuf::from(env::var("DIP_WORKDIR")?);
+        serde_json::from_str(&payload.body).expect("Could not parse Github input into json");
+    let mut target_path =
+        PathBuf::from(env::var("DIP_WORKDIR").expect("Could not determine working directory"));
     target_path.push(&config.path);
     Command::new("git")
         .arg("clone")
@@ -100,6 +111,6 @@ fn main() -> Result<(), Error> {
         .arg("--depth")
         .arg("1")
         .arg(&target_path)
-        .output()?;
+        .output()
-    Ok(())
+        .expect("Could not spawn process to clone");
 }

src/config.rs

@@ -65,7 +65,8 @@ where
                 if !path.is_file() {
                     continue;
                 }
-                match path.file_name()
+                match path
+                    .file_name()
                     .and_then(|s| s.to_str())
                     .ok_or(err_msg("???"))
                     .map(|s| {

src/handler.rs

@@ -1,19 +1,26 @@
-use std::io::Write;
 use std::path::PathBuf;
 use std::process::{Command, Stdio};
 
 use failure::{err_msg, Error};
+use futures::{
+    future::{self, Either},
+    Future,
+};
 use serde::Serialize;
 use serde_json::{Serializer as JsonSerializer, Value as JsonValue};
+use tokio::io::write_all;
+use tokio_process::CommandExt;
 use toml::Value as TomlValue;
 
 use PROGRAMS;
 
+#[derive(Clone, Debug)]
 pub struct Handler {
-    config: TomlValue,
+    pub config: TomlValue,
-    action: Action,
+    pub action: Action,
 }
 
+#[derive(Clone, Debug)]
 pub enum Action {
     Command(String),
     Exec(PathBuf),
@@ -47,8 +54,7 @@ impl Handler {
                         value
                             .canonicalize()
                             .map_err(|_| err_msg("failed to canonicalize the path"))
-                    })
+                    }).map(|value| value.clone())?;
-                    .map(|value| value.clone())?;
                 Action::Exec(program)
             }
         };
@@ -56,81 +62,89 @@ impl Handler {
         Ok(Handler { config, action })
     }
 
-    pub fn run(&self, temp_path: &PathBuf, input: JsonValue) -> Result<JsonValue, Error> {
+    pub fn run(
+        config: TomlValue,
+        action: Action,
+        temp_path: PathBuf,
+        input: JsonValue,
+    ) -> impl Future<Item = (PathBuf, JsonValue), Error = Error> {
         let config = {
             let mut buf: Vec<u8> = Vec::new();
             {
                 let mut serializer = JsonSerializer::new(&mut buf);
-                TomlValue::serialize(&self.config, &mut serializer)?;
+                TomlValue::serialize(&config, &mut serializer).unwrap();
             }
             String::from_utf8(buf).unwrap()
         };
 
-        let output = match &self.action {
+        let output: Box<Future<Item = JsonValue, Error = Error> + Send> = match action {
             Action::Command(ref cmd) => {
                 // TODO: allow some kind of simple variable replacement
-                let child = Command::new("/bin/bash")
+                let mut child = Command::new("/bin/bash");
+                let child = child
                     .current_dir(&temp_path)
                     .env("DIP_ROOT", "lol")
-                    .env("DIP_WORKDIR", temp_path)
+                    .env("DIP_WORKDIR", &temp_path)
                     .arg("-c")
                     .arg(cmd)
                     .stdin(Stdio::piped())
                     .stdout(Stdio::piped())
-                    .stderr(Stdio::piped())
+                    .stderr(Stdio::piped());
-                    .spawn()?;
+                let result = child
-                let output = child.wait_with_output()?;
+                    .output_async()
-                if !output.status.success() {
+                    .map_err(|err| err_msg(format!("failed to spawn child: {}", err)))
-                    // TODO: get rid of unwraps
+                    .and_then(|output| {
-                    return Err(err_msg(format!(
+                        let stdout =
-                        "Command '{}' returned with a non-zero status code: {}\nstdout:\n{}\nstderr:\n{}",
+                            String::from_utf8(output.stdout).unwrap_or_else(|_| String::new());
-                        cmd,
+                        let stderr =
-                        output.status,
+                            String::from_utf8(output.stderr).unwrap_or_else(|_| String::new());
-                        String::from_utf8(output.stdout).unwrap_or_else(|_| String::new()),
+                        future::ok(json!({
-                        String::from_utf8(output.stderr).unwrap_or_else(|_| String::new())
+                            "stdout": stdout,
-                    )));
+                            "stderr": stderr,
-                }
+                        }))
-                output
+                    });
+                Box::new(result)
             }
             Action::Exec(ref path) => {
                 let mut child = Command::new(&path)
                     .current_dir(&temp_path)
                     .env("DIP_ROOT", "")
-                    .env("DIP_WORKDIR", temp_path)
+                    .env("DIP_WORKDIR", &temp_path)
                     .arg("--config")
                     .arg(config)
                     .stdin(Stdio::piped())
                     .stdout(Stdio::piped())
                     .stderr(Stdio::piped())
-                    .spawn()?;
+                    .spawn_async()
-                {
+                    .expect("could not spawn child");
-                    match child.stdin {
-                        Some(ref mut stdin) => {
-                            write!(stdin, "{}", input)?;
-                        }
-                        None => bail!("done fucked"),
-                    };
-                }
-                let output = child.wait_with_output()?;
-                if !output.status.success() {
-                    // TODO: get rid of unwraps
-                    return Err(err_msg(format!(
-                        "'{:?}' returned with a non-zero status code: {}\nstdout:\n{}\nstderr:\n{}",
-                        path,
-                        output.status,
-                        String::from_utf8(output.stdout).unwrap_or_else(|_| String::new()),
-                        String::from_utf8(output.stderr).unwrap_or_else(|_| String::new())
-                    )));
-                }
-                output
-            }
-        };
 
-        let stdout = String::from_utf8(output.stdout).unwrap_or_else(|_| String::new());
+                let stdin = child.stdin().take().unwrap();
-        let stderr = String::from_utf8(output.stderr).unwrap_or_else(|_| String::new());
+
-        Ok(json!({
+                let input = format!("{}", input);
+                let result = write_all(stdin, input)
+                    .and_then(|_| child.wait_with_output())
+                    .map_err(|err| err_msg(format!("error: {}", err)))
+                    .and_then(|output| {
+                        let stdout =
+                            String::from_utf8(output.stdout).unwrap_or_else(|_| String::new());
+                        let stderr =
+                            String::from_utf8(output.stderr).unwrap_or_else(|_| String::new());
+                        if output.status.success() {
+                            Either::A(future::ok(json!({
                                 "stdout": stdout,
                                 "stderr": stderr,
-            }))
+                            })))
+                        } else {
+                            Either::B(future::err(err_msg(format!(
+                                "Failed, stdout: '{}', stderr: '{}'",
+                                stdout, stderr
+                            ))))
+                        }
+                    });
+
+                Box::new(result)
+            }
+        };
+        output.map(|x| (temp_path, x))
     }
 }

src/hook.rs

@@ -1,11 +1,11 @@
 use std::fs::File;
 use std::io::Read;
 use std::path::{Path, PathBuf};
-use std::slice::Iter;
 
 use failure::{err_msg, Error};
-use hyper::StatusCode;
+use futures::{stream, Future};
 use serde_json::Value as JsonValue;
+use tokio::{self, prelude::*};
 use toml::Value;
 
 use Handler;
@@ -32,7 +32,8 @@ impl Hook {
     where
         P: AsRef<Path>,
     {
-        let filename = path.as_ref()
+        let filename = path
+            .as_ref()
             .file_name()
             .ok_or(err_msg("what the fuck bro"))?
             .to_str()
@@ -49,33 +50,20 @@ impl Hook {
     pub fn get_name(&self) -> String {
         self.name.clone()
     }
-    pub fn iter(&self) -> Iter<Handler> {
+    pub fn handle(&self, req: JsonValue, temp_path: PathBuf) -> Result<String, String> {
-        self.handlers.iter()
+        let handlers = self
-    }
+            .handlers
-    pub fn handle(
+            .iter()
-        &self,
+            .map(|handler| (handler.config.clone(), handler.action.clone()))
-        req: JsonValue,
+            .collect::<Vec<_>>();
-        temp_path: &PathBuf,
+        let st = stream::iter_ok::<_, Error>(handlers.into_iter())
-    ) -> Result<(StatusCode, String), Error> {
+            .fold((temp_path, req), |(path, prev), (config, action)| {
-        Ok(self.iter()
+                Handler::run(config, action, path, prev)
-            .fold(Ok(req), |prev, handler| {
+            }).map(|_| ())
-                prev.and_then(|val| {
+            .map_err(|err: Error| {
-                    println!("Running {}...", handler.config());
+                println!("Error from stream: {}", err);
-                    let result = handler.run(&temp_path, val);
+            });
-                    println!("{:?}", result);
+        tokio::executor::spawn(st);
-                    result
+        Ok("success".to_owned())
-                })
-            })
-            .map(|res| {
-                (
-                    StatusCode::ACCEPTED,
-                    format!(
-                        "stdout:\n{}\n\nstderr:\n{}",
-                        res.get("stdout").and_then(|v| v.as_str()).unwrap_or(""),
-                        res.get("stderr").and_then(|v| v.as_str()).unwrap_or(""),
-                    ),
-                )
-            })
-            .unwrap_or_else(|err| (StatusCode::BAD_REQUEST, format!("Error: {:?}", err))))
     }
 }

src/lib.rs

@@ -1,11 +1,13 @@
 //! # Dip
 
-#[macro_use]
 extern crate failure;
 extern crate futures;
 extern crate hyper;
 extern crate mktemp;
+extern crate owning_ref;
 extern crate serde;
+extern crate tokio;
+extern crate tokio_process;
 #[macro_use]
 extern crate serde_json;
 #[macro_use]
@@ -46,13 +48,12 @@ const URIPATTERN_STR: &str = r"/webhook/(?P<name>[A-Za-z._][A-Za-z0-9._]*)";
 
 lazy_static! {
     static ref URIPATTERN: Regex = Regex::new(URIPATTERN_STR).unwrap();
-    // static ref HANDLERS: Mutex<HashMap<String, Box<Handler>>> = Mutex::new(HashMap::new());
+    static ref HANDLERS: Arc<Mutex<HashMap<String, Handler>>> =
+        Arc::new(Mutex::new(HashMap::new()));
     static ref PROGRAMS: Mutex<HashMap<String, PathBuf>> = Mutex::new(HashMap::new());
     static ref HOOKS: Arc<Mutex<HashMap<String, Hook>>> = Arc::new(Mutex::new(HashMap::new()));
 }
 
-// const NOTFOUND: &str = "<html> <head> <style> * { font-family: sans-serif; } body { padding: 20px 60px; } </style> </head> <body> <h1>Looks like you took a wrong turn!</h1> <p>There's nothing to see here.</p> </body> </html>";
-
 fn watch<P>(root: P) -> notify::Result<()>
 where
     P: AsRef<Path>,
@@ -68,7 +69,7 @@ where
                 // TODO: don't do this
                 config::load_config(root.as_ref())
             }
-            Err(e) => println!("watch error: {:?}", e),
+            Err(e) => eprintln!("watch error: {:?}", e),
         }
     }
 }

src/service.rs

@@ -1,5 +1,4 @@
 use std::collections::HashMap;
-use std::thread;
 
 use futures::{future, Future, Stream};
 use hyper::{Body, Error, Request, Response, StatusCode};
@@ -34,19 +33,18 @@ pub fn dip_service(req: Request<Body>) -> Box<Future<Item = Response<Body>, Erro
 
     // TODO: filter by method as well
 
-    let headers = req.headers()
+    let headers = req
+        .headers()
         .clone()
         .into_iter()
         .filter_map(|(k, v)| {
             let key = k.unwrap().as_str().to_owned();
             v.to_str().map(|value| (key, value.to_owned())).ok()
-        })
+        }).collect::<HashMap<_, _>>();
-        .collect::<HashMap<_, _>>();
     let method = req.method().as_str().to_owned();
 
     // spawn job
-    thread::spawn(move || {
+    Box::new(req.into_body().concat2().map(move |body| {
-        req.into_body().concat2().map(move |body| {
         let body = String::from_utf8(body.to_vec()).unwrap();
         let req_obj = json!({
             "body": body, 
@@ -60,7 +58,10 @@ pub fn dip_service(req: Request<Body>) -> Box<Future<Item = Response<Body>, Erro
             assert!(temp_path.exists());
 
             let hook = hooks.get(&name).unwrap();
-                let (code, msg) = hook.handle(req_obj, &temp_path).unwrap();
+            let (code, msg) = match hook.handle(req_obj, temp_path) {
+                Ok(msg) => (StatusCode::ACCEPTED, msg),
+                Err(msg) => (StatusCode::BAD_REQUEST, msg),
+            };
 
             temp_dir.release();
             Response::builder()
@@ -68,14 +69,5 @@ pub fn dip_service(req: Request<Body>) -> Box<Future<Item = Response<Body>, Erro
                 .body(Body::from(msg))
                 .unwrap_or_else(|err| Response::new(Body::from(format!("{}", err))))
         }
-        })
+    }))
-    });
-
-    Box::new(future::ok(
-        Response::builder()
-            .status(StatusCode::ACCEPTED)
-            // TODO: assign job a uuid and do some logging
-            .body(Body::from(format!("job {} started", -1)))
-            .unwrap(),
-    ))
 }