Minimal Agent with issues - hardcoded bind address, no error handling

Author: petehayes102

agent/Cargo.toml

@@ -7,7 +7,7 @@ description = "Tiny daemon that wraps the core API"
 homepage = "https://intecture.io"
 repository = "https://github.com/intecture/api"
 documentation = "https://intecture.io/rust/inapi/"
-keywords = ["intecture", "remote"]
+keywords = ["intecture", "agent"]
 categories = ["servers"]
 readme = "README.md"
 include = ["Cargo.toml", "src/**/*.rs", "README.md", "LICENSE"]
@@ -16,8 +16,10 @@ include = ["Cargo.toml", "src/**/*.rs", "README.md", "LICENSE"]
 travis-ci = { repository = "intecture/api" }
 
 [dependencies]
+env_logger = "0.4"
 error-chain = "0.11"
+futures = "0.1"
 intecture_api = { version = "0.4.0", path = "../core" }
-serde = "1.0"
-serde_derive = "1.0"
 serde_json = "1.0"
+tokio-proto = "0.1"
+tokio-service = "0.1"

agent/src/main.rs

@@ -6,16 +6,59 @@
 
 #![recursion_limit = "1024"]
 
+extern crate env_logger;
 #[macro_use] extern crate error_chain;
+extern crate futures;
 extern crate intecture_api;
-extern crate serde;
-#[macro_use] extern crate serde_derive;
 extern crate serde_json;
+extern crate tokio_proto;
+extern crate tokio_service;
 
 mod errors;
 
 use errors::*;
+use futures::{future, Future};
+use intecture_api::{Executable, Runnable};
+use intecture_api::host::JsonProto;
+use std::io;
+use tokio_proto::TcpServer;
+use tokio_service::Service;
+
+pub struct Api;
+
+impl Service for Api {
+    type Request = serde_json::Value;
+    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 {
+        let runnable: Runnable = match serde_json::from_value(req).chain_err(|| "Received invalid Runnable") {
+            Ok(r) => r,
+            Err(e) => return Box::new(
+                future::err(
+                    io::Error::new(
+                        // @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
+                        io::ErrorKind::Other, e.description()
+                    ))),
+        };
+        Box::new(runnable.exec()
+            // @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()))
+            .and_then(|ser| match serde_json::to_value(ser).chain_err(|| "Could not serialize result") {
+                Ok(v) => future::ok(v),
+                Err(e) => future::err(io::Error::new(io::ErrorKind::Other, e.description())),
+            }))
+    }
+}
 
 quick_main!(|| -> Result<()> {
+    env_logger::init().chain_err(|| "Could not start logging")?;
+
+    let addr = "127.0.0.1:7101".parse().chain_err(|| "Invalid server address")?;
+    let server = TcpServer::new(JsonProto, addr);
+    server.serve(|| Ok(Api));
     Ok(())
 });

core/Cargo.toml

@@ -22,6 +22,7 @@ error-chain = "0.11"
 futures = "0.1"
 hostname = "0.1"
 ipnetwork = "0.12"
+log = "0.3"
 pnet = "0.19"
 regex = "0.2"
 serde = "1.0"

core/src/host.rs

@@ -20,7 +20,7 @@ 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::pipeline::{ClientProto, ClientService, ServerProto};
 use tokio_proto::TcpClient;
 use tokio_service::Service;
 
@@ -78,16 +78,27 @@ pub struct RemoteHost {
     telemetry: Option<Telemetry>,
 }
 
-struct JsonCodec;
-struct JsonProto;
+#[doc(hidden)]
+pub struct JsonCodec;
+#[doc(hidden)]
+pub struct JsonProto;
 
 impl RemoteHost {
     /// Create a new Host connected to addr.
-    pub fn connect(addr: &SocketAddr, handle: &Handle) -> Box<Future<Item = Arc<RemoteHost>, Error = Error>> {
+    pub fn connect(addr: &str, handle: &Handle) -> Box<Future<Item = Arc<RemoteHost>, Error = Error>> {
+        let addr: SocketAddr = match addr.parse().chain_err(|| "Invalid host address") {
+            Ok(addr) => addr,
+            Err(e) => return Box::new(future::err(e)),
+        };
+
+        info!("Connecting to host {}", addr);
+
         Box::new(TcpClient::new(JsonProto)
-            .connect(addr, handle)
+            .connect(&addr, handle)
             .chain_err(|| "Could not connect to host")
             .and_then(|client_service| {
+                info!("Connected!");
+
                 let mut host = Arc::new(RemoteHost {
                     inner: client_service,
                     telemetry: None,
@@ -111,17 +122,21 @@ impl Host for RemoteHost {
     fn run<D: 'static>(&self, provider: Runnable) -> Box<Future<Item = D, Error = Error>>
         where for<'de> D: Deserialize<'de>
     {
-        Box::new(self.inner.call(provider)
-                           .chain_err(|| "Could not run provider")
-                           .and_then(|v| match serde_json::from_value::<D>(v).chain_err(|| "Could not run provider") {
+        let value = match serde_json::to_value(provider).chain_err(|| "Could not encode provider to send to host") {
+            Ok(v) => v,
+            Err(e) => return Box::new(future::err(e))
+        };
+        Box::new(self.inner.call(value)
+                           .chain_err(|| "Error while running provider on host")
+                           .and_then(|v| match serde_json::from_value::<D>(v).chain_err(|| "Could not understand response from host") {
                                Ok(d) => future::ok(d),
                                Err(e) => future::err(e)
                            }))
     }
 }
 
 impl Service for RemoteHost {
-    type Request = Runnable;
+    type Request = serde_json::Value;
     type Response = serde_json::Value;
     type Error = io::Error;
     type Future = Box<Future<Item = Self::Response, Error = Self::Error>>;
@@ -152,11 +167,11 @@ impl Decoder for JsonCodec {
 }
 
 impl Encoder for JsonCodec {
-    type Item = Runnable;
+    type Item = serde_json::Value;
     type Error = io::Error;
 
-    fn encode(&mut self, provider: Self::Item, buf: &mut BytesMut) -> io::Result<()> {
-        let json = serde_json::to_string(&provider).unwrap();
+    fn encode(&mut self, value: Self::Item, buf: &mut BytesMut) -> io::Result<()> {
+        let json = serde_json::to_string(&value).unwrap();
         buf.reserve(json.len() + 1);
         buf.extend(json.as_bytes());
         buf.put_u8(b'\n');
@@ -166,7 +181,18 @@ impl Encoder for JsonCodec {
 }
 
 impl<T: AsyncRead + AsyncWrite + 'static> ClientProto<T> for JsonProto {
-    type Request = Runnable;
+    type Request = serde_json::Value;
+    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))
+    }
+}
+
+impl<T: AsyncRead + AsyncWrite + 'static> ServerProto<T> for JsonProto {
+    type Request = serde_json::Value;
     type Response = serde_json::Value;
     type Transport = Framed<T, JsonCodec>;
     type BindTransport = result::Result<Self::Transport, io::Error>;

core/src/lib.rs

@@ -17,6 +17,7 @@ extern crate erased_serde;
 extern crate futures;
 extern crate hostname;
 extern crate ipnetwork;
+#[macro_use] extern crate log;
 extern crate pnet;
 extern crate regex;
 extern crate serde;