Switch the client to new futures (#3103)

* Switch the client to new futures * No need for compat in the client * Fix client tests * Address review
paritytech · Jul 11, 2019 · 9505713 · 9505713
1 parent 98d866d
commit 9505713
Show file tree

Hide file tree

Showing 28 changed files with 249 additions and 112 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/core/cli/Cargo.toml b/core/cli/Cargo.toml
@@ -18,6 +18,7 @@ lazy_static = "1.3"
 app_dirs = "1.2"
 tokio = "0.1.7"
 futures = "0.1.17"
+futures03 = { package = "futures-preview", version = "0.3.0-alpha.17", features = ["compat"] }
 fdlimit = "0.1"
 exit-future = "0.1"
 serde_json = "1.0"

diff --git a/core/cli/src/informant.rs b/core/cli/src/informant.rs
@@ -20,6 +20,7 @@ use ansi_term::Colour;
 use std::fmt;
 use std::time;
 use futures::{Future, Stream};
+use futures03::{StreamExt as _, TryStreamExt as _};
 use service::{Service, Components};
 use tokio::runtime::TaskExecutor;
 use network::SyncState;
@@ -81,7 +82,7 @@ where C: Components {
 		Some((info.chain.best_number, info.chain.best_hash))
 	};
 
-	let display_block_import = client.import_notification_stream().for_each(move |n| {
+	let display_block_import = client.import_notification_stream().map(|v| Ok::<_, ()>(v)).compat().for_each(move |n| {
 		// detect and log reorganizations.
 		if let Some((ref last_num, ref last_hash)) = last {
 			if n.header.parent_hash() != last_hash {

diff --git a/core/client/Cargo.toml b/core/client/Cargo.toml
@@ -10,7 +10,7 @@ fnv = { version = "1.0", optional = true }
 log = { version = "0.4", optional = true }
 parking_lot = { version = "0.8.0", optional = true }
 hex = { package = "hex-literal", version = "0.1", optional = true }
-futures = { version = "0.1.17", optional = true }
+futures-preview = { version = "0.3.0-alpha.17", optional = true }
 consensus = { package = "substrate-consensus-common", path = "../consensus/common", optional = true }
 executor = { package = "substrate-executor", path = "../executor", optional = true }
 state-machine = { package = "substrate-state-machine", path = "../state-machine", optional = true }
@@ -47,7 +47,7 @@ std = [
 	"fnv",
 	"log",
 	"hex",
-	"futures",
+	"futures-preview",
 	"executor",
 	"state-machine",
 	"keyring",

diff --git a/core/client/src/client.rs b/core/client/src/client.rs
@@ -21,7 +21,7 @@ use std::{
 	panic::UnwindSafe, result, cell::RefCell, rc::Rc,
 };
 use crate::error::Error;
-use futures::sync::mpsc;
+use futures::channel::mpsc;
 use parking_lot::{Mutex, RwLock};
 use primitives::NativeOrEncoded;
 use runtime_primitives::{

diff --git a/core/client/src/lib.rs b/core/client/src/lib.rs
@@ -58,7 +58,7 @@ pub use crate::client::{
 	new_with_backend,
 	new_in_mem,
 	BlockBody, BlockStatus, ImportNotifications, FinalityNotifications, BlockchainEvents,
-	BlockImportNotification, Client, ClientInfo, ExecutionStrategies,
+	BlockImportNotification, Client, ClientInfo, ExecutionStrategies, FinalityNotification,
 	LongestChain,
 };
 #[cfg(feature = "std")]

diff --git a/core/client/src/light/backend.rs b/core/client/src/light/backend.rs
@@ -19,7 +19,6 @@
 
 use std::collections::HashMap;
 use std::sync::{Arc, Weak};
-use futures::{Future, IntoFuture};
 use parking_lot::{RwLock, Mutex};
 
 use runtime_primitives::{generic::BlockId, Justification, StorageOverlay, ChildrenStorageOverlay};
@@ -359,14 +358,15 @@ where
 			*self.cached_header.write() = Some(cached_header);
 		}
 
-		self.fetcher.upgrade().ok_or(ClientError::NotAvailableOnLightClient)?
-			.remote_read(RemoteReadRequest {
-				block: self.block,
-				header: header.expect("if block above guarantees that header is_some(); qed"),
-				key: key.to_vec(),
-				retry_count: None,
-			})
-			.into_future().wait()
+		futures::executor::block_on(
+			self.fetcher.upgrade().ok_or(ClientError::NotAvailableOnLightClient)?
+				.remote_read(RemoteReadRequest {
+					block: self.block,
+					header: header.expect("if block above guarantees that header is_some(); qed"),
+					key: key.to_vec(),
+					retry_count: None,
+				})
+		)
 	}
 
 	fn child_storage(&self, _storage_key: &[u8], _key: &[u8]) -> ClientResult<Option<Vec<u8>>> {

diff --git a/core/client/src/light/blockchain.rs b/core/client/src/light/blockchain.rs
@@ -18,7 +18,6 @@
 //! blocks. CHT roots are stored for headers of ancient blocks.
 
 use std::{sync::{Weak, Arc}, collections::HashMap};
-use futures::{Future, IntoFuture};
 use parking_lot::Mutex;
 
 use runtime_primitives::{Justification, generic::BlockId};
@@ -122,14 +121,15 @@ impl<S, F, Block> BlockchainHeaderBackend<Block> for Blockchain<S, F> where Bloc
 					return Ok(None);
 				}
 
-				self.fetcher().upgrade().ok_or(ClientError::NotAvailableOnLightClient)?
-					.remote_header(RemoteHeaderRequest {
-						cht_root: self.storage.header_cht_root(cht::size(), number)?,
-						block: number,
-						retry_count: None,
+				futures::executor::block_on(
+					self.fetcher().upgrade()
+						.ok_or(ClientError::NotAvailableOnLightClient)?
+						.remote_header(RemoteHeaderRequest {
+							cht_root: self.storage.header_cht_root(cht::size(), number)?,
+							block: number,
+							retry_count: None,
 					})
-					.into_future().wait()
-					.map(Some)
+				).map(Some)
 			}
 		}
 	}
@@ -158,13 +158,13 @@ impl<S, F, Block> BlockchainBackend<Block> for Blockchain<S, F> where Block: Blo
 			None => return Ok(None),
 		};
 
-		self.fetcher().upgrade().ok_or(ClientError::NotAvailableOnLightClient)?
-			.remote_body(RemoteBodyRequest {
-				header,
-				retry_count: None,
-			})
-			.into_future().wait()
-			.map(Some)
+		futures::executor::block_on(
+			self.fetcher().upgrade().ok_or(ClientError::NotAvailableOnLightClient)?
+				.remote_body(RemoteBodyRequest {
+					header,
+					retry_count: None,
+				})
+		).map(Some)
 	}
 
 	fn justification(&self, _id: BlockId<Block>) -> ClientResult<Option<Justification>> {

diff --git a/core/client/src/light/call_executor.rs b/core/client/src/light/call_executor.rs
@@ -21,7 +21,6 @@ use std::{
 	collections::HashSet, sync::Arc, panic::UnwindSafe, result,
 	marker::PhantomData, cell::RefCell, rc::Rc,
 };
-use futures::{IntoFuture, Future};
 
 use parity_codec::{Encode, Decode};
 use primitives::{offchain, H256, Blake2Hasher, convert_hash, NativeOrEncoded};
@@ -100,13 +99,13 @@ where
 		let block_hash = self.blockchain.expect_block_hash_from_id(id)?;
 		let block_header = self.blockchain.expect_header(id.clone())?;
 
-		self.fetcher.remote_call(RemoteCallRequest {
+		futures::executor::block_on(self.fetcher.remote_call(RemoteCallRequest {
 			block: block_hash,
 			header: block_header,
 			method: method.into(),
 			call_data: call_data.to_vec(),
 			retry_count: None,
-		}).into_future().wait()
+		}))
 	}
 
 	fn contextual_call<