Add "server mode" to Factotum (closes #98) - WIP

executor/.gitignore

@@ -0,0 +1,18 @@
+# Compiled files
+*.o
+*.so
+*.rlib
+*.dll
+
+# Executables
+*.exe
+
+# Generated by Cargo
+/target/
+
+# Remove Cargo.lock from gitignore if creating an executable, leave it for libraries
+# More information here http://doc.crates.io/guide.html#cargotoml-vs-cargolock
+Cargo.lock
+.factotum
+.vagrant
+

executor/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "executor"
+version = "0.1.0"
+authors = ["Nicholas Ung <support@snowplowanalytics.com>"]
+
+[dependencies]
+log = "0.3"
+log4rs = "0.6"
+docopt = "0.7"
+uuid = { version = "0.4", features = ["serde", "v4"] }
+threadpool = "1.3"
+iron = "0.5"
+router = "0.5"
+bodyparser = "0.6"
+persistent = "0.3"
+logger = "0.3"
+rustc-serialize = "0.3"
+serde = "0.9"
+serde_derive = "0.9"
+serde_json = "0.9"

executor/src/executor/commander/mod.rs

@@ -0,0 +1,68 @@
+// Copyright (c) 2016-2017 Snowplow Analytics Ltd. All rights reserved.
+//
+// This program is licensed to you under the Apache License Version 2.0, and
+// you may not use this file except in compliance with the Apache License
+// Version 2.0.  You may obtain a copy of the Apache License Version 2.0 at
+// http://www.apache.org/licenses/LICENSE-2.0.
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the Apache License Version 2.0 is distributed on an "AS
+// IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied.  See the Apache License Version 2.0 for the specific language
+// governing permissions and limitations there under.
+//
+
+use std::collections::HashMap;
+use std::process::Command;
+
+macro_rules! commands {
+    ($( $key: expr => $val: expr ),*) => {{
+         let mut map = ::std::collections::HashMap::new();
+         $( map.insert($key, $val); )*
+         Commander::new(map)
+    }}
+}
+
+#[cfg(test)]
+mod tests;
+
+#[derive(Debug)]
+pub struct Commander {
+    pub command_map: HashMap<String, String>
+}
+
+impl Commander {
+    pub fn new(commands: HashMap<String, String>) -> Commander {
+        Commander {
+            command_map: commands
+        }
+    }
+
+    pub fn get_command(&self, command: &str) -> Result<String, String> {
+        match self.command_map.get(command) {
+            Some(command) => Ok(command.to_owned()),
+            None => Err(format!("Command <{}> not found in map.", command))
+        }
+    }
+}
+
+pub fn execute(cmd_path: String, cmd_args: Vec<String>) -> Result<String, String> {
+    let command_str = format!("{} {}", cmd_path, cmd_args.join(" "));
+    info!("Executing: [{}]", command_str);
+    let failed_command_msg = format!("Failed to execute command: [{}]", command_str);
+    match Command::new(cmd_path)
+                .args(&cmd_args)
+                .output()
+                {
+                    Ok(output) => {
+                        if output.status.success() {
+                            let stdout = String::from_utf8_lossy(&output.stdout);
+                            Ok(stdout.into_owned())
+                        } else {
+                            let stderr = String::from_utf8_lossy(&output.stderr);
+                            Err(format!("{} - {}", failed_command_msg, stderr.into_owned()))
+                        }
+                    }
+                    Err(e) => Err(format!("{} - {}", failed_command_msg, e))
+                }
+}

executor/src/executor/commander/tests.rs

@@ -0,0 +1,46 @@
+// Copyright (c) 2016-2017 Snowplow Analytics Ltd. All rights reserved.
+//
+// This program is licensed to you under the Apache License Version 2.0, and
+// you may not use this file except in compliance with the Apache License
+// Version 2.0.  You may obtain a copy of the Apache License Version 2.0 at
+// http://www.apache.org/licenses/LICENSE-2.0.
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the Apache License Version 2.0 is distributed on an "AS
+// IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied.  See the Apache License Version 2.0 for the specific language
+// governing permissions and limitations there under.
+//
+
+use super::*;
+
+#[test]
+fn create_commander_macro() {
+    let commander = commands!["dummy".to_string() => "/tmp/fake_command".to_string()];
+    assert_eq!(commander.command_map.contains_key("dummy"), true);
+    assert_eq!(commander.command_map.contains_key("other_dummy"), false);
+}
+
+#[test]
+fn commander_get_command_success() {
+    let commander = commands!["dummy".to_string() => "/tmp/fake_command".to_string()];
+    assert_eq!(commander.get_command("dummy"), Ok("/tmp/fake_command".to_string()));
+}
+
+#[test]
+fn commander_get_command_error() {
+    let commander = Commander::new(HashMap::new());
+    assert_eq!(commander.get_command("dummy"), Err("Command <dummy> not found in map.".to_string()));
+}
+
+#[test]
+fn commander_execute_fail() {
+    let output = execute("/tmp/fake_command".to_string(), vec!["--random_arg".to_string()]).unwrap_err();
+    assert_eq!(output, "Failed to execute command: [/tmp/fake_command --random_arg] - No such file or directory (os error 2)");
+}
+
+#[test]
+fn commander_execute_illegal_option() {
+    let output = execute("pwd".to_string(), vec!["--random_arg".to_string()]).unwrap_err();
+    assert_eq!(output, "Failed to execute command: [pwd --random_arg] - pwd: illegal option -- -\nusage: pwd [-L | -P]\n");
+}

executor/src/executor/dispatcher/mod.rs

@@ -0,0 +1,70 @@
+// Copyright (c) 2016-2017 Snowplow Analytics Ltd. All rights reserved.
+//
+// This program is licensed to you under the Apache License Version 2.0, and
+// you may not use this file except in compliance with the Apache License
+// Version 2.0.  You may obtain a copy of the Apache License Version 2.0 at
+// http://www.apache.org/licenses/LICENSE-2.0.
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the Apache License Version 2.0 is distributed on an "AS
+// IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied.  See the Apache License Version 2.0 for the specific language
+// governing permissions and limitations there under.
+//
+
+#[cfg(test)]
+mod tests;
+
+use std::collections::VecDeque;
+use std::sync::mpsc::Sender;
+use executor::server::JobRequest;
+use executor::responder::DispatcherStatus;
+
+#[derive(Debug, PartialEq)]
+pub enum Dispatch {
+    StatusUpdate(Query<DispatcherStatus>),
+    CheckQueue(Query<bool>),
+    NewRequest(JobRequest),
+    ProcessRequest,
+    RequestComplete(JobRequest),
+    RequestFailure(JobRequest),
+    StopProcessing,
+}
+
+#[derive(Debug)]
+pub struct Dispatcher {
+    pub max_jobs: usize,
+    pub max_workers: usize,
+    pub requests_queue: VecDeque<JobRequest>,
+}
+
+impl Dispatcher {
+    pub fn new(queue_size: usize, workers_size: usize) -> Dispatcher {
+        Dispatcher {
+            max_jobs: queue_size,
+            max_workers: workers_size,
+            requests_queue: VecDeque::with_capacity(queue_size),
+        }
+    }
+}
+
+#[derive(Debug)]
+pub struct Query<T> {
+    pub name: String,
+    pub status_tx: Sender<T>,
+}
+
+impl<T> Query<T> {
+    pub fn new(name: String, status_tx: Sender<T>) -> Query<T> {
+        Query {
+            name: name,
+            status_tx: status_tx,
+        }
+    }
+}
+
+impl<T> PartialEq for Query<T> {
+    fn eq(&self, other: &Query<T>) -> bool {
+        self.name == other.name
+    }
+}

executor/src/executor/dispatcher/tests.rs

@@ -0,0 +1,24 @@
+// Copyright (c) 2016-2017 Snowplow Analytics Ltd. All rights reserved.
+//
+// This program is licensed to you under the Apache License Version 2.0, and
+// you may not use this file except in compliance with the Apache License
+// Version 2.0.  You may obtain a copy of the Apache License Version 2.0 at
+// http://www.apache.org/licenses/LICENSE-2.0.
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the Apache License Version 2.0 is distributed on an "AS
+// IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied.  See the Apache License Version 2.0 for the specific language
+// governing permissions and limitations there under.
+//
+
+use super::*;
+
+#[test]
+fn create_new_dispatcher() {
+    let dispatcher = Dispatcher::new(10, 2);
+
+    assert_eq!(dispatcher.max_jobs, 10);
+    assert_eq!(dispatcher.max_workers, 2);
+    assert!(dispatcher.requests_queue.is_empty());
+}