Merge 4485e8f into ce8ab15

Author: sanlee42

Cargo.lock

@@ -20,7 +20,7 @@ starcoin-types = {path= "../../types" }
 starcoin-rpc-client = { path= "../../rpc/client" }
 starcoin-rpc-api = {path = "../../rpc/api" }
 starcoin-logger = {path = "../../commons/logger"}
-jsonrpc-core-client="~15"
+jsonrpc-core-client="~17"
 tokio={version="0.2", features=["full"]}
 tokio-compat = "0.1.6"
 futures-util = "~0.3"

cmd/indexer/Cargo.toml

@@ -1,6 +1,5 @@
 use crate::{BlockData, TransactionData};
 use anyhow::Result;
-use futures_util::compat::Future01CompatExt;
 use jsonrpc_core_client::RpcError;
 use starcoin_rpc_api::chain::ChainClient;
 use starcoin_rpc_api::types::{
@@ -19,21 +18,15 @@ impl BlockClient {
         }
     }
     pub async fn get_block_whole_by_height(&self, height: u64) -> Result<BlockData, RpcError> {
-        let block: Option<BlockView> = self
-            .node_client
-            .get_block_by_number(height)
-            .compat()
-            .await?;
+        let block: Option<BlockView> = self.node_client.get_block_by_number(height).await?;
         let block = block.ok_or_else(|| {
-            RpcError::Other(failure::err_msg(format!(
-                "cannot find block of height {}",
-                height
-            )))
+            RpcError::Other(Box::new(
+                failure::err_msg(format!("cannot find block of height {}", height)).compat(),
+            ))
         })?;
         let mut txn_infos: Vec<TransactionInfoView> = self
             .node_client
             .get_block_txn_infos(block.header.block_hash)
-            .compat()
             .await?;
         let mut txns_data = vec![];
 
@@ -42,19 +35,20 @@ impl BlockClient {
             let txn: Option<TransactionView> = self
                 .node_client
                 .get_transaction(txn_info.transaction_hash)
-                .compat()
                 .await?;
             let txn = txn.ok_or_else(|| {
-                RpcError::Other(failure::err_msg(format!(
-                    "cannot find txn with id {}",
-                    txn_info.transaction_hash
-                )))
+                RpcError::Other(Box::new(
+                    failure::err_msg(format!(
+                        "cannot find txn with id {}",
+                        txn_info.transaction_hash
+                    ))
+                    .compat(),
+                ))
             })?;
 
             let events: Vec<TransactionEventView> = self
                 .node_client
                 .get_events_by_txn_hash(txn_info.transaction_hash)
-                .compat()
                 .await?;
             txns_data.push(TransactionData {
                 info: txn_info,
@@ -71,7 +65,7 @@ impl BlockClient {
         let fetch_events_tasks = txn_infos
             .iter()
             .map(|txn_info| txn_info.transaction_hash)
-            .map(|txn_hash| self.node_client.get_events_by_txn_hash(txn_hash).compat());
+            .map(|txn_hash| self.node_client.get_events_by_txn_hash(txn_hash));
 
         let events = futures_util::future::try_join_all(fetch_events_tasks).await?;
 
@@ -89,7 +83,7 @@ impl BlockClient {
         Ok(BlockData { block, txns_data })
     }
     pub async fn get_chain_head(&self) -> Result<BlockHeaderView, RpcError> {
-        let chain_info: ChainInfoView = self.node_client.info().compat().await?;
+        let chain_info: ChainInfoView = self.node_client.info().await?;
         Ok(chain_info.head)
     }
 }

cmd/indexer/src/block_client.rs

@@ -1,18 +1,17 @@
-use anyhow::Result;
+use anyhow::{anyhow, Result};
 use clap::Clap;
 use elasticsearch::auth::Credentials;
 use elasticsearch::http::transport::SingleNodeConnectionPool;
 use elasticsearch::http::Url;
 use elasticsearch::Elasticsearch;
-use failure::Fail;
 use futures_retry::{FutureRetry, RetryPolicy};
 use futures_util::TryFutureExt;
 use jsonrpc_core_client::transports::http;
 use starcoin_indexer::{BlockClient, BlockData, EsSinker, IndexConfig};
 use starcoin_logger::prelude::*;
 use starcoin_rpc_api::chain::ChainClient;
 use std::time::Duration;
-use tokio_compat::runtime;
+use tokio::runtime;
 
 #[derive(Clap, Debug, Clone)]
 #[clap(version = "0.1.0", author = "Starcoin Core Dev <dev@starcoin.org>")]
@@ -38,7 +37,7 @@ async fn start_loop(block_client: BlockClient, sinker: EsSinker) -> Result<()> {
 
     loop {
         let remote_tip_header = FutureRetry::new(
-            || block_client.get_chain_head().map_err(|e| e.compat()),
+            || block_client.get_chain_head().map_err(|e| e),
             |e| {
                 warn!("[Retry]: get chain head, err: {}", &e);
                 RetryPolicy::<anyhow::Error>::WaitRetry(Duration::from_secs(1))
@@ -68,9 +67,8 @@ async fn start_loop(block_client: BlockClient, sinker: EsSinker) -> Result<()> {
         } else {
             let next_block: BlockData = FutureRetry::new(
                 || {
-                    block_client
-                        .get_block_whole_by_height(next_block_number)
-                        .map_err(|e| e.compat())
+                    block_client.get_block_whole_by_height(next_block_number)
+                    //.map_err(|e| e.compat())
                 },
                 |e| {
                     warn!("[Retry]: get chain block data, err: {}", &e);
@@ -125,11 +123,13 @@ fn main() -> anyhow::Result<()> {
     let opts: Options = Options::parse();
     info!("opts: {:?}", &opts);
     let mut rt = runtime::Builder::new()
-        .name_prefix("starcoin-indexer")
+        .thread_name("starcoin-indexer")
+        .threaded_scheduler()
+        .enable_all()
         .build()?;
     let channel: ChainClient = rt
         .block_on(http::connect(opts.node_url.as_str()))
-        .map_err(|e| e.compat())?;
+        .map_err(|e| anyhow!(format!("{}", e)))?;
     let block_client = BlockClient::new(channel);
     let mut transport = elasticsearch::http::transport::TransportBuilder::new(
         SingleNodeConnectionPool::new(opts.es_url),
@@ -145,7 +145,7 @@ fn main() -> anyhow::Result<()> {
     let index_config = IndexConfig::new_with_prefix(opts.es_index_prefix.as_str());
     let sinker = EsSinker::new(es, index_config);
 
-    rt.block_on_std(start_loop(block_client, sinker))?;
+    rt.block_on(start_loop(block_client, sinker))?;
 
     Ok(())
 }

cmd/indexer/src/main.rs

@@ -18,8 +18,8 @@ tokio = { version = "0.2", features = ["full"] }
 logger = { path = "../../commons/logger", package = "starcoin-logger" }
 futures-channel = "0.3"
 crypto = { package = "starcoin-crypto", path = "../../commons/crypto" }
-jsonrpc-core = { version = "15.1.0", features = ["arbitrary_precision"] }
-jsonrpc-tcp-server = "15.1.0"
+jsonrpc-core = { version = "17.0.0", features = ["arbitrary_precision"] }
+jsonrpc-tcp-server = "17.0.0"
 serde_json = { version = "1.0", features = ["arbitrary_precision"] }
 serde = { version = "1.0", features = ["derive"] }
 rust-argon2 = "0.8"

cmd/miner_client/Cargo.toml

@@ -15,11 +15,11 @@ hex = "0.4.2"
 async-trait = "0.1"
 actix = "0.10.0"
 actix-rt = "1.1"
-jsonrpc-core = { version = "15.1.0", features = ["arbitrary_precision"] }
-jsonrpc-derive = "15.1.0"
-jsonrpc-server-utils = "15.1.0"
-jsonrpc-pubsub = "15.1.0"
-jsonrpc-core-client = { version = "15.1.0", features = ["http", "ipc", "ws", "arbitrary_precision"]}
+jsonrpc-core = { version = "17.0.0", features = ["arbitrary_precision"] }
+jsonrpc-derive = "17.0.0"
+jsonrpc-server-utils = "17.0.0"
+jsonrpc-pubsub = "17.0.0"
+jsonrpc-core-client = { version = "17.0.0", features = ["http", "ipc", "ws", "arbitrary_precision"]}
 futures = { version = "0.3", features = ["compat"] }
 scs = { package="starcoin-canonical-serialization", path = "../../commons/scs"}
 starcoin-rpc-middleware = { path = "../middleware" }

rpc/api/Cargo.toml

@@ -1,9 +1,10 @@
 // Copyright (c) The Starcoin Core Contributors
 // SPDX-License-Identifier: Apache-2
 
-use jsonrpc_core::Error;
+use jsonrpc_core::{BoxFuture, Error};
 
-pub type FutureResult<T> = Box<dyn jsonrpc_core::futures::Future<Item = T, Error = Error> + Send>;
+pub type FutureResult<T> = BoxFuture<Result<T, Error>>;
+//pub type FutureResult<T> = Pin<Box<dyn futures::Future<Output = Result<T, Error>> + Send>>;
 
 pub mod account;
 pub mod chain;

rpc/api/src/lib.rs

@@ -22,12 +22,12 @@ actix-rt = "1.1"
 tokio = "0.2"
 log = "0.4.13"
 parking_lot = "0.11.1"
-jsonrpc-core = { version = "15.1.0", features = ["arbitrary_precision"] }
-jsonrpc-derive = "15.1.0"
-jsonrpc-server-utils = "15.1.0"
-jsonrpc-pubsub = "15.1.0"
-jsonrpc-core-client = { version = "15.1.0", features = ["http", "ipc", "ws", "arbitrary_precision"]}
-jsonrpc-client-transports = { version = "15.1.0", features = ["http", "ipc", "ws", "arbitrary_precision"] }
+jsonrpc-core = { version = "17.0.0", features = ["arbitrary_precision"] }
+jsonrpc-derive = "17.0.0"
+jsonrpc-server-utils = "17.0.0"
+jsonrpc-pubsub = "17.0.0"
+jsonrpc-core-client = { version = "17.0.0", features = ["http", "ipc", "ws", "arbitrary_precision"]}
+jsonrpc-client-transports = { version = "17.0.0", features = ["http", "ipc", "ws", "arbitrary_precision"] }
 futures03 = { package="futures",version = "0.3", features = ["compat"] }
 futures = "0.1.29"
 parity-tokio-ipc = { version = "0.2"}

rpc/client/Cargo.toml

@@ -5,7 +5,6 @@ use crate::pubsub_client::PubSubClient;
 use actix::prelude::*;
 use actix::AsyncContext;
 use futures03::channel::oneshot;
-use futures03::compat::Stream01CompatExt;
 use jsonrpc_core_client::RpcError;
 use serde::{Deserialize, Serialize};
 use starcoin_crypto::HashValue;
@@ -54,7 +53,7 @@ impl ChainWatcher {
             .then(|res, act, ctx| {
                 match res {
                     Ok(s) => {
-                        ctx.add_stream(s.compat());
+                        ctx.add_stream(s);
                     }
                     Err(e) => {
                         // TODO: figure out why this error cannot printed.