Refactored jsonrpc to use futures 0.3 + async/await

.gitlab-ci.yml

@@ -1,4 +1,4 @@
-image: parity/rust:nightly-20190315
+image: parity/rust:a811bb14-20190522 
 
 variables:
     CI_SERVER_NAME:  "GitLab CI"
@@ -26,7 +26,7 @@ variables:
 
 .install_protos: &install_protos
     sudo apt-get install -y unzip &&
-    wget -O /tmp/protoc.zip https://github.com/protocolbuffers/protobuf/releases/download/v3.6.1/protoc-3.6.1-linux-x86_64.zip &&
+    curl -Lo /tmp/protoc.zip https://github.com/protocolbuffers/protobuf/releases/download/v3.6.1/protoc-3.6.1-linux-x86_64.zip &&
     sudo unzip /tmp/protoc.zip -d protoc &&
     sudo mv protoc/bin/* /usr/local/bin/ &&
     sudo chmod 755 /usr/local/bin/protoc

Cargo.toml

@@ -8,6 +8,7 @@ edition = "2018"
 
 [workspace]
 members = [
+    "async-utils/",
     "protos/builder",
     "core/primitives",
     "core/store",

async-utils/Cargo.toml

@@ -0,0 +1,9 @@
+[package]
+name = "async-utils"
+version = "0.1.0"
+authors = ["Near Inc <hello@nearprotocol.com>"]
+edition = "2018"
+
+[dependencies]
+futures03 = { package = "futures-preview", version = "0.3.0-alpha.16", features = ["compat", "async-await", "nightly"] }
+tokio = "0.1.15"

async-utils/src/lib.rs

@@ -0,0 +1,49 @@
+#![feature(await_macro, async_await)]
+
+use std::time::{Duration, Instant};
+
+use futures03::{compat::Future01CompatExt as _, FutureExt as _};
+
+/// This macro is extracted from
+/// https://github.com/rust-lang-nursery/futures-rs/blob/c30adf513b9eea35ab385c0797210c77986fc82f/futures/src/lib.rs#L503-L510
+///
+/// It is useful when the `futures-preview` is imported as `futures03`.
+macro_rules! select { // replace `::futures_util` with `::futures03` as the crate path
+    ($($tokens:tt)*) => {
+        futures03::inner_select::select! {
+            futures_crate_path ( ::futures03 )
+            $( $tokens )*
+        }
+    }
+}
+
+/// An async/await helper to delay the execution:
+///
+/// ```ignore
+/// let _ = delay(std::time::Duration::from_secs(1)).await;
+/// ```
+pub async fn delay(duration: Duration) -> Result<(), tokio::timer::Error> {
+    tokio::timer::Delay::new(Instant::now() + duration).compat().await
+}
+
+pub struct TimeoutError;
+
+/// An async/await helper to timeout a given async context:
+///
+/// ```ignore
+/// timeout(
+///     std::time::Duration::from_secs(1),
+///     async {
+///         let _ = delay(std::time::Duration::from_secs(2)).await;
+///     }
+/// ).await
+/// ```
+pub async fn timeout<T, Fut>(timeout: Duration, f: Fut) -> Result<T, TimeoutError>
+where
+    Fut: futures03::Future<Output = T> + Send,
+{
+    select! {
+        result = f.boxed().fuse() => Ok(result),
+        _ = delay(timeout).boxed().fuse() => Err(TimeoutError {})
+    }
+}

chain/chain/src/types.rs

@@ -305,7 +305,7 @@ pub type ReceiptResult = HashMap<ShardId, Vec<ReceiptTransaction>>;
 /// Bridge between the chain and the runtime.
 /// Main function is to update state given transactions.
 /// Additionally handles authorities and block weight computation.
-pub trait RuntimeAdapter {
+pub trait RuntimeAdapter : Send + Sync {
     /// Initialize state to genesis state and returns StoreUpdate and state root.
     /// StoreUpdate can be discarded if the chain past the genesis.
     fn genesis_state(&self, shard_id: ShardId) -> (StoreUpdate, MerkleHash);
@@ -453,4 +453,4 @@ mod tests {
         assert!(signer.verify(b2.hash().as_ref(), &b2.header.signature));
         assert_eq!(b2.header.total_weight.to_num(), 3);
     }
-}
+}

chain/client/src/client.rs

@@ -47,7 +47,7 @@ pub struct ClientActor {
     config: ClientConfig,
     sync_status: SyncStatus,
     chain: Chain,
-    runtime_adapter: Arc<RuntimeAdapter>,
+    runtime_adapter: Arc<dyn RuntimeAdapter>,
     tx_pool: TransactionPool,
     network_actor: Recipient<NetworkRequests>,
     block_producer: Option<BlockProducer>,
@@ -72,7 +72,7 @@ impl ClientActor {
         config: ClientConfig,
         store: Arc<Store>,
         genesis_time: DateTime<Utc>,
-        runtime_adapter: Arc<RuntimeAdapter>,
+        runtime_adapter: Arc<dyn RuntimeAdapter>,
         network_actor: Recipient<NetworkRequests>,
         block_producer: Option<BlockProducer>,
     ) -> Result<Self, Error> {

chain/client/src/view_client.rs

@@ -21,14 +21,14 @@ use crate::TxDetails;
 /// View client provides currently committed (to the storage) view of the current chain and state.
 pub struct ViewClientActor {
     chain: Chain,
-    runtime_adapter: Arc<RuntimeAdapter>,
+    runtime_adapter: Arc<dyn RuntimeAdapter>,
 }
 
 impl ViewClientActor {
     pub fn new(
         store: Arc<Store>,
         genesis_time: DateTime<Utc>,
-        runtime_adapter: Arc<RuntimeAdapter>,
+        runtime_adapter: Arc<dyn RuntimeAdapter>,
     ) -> Result<Self, Error> {
         // TODO: should we create shared ChainStore that is passed to both Client and ViewClient?
         let chain = Chain::new(store, runtime_adapter.clone(), genesis_time)?;

chain/jsonrpc/Cargo.toml

@@ -7,9 +7,11 @@ edition = "2018"
 [dependencies]
 ansi_term = "0.11.0"
 actix = "0.8.1"
-actix-web = "1.0.0-beta.3"
+actix-web = "1.0.0-rc"
+base64 = "0.10.0"
 bytes = "0.4.11"
 futures = "0.1"
+futures03 = { package = "futures-preview", version = "0.3.0-alpha.16", features = ["compat", "async-await", "nightly"] }
 chrono = { version = "0.4.4", features = ["serde"] }
 log = "0.4"
 serde_derive = "1.0"
@@ -19,6 +21,7 @@ tokio = "0.1.15"
 uuid = { version = "~0.6", features = ["v4"] }
 protobuf = "2.4"
 
+async-utils = { path = "../../async-utils" }
 near-primitives = { path = "../../core/primitives" }
 near-protos = { path = "../../core/protos" }
 near-store = { path = "../../core/store" }