Tokio/Futures - First working version!

Author: petehayes102

core/src/command/mod.rs

@@ -10,15 +10,29 @@ pub mod providers;
 
 use erased_serde::Serialize;
 use errors::*;
-use ExecutableProvider;
+use Executable;
+use futures::{future, Future};
 use host::Host;
-use self::providers::{Nix, NixRemoteProvider};
+use self::providers::{Nix, NixRunnable};
+use std::sync::Arc;
+
+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)
+}
+
+#[doc(hidden)]
+#[derive(Clone, Serialize, Deserialize)]
 pub struct Command {
     shell: Vec<String>,
-    cmd: Vec<String>,
+    cmd: String,
 }
 
 #[derive(Debug, Serialize, Deserialize)]
@@ -29,30 +43,18 @@ pub struct CommandResult {
     pub stderr: Vec<u8>,
 }
 
-pub trait CommandProvider<'a> {
-    fn available(&Host) -> bool where Self: Sized;
-    fn try_new(&'a Host, &[&str], Option<&[&str]>) -> Option<Self> where Self: Sized;
-    fn exec(&self) -> Result<CommandResult>;
-}
-
-#[doc(hidden)]
-#[derive(Serialize, Deserialize)]
-pub enum RemoteProvider {
-    Nix(NixRemoteProvider)
-}
-
-impl <'de>ExecutableProvider<'de> for RemoteProvider {
-    fn exec(self, host: &Host) -> Result<Box<Serialize>> {
+impl Executable for CommandRunnable {
+    fn exec(self) -> Box<Future<Item = Box<Serialize>, Error = Error>> {
         match self {
-            RemoteProvider::Nix(p) => p.exec(host)
+            CommandRunnable::Nix(p) => p.exec()
         }
     }
 }
 
-pub fn factory<'a>(host: &'a Host, cmd: &[&str], shell: Option<&[&str]>) -> Result<Box<CommandProvider<'a> + 'a>> {
-    if let Some(p) = Nix::try_new(host, cmd, shell) {
-        Ok(Box::new(p))
-    } else {
-        Err(ErrorKind::ProviderUnavailable("Command").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())
+                     }))
 }

core/src/command/providers/mod.rs

@@ -6,4 +6,4 @@
 
 mod nix;
 
-pub use self::nix::{Nix, RemoteProvider as NixRemoteProvider};
+pub use self::nix::{Nix, NixRunnable};

core/src/command/providers/nix.rs

@@ -4,84 +4,90 @@
 // https://www.tldrlegal.com/l/mpl-2.0>. This file may not be copied,
 // modified, or distributed except according to those terms.
 
-use command::{Command, CommandProvider, CommandResult};
+use command::{Command, CommandProvider, CommandResult, CommandRunnable};
 use erased_serde::Serialize;
 use errors::*;
-use ExecutableProvider;
-use host::*;
+use Executable;
+use futures::{future, Future};
+use host::Host;
+use Runnable;
 use std::process;
+use std::sync::Arc;
 
 const DEFAULT_SHELL: [&'static str; 2] = ["/bin/sh", "-c"];
 
-pub struct Nix<'a> {
-    host: &'a Host,
+pub struct Nix<H: Host> {
+    host: Arc<H>,
     inner: Command
 }
 
 #[doc(hidden)]
 #[derive(Serialize, Deserialize)]
-pub enum RemoteProvider {
+pub enum NixRunnable {
     Available,
     Exec(Command),
 }
 
-impl <'de>ExecutableProvider<'de> for RemoteProvider {
-    fn exec(self, host: &Host) -> Result<Box<Serialize>> {
-        match self {
-            RemoteProvider::Available => Ok(Box::new(Nix::available(host))),
-            RemoteProvider::Exec(inner) => {
-                let p = Nix { host, inner };
-                Ok(Box::new(p.exec()?))
-            }
-        }
+impl <H: Host + 'static>CommandProvider<H> for Nix<H> {
+    fn available(host: &Arc<H>) -> Box<Future<Item = bool, Error = Error>> {
+        host.run(Runnable::Command(CommandRunnable::Nix(NixRunnable::Available)))
+            .chain_err(|| ErrorKind::Runnable { endpoint: "Command::Nix", func: "available" })
     }
-}
 
-impl <'a>CommandProvider<'a> for Nix<'a> {
-    fn available(host: &Host) -> bool {
-        if host.is_local() {
-            cfg!(not(windows))
-        } else {
-            unimplemented!();
-            // let r = RemoteProvider::Available;
-            // self.host.send(r).chain_err(|| ErrorKind::RemoteProvider("Command", "available"))?;
-            // Ok(self.host.recv()?)
-        }
+    fn try_new(host: &Arc<H>, cmd: &str, shell: Option<&[&str]>) -> Box<Future<Item = Option<Nix<H>>, Error = Error>> {
+        let cmd_owned = cmd.to_owned();
+        let shell_owned: Vec<String> = shell.unwrap_or(&DEFAULT_SHELL).to_owned().iter().map(|s| s.to_string()).collect();
+        let host = host.clone();
+
+        Box::new(Self::available(&host).and_then(move |available| {
+            if available {
+                let inner = Command {
+                    shell: shell_owned,
+                    cmd: cmd_owned,
+                };
+                future::ok(Some(Nix { host, inner }))
+            } else {
+                future::ok(None)
+            }
+        }))
     }
 
-    fn try_new(host: &'a Host, cmd: &[&str], shell: Option<&[&str]>) -> Option<Nix<'a>> {
-        if Self::available(host) {
-            let inner = Command {
-                shell: shell.unwrap_or(&DEFAULT_SHELL).to_owned().iter().map(|s| s.to_string()).collect(),
-                cmd: cmd.to_owned().iter().map(|s| s.to_string()).collect(),
-            };
-            Some(Nix { host, inner })
-        } else {
-            None
-        }
+    fn exec(&mut self) -> Box<Future<Item = CommandResult, Error = Error>> {
+        self.host.run(Runnable::Command(CommandRunnable::Nix(NixRunnable::Exec(self.inner.clone()))))
+                 .chain_err(|| ErrorKind::Runnable { endpoint: "Command::Nix", func: "exec" })
     }
+}
 
-    fn exec(&self) -> Result<CommandResult> {
-        if self.host.is_local() {
-            let (shell, shell_args) = self.inner.shell.split_first()
-                                                      .ok_or("Invalid shell provided")?;
-            let out = process::Command::new(shell)
-                                       .args(shell_args)
-                                       .args(&self.inner.cmd)
-                                       .output()
-                                       .chain_err(|| "Command execution failed")?;
-            Ok(CommandResult {
-                success: out.status.success(),
-                exit_code: out.status.code(),
-                stdout: out.stdout,
-                stderr: out.stderr
-            })
-        } else {
-            unimplemented!();
-            // let r = RemoteProvider::Load;
-            // self.host.send(r).chain_err(|| ErrorKind::RemoteProvider("Command", "exec"))?;
-            // let result: CommandResult = self.host.recv()?;
-            // Ok(result)
+impl Executable for NixRunnable {
+    fn exec(self) -> Box<Future<Item = Box<Serialize>, Error = Error>> {
+        match self {
+            NixRunnable::Available =>
+                Box::new(future::ok(Box::new(
+                    cfg!(unix)
+                ) as Box<Serialize>)),
+            NixRunnable::Exec(inner) => {
+                Box::new(future::lazy(move || -> future::FutureResult<Box<Serialize>, Error> {
+                    let (shell, shell_args) = match inner.shell.split_first() {
+                        Some((s, a)) => (s, a),
+                        None => return future::err("Invalid shell provided".into()),
+                    };
+
+                    let out = process::Command::new(shell)
+                                               .args(shell_args)
+                                               .arg(&inner.cmd)
+                                               .output()
+                                               .chain_err(|| "Command execution failed");
+                    match out {
+                        Ok(output) => future::ok(Box::new(CommandResult {
+                            success: output.status.success(),
+                            exit_code: output.status.code(),
+                            stdout: output.stdout,
+                            stderr: output.stderr,
+                        }) as Box<Serialize>),
+                        Err(e) => future::err(e),
+                    }
+                }))
+            }
         }
     }
 }

core/src/errors.rs

@@ -4,7 +4,14 @@
 // https://www.tldrlegal.com/l/mpl-2.0>. This file may not be copied,
 // modified, or distributed except according to those terms.
 
+use futures::Future;
+use std::{error, io};
+
 error_chain! {
+    foreign_links {
+        Io(io::Error);
+    }
+
     errors {
         InvalidTelemetryKey {
             cmd: &'static str,
@@ -19,7 +26,7 @@ error_chain! {
             display("No providers available for {}", p),
         }
 
-        RemoteProvider {
+        Runnable {
             endpoint: &'static str,
             func: &'static str,
         } {
@@ -48,3 +55,26 @@ error_chain! {
         }
     }
 }
+
+// @todo This should disappear once Futures are officially supported
+// by error_chain.
+// See: https://github.com/rust-lang-nursery/error-chain/issues/90
+pub type SFuture<T> = Box<Future<Item = T, Error = Error>>;
+
+pub trait FutureChainErr<T> {
+    fn chain_err<F, E>(self, callback: F) -> SFuture<T>
+        where F: FnOnce() -> E + 'static,
+              E: Into<ErrorKind>;
+}
+
+impl<F> FutureChainErr<F::Item> for F
+    where F: Future + 'static,
+          F::Error: error::Error + Send + 'static,
+{
+    fn chain_err<C, E>(self, callback: C) -> SFuture<F::Item>
+        where C: FnOnce() -> E + 'static,
+              E: Into<ErrorKind>,
+    {
+        Box::new(self.then(|r| r.chain_err(callback)))
+    }
+}