Cargo.toml

@@ -1,33 +1,28 @@
 [package]
 name = "dip"
 description = "Configurable webhook server."
-version = "0.2.0"
-authors = ["Michael Zhang <mail@mzhang.io>"]
-edition = "2018"
-
-[features]
-github = []
-default = ["github"]
+version = "0.1.4"
+authors = ["Michael Zhang <failed.down@gmail.com>"]
 
 [dependencies]
-anyhow = "1.0.34"
-futures = "0.3.7"
-generic-array = "0.14.4"
-hmac = "0.10.1"
-hyper = { version = "0.13", features = ["stream"] }
-lazy_static = "1.4.0"
-mktemp = "0.4.0"
-notify = "4.0.15"
-owning_ref = "0.4.1"
-parking_lot = "0.11.0"
-regex = "1.4.2"
-secstr = "0.4.0"
-serde = "1.0.117"
-serde_derive = "1.0.117"
-serde_json = "1.0.59"
-sha-1 = "0.9.1"
-structopt = "0.3.20"
-tokio = { version = "0.3", features = ["full"] }
-toml = "0.5.7"
-walkdir = "2.3.1"
+failure = "0.1"
+futures = "0.1"
+generic-array = "0.9"
+hmac = "0.6"
+hyper = "0.12"
+mktemp = "0.3"
+lazy_static = "1.1"
+notify = "4.0"
+owning_ref = "0.3"
+regex = "1.0"
+secstr = "0.3"
+serde = "1.0"
+serde_derive = "1.0"
+serde_json = "1.0"
+sha-1 = "0.7"
+structopt = "0.2"
+tokio = "0.1"
+tokio-process = "0.2"
+toml = "0.4"
+walkdir = "2.2"
 

src/builtins/github.rs

@@ -1,13 +0,0 @@
-use serde_json::Value;
-use anyhow::Result;
-
-use super::Builtin;
-
-#[derive(Default)]
-pub struct Github;
-
-impl Builtin for Github {
-    fn handle(&self, env: (), config: (), input: ()) -> Result<Value> {
-        todo!()
-    }
-}

src/builtins/mod.rs

@@ -1,10 +0,0 @@
-mod github;
-
-use serde_json::Value;
-use anyhow::Result;
-
-pub use self::github::Github;
-
-pub trait Builtin {
-    fn handle(&self, env: (), config: (), input: ()) -> Result<Value>;
-}

src/config.rs

@@ -6,7 +6,7 @@ use std::path::{Path, PathBuf};
 use std::sync::mpsc;
 use std::time::Duration;
 
-use anyhow::Result;
+use failure::{err_msg, Error};
 use notify::{self, RecommendedWatcher, RecursiveMode, Watcher};
 use walkdir::WalkDir;
 
@@ -19,12 +19,10 @@ pub struct Config {
     /// The root configuration directory for dip. This argument is required.
     #[structopt(short = "d", long = "root", parse(from_os_str))]
     pub root: PathBuf,
-
     /// A string containing the address to bind to. This defaults to
     /// "0.0.0.0:5000".
     #[structopt(short = "b", long = "bind", default_value = "0.0.0.0:5000")]
     pub bind: String,
-
     /// If a hook is specified here, it will be triggered manually exactly once
     /// and then the program will exit rather than running as a server.
     #[structopt(short = "h", long = "hook")]
@@ -71,7 +69,13 @@ where
     println!("Reloading config...");
     // hold on to the lock while config is being reloaded
     {
-        let mut programs = PROGRAMS.lock();
+        let mut programs = match PROGRAMS.lock() {
+            Ok(programs) => programs,
+            Err(err) => {
+                eprintln!("Could not acquire programs lock: {}", err);
+                return;
+            }
+        };
         // TODO: some kind of smart diff
         programs.clear();
 
@@ -92,7 +96,7 @@ where
                 match path
                     .file_name()
                     .and_then(|s| s.to_str())
-                    .ok_or(anyhow!("???"))
+                    .ok_or(err_msg("???"))
                     .map(|s| {
                         let filename = s.to_owned();
                         programs.insert(filename, path.to_path_buf())
@@ -102,9 +106,14 @@ where
             }
         }
     }
-
     {
-        let mut hooks = HOOKS.lock();
+        let mut hooks = match HOOKS.lock() {
+            Ok(hooks) => hooks,
+            Err(err) => {
+                eprintln!("Could not acquire hooks lock: {}", err);
+                return;
+            }
+        };
         hooks.clear();
         let hooks_dir = {
             let mut p = root.as_ref().to_path_buf();
@@ -120,7 +129,7 @@ where
                 if !path.is_file() {
                     continue;
                 }
-                match (|path: &Path| -> Result<()> {
+                match (|path: &Path| -> Result<(), Error> {
                     let hook = Hook::from_file(path)?;
                     let name = hook.get_name();
                     hooks.insert(name, hook);

src/github.rs

@@ -3,9 +3,9 @@ use std::iter::FromIterator;
 use std::path::PathBuf;
 use std::process::Command;
 
-use anyhow::Result;
+use failure::{err_msg, Error};
 use generic_array::GenericArray;
-use hmac::{Hmac, Mac, NewMac};
+use hmac::{Hmac, Mac};
 use secstr::*;
 use serde::Serialize;
 use serde_json::{self, Serializer as JsonSerializer, Value as JsonValue};
@@ -55,7 +55,7 @@ pub(crate) fn main(
     env: &Environment,
     config: &TomlValue,
     input: &JsonValue,
-) -> Result<JsonValue> {
+) -> Result<JsonValue, Error> {
     let config_str = {
         let mut buf: Vec<u8> = Vec::new();
         {
@@ -72,10 +72,10 @@ pub(crate) fn main(
     if !config.disable_hmac_verify {
         let secret = GenericArray::from_iter(config.secret.bytes());
         let mut mac = Hmac::<Sha1>::new(&secret);
-        mac.update(payload.body.as_bytes());
+        mac.input(payload.body.as_bytes());
         let signature = mac
-            .finalize()
-            .into_bytes()
+            .result()
+            .code()
             .into_iter()
             .map(|b| format!("{:02x}", b))
             .collect::<Vec<_>>()
@@ -84,7 +84,7 @@ pub(crate) fn main(
         let auth = payload
             .headers
             .get("x-hub-signature")
-            .ok_or(anyhow!("Missing auth header"))
+            .ok_or(err_msg("Missing auth header"))
             .expect("Missing auth header");
 
         let left = SecStr::from(format!("sha1={}", signature));

src/handler.rs

@@ -1,31 +1,36 @@
 use std::fmt;
 use std::path::PathBuf;
-use std::sync::Arc;
-use std::process::Stdio;
+use std::process::{Command, Stdio};
 
-use anyhow::Result;
+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::AsyncWriteExt, process::Command};
+use tokio::io::write_all;
+use tokio_process::CommandExt;
 use toml::Value as TomlValue;
 
-use crate::builtins::{Builtin, Github};
+use crate::github;
 
 /// A single instance of handler as defined by the config.
+#[derive(Clone, Debug)]
 pub struct Handler {
     pub(crate) config: TomlValue,
     pub(crate) action: Action,
 }
 
 /// Describes an action that a hook can take.
+#[derive(Clone)]
 pub enum Action {
     /// A builtin function (for example, the Github handler).
-    Builtin(Box<dyn Builtin + Send>),
-    // Fn(&Environment, &TomlValue, &JsonValue) -> Result<JsonValue>>),
-
+    Builtin(
+        fn(&Environment, &TomlValue, &JsonValue) -> Result<JsonValue, Error>,
+    ),
     /// A command represents a string to be executed by `bash -c`.
     Command(String),
-
     /// A program represents one of the handlers specified in the `handlers`
     /// directory.
     Program(String),
@@ -48,23 +53,22 @@ impl fmt::Debug for Action {
 }
 
 impl Handler {
-    pub(crate) fn from(config: &TomlValue) -> Result<Self> {
+    pub(crate) fn from(config: &TomlValue) -> Result<Self, Error> {
         let handler = config
             .get("type")
-            .ok_or(anyhow!("No 'type' found."))?
+            .ok_or(err_msg("No 'type' found."))?
             .as_str()
-            .ok_or(anyhow!("'type' is not a string."))?;
-
+            .ok_or(err_msg("'type' is not a string."))?;
         let action = match handler {
             "command" => {
                 let command = config
                     .get("command")
-                    .ok_or(anyhow!("No 'command' found"))?
+                    .ok_or(err_msg("No 'command' found"))?
                     .as_str()
-                    .ok_or(anyhow!("'command' is not a string."))?;
+                    .ok_or(err_msg("'command' is not a string."))?;
                 Action::Command(command.to_owned())
             }
-            "github" => Action::Builtin(Box::new(Github::default())),
+            "github" => Action::Builtin(github::main),
             handler => Action::Program(handler.to_owned()),
         };
         let config = config.clone();
@@ -72,12 +76,12 @@ impl Handler {
     }
 
     /// Runs the given [action](Action) and produces a [Future](Future).
-    pub async fn run(
+    pub fn run(
         config: TomlValue,
         action: Action,
         temp_path: PathBuf,
         input: JsonValue,
-    ) -> Result<(PathBuf, JsonValue)> {
+    ) -> impl Future<Item = (PathBuf, JsonValue), Error = Error> {
         let temp_path_cp = temp_path.clone();
         let config_str = {
             let mut buf: Vec<u8> = Vec::new();
@@ -98,57 +102,72 @@ impl Handler {
                 .stderr(Stdio::piped());
         };
 
-        let output = match action {
-            Action::Builtin(ref builtin) => {
-                let workdir = temp_path_cp.clone();
-                let env = Environment { workdir };
-                builtin.handle(&env, &config, &input)
-            }
-            Action::Command(ref cmd) => {
-                // TODO: allow some kind of simple variable replacement
-                let mut command = Command::new("/bin/bash");
-                command_helper(&mut command);
-                let child = command.arg("-c").arg(cmd);
-                let output = child.output().await?;
-
-                let stdout = String::from_utf8(output.stdout)?;
-                let stderr = String::from_utf8(output.stderr)?;
-
-                Ok(json!({
-                    "stdout": stdout,
-                    "stderr": stderr,
-                }))
-            }
-            Action::Program(ref path) => {
-                let mut command = Command::new(&path);
-                command_helper(&mut command);
-                let mut child =
-                    command.arg("--config").arg(config_str).spawn()?;
-                let mut stdin = child.stdin.take().unwrap();
-
-                let input = format!("{}", input);
-                stdin.write_all(input.as_bytes()).await?;
-
-                let output = child.wait_with_output().await?;
-
-                let stdout = String::from_utf8(output.stdout)?;
-                let stderr = String::from_utf8(output.stderr)?;
-
-                if output.status.success() {
-                    Ok(json!({
-                        "stdout": stdout,
-                        "stderr": stderr,
-                    }))
-                } else {
-                    Err(anyhow!(
-                        "Failed, stdout: '{}', stderr: '{}'",
-                        stdout,
-                        stderr
-                    ))
+        let output: Box<dyn Future<Item = JsonValue, Error = Error> + Send> =
+            match action {
+                Action::Builtin(ref func) => {
+                    let workdir = temp_path_cp.clone();
+                    let env = Environment { workdir };
+                    let result = func(&env, &config, &input);
+                    Box::new(future::result(result))
                 }
-            }
-        };
+                Action::Command(ref cmd) => {
+                    // TODO: allow some kind of simple variable replacement
+                    let mut command = Command::new("/bin/bash");
+                    command_helper(&mut command);
+                    let child = command.arg("-c").arg(cmd);
+                    let result = child
+                        .output_async()
+                        .map_err(|err| {
+                            err_msg(format!("failed to spawn child: {}", 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());
+                            future::ok(json!({
+                                "stdout": stdout,
+                                "stderr": stderr,
+                            }))
+                        });
+                    Box::new(result)
+                }
+                Action::Program(ref path) => {
+                    let mut command = Command::new(&path);
+                    command_helper(&mut command);
+                    let mut child = command
+                        .arg("--config")
+                        .arg(config_str)
+                        .spawn_async()
+                        .expect("could not spawn child");
 
+                    let stdin = child.stdin().take().unwrap();
+
+                    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_cp, x))
     }
 }

src/hook.rs

@@ -4,8 +4,10 @@ use std::fs::File;
 use std::io::Read;
 use std::path::{Path, PathBuf};
 
-use anyhow::Result;
+use failure::{err_msg, Error};
+use futures::{stream, Future};
 use serde_json::Value as JsonValue;
+use tokio::{self, prelude::*};
 use toml::Value;
 
 use crate::Handler;
@@ -18,13 +20,16 @@ pub struct Hook {
 
 impl Hook {
     /// Creates a hook from a (name, config) pair.
-    pub fn from(name: impl Into<String>, config: &Value) -> Result<Self> {
+    pub fn from(
+        name: impl Into<String>,
+        config: &Value,
+    ) -> Result<Self, Error> {
         let name = name.into();
         let handlers = config
             .get("handlers")
-            .ok_or(anyhow!("No 'handlers' found."))?
+            .ok_or(err_msg("No 'handlers' found."))?
             .as_array()
-            .ok_or(anyhow!("'handlers' is not an array."))?
+            .ok_or(err_msg("'handlers' is not an array."))?
             .iter()
             .map(|value: &Value| Handler::from(value))
             .collect::<Result<Vec<_>, _>>()?;
@@ -32,16 +37,16 @@ impl Hook {
     }
 
     /// Creates a hook from a configuration file.
-    pub fn from_file<P>(path: P) -> Result<Hook>
+    pub fn from_file<P>(path: P) -> Result<Hook, Error>
     where
         P: AsRef<Path>,
     {
         let filename = path
             .as_ref()
             .file_name()
-            .ok_or(anyhow!("what the fuck bro"))?
+            .ok_or(err_msg("what the fuck bro"))?
             .to_str()
-            .ok_or(anyhow!("???"))?
+            .ok_or(err_msg("???"))?
             .to_owned();
         let mut file = File::open(path)?;
         let mut contents = String::new();
@@ -57,25 +62,25 @@ impl Hook {
         self.name.clone()
     }
 
-    pub(crate) async fn handle(
+    pub(crate) fn handle(
         &self,
         req: JsonValue,
         temp_path: PathBuf,
-    ) -> Result<String> {
+    ) -> Result<String, String> {
         let handlers = self
             .handlers
             .iter()
-            .map(|handler| (handler.config, handler.action))
+            .map(|handler| (handler.config.clone(), handler.action.clone()))
             .collect::<Vec<_>>();
-
-        let mut path = temp_path;
-        let mut prev = req;
-        for (config, action) in handlers {
-            let result = Handler::run(config, action, path, prev).await?;
-            path = result.0;
-            prev = result.1;
-        }
-
+        let st = stream::iter_ok::<_, Error>(handlers.into_iter())
+            .fold((temp_path, req), |(path, prev), (config, action)| {
+                Handler::run(config, action, path, prev)
+            })
+            .map(|_| ())
+            .map_err(|err: Error| {
+                println!("Error from stream: {}", err);
+            });
+        tokio::executor::spawn(st);
         Ok("success".to_owned())
     }
 }

src/lib.rs

@@ -6,18 +6,31 @@ The configurable webhook server. Latest stable binary releases for Linux are ava
 
 #![deny(missing_docs)]
 
-#[macro_use]
-extern crate anyhow;
+extern crate hmac;
+extern crate secstr;
+extern crate sha1;
 #[macro_use]
 extern crate serde_derive;
+extern crate failure;
+extern crate futures;
+extern crate generic_array;
+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]
 extern crate lazy_static;
+extern crate notify;
 #[macro_use]
 extern crate structopt;
+extern crate regex;
+extern crate toml;
+extern crate walkdir;
 
-mod builtins;
 pub mod config;
 mod github;
 mod handler;
@@ -28,15 +41,13 @@ use std::collections::HashMap;
 use std::net::SocketAddrV4;
 use std::path::PathBuf;
 use std::str::FromStr;
-use std::sync::Arc;
+use std::sync::{Arc, Mutex};
 use std::thread;
 
-use anyhow::{Error, Result};
-use hyper::{
-    service::{make_service_fn, service_fn},
-    Server,
-};
-use parking_lot::Mutex;
+use failure::Error;
+use hyper::rt::Future;
+use hyper::service::service_fn;
+use hyper::Server;
 use regex::Regex;
 
 pub use crate::config::Config;
@@ -56,23 +67,17 @@ lazy_static! {
 }
 
 /// Main entry point of the entire application.
-pub async fn run(config: &Config) -> Result<()> {
+pub fn run(config: &Config) -> Result<(), Error> {
     config::load_config(&config.root);
 
     let v = config.root.clone();
     thread::spawn(|| config::watch(v));
 
     let addr: SocketAddrV4 = SocketAddrV4::from_str(config.bind.as_ref())?;
-    let make_svc = make_service_fn(|conn| async {
-        Ok::<_, Error>(service_fn(dip_service))
-    });
-
-    let server = Server::bind(&addr.into()).serve(make_svc);
+    let server = Server::bind(&addr.into())
+        .serve(|| service_fn(dip_service))
+        .map_err(|e| eprintln!("server error: {}", e));
     println!("Listening on {:?}", addr);
-
-    if let Err(e) = server.await {
-        eprintln!("server error: {}", e);
-    }
-
+    hyper::rt::run(server);
     Ok(())
 }

src/main.rs

@@ -1,9 +1,12 @@
-use anyhow::Result;
+extern crate dip;
+extern crate failure;
+extern crate structopt;
+
 use dip::Config;
+use failure::Error;
 use structopt::StructOpt;
 
-#[tokio::main]
-async fn main() -> Result<()> {
+fn main() -> Result<(), Error> {
     let config = Config::from_args();
     dip::run(&config)
 }

src/service.rs

@@ -1,34 +1,35 @@
 use std::collections::HashMap;
 
-use anyhow::Result;
-use futures::stream::TryStreamExt;
-use hyper::{
-    body::{to_bytes, Body},
-    Request, Response, StatusCode,
-};
+use futures::{future, Future, Stream};
+use hyper::{Body, Error, Request, Response, StatusCode};
 use mktemp::Temp;
 
 use crate::{HOOKS, URIPATTERN};
 
-pub(crate) async fn dip_service(req: Request<Body>) -> Result<Response<Body>> {
+pub(crate) fn dip_service(
+    req: Request<Body>,
+) -> Box<dyn Future<Item = Response<Body>, Error = Error> + Send> {
     let path = req.uri().path().to_owned();
     let captures = match URIPATTERN.captures(path.as_ref()) {
         Some(value) => value,
         None => {
-            return Ok(Response::builder()
-                .status(StatusCode::NOT_FOUND)
-                .body(Body::from("not found"))
-                .unwrap());
+            return Box::new(future::ok(
+                Response::builder()
+                    .status(StatusCode::NOT_FOUND)
+                    .body(Body::from("not found"))
+                    .unwrap(),
+            ))
         }
     };
-
     let name = match captures.name("name") {
         Some(value) => value.as_str().to_owned(),
         None => {
-            return Ok(Response::builder()
-                .status(StatusCode::NOT_FOUND)
-                .body(Body::from("not found"))
-                .unwrap());
+            return Box::new(future::ok(
+                Response::builder()
+                    .status(StatusCode::NOT_FOUND)
+                    .body(Body::from("not found"))
+                    .unwrap(),
+            ))
         }
     };
 
@@ -46,45 +47,43 @@ pub(crate) async fn dip_service(req: Request<Body>) -> Result<Response<Body>> {
     let method = req.method().as_str().to_owned();
 
     // spawn job
-    let body = to_bytes(req.into_body()).await?;
+    Box::new(req.into_body().concat2().map(move |body| {
+        let body = String::from_utf8(body.to_vec()).unwrap();
+        let req_obj = json!({
+            "body": body,
+            "headers": headers,
+            "method": method,
+        });
+        let hooks = HOOKS.lock().unwrap();
+        {
+            let mut temp_dir = Temp::new_dir().unwrap();
+            let temp_path = temp_dir.to_path_buf();
+            assert!(temp_path.exists());
 
-    let body = String::from_utf8(body.to_vec()).unwrap();
-    let req_obj = json!({
-        "body": body,
-        "headers": headers,
-        "method": method,
-    });
+            let hook = match hooks.get(&name) {
+                Some(hook) => hook,
+                None => {
+                    temp_dir.release();
+                    return Response::builder()
+                        .status(StatusCode::NOT_FOUND)
+                        .body(Body::from("not found"))
+                        .unwrap_or_else(|_| {
+                            Response::new(Body::from("not found"))
+                        });
+                }
+            };
+            let (code, msg) = match hook.handle(req_obj, temp_path) {
+                Ok(msg) => (StatusCode::ACCEPTED, msg),
+                Err(msg) => (StatusCode::BAD_REQUEST, msg),
+            };
 
-    let mut temp_dir = Temp::new_dir().unwrap();
-    let temp_path = temp_dir.to_path_buf();
-    assert!(temp_path.exists());
-
-    let (code, msg) = {
-        let hooks = HOOKS.lock();
-
-        let hook = match hooks.get(&name) {
-            Some(hook) => hook,
-            None => {
-                temp_dir.release();
-                return Ok(Response::builder()
-                    .status(StatusCode::NOT_FOUND)
-                    .body(Body::from("not found"))
-                    .unwrap_or_else(|_| {
-                        Response::new(Body::from("not found"))
-                    }));
-            }
-        };
-
-        match hook.handle(req_obj, temp_path).await {
-            Ok(msg) => (StatusCode::ACCEPTED, msg),
-            Err(msg) => (StatusCode::BAD_REQUEST, msg.to_string()),
+            temp_dir.release();
+            Response::builder()
+                .status(code)
+                .body(Body::from(msg))
+                .unwrap_or_else(|err| {
+                    Response::new(Body::from(format!("{}", err)))
+                })
         }
-    };
-
-    temp_dir.release();
-
-    Ok(Response::builder()
-        .status(code)
-        .body(Body::from(msg))
-        .unwrap_or_else(|err| Response::new(Body::from(format!("{}", err)))))
+    }))
 }