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

substrate/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"

substrate/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 {

substrate/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",

substrate/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::{

substrate/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")]

substrate/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>>> {

substrate/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>> {

substrate/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<

substrate/core/client/src/light/fetcher.rs

@@ -19,7 +19,7 @@
 use std::sync::Arc;
 use std::collections::BTreeMap;
 use std::marker::PhantomData;
-use futures::IntoFuture;
+use std::future::Future;
 
 use hash_db::{HashDB, Hasher};
 use parity_codec::{Decode, Encode};
@@ -141,15 +141,15 @@ pub struct RemoteBodyRequest<Header: HeaderT> {
 /// is correct (see FetchedDataChecker) and return already checked data.
 pub trait Fetcher<Block: BlockT>: Send + Sync {
 	/// Remote header future.
-	type RemoteHeaderResult: IntoFuture<Item = Block::Header, Error = ClientError>;
+	type RemoteHeaderResult: Future<Output = Result<Block::Header, ClientError>>;
 	/// Remote storage read future.
-	type RemoteReadResult: IntoFuture<Item = Option<Vec<u8>>, Error = ClientError>;
+	type RemoteReadResult: Future<Output = Result<Option<Vec<u8>>, ClientError>>;
 	/// Remote call result future.
-	type RemoteCallResult: IntoFuture<Item = Vec<u8>, Error = ClientError>;
+	type RemoteCallResult: Future<Output = Result<Vec<u8>, ClientError>>;
 	/// Remote changes result future.
-	type RemoteChangesResult: IntoFuture<Item = Vec<(NumberFor<Block>, u32)>, Error = ClientError>;
+	type RemoteChangesResult: Future<Output = Result<Vec<(NumberFor<Block>, u32)>, ClientError>>;
 	/// Remote block body result future.
-	type RemoteBodyResult: IntoFuture<Item = Vec<Block::Extrinsic>, Error = ClientError>;
+	type RemoteBodyResult: Future<Output = Result<Vec<Block::Extrinsic>, ClientError>>;
 
 	/// Fetch remote header.
 	fn remote_header(&self, request: RemoteHeaderRequest<Block::Header>) -> Self::RemoteHeaderResult;
@@ -484,7 +484,7 @@ impl<'a, H, Number, Hash> ChangesTrieRootsStorage<H, Number> for RootsStorage<'a
 
 #[cfg(test)]
 pub mod tests {
-	use futures::future::{ok, err, FutureResult};
+	use futures::future::Ready;
 	use parking_lot::Mutex;
 	use parity_codec::Decode;
 	use crate::client::tests::prepare_client_with_key_changes;
@@ -508,19 +508,19 @@ pub mod tests {
 
 	pub type OkCallFetcher = Mutex<Vec<u8>>;
 
-	fn not_implemented_in_tests<T, E>() -> FutureResult<T, E>
+	fn not_implemented_in_tests<T, E>() -> Ready<Result<T, E>>
 	where
 		E: std::convert::From<&'static str>,
 	{
-		err("Not implemented on test node".into())
+		futures::future::ready(Err("Not implemented on test node".into()))
 	}
 
 	impl Fetcher<Block> for OkCallFetcher {
-		type RemoteHeaderResult = FutureResult<Header, ClientError>;
-		type RemoteReadResult = FutureResult<Option<Vec<u8>>, ClientError>;
-		type RemoteCallResult = FutureResult<Vec<u8>, ClientError>;
-		type RemoteChangesResult = FutureResult<Vec<(NumberFor<Block>, u32)>, ClientError>;
-		type RemoteBodyResult = FutureResult<Vec<Extrinsic>, ClientError>;
+		type RemoteHeaderResult = Ready<Result<Header, ClientError>>;
+		type RemoteReadResult = Ready<Result<Option<Vec<u8>>, ClientError>>;
+		type RemoteCallResult = Ready<Result<Vec<u8>, ClientError>>;
+		type RemoteChangesResult = Ready<Result<Vec<(NumberFor<Block>, u32)>, ClientError>>;
+		type RemoteBodyResult = Ready<Result<Vec<Extrinsic>, ClientError>>;
 
 		fn remote_header(&self, _request: RemoteHeaderRequest<Header>) -> Self::RemoteHeaderResult {
 			not_implemented_in_tests()
@@ -535,7 +535,7 @@ pub mod tests {
 		}
 
 		fn remote_call(&self, _request: RemoteCallRequest<Header>) -> Self::RemoteCallResult {
-			ok((*self.lock()).clone())
+			futures::future::ready(Ok((*self.lock()).clone()))
 		}
 
 		fn remote_changes(&self, _request: RemoteChangesRequest<Header>) -> Self::RemoteChangesResult {

substrate/core/client/src/notifications.rs

@@ -22,7 +22,7 @@ use std::{
 };
 
 use fnv::{FnvHashSet, FnvHashMap};
-use futures::sync::mpsc;
+use futures::channel::mpsc;
 use primitives::storage::{StorageKey, StorageData};
 use runtime_primitives::traits::Block as BlockT;
 
@@ -309,7 +309,6 @@ impl<Block: BlockT> StorageNotifications<Block> {
 mod tests {
 	use runtime_primitives::testing::{H256 as Hash, Block as RawBlock, ExtrinsicWrapper};
 	use super::*;
-	use futures::Stream;
 	use std::iter::{empty, Empty};
 
 	type TestChangeSet = (
@@ -348,7 +347,9 @@ mod tests {
 		// given
 		let mut notifications = StorageNotifications::<Block>::default();
 		let child_filter = [(StorageKey(vec![4]), None)];
-		let mut recv = notifications.listen(None, Some(&child_filter[..])).wait();
+		let mut recv = futures::executor::block_on_stream(
+			notifications.listen(None, Some(&child_filter[..]))
+		);
 
 		// when
 		let changeset = vec![
@@ -367,13 +368,13 @@ mod tests {
 		);
 
 		// then
-		assert_eq!(recv.next().unwrap(), Ok((Hash::from_low_u64_be(1), (vec![
+		assert_eq!(recv.next().unwrap(), (Hash::from_low_u64_be(1), (vec![
 			(StorageKey(vec![2]), Some(StorageData(vec![3]))),
 			(StorageKey(vec![3]), None),
 		], vec![(StorageKey(vec![4]), vec![
 			(StorageKey(vec![5]), Some(StorageData(vec![4]))),
 			(StorageKey(vec![6]), None),
-		])]).into())));
+		])]).into()));
 	}
 
 	#[test]
@@ -381,9 +382,15 @@ mod tests {
 		// given
 		let mut notifications = StorageNotifications::<Block>::default();
 		let child_filter = [(StorageKey(vec![4]), Some(vec![StorageKey(vec![5])]))];
-		let mut recv1 = notifications.listen(Some(&[StorageKey(vec![1])]), None).wait();
-		let mut recv2 = notifications.listen(Some(&[StorageKey(vec![2])]), None).wait();
-		let mut recv3 = notifications.listen(Some(&[]), Some(&child_filter)).wait();
+		let mut recv1 = futures::executor::block_on_stream(
+			notifications.listen(Some(&[StorageKey(vec![1])]), None)
+		);
+		let mut recv2 = futures::executor::block_on_stream(
+			notifications.listen(Some(&[StorageKey(vec![2])]), None)
+		);
+		let mut recv3 = futures::executor::block_on_stream(
+			notifications.listen(Some(&[]), Some(&child_filter))
+		);
 
 		// when
 		let changeset = vec![
@@ -403,16 +410,16 @@ mod tests {
 		);
 
 		// then
-		assert_eq!(recv1.next().unwrap(), Ok((Hash::from_low_u64_be(1), (vec![
+		assert_eq!(recv1.next().unwrap(), (Hash::from_low_u64_be(1), (vec![
 			(StorageKey(vec![1]), None),
-		], vec![]).into())));
-		assert_eq!(recv2.next().unwrap(), Ok((Hash::from_low_u64_be(1), (vec![
+		], vec![]).into()));
+		assert_eq!(recv2.next().unwrap(), (Hash::from_low_u64_be(1), (vec![
 			(StorageKey(vec![2]), Some(StorageData(vec![3]))),
-		], vec![]).into())));
-		assert_eq!(recv3.next().unwrap(), Ok((Hash::from_low_u64_be(1), (vec![],
+		], vec![]).into()));
+		assert_eq!(recv3.next().unwrap(), (Hash::from_low_u64_be(1), (vec![],
 		vec![
 			(StorageKey(vec![4]), vec![(StorageKey(vec![5]), Some(StorageData(vec![4])))]),
-		]).into())));
+		]).into()));
 
 	}
 
@@ -422,10 +429,18 @@ mod tests {
 		let mut notifications = StorageNotifications::<Block>::default();
 		{
 			let child_filter = [(StorageKey(vec![4]), Some(vec![StorageKey(vec![5])]))];
-			let _recv1 = notifications.listen(Some(&[StorageKey(vec![1])]), None).wait();
-			let _recv2 = notifications.listen(Some(&[StorageKey(vec![2])]), None).wait();
-			let _recv3 = notifications.listen(None, None).wait();
-			let _recv4 = notifications.listen(None, Some(&child_filter)).wait();
+			let _recv1 = futures::executor::block_on_stream(
+				notifications.listen(Some(&[StorageKey(vec![1])]), None)
+			);
+			let _recv2 = futures::executor::block_on_stream(
+				notifications.listen(Some(&[StorageKey(vec![2])]), None)
+			);
+			let _recv3 = futures::executor::block_on_stream(
+				notifications.listen(None, None)
+			);
+			let _recv4 = futures::executor::block_on_stream(
+				notifications.listen(None, Some(&child_filter))
+			);
 			assert_eq!(notifications.listeners.len(), 2);
 			assert_eq!(notifications.wildcard_listeners.len(), 2);
 			assert_eq!(notifications.child_listeners.len(), 1);
@@ -450,7 +465,7 @@ mod tests {
 		// given
 		let mut recv = {
 			let mut notifications = StorageNotifications::<Block>::default();
-			let recv = notifications.listen(None, None).wait();
+			let recv = futures::executor::block_on_stream(notifications.listen(None, None));
 
 			// when
 			let changeset = vec![];

substrate/core/consensus/aura/Cargo.toml

@@ -25,6 +25,7 @@ parking_lot = "0.8.0"
 log = "0.4"
 
 [dev-dependencies]
+futures03 = { package = "futures-preview", version = "0.3.0-alpha.17", features = ["compat"] }
 keyring = { package = "substrate-keyring", path = "../../keyring" }
 substrate-executor = { path = "../../executor" }
 network = { package = "substrate-network", path = "../../network", features = ["test-helpers"]}

substrate/core/consensus/aura/src/lib.rs

@@ -709,6 +709,7 @@ pub fn import_queue<B, C, P>(
 mod tests {
 	use super::*;
 	use futures::{Async, stream::Stream as _};
+	use futures03::{StreamExt as _, TryStreamExt as _};
 	use consensus_common::NoNetwork as DummyOracle;
 	use network::test::*;
 	use network::test::{Block as TestBlock, PeersClient, PeersFullClient};
@@ -838,6 +839,7 @@ mod tests {
 			let environ = Arc::new(DummyFactory(client.clone()));
 			import_notifications.push(
 				client.import_notification_stream()
+					.map(|v| Ok::<_, ()>(v)).compat()
 					.take_while(|n| Ok(!(n.origin != BlockOrigin::Own && n.header.number() < &5)))
 					.for_each(move |_| Ok(()))
 			);

substrate/core/consensus/babe/Cargo.toml

@@ -28,6 +28,7 @@ rand = "0.6.5"
 merlin = "1.0.3"
 
 [dev-dependencies]
+futures03 = { package = "futures-preview", version = "0.3.0-alpha.17", features = ["compat"] }
 keyring = { package = "substrate-keyring", path = "../../keyring" }
 substrate-executor = { path = "../../executor" }
 network = { package = "substrate-network", path = "../../network", features = ["test-helpers"]}

substrate/core/consensus/babe/src/lib.rs

@@ -878,6 +878,7 @@ mod tests {
 	use client::BlockchainEvents;
 	use test_client;
 	use futures::{Async, stream::Stream as _};
+	use futures03::{StreamExt as _, TryStreamExt as _};
 	use log::debug;
 	use std::time::Duration;
 	type Item = generic::DigestItem<Hash>;
@@ -1005,6 +1006,7 @@ mod tests {
 			let environ = Arc::new(DummyFactory(client.clone()));
 			import_notifications.push(
 				client.import_notification_stream()
+					.map(|v| Ok::<_, ()>(v)).compat()
 					.take_while(|n| Ok(!(n.origin != BlockOrigin::Own && n.header.number() < &5)))
 					.for_each(move |_| Ok(()))
 			);

substrate/core/finality-grandpa/Cargo.toml

@@ -7,6 +7,7 @@ edition = "2018"
 [dependencies]
 fork-tree = { path = "../../core/utils/fork-tree" }
 futures = "0.1"
+futures03 = { package = "futures-preview", version = "0.3.0-alpha.17", features = ["compat"] }
 log = "0.4"
 parking_lot = "0.8.0"
 tokio-executor = "0.1.7"

substrate/core/finality-grandpa/src/tests.rs

@@ -21,6 +21,7 @@ use network::test::{Block, DummySpecialization, Hash, TestNetFactory, Peer, Peer
 use network::test::{PassThroughVerifier};
 use network::config::{ProtocolConfig, Roles, BoxFinalityProofRequestBuilder};
 use parking_lot::Mutex;
+use futures03::{StreamExt as _, TryStreamExt as _};
 use tokio::runtime::current_thread;
 use keyring::ed25519::{Keyring as AuthorityKeyring};
 use client::{
@@ -385,6 +386,7 @@ fn run_to_completion_with<F>(
 		wait_for.push(
 			Box::new(
 				client.finality_notification_stream()
+					.map(|v| Ok::<_, ()>(v)).compat()
 					.take_while(move |n| {
 						let mut highest_finalized = highest_finalized.write();
 						if *n.header.number() > *highest_finalized {
@@ -495,6 +497,7 @@ fn finalize_3_voters_1_full_observer() {
 		};
 		finality_notifications.push(
 			client.finality_notification_stream()
+				.map(|v| Ok::<_, ()>(v)).compat()
 				.take_while(|n| Ok(n.header.number() < &20))
 				.for_each(move |_| Ok(()))
 		);
@@ -585,6 +588,7 @@ fn transition_3_voters_twice_1_full_observer() {
 
 		// wait for blocks to be finalized before generating new ones
 		let block_production = client.finality_notification_stream()
+			.map(|v| Ok::<_, ()>(v)).compat()
 			.take_while(|n| Ok(n.header.number() < &30))
 			.for_each(move |n| {
 				match n.header.number() {
@@ -652,6 +656,7 @@ fn transition_3_voters_twice_1_full_observer() {
 
 		finality_notifications.push(
 			client.finality_notification_stream()
+				.map(|v| Ok::<_, ()>(v)).compat()
 				.take_while(|n| Ok(n.header.number() < &30))
 				.for_each(move |_| Ok(()))
 				.map(move |()| {
@@ -1275,6 +1280,7 @@ fn finalize_3_voters_1_light_observer() {
 	let link = net.lock().peer(3).data.lock().take().expect("link initialized on startup; qed");
 
 	let finality_notifications = net.lock().peer(3).client().finality_notification_stream()
+		.map(|v| Ok::<_, ()>(v)).compat()
 		.take_while(|n| Ok(n.header.number() < &20))
 		.collect();
 
@@ -1436,6 +1442,7 @@ fn voter_catches_up_to_latest_round_when_behind() {
 
 		finality_notifications.push(
 			client.finality_notification_stream()
+				.map(|v| Ok::<_, ()>(v)).compat()
 				.take_while(|n| Ok(n.header.number() < &50))
 				.for_each(move |_| Ok(()))
 		);
@@ -1471,6 +1478,7 @@ fn voter_catches_up_to_latest_round_when_behind() {
 			let set_state = link.persistent_data.set_state.clone();
 
 			let wait = client.finality_notification_stream()
+				.map(|v| Ok::<_, ()>(v)).compat()
 				.take_while(|n| Ok(n.header.number() < &50))
 				.collect()
 				.map(|_| set_state);

substrate/core/finality-grandpa/src/until_imported.rs

@@ -23,9 +23,10 @@
 use super::{BlockStatus, CommunicationIn, Error, SignedMessage};
 
 use log::{debug, warn};
-use client::ImportNotifications;
+use client::{BlockImportNotification, ImportNotifications};
 use futures::prelude::*;
 use futures::stream::Fuse;
+use futures03::{StreamExt as _, TryStreamExt as _};
 use grandpa::voter;
 use parking_lot::Mutex;
 use runtime_primitives::traits::{Block as BlockT, Header as HeaderT, NumberFor};
@@ -64,7 +65,7 @@ pub(crate) trait BlockUntilImported<Block: BlockT>: Sized {
 
 /// Buffering imported messages until blocks with given hashes are imported.
 pub(crate) struct UntilImported<Block: BlockT, Status, I, M: BlockUntilImported<Block>> {
-	import_notifications: Fuse<ImportNotifications<Block>>,
+	import_notifications: Fuse<Box<dyn Stream<Item = BlockImportNotification<Block>, Error = ()> + Send>>,
 	status_check: Status,
 	inner: Fuse<I>,
 	ready: VecDeque<M::Blocked>,
@@ -91,7 +92,10 @@ impl<Block: BlockT, Status, I: Stream, M> UntilImported<Block, Status, I, M>
 
 		let check_pending = Interval::new(now + CHECK_PENDING_INTERVAL, CHECK_PENDING_INTERVAL);
 		UntilImported {
-			import_notifications: import_notifications.fuse(),
+			import_notifications: {
+				let stream = import_notifications.map::<_, fn(_) -> _>(|v| Ok::<_, ()>(v)).compat();
+				Box::new(stream) as Box<dyn Stream<Item = _, Error = _> + Send>
+			}.fuse(),
 			status_check,
 			inner: stream.fuse(),
 			ready: VecDeque::new(),
@@ -194,7 +198,6 @@ impl<Block: BlockT, Status, I, M> Stream for UntilImported<Block, Status, I, M>
 		if self.import_notifications.is_done() && self.inner.is_done() {
 			Ok(Async::Ready(None))
 		} else {
-
 			Ok(Async::NotReady)
 		}
 	}
@@ -435,7 +438,7 @@ mod tests {
 	use consensus_common::BlockOrigin;
 	use client::BlockImportNotification;
 	use futures::future::Either;
-	use futures::sync::mpsc;
+	use futures03::channel::mpsc;
 	use grandpa::Precommit;
 
 	#[derive(Clone)]
@@ -523,7 +526,7 @@ mod tests {
 		// enact all dependencies before importing the message
 		enact_dependencies(&chain_state);
 
-		let (global_tx, global_rx) = mpsc::unbounded();
+		let (global_tx, global_rx) = futures::sync::mpsc::unbounded();
 
 		let until_imported = UntilGlobalMessageBlocksImported::new(
 			import_notifications,
@@ -548,7 +551,7 @@ mod tests {
 		let (chain_state, import_notifications) = TestChainState::new();
 		let block_status = chain_state.block_status();
 
-		let (global_tx, global_rx) = mpsc::unbounded();
+		let (global_tx, global_rx) = futures::sync::mpsc::unbounded();
 
 		let until_imported = UntilGlobalMessageBlocksImported::new(
 			import_notifications,

substrate/core/network/Cargo.toml

@@ -14,6 +14,7 @@ parking_lot = "0.8.0"
 bitflags = "1.0"
 fnv = "1.0"
 futures = "0.1.17"
+futures03 = { package = "futures-preview", version = "0.3.0-alpha.17", features = ["compat"] }
 linked-hash-map = "0.5"
 linked_hash_set = "0.1.3"
 lru-cache = "0.1.1"

substrate/core/network/src/on_demand_layer.rs

@@ -19,6 +19,7 @@
 use crate::protocol::on_demand::RequestData;
 use std::sync::Arc;
 use futures::{prelude::*, sync::mpsc, sync::oneshot};
+use futures03::compat::{Compat01As03, Future01CompatExt as _};
 use parking_lot::Mutex;
 use client::error::Error as ClientError;
 use client::light::fetcher::{Fetcher, FetchChecker, RemoteHeaderRequest,
@@ -82,22 +83,22 @@ impl<B> Fetcher<B> for OnDemand<B> where
 	B: BlockT,
 	B::Header: HeaderT,
 {
-	type RemoteHeaderResult = RemoteResponse<B::Header>;
-	type RemoteReadResult = RemoteResponse<Option<Vec<u8>>>;
-	type RemoteCallResult = RemoteResponse<Vec<u8>>;
-	type RemoteChangesResult = RemoteResponse<Vec<(NumberFor<B>, u32)>>;
-	type RemoteBodyResult = RemoteResponse<Vec<B::Extrinsic>>;
+	type RemoteHeaderResult = Compat01As03<RemoteResponse<B::Header>>;
+	type RemoteReadResult = Compat01As03<RemoteResponse<Option<Vec<u8>>>>;
+	type RemoteCallResult = Compat01As03<RemoteResponse<Vec<u8>>>;
+	type RemoteChangesResult = Compat01As03<RemoteResponse<Vec<(NumberFor<B>, u32)>>>;
+	type RemoteBodyResult = Compat01As03<RemoteResponse<Vec<B::Extrinsic>>>;
 
 	fn remote_header(&self, request: RemoteHeaderRequest<B::Header>) -> Self::RemoteHeaderResult {
 		let (sender, receiver) = oneshot::channel();
 		let _ = self.requests_send.unbounded_send(RequestData::RemoteHeader(request, sender));
-		RemoteResponse { receiver }
+		RemoteResponse { receiver }.compat()
 	}
 
 	fn remote_read(&self, request: RemoteReadRequest<B::Header>) -> Self::RemoteReadResult {
 		let (sender, receiver) = oneshot::channel();
 		let _ = self.requests_send.unbounded_send(RequestData::RemoteRead(request, sender));
-		RemoteResponse { receiver }
+		RemoteResponse { receiver }.compat()
 	}
 
 	fn remote_read_child(
@@ -106,25 +107,25 @@ impl<B> Fetcher<B> for OnDemand<B> where
 	) -> Self::RemoteReadResult {
 		let (sender, receiver) = oneshot::channel();
 		let _ = self.requests_send.unbounded_send(RequestData::RemoteReadChild(request, sender));
-		RemoteResponse { receiver }
+		RemoteResponse { receiver }.compat()
 	}
 
 	fn remote_call(&self, request: RemoteCallRequest<B::Header>) -> Self::RemoteCallResult {
 		let (sender, receiver) = oneshot::channel();
 		let _ = self.requests_send.unbounded_send(RequestData::RemoteCall(request, sender));
-		RemoteResponse { receiver }
+		RemoteResponse { receiver }.compat()
 	}
 
 	fn remote_changes(&self, request: RemoteChangesRequest<B::Header>) -> Self::RemoteChangesResult {
 		let (sender, receiver) = oneshot::channel();
 		let _ = self.requests_send.unbounded_send(RequestData::RemoteChanges(request, sender));
-		RemoteResponse { receiver }
+		RemoteResponse { receiver }.compat()
 	}
 
 	fn remote_body(&self, request: RemoteBodyRequest<B::Header>) -> Self::RemoteBodyResult {
 		let (sender, receiver) = oneshot::channel();
 		let _ = self.requests_send.unbounded_send(RequestData::RemoteBody(request, sender));
-		RemoteResponse { receiver }
+		RemoteResponse { receiver }.compat()
 	}
 }
 

substrate/core/network/src/test/mod.rs

@@ -27,7 +27,7 @@ use std::sync::Arc;
 use crate::config::build_multiaddr;
 use log::trace;
 use crate::chain::FinalityProofProvider;
-use client::{self, ClientInfo, BlockchainEvents, ImportNotifications, FinalityNotifications};
+use client::{self, ClientInfo, BlockchainEvents, BlockImportNotification, FinalityNotifications, FinalityNotification};
 use client::{in_mem::Backend as InMemoryBackend, error::Result as ClientResult};
 use client::block_builder::BlockBuilder;
 use client::backend::AuxStore;
@@ -40,6 +40,7 @@ use consensus::block_import::{BlockImport, ImportResult};
 use consensus::{Error as ConsensusError, well_known_cache_keys::{self, Id as CacheKeyId}};
 use consensus::{BlockOrigin, ForkChoiceStrategy, ImportBlock, JustificationImport};
 use futures::prelude::*;
+use futures03::{StreamExt as _, TryStreamExt as _};
 use crate::{NetworkWorker, NetworkService, config::ProtocolId};
 use crate::config::{NetworkConfiguration, TransportConfig, BoxFinalityProofRequestBuilder};
 use libp2p::PeerId;
@@ -216,8 +217,8 @@ pub struct Peer<D, S: NetworkSpecialization<Block>> {
 	/// instead of going through the import queue.
 	block_import: Box<dyn BlockImport<Block, Error = ConsensusError>>,
 	network: NetworkWorker<Block, S, <Block as BlockT>::Hash>,
-	imported_blocks_stream: futures::stream::Fuse<ImportNotifications<Block>>,
-	finality_notification_stream: futures::stream::Fuse<FinalityNotifications<Block>>,
+	imported_blocks_stream: Box<dyn Stream<Item = BlockImportNotification<Block>, Error = ()> + Send>,
+	finality_notification_stream: Box<dyn Stream<Item = FinalityNotification<Block>, Error = ()> + Send>,
 }
 
 impl<D, S: NetworkSpecialization<Block>> Peer<D, S> {
@@ -482,8 +483,10 @@ pub trait TestNetFactory: Sized {
 				peer.network.add_known_address(network.service().local_peer_id(), listen_addr.clone());
 			}
 
-			let imported_blocks_stream = client.import_notification_stream().fuse();
-			let finality_notification_stream = client.finality_notification_stream().fuse();
+			let imported_blocks_stream = Box::new(client.import_notification_stream()
+				.map(|v| Ok::<_, ()>(v)).compat().fuse());
+			let finality_notification_stream = Box::new(client.finality_notification_stream()
+				.map(|v| Ok::<_, ()>(v)).compat().fuse());
 
 			peers.push(Peer {
 				data,
@@ -539,8 +542,10 @@ pub trait TestNetFactory: Sized {
 				peer.network.add_known_address(network.service().local_peer_id(), listen_addr.clone());
 			}
 
-			let imported_blocks_stream = client.import_notification_stream().fuse();
-			let finality_notification_stream = client.finality_notification_stream().fuse();
+			let imported_blocks_stream = Box::new(client.import_notification_stream()
+				.map(|v| Ok::<_, ()>(v)).compat().fuse());
+			let finality_notification_stream = Box::new(client.finality_notification_stream()
+				.map(|v| Ok::<_, ()>(v)).compat().fuse());
 
 			peers.push(Peer {
 				data,

substrate/core/rpc/Cargo.toml

@@ -7,6 +7,7 @@ edition = "2018"
 [dependencies]
 derive_more = "0.14.0"
 futures = "0.1"
+futures03 = { package = "futures-preview", version = "0.3.0-alpha.17", features = ["compat"] }
 jsonrpc-core = "12.0.0"
 jsonrpc-core-client = "12.0.0"
 jsonrpc-pubsub = "12.0.0"

substrate/core/rpc/src/chain/mod.rs

@@ -23,6 +23,7 @@ pub mod number;
 mod tests;
 
 use std::sync::Arc;
+use futures03::{future, StreamExt as _, TryStreamExt as _};
 
 use client::{self, Client, BlockchainEvents};
 use crate::rpc::Result as RpcResult;
@@ -203,8 +204,9 @@ impl<B, E, Block, RA> ChainApi<NumberFor<Block>, Block::Hash, Block::Header, Sig
 			subscriber,
 			|| self.block_hash(None.into()),
 			|| self.client.import_notification_stream()
-				.filter(|notification| notification.is_new_best)
-				.map(|notification| notification.header),
+				.filter(|notification| future::ready(notification.is_new_best))
+				.map(|notification| Ok::<_, ()>(notification.header))
+				.compat(),
 		)
 	}
 
@@ -217,7 +219,8 @@ impl<B, E, Block, RA> ChainApi<NumberFor<Block>, Block::Hash, Block::Header, Sig
 			subscriber,
 			|| Ok(Some(self.client.info().chain.finalized_hash)),
 			|| self.client.finality_notification_stream()
-				.map(|notification| notification.header),
+				.map(|notification| Ok::<_, ()>(notification.header))
+				.compat(),
 		)
 	}
 

substrate/core/rpc/src/state/mod.rs

@@ -26,6 +26,7 @@ use std::{
 	ops::Range,
 	sync::Arc,
 };
+use futures03::{future, StreamExt as _, TryStreamExt as _};
 
 use client::{self, Client, CallExecutor, BlockchainEvents, runtime_api::Metadata};
 use crate::rpc::Result as RpcResult;
@@ -484,14 +485,14 @@ impl<B, E, Block, RA> StateApi<Block::Hash> for State<B, E, Block, RA> where
 
 		self.subscriptions.add(subscriber, |sink| {
 			let stream = stream
-				.map_err(|e| warn!("Error creating storage notification stream: {:?}", e))
-				.map(|(block, changes)| Ok(StorageChangeSet {
+				.map(|(block, changes)| Ok::<_, ()>(Ok(StorageChangeSet {
 					block,
 					changes: changes.iter()
 						.filter_map(|(o_sk, k, v)| if o_sk.is_none() {
 							Some((k.clone(),v.cloned()))
 						} else { None }).collect(),
-				}));
+				})))
+				.compat();
 
 			sink
 				.sink_map_err(|e| warn!("Error sending notifications: {:?}", e))
@@ -530,7 +531,6 @@ impl<B, E, Block, RA> StateApi<Block::Hash> for State<B, E, Block, RA> where
 			let mut previous_version = version.clone();
 
 			let stream = stream
-				.map_err(|e| warn!("Error creating storage notification stream: {:?}", e))
 				.filter_map(move |_| {
 					let info = client.info();
 					let version = client
@@ -539,11 +539,12 @@ impl<B, E, Block, RA> StateApi<Block::Hash> for State<B, E, Block, RA> where
 						.map_err(Into::into);
 					if previous_version != version {
 						previous_version = version.clone();
-						Some(version)
+						future::ready(Some(Ok::<_, ()>(version)))
 					} else {
-						None
+						future::ready(None)
 					}
-				});
+				})
+				.compat();
 
 			sink
 				.sink_map_err(|e| warn!("Error sending notifications: {:?}", e))

substrate/core/service/Cargo.toml

@@ -7,6 +7,7 @@ edition = "2018"
 [dependencies]
 derive_more = "0.14.0"
 futures = "0.1.17"
+futures03 = { package = "futures-preview", version = "0.3.0-alpha.17", features = ["compat"] }
 parking_lot = "0.8.0"
 lazy_static = "1.0"
 log = "0.4"

substrate/core/service/src/lib.rs

@@ -30,6 +30,7 @@ use std::net::SocketAddr;
 use std::collections::HashMap;
 use std::time::Duration;
 use futures::sync::mpsc;
+use futures03::{StreamExt as _, TryStreamExt as _};
 use parking_lot::Mutex;
 
 use client::{BlockchainEvents, backend::Backend, runtime_api::BlockT};
@@ -294,6 +295,7 @@ impl<Components: components::Components> Service<Components> {
 			let to_spawn_tx_ = to_spawn_tx.clone();
 
 			let events = client.import_notification_stream()
+				.map(|v| Ok::<_, ()>(v)).compat()
 				.for_each(move |notification| {
 					let number = *notification.header.number();
 
@@ -623,8 +625,10 @@ fn build_network_future<
 	const STATUS_INTERVAL: Duration = Duration::from_millis(5000);
 	let mut status_interval = tokio_timer::Interval::new_interval(STATUS_INTERVAL);
 
-	let mut imported_blocks_stream = client.import_notification_stream().fuse();
-	let mut finality_notification_stream = client.finality_notification_stream().fuse();
+	let mut imported_blocks_stream = client.import_notification_stream().fuse()
+		.map(|v| Ok::<_, ()>(v)).compat();
+	let mut finality_notification_stream = client.finality_notification_stream().fuse()
+		.map(|v| Ok::<_, ()>(v)).compat();
 
 	futures::future::poll_fn(move || {
 		// We poll `imported_blocks_stream`.

substrate/core/test-client/Cargo.toml

@@ -9,7 +9,7 @@ client = { package = "substrate-client", path = "../client" }
 client-db = { package = "substrate-client-db", path = "../client/db", features = ["test-helpers"] }
 consensus = { package = "substrate-consensus-common", path = "../consensus/common" }
 executor = { package = "substrate-executor", path = "../executor" }
-futures = { version = "0.1.27" }
+futures-preview = "0.3.0-alpha.17"
 hash-db = "0.14.0"
 keyring = { package = "substrate-keyring", path = "../keyring" }
 parity-codec = "4.1.1"

substrate/core/test-client/src/lib.rs

@@ -32,7 +32,7 @@ pub use state_machine::ExecutionStrategy;
 
 use std::sync::Arc;
 use std::collections::HashMap;
-use futures::future::FutureResult;
+use futures::future::Ready;
 use hash_db::Hasher;
 use primitives::storage::well_known_keys;
 use runtime_primitives::traits::{
@@ -220,11 +220,11 @@ impl<E, Backend, G: GenesisInit> TestClientBuilder<
 }
 
 impl<Block: BlockT> client::light::fetcher::Fetcher<Block> for LightFetcher {
-	type RemoteHeaderResult = FutureResult<Block::Header, client::error::Error>;
-	type RemoteReadResult = FutureResult<Option<Vec<u8>>, client::error::Error>;
-	type RemoteCallResult = FutureResult<Vec<u8>, client::error::Error>;
-	type RemoteChangesResult = FutureResult<Vec<(NumberFor<Block>, u32)>, client::error::Error>;
-	type RemoteBodyResult = FutureResult<Vec<Block::Extrinsic>, client::error::Error>;
+	type RemoteHeaderResult = Ready<Result<Block::Header, client::error::Error>>;
+	type RemoteReadResult = Ready<Result<Option<Vec<u8>>, client::error::Error>>;
+	type RemoteCallResult = Ready<Result<Vec<u8>, client::error::Error>>;
+	type RemoteChangesResult = Ready<Result<Vec<(NumberFor<Block>, u32)>, client::error::Error>>;
+	type RemoteBodyResult = Ready<Result<Vec<Block::Extrinsic>, client::error::Error>>;
 
 	fn remote_header(
 		&self,