Upgrade from futures-preview to futures 0.3.1, and remove futures 0.1 where currently possible

core/authority-discovery/Cargo.toml

@@ -14,15 +14,15 @@ bytes = "0.4.12"
 client = { package = "substrate-client", path = "../../core/client" }
 codec = { package = "parity-scale-codec", default-features = false, version = "1.0.3" }
 derive_more = "0.15.0"
-futures-preview = "0.3.0-alpha.19"
+futures = "0.3.1"
 libp2p = { version = "0.13.0", default-features = false, features = ["secp256k1", "libp2p-websocket"] }
 log = "0.4.8"
 network = { package = "substrate-network", path = "../../core/network" }
 primitives = { package = "substrate-primitives", path = "../primitives" }
 prost = "0.5.0"
 serde_json = "1.0.41"
 sr-primitives = { path = "../../core/sr-primitives" }
-futures-timer = "0.4"
+async-std = { version = "0.99.12", features = ["unstable"] }
 
 [dev-dependencies]
 parking_lot = "0.9.0"

core/authority-discovery/src/lib.rs

@@ -54,7 +54,7 @@ use futures::channel::mpsc::Receiver;
 use futures::stream::StreamExt;
 use futures::task::{Context, Poll};
 use futures::Future;
-use futures_timer::Interval;
+use async_std::stream::{interval, Interval};
 
 use authority_discovery_primitives::{AuthorityDiscoveryApi, AuthorityId, Signature};
 use client::blockchain::HeaderBackend;
@@ -117,11 +117,11 @@ where
 		// Kademlia's default time-to-live for Dht records is 36h, republishing records every 24h. Given that a node
 		// could restart at any point in time, one can not depend on the republishing process, thus publishing own
 		// external addresses should happen on an interval < 36h.
-		let publish_interval = Interval::new(Duration::from_secs(12 * 60 * 60));
+		let publish_interval = interval(Duration::from_secs(12 * 60 * 60));
 
 		// External addresses of other authorities can change at any given point in time. The interval on which to query
 		// for external addresses of other authorities is a trade off between efficiency and performance.
-		let query_interval = Interval::new(Duration::from_secs(10 * 60));
+		let query_interval = interval(Duration::from_secs(10 * 60));
 
 		let address_cache = HashMap::new();
 

core/basic-authorship/Cargo.toml

@@ -6,7 +6,7 @@ edition = "2018"
 
 [dependencies]
 log = "0.4.8"
-futures-preview = "0.3.0-alpha.19"
+futures = "0.3.1"
 codec = { package = "parity-scale-codec", version = "1.0.0" }
 sr-primitives = { path = "../../core/sr-primitives" }
 primitives = { package = "substrate-primitives", path = "../../core/primitives" }

core/cli/Cargo.toml

@@ -17,10 +17,8 @@ ansi_term = "0.12.1"
 lazy_static = "1.4.0"
 app_dirs = "1.2.1"
 tokio = "0.1.22"
-futures = "0.1.29"
-futures03 = { package = "futures-preview", version = "=0.3.0-alpha.19", features = ["compat"] }
+futures = { version = "0.3.1", features = ["compat"] }
 fdlimit = "0.1.1"
-exit-future = "0.1.4"
 serde_json = "1.0.41"
 panic-handler = { package = "substrate-panic-handler", path = "../../core/panic-handler" }
 client = { package = "substrate-client", path = "../../core/client" }

core/cli/src/informant.rs

@@ -17,8 +17,7 @@
 //! Console informant. Prints sync progress and block events. Runs on the calling thread.
 
 use client::BlockchainEvents;
-use futures::{Future, Stream};
-use futures03::{StreamExt as _, TryStreamExt as _};
+use futures::{StreamExt, TryStreamExt, FutureExt, future, compat::Stream01CompatExt};
 use log::{info, warn};
 use sr_primitives::traits::Header;
 use service::AbstractService;
@@ -27,17 +26,18 @@ use std::time::Duration;
 mod display;
 
 /// Creates an informant in the form of a `Future` that must be polled regularly.
-pub fn build(service: &impl AbstractService) -> impl Future<Item = (), Error = ()> {
+pub fn build(service: &impl AbstractService) -> impl futures::Future<Output = ()> {
 	let client = service.client();
 
 	let mut display = display::InformantDisplay::new();
 
 	let display_notifications = service
 		.network_status(Duration::from_millis(5000))
-		.for_each(move |(net_status, _)| {
+		.compat()
+		.try_for_each(move |(net_status, _)| {
 			let info = client.info();
 			display.display(&info, net_status);
-			Ok(())
+			future::ok(())
 		});
 
 	let client = service.client();
@@ -46,7 +46,7 @@ pub fn build(service: &impl AbstractService) -> impl Future<Item = (), Error = (
 		Some((info.chain.best_number, info.chain.best_hash))
 	};
 
-	let display_block_import = client.import_notification_stream().map(|v| Ok::<_, ()>(v)).compat().for_each(move |n| {
+	let display_block_import = client.import_notification_stream().for_each(move |n| {
 		// detect and log reorganizations.
 		if let Some((ref last_num, ref last_hash)) = last_best {
 			if n.header.parent_hash() != last_hash && n.is_new_best  {
@@ -74,9 +74,11 @@ pub fn build(service: &impl AbstractService) -> impl Future<Item = (), Error = (
 		}
 
 		info!(target: "substrate", "Imported #{} ({})", n.header.number(), n.hash);
-		Ok(())
+		future::ready(())
 	});
 
-	display_notifications.join(display_block_import)
-		.map(|((), ())| ())
+	future::join(
+		display_notifications,
+		display_block_import
+	).map(|_| ())
 }

core/cli/src/lib.rs

@@ -61,8 +61,7 @@ pub use traits::{GetLogFilter, AugmentClap};
 use app_dirs::{AppInfo, AppDataType};
 use log::info;
 use lazy_static::lazy_static;
-
-use futures::Future;
+use futures::{Future, FutureExt, TryFutureExt};
 use substrate_telemetry::TelemetryEndpoints;
 
 /// default sub directory to store network config
@@ -102,7 +101,7 @@ pub struct VersionInfo {
 /// Something that can be converted into an exit signal.
 pub trait IntoExit {
 	/// Exit signal type.
-	type Exit: Future<Item=(),Error=()> + Send + 'static;
+	type Exit: Future<Output=()> + Unpin + Send + 'static;
 	/// Convert into exit signal.
 	fn into_exit(self) -> Self::Exit;
 }
@@ -388,7 +387,11 @@ impl<'a> ParseAndPrepareExport<'a> {
 			None => Box::new(stdout()),
 		};
 
-		builder(config)?.export_blocks(exit.into_exit(), file, from.into(), to.map(Into::into), json)?;
+		let exit = exit.into_exit()
+			.map(|_| Ok(()))
+			.compat();
+
+		builder(config)?.export_blocks(exit, file, from.into(), to.map(Into::into), json)?;
 		Ok(())
 	}
 }
@@ -432,7 +435,11 @@ impl<'a> ParseAndPrepareImport<'a> {
 			},
 		};
 
-		let fut = builder(config)?.import_blocks(exit.into_exit(), file)?;
+		let exit = exit.into_exit()
+			.map(|_| Ok(()))
+			.compat();
+
+		let fut = builder(config)?.import_blocks(exit, file)?;
 		tokio::run(fut);
 		Ok(())
 	}

core/client/Cargo.toml

@@ -10,8 +10,7 @@ fnv = { version = "1.0.6", optional = true }
 log = { version = "0.4.8", optional = true }
 parking_lot = { version = "0.9.0", optional = true }
 hex-literal = { version = "0.2.1", optional = true }
-futures = { version = "0.1.29", optional = true }
-futures03 = { package = "futures-preview", version = "0.3.0-alpha.19", features = ["compat"], optional = true }
+futures = { version = "0.3.1", features = ["compat"], 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 }
@@ -55,7 +54,6 @@ std = [
 	"log",
 	"hex-literal",
 	"futures",
-	"futures03",
 	"executor",
 	"state-machine",
 	"keyring",

core/client/src/client.rs

@@ -21,7 +21,7 @@ use std::{
 	panic::UnwindSafe, result, cell::RefCell, rc::Rc,
 };
 use log::{info, trace, warn};
-use futures03::channel::mpsc;
+use futures::channel::mpsc;
 use parking_lot::{Mutex, RwLock};
 use codec::{Encode, Decode};
 use hash_db::{Hasher, Prefix};

core/client/src/light/blockchain.rs

@@ -231,7 +231,7 @@ pub fn future_header<Block: BlockT, F: Fetcher<Block>>(
 	fetcher: &F,
 	id: BlockId<Block>,
 ) -> impl Future<Output = Result<Option<Block::Header>, ClientError>> {
-	use futures03::future::{ready, Either, FutureExt};
+	use futures::future::{ready, Either, FutureExt};
 
 	match blockchain.header(id) {
 		Ok(LocalOrRemote::Remote(request)) => Either::Left(

core/client/src/light/fetcher.rs

@@ -499,7 +499,7 @@ impl<'a, H, Number, Hash> ChangesTrieRootsStorage<H, Number> for RootsStorage<'a
 
 #[cfg(test)]
 pub mod tests {
-	use futures03::future::Ready;
+	use futures::future::Ready;
 	use parking_lot::Mutex;
 	use codec::Decode;
 	use crate::client::tests::prepare_client_with_key_changes;
@@ -527,7 +527,7 @@ pub mod tests {
 	where
 		E: std::convert::From<&'static str>,
 	{
-		futures03::future::ready(Err("Not implemented on test node".into()))
+		futures::future::ready(Err("Not implemented on test node".into()))
 	}
 
 	impl Fetcher<Block> for OkCallFetcher {
@@ -550,7 +550,7 @@ pub mod tests {
 		}
 
 		fn remote_call(&self, _request: RemoteCallRequest<Header>) -> Self::RemoteCallResult {
-			futures03::future::ready(Ok((*self.lock()).clone()))
+			futures::future::ready(Ok((*self.lock()).clone()))
 		}
 
 		fn remote_changes(&self, _request: RemoteChangesRequest<Header>) -> Self::RemoteChangesResult {

core/client/src/notifications.rs

@@ -22,7 +22,7 @@ use std::{
 };
 
 use fnv::{FnvHashSet, FnvHashMap};
-use futures03::channel::mpsc;
+use futures::channel::mpsc;
 use primitives::storage::{StorageKey, StorageData};
 use sr_primitives::traits::Block as BlockT;
 
@@ -347,7 +347,7 @@ mod tests {
 		// given
 		let mut notifications = StorageNotifications::<Block>::default();
 		let child_filter = [(StorageKey(vec![4]), None)];
-		let mut recv = futures03::executor::block_on_stream(
+		let mut recv = futures::executor::block_on_stream(
 			notifications.listen(None, Some(&child_filter[..]))
 		);
 
@@ -382,13 +382,13 @@ mod tests {
 		// given
 		let mut notifications = StorageNotifications::<Block>::default();
 		let child_filter = [(StorageKey(vec![4]), Some(vec![StorageKey(vec![5])]))];
-		let mut recv1 = futures03::executor::block_on_stream(
+		let mut recv1 = futures::executor::block_on_stream(
 			notifications.listen(Some(&[StorageKey(vec![1])]), None)
 		);
-		let mut recv2 = futures03::executor::block_on_stream(
+		let mut recv2 = futures::executor::block_on_stream(
 			notifications.listen(Some(&[StorageKey(vec![2])]), None)
 		);
-		let mut recv3 = futures03::executor::block_on_stream(
+		let mut recv3 = futures::executor::block_on_stream(
 			notifications.listen(Some(&[]), Some(&child_filter))
 		);
 
@@ -429,16 +429,16 @@ mod tests {
 		let mut notifications = StorageNotifications::<Block>::default();
 		{
 			let child_filter = [(StorageKey(vec![4]), Some(vec![StorageKey(vec![5])]))];
-			let _recv1 = futures03::executor::block_on_stream(
+			let _recv1 = futures::executor::block_on_stream(
 				notifications.listen(Some(&[StorageKey(vec![1])]), None)
 			);
-			let _recv2 = futures03::executor::block_on_stream(
+			let _recv2 = futures::executor::block_on_stream(
 				notifications.listen(Some(&[StorageKey(vec![2])]), None)
 			);
-			let _recv3 = futures03::executor::block_on_stream(
+			let _recv3 = futures::executor::block_on_stream(
 				notifications.listen(None, None)
 			);
-			let _recv4 = futures03::executor::block_on_stream(
+			let _recv4 = futures::executor::block_on_stream(
 				notifications.listen(None, Some(&child_filter))
 			);
 			assert_eq!(notifications.listeners.len(), 2);
@@ -465,7 +465,7 @@ mod tests {
 		// given
 		let mut recv = {
 			let mut notifications = StorageNotifications::<Block>::default();
-			let recv = futures03::executor::block_on_stream(notifications.listen(None, None));
+			let recv = futures::executor::block_on_stream(notifications.listen(None, None));
 
 			// when
 			let changeset = vec![];

core/consensus/aura/Cargo.toml

@@ -21,7 +21,7 @@ substrate-telemetry = { path = "../../telemetry" }
 keystore = { package = "substrate-keystore", path = "../../keystore" }
 consensus_common = { package = "substrate-consensus-common", path = "../common" }
 sr-primitives = {  path = "../../sr-primitives" }
-futures-preview = { version = "0.3.0-alpha.19", features = ["compat"] }
+futures = { version = "0.3.1", features = ["compat"] }
 futures01 = { package = "futures", version = "0.1" }
 futures-timer = "0.4.0"
 parking_lot = "0.9.0"

core/consensus/babe/Cargo.toml

@@ -27,7 +27,7 @@ uncles = { package = "substrate-consensus-uncles", path = "../uncles" }
 slots = { package = "substrate-consensus-slots", path = "../slots"  }
 sr-primitives = {  path = "../../sr-primitives" }
 fork-tree = { path = "../../utils/fork-tree" }
-futures-preview = { version = "0.3.0-alpha.19", features = ["compat"] }
+futures = { version = "0.3.1", features = ["compat"] }
 futures01 = { package = "futures", version = "0.1" }
 futures-timer = "0.4.0"
 parking_lot = "0.9.0"

core/consensus/common/Cargo.toml

@@ -11,7 +11,7 @@ libp2p = { version = "0.13.0", default-features = false }
 log = "0.4.8"
 primitives = { package = "substrate-primitives", path= "../../primitives" }
 inherents = { package = "substrate-inherents", path = "../../inherents" }
-futures-preview = "0.3.0-alpha.19"
+futures = { version = "0.3.1", features = ["thread-pool"] }
 futures-timer = "0.4.0"
 rstd = { package = "sr-std", path = "../../sr-std" }
 runtime_version = { package = "sr-version", path = "../../sr-version" }

core/consensus/pow/Cargo.toml

@@ -15,5 +15,5 @@ inherents = { package = "substrate-inherents", path = "../../inherents" }
 pow-primitives = { package = "substrate-consensus-pow-primitives", path = "primitives" }
 consensus-common = { package = "substrate-consensus-common", path = "../common" }
 log = "0.4.8"
-futures-preview = { version = "0.3.0-alpha.19", features = ["compat"] }
+futures = { version = "0.3.1", features = ["compat"] }
 derive_more = "0.15.0"

core/consensus/slots/Cargo.toml

@@ -14,8 +14,8 @@ sr-primitives = {  path = "../../sr-primitives" }
 substrate-telemetry = { path = "../../telemetry" }
 consensus_common = { package = "substrate-consensus-common", path = "../common" }
 inherents = { package = "substrate-inherents", path = "../../inherents" }
-futures-preview = "0.3.0-alpha.19"
-futures-timer = "0.4.0"
+futures = "0.3.1"
+futures-timer = "2.0"
 parking_lot = "0.9.0"
 log = "0.4.8"
 

core/consensus/slots/src/lib.rs

@@ -191,12 +191,10 @@ pub trait SimpleSlotWorker<B: BlockT> {
 				remaining_duration,
 			).map_err(|e| consensus_common::Error::ClientImport(format!("{:?}", e))),
 			Delay::new(remaining_duration)
-				.map_err(consensus_common::Error::FaultyTimer)
 		).map(|v| match v {
 			futures::future::Either::Left((b, _)) => b.map(|b| (b, claim)),
-			futures::future::Either::Right((Ok(_), _)) =>
+			futures::future::Either::Right(_) =>
 				Err(consensus_common::Error::ClientImport("Timeout in the Slots proposer".into())),
-			futures::future::Either::Right((Err(err), _)) => Err(err),
 		});
 
 		let block_import_params_maker = self.block_import_params();

core/consensus/slots/src/slots.rs

@@ -137,8 +137,7 @@ impl<SC: SlotCompatible + Unpin> Stream for Slots<SC> {
 			if let Some(ref mut inner_delay) = self.inner_delay {
 				match Future::poll(Pin::new(inner_delay), cx) {
 					Poll::Pending => return Poll::Pending,
-					Poll::Ready(Err(err)) => return Poll::Ready(Some(Err(Error::FaultyTimer(err)))),
-					Poll::Ready(Ok(())) => {}
+					Poll::Ready(()) => {}
 				}
 			}
 

core/finality-grandpa/Cargo.toml

@@ -7,7 +7,7 @@ edition = "2018"
 [dependencies]
 fork-tree = { path = "../../core/utils/fork-tree" }
 futures = "0.1.29"
-futures03 = { package = "futures-preview", version = "0.3.0-alpha.19", features = ["compat"] }
+futures03 = { package = "futures", version = "0.3.1", features = ["compat"] }
 log = "0.4.8"
 parking_lot = "0.9.0"
 tokio-executor = "0.1.8"

core/network/Cargo.toml

@@ -15,8 +15,9 @@ parking_lot = "0.9.0"
 bitflags = "1.2.0"
 fnv = "1.0.6"
 futures = "0.1.29"
-futures03 = { package = "futures-preview", version = "0.3.0-alpha.19", features = ["compat"] }
+futures03 = { package = "futures", version = "0.3.1", features = ["compat"] }
 futures-timer = "0.4.0"
+async-std = { version = "0.99.12", features = ["unstable"] }
 linked-hash-map = "0.5.2"
 linked_hash_set = "0.1.3"
 lru-cache = "0.1.2"

core/network/src/debug_info.rs

@@ -27,7 +27,7 @@ use log::{debug, trace, error};
 use std::collections::hash_map::Entry;
 use std::time::{Duration, Instant};
 use tokio_io::{AsyncRead, AsyncWrite};
-use futures_timer::Interval;
+use async_std::stream::interval;
 
 /// Time after we disconnect from a node before we purge its information from the cache.
 const CACHE_EXPIRE: Duration = Duration::from_secs(10 * 60);
@@ -76,7 +76,7 @@ impl<TSubstream> DebugInfoBehaviour<TSubstream> {
 			ping: Ping::new(PingConfig::new()),
 			identify,
 			nodes_info: FnvHashMap::default(),
-			garbage_collect: Box::new(Interval::new(GARBAGE_COLLECT_INTERVAL).map(|()| Ok(())).compat()),
+			garbage_collect: Box::new(interval(GARBAGE_COLLECT_INTERVAL).map(|()| Ok(())).compat()),
 		}
 	}