Merge pull request #45 from petehayes102/master

Restructure endpoints to pave way for streaming API

Author: petehayes102

agent/Cargo.toml

@@ -24,6 +24,7 @@ intecture_api = { version = "0.4.0", path = "../core" }
 serde = "1.0"
 serde_derive = "1.0"
 serde_json = "1.0"
+tokio-core = "0.1"
 tokio-proto = "0.1"
 tokio-service = "0.1"
 toml = "0.4"

agent/src/main.rs

@@ -11,6 +11,7 @@ extern crate futures;
 extern crate intecture_api;
 #[macro_use] extern crate serde_derive;
 extern crate serde_json;
+extern crate tokio_core;
 extern crate tokio_proto;
 extern crate tokio_service;
 extern crate toml;
@@ -19,15 +20,21 @@ mod errors;
 
 use errors::*;
 use futures::{future, Future};
-use intecture_api::{Executable, Runnable};
+use intecture_api::remote::{Executable, Runnable};
+use intecture_api::host::local::Local;
 use intecture_api::host::remote::JsonProto;
 use std::fs::File;
 use std::io::{self, Read};
 use std::net::SocketAddr;
+use std::sync::Arc;
+use tokio_core::reactor::Remote;
 use tokio_proto::TcpServer;
-use tokio_service::Service;
+use tokio_service::{NewService, Service};
 
-pub struct Api;
+pub struct Api {
+    host: Local,
+    remote: Remote,
+}
 
 impl Service for Api {
     type Request = serde_json::Value;
@@ -46,7 +53,14 @@ impl Service for Api {
                         io::ErrorKind::Other, e.description()
                     ))),
         };
-        Box::new(runnable.exec()
+
+        // XXX Danger zone! If we're running multiple threads, this `unwrap()`
+        // will explode. The API requires a `Handle`, but we can only send a
+        // `Remote` to this Service. Currently we force the `Handle`, which is
+        // only safe for the current thread.
+        // See https://github.com/alexcrichton/tokio-process/issues/23
+        let handle = self.remote.handle().unwrap();
+        Box::new(runnable.exec(&self.host, &handle)
             // @todo Can't wrap 'e' as error_chain Error doesn't derive Sync.
             // Waiting for https://github.com/rust-lang-nursery/error-chain/pull/163
             .map_err(|e| io::Error::new(io::ErrorKind::Other, e.description()))
@@ -57,6 +71,19 @@ impl Service for Api {
     }
 }
 
+impl NewService for Api {
+    type Request = serde_json::Value;
+    type Response = serde_json::Value;
+    type Error = io::Error;
+    type Instance = Api;
+    fn new_service(&self) -> io::Result<Self::Instance> {
+        Ok(Api {
+            host: self.host.clone(),
+            remote: self.remote.clone(),
+        })
+    }
+}
+
 #[derive(Deserialize)]
 struct Config {
     address: SocketAddr,
@@ -96,7 +123,19 @@ quick_main!(|| -> Result<()> {
         Config { address }
     };
 
+    let host = Local::new().wait()?;
+    // XXX We can only run a single thread here, or big boom!!
+    // The API requires a `Handle`, but we can only send a `Remote`.
+    // Currently we force the issue (`unwrap()`), which is only safe
+    // for the current thread.
+    // See https://github.com/alexcrichton/tokio-process/issues/23
     let server = TcpServer::new(JsonProto, config.address);
-    server.serve(|| Ok(Api));
+    server.with_handle(move |handle| {
+        let api = Api {
+            host: host.clone(),
+            remote: handle.remote().clone(),
+        };
+        Arc::new(api)
+    });
     Ok(())
 });

core/Cargo.toml

@@ -30,6 +30,7 @@ serde_derive = "1.0"
 serde_json = "1.0"
 tokio-core = "0.1"
 tokio-io = "0.1"
+tokio-process = "0.1"
 tokio-proto = "0.1"
 tokio-service = "0.1"
 

core/examples/basic.rs

@@ -14,14 +14,16 @@ use tokio_core::reactor::Core;
 
 fn main() {
     let mut core = Core::new().unwrap();
+    let handle = core.handle();
 
     let host = Local::new().and_then(|host| {
-        command::factory(&host, "whoami", None).and_then(|mut cmd| {
-            cmd.exec().map(|out| {
+        Command::new(&host, "whoami", None).and_then(|mut cmd| {
+            cmd.exec(&handle).map(|out| {
                 println!("I'm currently running as {}", String::from_utf8_lossy(&out.stdout).trim());
             })
         })
     });
 
     core.run(host).unwrap();
 }
+

core/src/command/mod.rs

@@ -8,29 +8,20 @@
 
 pub mod providers;
 
-use erased_serde::Serialize;
 use errors::*;
-use Executable;
 use futures::{future, Future};
 use host::Host;
-use self::providers::{Nix, NixRunnable};
-use std::sync::Arc;
+use self::providers::CommandProvider;
+use tokio_core::reactor::Handle;
 
-pub trait CommandProvider<H: Host> {
-    fn available(&Arc<H>) -> Box<Future<Item = bool, Error = Error>> where Self: Sized;
-    fn try_new(&Arc<H>, &str, Option<&[&str]>) -> Box<Future<Item = Option<Self>, Error = Error>> where Self: Sized;
-    fn exec(&mut self) -> Box<Future<Item = CommandResult, Error = Error>>;
-}
-
-#[doc(hidden)]
-#[derive(Serialize, Deserialize)]
-pub enum CommandRunnable {
-    Nix(NixRunnable)
-}
+#[cfg(not(windows))]
+const DEFAULT_SHELL: [&'static str; 2] = ["/bin/sh", "-c"];
+#[cfg(windows)]
+const DEFAULT_SHELL: [&'static str; 1] = ["yeah...we don't currently support windows :("];
 
-#[doc(hidden)]
-#[derive(Clone, Serialize, Deserialize)]
-pub struct Command {
+pub struct Command<H: Host> {
+    host: H,
+    inner: Box<CommandProvider<H>>,
     shell: Vec<String>,
     cmd: String,
 }
@@ -43,18 +34,41 @@ pub struct CommandResult {
     pub stderr: Vec<u8>,
 }
 
-impl Executable for CommandRunnable {
-    fn exec(self) -> Box<Future<Item = Box<Serialize>, Error = Error>> {
-        match self {
-            CommandRunnable::Nix(p) => p.exec()
+impl<H: Host + 'static> Command<H> {
+    pub fn new(host: &H, cmd: &str, shell: Option<&[&str]>) -> Box<Future<Item = Command<H>, Error = Error>> {
+        let host = host.clone();
+        let cmd = cmd.to_owned();
+        let shell: Vec<String> = shell.unwrap_or(&DEFAULT_SHELL)
+            .to_owned()
+            .iter()
+            .map(|s| s.to_string())
+            .collect();
+
+        Box::new(host.command_provider()
+            .and_then(|provider| future::ok(Command {
+                host: host,
+                inner: provider,
+                shell: shell,
+                cmd: cmd,
+            })))
+    }
+
+    pub fn with_provider<P>(host: &H, provider: P, cmd: &str, shell: Option<&[&str]>) -> Command<H>
+        where P: CommandProvider<H> + 'static
+    {
+        Command {
+            host: host.clone(),
+            inner: Box::new(provider),
+            shell: shell.unwrap_or(&DEFAULT_SHELL)
+                        .to_owned()
+                        .iter()
+                        .map(|s| s.to_string())
+                        .collect(),
+            cmd: cmd.into(),
         }
     }
-}
 
-pub fn factory<H: Host + 'static>(host: &Arc<H>, cmd: &str, shell: Option<&[&str]>) -> Box<Future<Item = Box<CommandProvider<H>>, Error = Error>> {
-    Box::new(Nix::try_new(host, cmd, shell)
-                     .and_then(|opt| match opt {
-                         Some(provider) => future::ok(Box::new(provider) as Box<CommandProvider<H>>),
-                         None => future::err(ErrorKind::ProviderUnavailable("Command").into())
-                     }))
+    pub fn exec(&mut self, handle: &Handle) -> Box<Future<Item = CommandResult, Error = Error>> {
+        self.inner.exec(&self.host, handle, &self.cmd, &self.shell)
+    }
 }

core/src/command/providers/generic.rs

@@ -0,0 +1,117 @@
+// Copyright 2015-2017 Intecture Developers.
+//
+// Licensed under the Mozilla Public License 2.0 <LICENSE or
+// https://www.tldrlegal.com/l/mpl-2.0>. This file may not be copied,
+// modified, or distributed except according to those terms.
+
+use command::CommandResult;
+use erased_serde::Serialize;
+use errors::*;
+use futures::{future, Future};
+use host::{Host, HostType};
+use host::local::Local;
+use host::remote::Plain;
+use provider::Provider;
+use remote::{Executable, Runnable};
+use std::process;
+use super::{CommandProvider, CommandRunnable};
+use tokio_core::reactor::Handle;
+use tokio_process::CommandExt;
+
+#[derive(Clone)]
+pub struct Generic;
+struct LocalGeneric;
+struct RemoteGeneric;
+
+#[doc(hidden)]
+#[derive(Serialize, Deserialize)]
+pub enum GenericRunnable {
+    Available,
+    Exec(String, Vec<String>),
+}
+
+impl<H: Host + 'static> Provider<H> for Generic {
+    fn available(host: &H) -> Box<Future<Item = bool, Error = Error>> {
+        match host.get_type() {
+            HostType::Local(_) => LocalGeneric::available(),
+            HostType::Remote(r) => RemoteGeneric::available(r),
+        }
+    }
+
+    fn try_new(host: &H) -> Box<Future<Item = Option<Generic>, Error = Error>> {
+        let host = host.clone();
+        Box::new(Self::available(&host)
+            .and_then(|available| {
+                if available {
+                    future::ok(Some(Generic))
+                } else {
+                    future::ok(None)
+                }
+            }))
+    }
+}
+
+impl<H: Host + 'static> CommandProvider<H> for Generic {
+    fn exec(&self, host: &H, handle: &Handle, cmd: &str, shell: &[String]) -> Box<Future<Item = CommandResult, Error = Error>> {
+        match host.get_type() {
+            HostType::Local(_) => LocalGeneric::exec(handle, cmd, shell),
+            HostType::Remote(r) => RemoteGeneric::exec(r, cmd, shell),
+        }
+    }
+}
+
+impl LocalGeneric {
+    fn available() -> Box<Future<Item = bool, Error = Error>> {
+        Box::new(future::ok(cfg!(unix)))
+    }
+
+    fn exec(handle: &Handle, cmd: &str, shell: &[String]) -> Box<Future<Item = CommandResult, Error = Error>> {
+        let cmd = cmd.to_owned();
+        let shell = shell.to_owned();
+        let (shell, shell_args) = match shell.split_first() {
+            Some((s, a)) => (s, a),
+            None => return Box::new(future::err("Invalid shell provided".into())),
+        };
+
+        Box::new(process::Command::new(shell)
+            .args(shell_args)
+            .arg(&cmd)
+            .output_async(handle)
+            .chain_err(|| "Command execution failed")
+            .and_then(|output| {
+                future::ok(CommandResult {
+                    success: output.status.success(),
+                    exit_code: output.status.code(),
+                    stdout: output.stdout,
+                    stderr: output.stderr,
+                })
+            }))
+    }
+}
+
+impl RemoteGeneric {
+    fn available(host: &Plain) -> Box<Future<Item = bool, Error = Error>> {
+        let runnable = Runnable::Command(
+                          CommandRunnable::Generic(
+                              GenericRunnable::Available));
+        host.run(runnable)
+            .chain_err(|| ErrorKind::Runnable { endpoint: "Command::Generic", func: "available" })
+    }
+
+    fn exec(host: &Plain, cmd: &str, shell: &[String]) -> Box<Future<Item = CommandResult, Error = Error>> {
+        let runnable = Runnable::Command(
+                          CommandRunnable::Generic(
+                              GenericRunnable::Exec(cmd.into(), shell.to_owned())));
+        host.run(runnable)
+            .chain_err(|| ErrorKind::Runnable { endpoint: "Command::Generic", func: "exec" })
+    }
+}
+
+impl Executable for GenericRunnable {
+    fn exec(self, _: &Local, handle: &Handle) -> Box<Future<Item = Box<Serialize>, Error = Error>> {
+        match self {
+            GenericRunnable::Available => Box::new(LocalGeneric::available().map(|b| Box::new(b) as Box<Serialize>)),
+            GenericRunnable::Exec(cmd, shell) => Box::new(LocalGeneric::exec(handle, &cmd, &shell).map(|r| Box::new(r) as Box<Serialize>)),
+        }
+    }
+}

core/src/command/providers/mod.rs

@@ -4,6 +4,41 @@
 // https://www.tldrlegal.com/l/mpl-2.0>. This file may not be copied,
 // modified, or distributed except according to those terms.
 
-mod nix;
+mod generic;
 
-pub use self::nix::{Nix, NixRunnable};
+use errors::*;
+use erased_serde::Serialize;
+use futures::{future, Future};
+use host::Host;
+use host::local::Local;
+use provider::Provider;
+use remote::Executable;
+pub use self::generic::{Generic, GenericRunnable};
+use super::CommandResult;
+use tokio_core::reactor::Handle;
+
+pub trait CommandProvider<H: Host>: Provider<H> {
+    fn exec(&self, &H, &Handle, &str, &[String]) -> Box<Future<Item = CommandResult, Error = Error>>;
+}
+
+#[doc(hidden)]
+#[derive(Serialize, Deserialize)]
+pub enum CommandRunnable {
+    Generic(GenericRunnable)
+}
+
+impl Executable for CommandRunnable {
+    fn exec(self, host: &Local, handle: &Handle) -> Box<Future<Item = Box<Serialize>, Error = Error>> {
+        match self {
+            CommandRunnable::Generic(p) => p.exec(host, handle)
+        }
+    }
+}
+
+pub fn factory<H: Host + 'static>(host: &H) -> Box<Future<Item = Box<CommandProvider<H>>, Error = Error>> {
+    Box::new(Generic::try_new(host)
+        .and_then(|opt| match opt {
+            Some(provider) => future::ok(Box::new(provider) as Box<CommandProvider<H>>),
+            None => future::err(ErrorKind::ProviderUnavailable("Command").into())
+        }))
+}