Update to futures 0.3 and async/await syntax (#20)

* Update to futures 0.3 and async/await syntax * Bump version and fix docs * Handle Ctrl-C interrupt * Use cargo+git for testing * Abort task when Ctrl-C interrupted * Use published golem-rpc-* crates

Author: Jakub Konka

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "gwasm-api"
-version = "0.1.1"
+version = "0.2.0"
 authors = ["Golem RnD Team <contact@golem.network>"]
 edition = "2018"
 license = "GPL-3.0"
@@ -11,23 +11,27 @@ documentation = "https://docs.rs/gwasm-rust-api"
 description = "gWasm API for Rust apps"
 
 [dependencies]
-actix = "0.8"
-thiserror = "1.0"
-futures = "0.1"
-actix-wamp = "0.1"
-golem-rpc-api = "0.1"
-golem-rpc-macros = "0.1"
+actix = "0.9"
+thiserror = "1"
+futures = "0.3"
+actix-wamp = "0.2"
+golem-rpc-api = "0.2"
+golem-rpc-macros = "0.2"
 serde_json = "1"
 serde = { version = "1", features = ["derive"] }
 chrono = "0.4"
-tempfile = "3.0"
-tokio-ctrlc-error = "0.1"
-tokio = "0.1"
+tempfile = "3"
+
+[dependencies.tokio]
+version = "0.2"
+features = [
+ "time",
+ "signal"
+]
 
 [dev-dependencies]
 indicatif = "0.11"
-tempfile = "3"
-anyhow = "1.0"
+anyhow = "1"
 
 [badges]
 maintenance = { status = "actively-developed" }

src/error.rs

@@ -1,7 +1,9 @@
 //! Errors that can be returned by the library
 use actix::MailboxError;
 use std::io;
-use tokio::timer;
+
+/// Convience wrapper for `Result` returned by fallible functions in the library
+pub type Result<T> = std::result::Result<T, Error>;
 
 /// Enum wrapping all possible errors that can be generated by the library
 #[derive(Debug, thiserror::Error)]
@@ -10,10 +12,6 @@ pub enum Error {
  #[error("internal gWasm API error: {0}")]
  MailboxError(#[from] MailboxError),
 
- /// Wraps tokio's `timer::Error` error
- #[error("internal gWasm API error: {0}")]
- TimerError(#[from] timer::Error),
-
  /// Wraps libstd's `std::io::Error` error
  #[error("internal gWasm API error: {0}")]
  IOError(#[from] io::Error),
@@ -26,19 +24,14 @@ pub enum Error {
  #[error("internal Golem error: {0}")]
  GolemRPCError(golem_rpc_api::Error),
 
- /// Wraps `tokio_ctrlc_error::KeyboardInterrupt` which is used to handle
- /// Ctrl-C interrupt event for the lib's client
- #[error("received Ctrl-C interrupt")]
- KeyboardInterrupt(tokio_ctrlc_error::KeyboardInterrupt),
-
- /// Wraps other `tokio_ctrlc_error::IoError` type errors
- #[error("internal gWasm API error: {0}")]
- CtrlcError(tokio_ctrlc_error::IoError),
-
  /// Wraps `chrono::ParseError` error
  #[error("error parsing Timeout value: {0}")]
  ChronoError(#[from] chrono::ParseError),
 
+ /// Received Ctrl-C interrupt
+ #[error("received Ctrl-C interrupt")]
+ KeyboardInterrupt,
+
  /// Error generated when trying to create a zero [`Timeout`](../timeout/struct.Timeout.html)
  /// value for a Golem Task
  #[error("zero timeout \"00:00:00\" is forbidden")]
@@ -73,15 +66,3 @@ impl From<golem_rpc_api::Error> for Error {
  Error::GolemRPCError(err)
  }
 }
-
-impl From<tokio_ctrlc_error::IoError> for Error {
- fn from(err: tokio_ctrlc_error::IoError) -> Self {
- Error::CtrlcError(err)
- }
-}
-
-impl From<tokio_ctrlc_error::KeyboardInterrupt> for Error {
- fn from(err: tokio_ctrlc_error::KeyboardInterrupt) -> Self {
- Error::KeyboardInterrupt(err)
- }
-}

src/golem.rs

@@ -1,26 +1,22 @@
 //! Convenience async functions for creating gWasm tasks, connecting to a
 //! Golem instance, and listening for task's progress as it's computed
 //! on Golem.
-use super::{
- error::Error,
- task::{ComputedTask, Task},
- Net, ProgressUpdate,
-};
+use super::error::{Error, Result};
+use super::task::{ComputedTask, Task};
+use super::{Net, ProgressUpdate};
 use actix::{Actor, ActorContext, Context, Handler, Message};
 use actix_wamp::RpcEndpoint;
-use futures::{
- future,
- stream::{self, Stream},
- Future,
-};
-use golem_rpc_api::{
- comp::{AsGolemComp, TaskStatus as GolemTaskStatus},
- connect_to_app,
-};
+use futures::future::FutureExt;
+use futures::stream::{self, Stream, StreamExt, TryStreamExt};
+use futures::{pin_mut, select};
+use golem_rpc_api::comp::{AsGolemComp, TaskStatus as GolemTaskStatus};
+use golem_rpc_api::connect_to_app;
 use serde_json::json;
-use std::{convert::TryInto, path::Path, time::Duration};
-use tokio::timer::Interval;
-use tokio_ctrlc_error::AsyncCtrlc;
+use std::convert::TryInto;
+use std::path::{Path, PathBuf};
+use std::pin::Pin;
+use std::time::Duration;
+use tokio::{signal, time};
 
 /// A convenience function for running a gWasm [`Task`] on Golem
 ///
@@ -34,46 +30,51 @@ use tokio_ctrlc_error::AsyncCtrlc;
 /// [`Task`]: ../task/struct.Task.html
 /// [`ComputedTask`]: ../task/struct.ComputedTask.html
 /// [`gwasm_api::compute`]: ../fn.compute.html
-pub fn compute<P, S>(
+pub async fn compute<P, S>(
  datadir: P,
  address: S,
  port: u16,
  task: Task,
  net: Net,
  progress_handler: impl ProgressUpdate + 'static,
  polling_interval: Option<Duration>,
-) -> impl Future<Item = ComputedTask, Error = Error> + 'static
+) -> Result<ComputedTask>
 where
- P: AsRef<Path>,
- S: AsRef<str>,
+ P: Into<PathBuf>,
+ S: Into<String>,
 {
- create_task(datadir.as_ref(), address.as_ref(), port, net, task.clone())
- .and_then(move |(endpoint, task_id)| {
- poll_task_progress(endpoint.clone(), task_id.clone(), polling_interval)
- .fold(
- ProgressActor::new(progress_handler).start(),
- |addr, task_status| {
- addr.send(Update {
- progress: task_status.progress,
- })
- .and_then(|_| Ok(addr))
- },
- )
- .and_then(|addr| addr.send(Finish).map_err(Error::from))
- .ctrlc_as_error()
- .or_else(move |e: Error| match e {
- Error::KeyboardInterrupt(e) => {
- future::Either::A(endpoint.as_golem_comp().abort_task(task_id).then(
- |res| match res {
- Ok(()) => future::err(Error::KeyboardInterrupt(e)),
- Err(e) => future::err(e.into()),
- },
- ))
- }
- e => future::Either::B(future::err(e)),
+ let (endpoint, task_id) =
+ create_task(&datadir.into(), &address.into(), port, net, task.clone()).await?;
+ let poll_stream = poll_task_progress(endpoint.clone(), task_id.clone(), polling_interval);
+ let progress = poll_stream
+ .try_fold(
+ ProgressActor::new(progress_handler).start(),
+ |addr, task_status| async move {
+ addr.send(Update {
+ progress: task_status.progress,
  })
- })
- .and_then(|()| task.try_into())
+ .await?;
+ Ok(addr)
+ },
+ )
+ .fuse();
+ let ctrlc = signal::ctrl_c().fuse();
+
+ pin_mut!(ctrlc, progress);
+
+ select! {
+ maybe_ctrlc = ctrlc => {
+ maybe_ctrlc?;
+ endpoint.as_golem_comp().abort_task(task_id).await?;
+ Err(Error::KeyboardInterrupt)
+ }
+ maybe_addr = progress => {
+ let addr = maybe_addr?;
+ addr.send(Finish).await?;
+ let task: ComputedTask = task.try_into()?;
+ Ok(task)
+ }
+ }
 }
 
 /// A convenience function for creating a gWasm [`Task`] on Golem
@@ -84,21 +85,16 @@ where
 /// [`Task`]: ../task/struct.Task.html
 /// [`RpcEndpoint`]:
 /// https://golemfactory.github.io/golem-client/latest/actix_wamp/trait.RpcEndpoint.html
-pub fn create_task(
+pub async fn create_task(
  datadir: &Path,
  address: &str,
  port: u16,
  net: Net,
  task: Task,
-) -> impl Future<Item = (impl Clone + Send + RpcEndpoint, String), Error = Error> + 'static {
- connect_to_app(datadir, Some(net), Some((address, port)))
- .and_then(move |endpoint| {
- endpoint
- .as_golem_comp()
- .create_task(json!(task))
- .map(|task_id| (endpoint, task_id))
- })
- .from_err()
+) -> Result<(impl Clone + Send + RpcEndpoint, String)> {
+ let endpoint = connect_to_app(datadir, Some(net), Some((address, port))).await?;
+ let task_id = endpoint.as_golem_comp().create_task(json!(task)).await?;
+ Ok((endpoint, task_id))
 }
 
 /// A convenience function for polling gWasm [`Task`]'s computation progress on Golem
@@ -114,72 +110,60 @@ pub fn poll_task_progress(
  endpoint: impl Clone + Send + RpcEndpoint + 'static,
  task_id: String,
  polling_interval: Option<Duration>,
-) -> impl Stream<Item = TaskStatus, Error = Error> + 'static {
- stream::unfold(TaskState::new(endpoint, task_id), |state| {
+) -> impl Stream<Item = Result<TaskStatus>> {
+ stream::try_unfold(TaskState::new(endpoint, task_id), |state| async move {
  if let Some(status) = state.task_status.status {
  match status {
- GolemTaskStatus::Finished => return None,
- GolemTaskStatus::Aborted => {
- return Some(future::Either::A(future::err(Error::TaskAborted)))
- }
- GolemTaskStatus::Timeout => {
- return Some(future::Either::A(future::err(Error::TaskTimedOut)))
- }
+ GolemTaskStatus::Finished => return Ok(None),
+ GolemTaskStatus::Aborted => return Err(Error::TaskAborted),
+ GolemTaskStatus::Timeout => return Err(Error::TaskTimedOut),
  _ => {}
  }
  }
 
  let mut next_state = TaskState::new(state.endpoint.clone(), state.task_id.clone());
- Some(future::Either::B(
- state
- .endpoint
- .as_golem_comp()
- .get_task(state.task_id.clone())
- .map_err(Error::from)
- .and_then(move |task_info| {
- let task_info = task_info.ok_or(Error::EmptyTaskInfo)?;
- next_state.task_status.status = Some(task_info.status);
- next_state.task_status.progress =
- task_info.progress.ok_or(Error::EmptyProgress)?;
- Ok((next_state.task_status.clone(), next_state))
- }),
- ))
+ let task_info = state
+ .endpoint
+ .as_golem_comp()
+ .get_task(state.task_id.clone())
+ .await?;
+ let task_info = task_info.ok_or(Error::EmptyTaskInfo)?;
+ next_state.task_status.status = Some(task_info.status);
+ next_state.task_status.progress = task_info.progress.ok_or(Error::EmptyProgress)?;
+ Ok(Some((next_state.task_status.clone(), next_state)))
  })
- .zip(
- Interval::new_interval(polling_interval.unwrap_or_else(|| Duration::from_secs(2)))
- .from_err(),
- )
+ .zip(time::interval(
+ polling_interval.unwrap_or_else(|| Duration::from_secs(2)),
+ ))
  .map(|(x, _)| x)
 }
 
-#[derive(Message)]
 struct Update {
  progress: f64,
 }
 
-#[derive(Message)]
+impl Message for Update {
+ type Result = ();
+}
+
 struct Finish;
 
-struct ProgressActor<T>
-where
- T: ProgressUpdate + 'static,
-{
- handler: T,
+impl Message for Finish {
+ type Result = ();
 }
 
-impl<T> ProgressActor<T>
-where
- T: ProgressUpdate + 'static,
-{
- fn new(handler: T) -> Self {
+struct ProgressActor {
+ handler: Pin<Box<dyn ProgressUpdate>>,
+}
+
+impl ProgressActor {
+ fn new<T: ProgressUpdate + 'static>(handler: T) -> Self {
+ let handler = Box::pin(handler);
  Self { handler }
  }
 }
 
-impl<T> Actor for ProgressActor<T>
-where
- T: ProgressUpdate + 'static,
-{
+impl Actor for ProgressActor {
  type Context = Context<Self>;
 
  fn started(&mut self, _ctx: &mut Self::Context) {
@@ -191,21 +175,15 @@ where
  }
 }
 
-impl<T> Handler<Update> for ProgressActor<T>
-where
- T: ProgressUpdate + 'static,
-{
+impl Handler<Update> for ProgressActor {
  type Result = ();
 
  fn handle(&mut self, msg: Update, _ctx: &mut Self::Context) -> Self::Result {
  self.handler.update(msg.progress);
  }
 }
 
-impl<T> Handler<Finish> for ProgressActor<T>
-where
- T: ProgressUpdate + 'static,
-{
+impl Handler<Finish> for ProgressActor {
  type Result = ();
 
  fn handle(&mut self, _msg: Finish, ctx: &mut Self::Context) -> Self::Result {