WIP - Remote host support with Tokio

Author: petehayes102

core/Cargo.toml

@@ -16,12 +16,18 @@ include = ["Cargo.toml", "src/**/*.rs", "README.md", "LICENSE"]
 travis-ci = { repository = "intecture/api" }
 
 [dependencies]
+bytes = "0.4"
 erased-serde = "0.3"
 error-chain = "0.11"
+futures = "0.1"
 hostname = "0.1"
 ipnetwork = "0.12"
 pnet = "0.19"
 regex = "0.2"
 serde = "1.0"
 serde_derive = "1.0"
 serde_json = "1.0"
+tokio-core = "0.1"
+tokio-io = "0.1"
+tokio-proto = "0.1"
+tokio-service = "0.1"

core/src/host.rs

@@ -6,41 +6,70 @@
 
 //! Manages the connection between the API and your host.
 
+use bytes::{BufMut, BytesMut};
 use errors::*;
+use futures::Future;
+use RemoteProvider;
+use serde::Deserialize;
+use serde_json;
+use std::{io, result};
+use std::net::SocketAddr;
 use telemetry::{self, Telemetry};
+use tokio_core::net::TcpStream;
+use tokio_core::reactor::Handle;
+use tokio_io::{AsyncRead, AsyncWrite};
+use tokio_io::codec::{Encoder, Decoder, Framed};
+use tokio_proto::pipeline::{ClientProto, ClientService};
+use tokio_proto::TcpClient;
+use tokio_service::Service;
 
 pub struct Host {
     telemetry: Option<Telemetry>,
-    sock: Option<u32>,
 }
 
+pub struct RemoteHost {
+    inner: ClientService<TcpStream, JsonProto>,
+    telemetry: Option<Telemetry>,
+}
+
+struct JsonCodec;
+struct JsonProto;
+
 impl Host {
     /// Create a new Host targeting the local machine.
     pub fn local() -> Result<Host> {
         let mut host = Host {
             telemetry: None,
-            sock: None
         };
 
         host.telemetry = Some(telemetry::load(&host)?);
 
         Ok(host)
     }
 
+    /// Retrieve Telemetry
+    pub fn telemetry(&self) -> &Telemetry {
+        self.telemetry.as_ref().unwrap()
+    }
+
+    pub fn is_local(&self) -> bool {
+        true
+    }
+}
+
+impl RemoteHost {
     /// Create a new Host connected to addr.
-    pub fn connect(_addr: &str) -> Result<Host> {
-        unimplemented!();
-        // @todo Connect to sock...
+    pub fn connect(addr: &SocketAddr, handle: &Handle) -> Box<Future<Item = RemoteHost, Error = io::Error>> {
+        let ret = TcpClient::new(JsonProto)
+            .connect(addr, handle)
+            .map(|client_service| {
+                RemoteHost {
+                    inner: client_service,
+                    telemetry: None,
+                }
+            });
 
-        // let mut host = Host {
-        //     telemetry: None,
-        //     sock: None
-        // };
-        //
-        // let provider = telemetry::factory(&host)?;
-        // host.telemetry = Some(provider.load()?);
-        //
-        // Ok(host)
+        Box::new(ret)
     }
 
     /// Retrieve Telemetry
@@ -49,6 +78,69 @@ impl Host {
     }
 
     pub fn is_local(&self) -> bool {
-        self.sock.is_none()
+        false
+    }
+
+    pub fn run<T>(&self, provider: RemoteProvider) -> Box<Future<Item = T, Error = io::Error>>
+        where for<'de> T: Deserialize<'de>
+    {
+        Box::new(self.inner.call(provider)
+                           .map(|v| serde_json::from_value::<T>(v).unwrap()))
+    }
+}
+
+impl Service for RemoteHost {
+    type Request = RemoteProvider;
+    type Response = serde_json::Value;
+    type Error = io::Error;
+    type Future = Box<Future<Item = Self::Response, Error = Self::Error>>;
+
+    fn call(&self, req: Self::Request) -> Self::Future {
+        Box::new(self.inner.call(req))
+    }
+}
+
+impl Decoder for JsonCodec {
+    type Item = serde_json::Value;
+    type Error = io::Error;
+
+    fn decode(&mut self, buf: &mut BytesMut) -> result::Result<Option<Self::Item>, Self::Error> {
+        // Check to see if the frame contains a new line
+        if let Some(n) = buf.as_ref().iter().position(|b| *b == b'\n') {
+            // remove the serialized frame from the buffer.
+            let line = buf.split_to(n);
+
+            // Also remove the '\n'
+            buf.split_to(1);
+
+            return Ok(Some(serde_json::from_slice(&line).unwrap()));
+        }
+
+        Ok(None)
+    }
+}
+
+impl Encoder for JsonCodec {
+    type Item = RemoteProvider;
+    type Error = io::Error;
+
+    fn encode(&mut self, provider: Self::Item, buf: &mut BytesMut) -> io::Result<()> {
+        let json = serde_json::to_string(&provider).unwrap();
+        buf.reserve(json.len() + 1);
+        buf.extend(json.as_bytes());
+        buf.put_u8(b'\n');
+
+        Ok(())
+    }
+}
+
+impl<T: AsyncRead + AsyncWrite + 'static> ClientProto<T> for JsonProto {
+    type Request = RemoteProvider;
+    type Response = serde_json::Value;
+    type Transport = Framed<T, JsonCodec>;
+    type BindTransport = result::Result<Self::Transport, io::Error>;
+
+    fn bind_transport(&self, io: T) -> Self::BindTransport {
+        Ok(io.framed(JsonCodec))
     }
 }

core/src/lib.rs

@@ -11,15 +11,21 @@
 
 #![recursion_limit = "1024"]
 
+extern crate bytes;
 extern crate erased_serde;
 #[macro_use] extern crate error_chain;
+extern crate futures;
 extern crate hostname;
 extern crate ipnetwork;
 extern crate pnet;
 extern crate regex;
 extern crate serde;
 #[macro_use] extern crate serde_derive;
 extern crate serde_json;
+extern crate tokio_core;
+extern crate tokio_io;
+extern crate tokio_proto;
+extern crate tokio_service;
 
 pub mod command;
 pub mod errors;

core/src/telemetry/providers/macos.rs

@@ -40,10 +40,8 @@ impl TelemetryProvider for Macos {
             cfg!(target_os="macos")
         } else {
             unimplemented!();
-            // let r = RemoteProvider::Available;
-            // self.host.send(r).chain_err(|| ErrorKind::RemoteProvider("Telemetry", "available"))?;
-            // let t: Telemetry = self.host.recv()?;
-            // Ok(t)
+            // host.call(RemoteProvider::Available)
+            //     .chain_err(|| ErrorKind::RemoteProvider { endpoint: "Telemetry", func: "available" })?
         }
     }