[api] use rain_core in rust client

Author: Kobzol

Cargo.lock

@@ -3,12 +3,15 @@ version = "0.3.0"
 [workspace]
 
 members = [
+    "rain_client",
+    "rain_client_test",
     "rain_core",
     "rain_server",
     "rain_task",
     "rain_task_test",
 ]
 
 [patch.crates-io]
+rain_client = { path = "rain_client" }
 rain_core = { path = "rain_core" }
 rain_task = { path = "rain_task" }

Cargo.toml

@@ -0,0 +1,31 @@
+[package]
+name = "rain_client"
+version = "0.3.0"
+
+description = "Distributed computational framework for large-scale task-based pipelines. Client library in Rust."
+# documentation = "https://docs.rs/rain_task/" # default docs.rs
+homepage = "https://github.com/substantic/rain"
+repository = "https://github.com/substantic/rain/"
+readme = "README.md"
+authors = [
+    "Stanislav Bohm <[email protected]>",
+    "Tomas Gavenciak <[email protected]>",
+    "Vojtech Cima <[email protected]>",
+    ]
+license = "MIT"
+
+exclude = ["testing/**/*"]
+
+[badges]
+travis-ci = { repository = "substantic/rain", branch = "master" }
+maintenance = { status = "actively-developed" }
+
+[dependencies]
+capnp-rpc = "0.8"
+error-chain = "0.11"
+futures = "0.1"
+log = "0.4"
+rain_core = "0.3.0"
+tokio-core = "0.1"
+serde = "1"
+serde_json = "1"

rain_client/Cargo.toml

@@ -1,9 +1,9 @@
 use std::error::Error;
 use std::net::SocketAddr;
 
-use CLIENT_PROTOCOL_VERSION;
-use super::session::Session;
 use super::communicator::Communicator;
+use super::session::Session;
+use rain_core::CLIENT_PROTOCOL_VERSION;
 use std::rc::Rc;
 
 pub struct Client {

src/client/client.rs renamed to rain_client/src/client/client.rs

@@ -1,21 +1,21 @@
-use tokio_core::reactor::Core;
-use std::net::SocketAddr;
-use tokio_core::net::TcpStream;
-use std::error::Error;
-use common::rpc::new_rpc_system;
 use capnp_rpc::rpc_twoparty_capnp;
 use futures::Future;
+use rain_core::comm::new_rpc_system;
+use std::error::Error;
+use std::net::SocketAddr;
+use tokio_core::net::TcpStream;
+use tokio_core::reactor::Core;
 
 use super::task::Task;
-use common::id::{DataObjectId, TaskId};
-use common::convert::{FromCapnp, ToCapnp};
 use client::dataobject::DataObject;
-use std::cell::RefCell;
-use std::cell::RefMut;
+use rain_core::types::{DataObjectId, TaskId};
+use rain_core::utils::{FromCapnp, ToCapnp};
+use rain_core::{client_capnp, common_capnp, server_capnp};
+use std::cell::{RefCell, RefMut};
 
 pub struct Communicator {
     core: RefCell<Core>,
-    service: ::client_capnp::client_service::Client,
+    service: client_capnp::client_service::Client,
 }
 
 impl Communicator {
@@ -28,7 +28,7 @@ impl Communicator {
         debug!("Connection to server {} established", scheduler);
 
         let mut rpc = Box::new(new_rpc_system(stream, None));
-        let bootstrap: ::server_capnp::server_bootstrap::Client =
+        let bootstrap: server_capnp::server_bootstrap::Client =
             rpc.bootstrap(rpc_twoparty_capnp::Side::Server);
         handle.spawn(rpc.map_err(|err| panic!("RPC error: {}", err)));
 
@@ -116,19 +116,28 @@ impl Communicator {
         ))
     }
 
-    pub fn fetch(&self, object_id: DataObjectId) -> Result<Vec<u8>, Box<Error>> {
+    pub fn fetch(&self, object_id: &DataObjectId) -> Result<Vec<u8>, Box<Error>> {
         let mut req = self.service.fetch_request();
         object_id.to_capnp(&mut req.get().get_id().unwrap());
         req.get().set_size(1024);
 
         let response = self.comm().run(req.send().promise)?;
 
+        // TODO: handle error states
         let reader = response.get()?;
         match reader.get_status().which()? {
-            ::common_capnp::fetch_result::status::Ok(()) => {
+            common_capnp::fetch_result::status::Ok(()) => {
                 let data = reader.get_data()?;
                 Ok(Vec::from(data))
             }
+            common_capnp::fetch_result::status::Removed(()) => {
+                print!("Removed");
+                Ok(vec![])
+            }
+            common_capnp::fetch_result::status::Error(err) => {
+                print!("Error: {:?}", err.unwrap().get_message());
+                Ok(vec![])
+            }
             _ => bail!("Non-ok status"),
         }
     }

src/client/communicator.rs renamed to rain_client/src/client/communicator.rs

@@ -1,19 +1,17 @@
-use common::Attributes;
-use common::id::DataObjectId;
-use common::DataType;
+use rain_core::types::{DataObjectId, ObjectSpec};
 use std::cell::Cell;
 
 pub struct DataObject {
-    pub id: DataObjectId,
-    pub label: String,
     pub keep: Cell<bool>,
     pub data: Option<Vec<u8>>,
-    pub attributes: Attributes,
-    pub data_type: DataType,
+    pub spec: ObjectSpec,
 }
 
 impl DataObject {
     pub fn keep(&self) {
         self.keep.set(true);
     }
+    pub fn id(&self) -> DataObjectId {
+        self.spec.id
+    }
 }

src/client/dataobject.rs renamed to rain_client/src/client/dataobject.rs

@@ -0,0 +1,46 @@
+use super::task::Task;
+use client::dataobject::DataObject;
+use rain_core::client_capnp;
+use rain_core::utils::ToCapnp;
+use serde_json;
+
+macro_rules! to_capnp_list {
+    ($builder:expr, $items:expr, $name:ident) => {{
+        let mut builder = $builder.$name($items.len() as u32);
+        for (i, obj) in $items.iter().enumerate() {
+            obj.to_capnp(&mut builder.reborrow().get(i as u32));
+        }
+    }};
+}
+macro_rules! from_capnp_list {
+    ($builder:expr, $items:ident, $obj:ident) => {{
+        $builder
+            .$items()?
+            .iter()
+            .map(|item| $obj::from_capnp(&item))
+            .collect()
+    }};
+}
+
+impl<'a> ToCapnp<'a> for Task {
+    type Builder = client_capnp::task::Builder<'a>;
+
+    fn to_capnp(&self, builder: &mut Self::Builder) {
+        builder.set_spec(&serde_json::to_string(&self.spec).unwrap());
+    }
+}
+impl<'a> ToCapnp<'a> for DataObject {
+    type Builder = client_capnp::data_object::Builder<'a>;
+
+    fn to_capnp(&self, builder: &mut Self::Builder) {
+        builder.set_spec(&serde_json::to_string(&self.spec).unwrap());
+        builder.set_keep(self.keep.get());
+
+        if let &Some(ref data) = &self.data {
+            builder.set_data(&data);
+            builder.set_has_data(true);
+        } else {
+            builder.set_has_data(false);
+        }
+    }
+}

src/client/mod.rs renamed to rain_client/src/client/mod.rs

@@ -1,15 +1,13 @@
 use super::communicator::Communicator;
 use client::dataobject::DataObject;
 use client::task::Task;
+use rain_core::common_capnp;
+use rain_core::types::{DataObjectId, DataType, ObjectSpec, Resources, SId, TaskId, TaskSpec,
+                       TaskSpecInput, UserAttrs};
+use serde_json;
+use std::cell::Cell;
 use std::collections::HashMap;
 use std::error::Error;
-use client::task::TaskInput;
-use common::Attributes;
-use common::id::TaskId;
-use common::id::DataObjectId;
-use common::id::SId;
-use common::DataType;
-use std::cell::Cell;
 use std::rc::Rc;
 
 pub type DataObjectPtr = Rc<DataObject>;
@@ -45,28 +43,36 @@ impl Session {
     }
 
     pub fn unkeep(&mut self, objects: &[DataObjectPtr]) -> Result<(), Box<Error>> {
-        self.comm
-            .unkeep(&objects.iter().map(|o| o.id).collect::<Vec<DataObjectId>>())
+        self.comm.unkeep(&objects
+            .iter()
+            .map(|o| o.id())
+            .collect::<Vec<DataObjectId>>())
     }
 
     pub fn wait(&mut self, tasks: &[TaskPtr], objects: &[DataObjectPtr]) -> Result<(), Box<Error>> {
         self.comm.wait(
-            &tasks.iter().map(|t| t.id).collect::<Vec<TaskId>>(),
-            &objects.iter().map(|o| o.id).collect::<Vec<DataObjectId>>(),
+            &tasks.iter().map(|t| t.id()).collect::<Vec<TaskId>>(),
+            &objects
+                .iter()
+                .map(|o| o.id())
+                .collect::<Vec<DataObjectId>>(),
         )
     }
     pub fn wait_some(
         &mut self,
         tasks: &[TaskPtr],
         objects: &[DataObjectPtr],
     ) -> Result<(Vec<TaskPtr>, Vec<DataObjectPtr>), Box<Error>> {
-        let task_map: HashMap<TaskId, &TaskPtr> = tasks.iter().map(|t| (t.id, t)).collect();
+        let task_map: HashMap<TaskId, &TaskPtr> = tasks.iter().map(|t| (t.id(), t)).collect();
         let object_map: HashMap<DataObjectId, &DataObjectPtr> =
-            objects.iter().map(|o| (o.id, o)).collect();
+            objects.iter().map(|o| (o.id(), o)).collect();
 
         let (task_ids, object_ids) = self.comm.wait_some(
-            &tasks.iter().map(|t| t.id).collect::<Vec<TaskId>>(),
-            &objects.iter().map(|o| o.id).collect::<Vec<DataObjectId>>(),
+            &tasks.iter().map(|t| t.id()).collect::<Vec<TaskId>>(),
+            &objects
+                .iter()
+                .map(|o| o.id())
+                .collect::<Vec<DataObjectId>>(),
         )?;
 
         Ok((
@@ -82,13 +88,13 @@ impl Session {
     }
     pub fn wait_all(&mut self) -> Result<(), Box<Error>> {
         self.comm.wait(
-            &vec![TaskId::new(self.id, ::common_capnp::ALL_TASKS_ID)],
+            &vec![TaskId::new(self.id, common_capnp::ALL_TASKS_ID)],
             &vec![],
         )
     }
 
-    pub fn fetch(&mut self, object: &DataObject) -> Result<Vec<u8>, Box<Error>> {
-        self.comm.fetch(object.id)
+    pub fn fetch(&mut self, o: &DataObjectPtr) -> Result<Vec<u8>, Box<Error>> {
+        self.comm.fetch(&o.id())
     }
 
     pub fn blob(&mut self, data: Vec<u8>) -> DataObjectPtr {
@@ -102,13 +108,17 @@ impl Session {
         DataObjectId::new(self.id, id)
     }
     pub(crate) fn create_object(&mut self, label: String, data: Option<Vec<u8>>) -> DataObjectPtr {
-        let object = DataObject {
+        let spec = ObjectSpec {
             id: self.create_object_id(),
-            keep: Cell::new(false),
             label,
-            data,
-            attributes: Attributes::new(),
             data_type: DataType::Blob,
+            content_type: "".to_owned(),
+            user: UserAttrs::new(),
+        };
+        let object = DataObject {
+            keep: Cell::new(false),
+            data,
+            spec,
         };
         let rc = Rc::new(object);
         self.data_objects.push(rc.clone());
@@ -124,27 +134,24 @@ impl Session {
     }
     pub fn create_task(
         &mut self,
-        command: String,
-        inputs: Vec<TaskInput>,
+        task_type: String,
+        inputs: Vec<TaskSpecInput>,
         outputs: Vec<DataObjectPtr>,
         config: HashMap<String, String>,
-        cpus: i32,
+        cpus: u32,
     ) -> TaskPtr {
-        let mut attributes = Attributes::new();
-        attributes.set("config", config).unwrap();
-
-        let mut resources: HashMap<String, i32> = HashMap::new();
-        resources.insert("cpus".to_owned(), cpus);
-        attributes.set("resources", resources).unwrap();
-
-        let task = Task {
+        let spec = TaskSpec {
             id: self.create_task_id(),
-            command,
             inputs,
-            outputs,
-            attributes,
+            task_type,
+            outputs: outputs.iter().map(|o| o.id()).collect(),
+            config: Some(serde_json::to_value(&config).unwrap()),
+            resources: Resources { cpus },
+            user: UserAttrs::new(),
         };
 
+        let task = Task { spec, outputs };
+
         let rc = Rc::new(task);
         self.tasks.push(rc.clone());