Upgrade from futures-preview to futures 0.3.1, and remove futures 0.1…

… where currently possible (#4083)

* Migrate node and node-template

* Migrate srml

* Simple changes

* Add async-std for interval

* Fix test-runtime warning

* Small changes

* move futures01 in core/rpc to dev-deps

* Change wasm CI builds

* Switch to async-std 1.0.1

* Remove async-std dep of network

* Add modified lockfile

* Fix node cli browser build

* Remove authority-discovery async-std dep

* Add Send + Sync to interval dyn stream

.gitlab-ci.yml

@@ -219,14 +219,14 @@ check-web-wasm:
   script:
     # WASM support is in progress. As more and more crates support WASM, we
     # should add entries here. See https://github.com/paritytech/substrate/issues/2416
-    - time cargo web build -p sr-io
-    - time cargo web build -p sr-primitives
-    - time cargo web build -p sr-std
-    - time cargo web build -p substrate-client
-    - time cargo web build -p substrate-consensus-aura
-    - time cargo web build -p substrate-consensus-babe
-    - time cargo web build -p substrate-consensus-common
-    - time cargo web build -p substrate-telemetry
+    - time cargo build --target=wasm32-unknown-unknown -p sr-io
+    - time cargo build --target=wasm32-unknown-unknown -p sr-primitives
+    - time cargo build --target=wasm32-unknown-unknown -p sr-std
+    - time cargo build --target=wasm32-unknown-unknown -p substrate-client
+    - time cargo build --target=wasm32-unknown-unknown -p substrate-consensus-aura
+    - time cargo build --target=wasm32-unknown-unknown -p substrate-consensus-babe
+    - time cargo build --target=wasm32-unknown-unknown -p substrate-consensus-common
+    - time cargo build --target=wasm32-unknown-unknown -p substrate-telemetry
     # Note: the command below is a bit weird because several Cargo issues prevent us from compiling the node in a more straight-forward way.
     - time cargo build --manifest-path=bin/node/cli/Cargo.toml --no-default-features --features "browser" --target=wasm32-unknown-unknown
     - sccache -s

bin/node-template/Cargo.toml

@@ -11,11 +11,11 @@ path = "src/main.rs"
 
 [dependencies]
 derive_more = "0.15.0"
-futures = "0.1.29"
+futures = "0.3.1"
+futures01 = { package = "futures", version = "0.1.29" }
 ctrlc = { version = "3.1.3", features = ["termination"] }
 log = "0.4.8"
 tokio = "0.1.22"
-exit-future = "0.1.4"
 parking_lot = "0.9.0"
 codec = { package = "parity-scale-codec", version = "1.0.0" }
 trie-root = "0.15.2"

bin/node-template/src/cli.rs

@@ -1,5 +1,5 @@
 use crate::service;
-use futures::{future, Future, sync::oneshot};
+use futures::{future::{select, Map}, FutureExt, TryFutureExt, channel::oneshot, compat::Future01CompatExt};
 use std::cell::RefCell;
 use tokio::runtime::Runtime;
 pub use substrate_cli::{VersionInfo, IntoExit, error};
@@ -69,25 +69,37 @@ where
 	T: AbstractService,
 	E: IntoExit,
 {
-	let (exit_send, exit) = exit_future::signal();
+	let (exit_send, exit) = oneshot::channel();
 
 	let informant = informant::build(&service);
-	runtime.executor().spawn(exit.until(informant).map(|_| ()));
+
+	let future = select(exit, informant)
+		.map(|_| Ok(()))
+		.compat();
+
+	runtime.executor().spawn(future);
 
 	// we eagerly drop the service so that the internal exit future is fired,
 	// but we need to keep holding a reference to the global telemetry guard
 	let _telemetry = service.telemetry();
 
 	let service_res = {
-		let exit = e.into_exit().map_err(|_| error::Error::Other("Exit future failed.".into()));
-		let service = service.map_err(|err| error::Error::Service(err));
-		let select = service.select(exit).map(|_| ()).map_err(|(err, _)| err);
+		let exit = e.into_exit();
+		let service = service
+			.map_err(|err| error::Error::Service(err))
+			.compat();
+		let select = select(service, exit)
+			.map(|_| Ok(()))
+			.compat();
 		runtime.block_on(select)
 	};
 
-	exit_send.fire();
+	let _ = exit_send.send(());
 
 	// TODO [andre]: timeout this future #1318
+
+	use futures01::Future;
+
 	let _ = runtime.shutdown_on_idle().wait();
 
 	service_res
@@ -96,7 +108,7 @@ where
 // handles ctrl-c
 pub struct Exit;
 impl IntoExit for Exit {
-	type Exit = future::MapErr<oneshot::Receiver<()>, fn(oneshot::Canceled) -> ()>;
+	type Exit = Map<oneshot::Receiver<()>, fn(Result<(), oneshot::Canceled>) -> ()>;
 	fn into_exit(self) -> Self::Exit {
 		// can't use signal directly here because CtrlC takes only `Fn`.
 		let (exit_send, exit) = oneshot::channel();
@@ -109,6 +121,6 @@ impl IntoExit for Exit {
 			}
 		}).expect("Error setting Ctrl-C handler");
 
-		exit.map_err(drop)
+		exit.map(drop)
 	}
 }

bin/node/cli/Cargo.toml

@@ -25,8 +25,8 @@ crate-type = ["cdylib", "rlib"]
 # third-party dependencies
 codec = { package = "parity-scale-codec", version = "1.0.6" }
 serde = { version = "1.0.102", features = [ "derive" ] }
-futures = "0.1.29"
-futures03 = { package = "futures-preview", version = "0.3.0-alpha.19", features = ["compat"] }
+futures01 = { package = "futures", version = "0.1.29" }
+futures = { version = "0.3.1", features = ["compat"] }
 hex-literal = "0.2.1"
 jsonrpc-core = "14.0.3"
 log = "0.4.8"
@@ -80,7 +80,6 @@ node-executor = { path = "../executor" }
 
 # CLI-specific dependencies
 tokio = { version = "0.1.22", optional = true }
-exit-future = { version = "0.1.4", optional = true }
 substrate-cli = { path = "../../../client/cli", optional = true }
 transaction-factory = { path = "../../../test/utils/transaction-factory", optional = true }
 ctrlc = { version = "3.1.3", features = ["termination"], optional = true }
@@ -101,7 +100,7 @@ keystore = { package = "substrate-keystore", path = "../../../client/keystore" }
 babe = { package = "substrate-consensus-babe", path = "../../../client/consensus/babe", features = ["test-helpers"] }
 consensus-common = { package = "substrate-consensus-common", path = "../../../primitives/consensus/common" }
 service-test = { package = "substrate-service-test", path = "../../../client/service/test" }
-futures03 = { package = "futures-preview", version = "0.3.0-alpha.19" }
+futures = "0.3.1"
 tempfile = "3.1.0"
 
 [build-dependencies]
@@ -128,7 +127,6 @@ cli = [
 	"substrate-cli",
 	"transaction-factory",
 	"tokio",
-	"exit-future",
 	"ctrlc",
 	"substrate-service/rocksdb",
 	"node-executor/wasmi-errno",

bin/node/cli/bin/main.rs

@@ -18,16 +18,16 @@
 
 #![warn(missing_docs)]
 
-use futures::sync::oneshot;
-use futures::{future, Future};
+use futures::channel::oneshot;
+use futures::{future, FutureExt};
 use substrate_cli::VersionInfo;
 
 use std::cell::RefCell;
 
 // handles ctrl-c
 struct Exit;
 impl substrate_cli::IntoExit for Exit {
-	type Exit = future::MapErr<oneshot::Receiver<()>, fn(oneshot::Canceled) -> ()>;
+	type Exit = future::Map<oneshot::Receiver<()>, fn(Result<(), oneshot::Canceled>) -> ()>;
 	fn into_exit(self) -> Self::Exit {
 		// can't use signal directly here because CtrlC takes only `Fn`.
 		let (exit_send, exit) = oneshot::channel();
@@ -39,7 +39,7 @@ impl substrate_cli::IntoExit for Exit {
 			}
 		}).expect("Error setting Ctrl-C handler");
 
-		exit.map_err(drop)
+		exit.map(|_| ())
 	}
 }
 

bin/node/cli/src/browser.rs

@@ -15,7 +15,7 @@
 // along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
 
 use crate::ChainSpec;
-use futures::{prelude::*, sync::oneshot, sync::mpsc};
+use futures01::{prelude::*, sync::oneshot, sync::mpsc};
 use libp2p::wasm_ext;
 use log::{debug, info};
 use std::sync::Arc;
@@ -71,7 +71,7 @@ fn start_inner(wasm_ext: wasm_ext::ffi::Transport) -> Result<Client, Box<dyn std
 	// `service.poll()`.
 	// The rest consists in handling RPC requests.
 	let (rpc_send_tx, mut rpc_send_rx) = mpsc::unbounded::<RpcMessage>();
-	wasm_bindgen_futures::spawn_local(futures::future::poll_fn(move || {
+	wasm_bindgen_futures::spawn_local(futures01::future::poll_fn(move || {
 		loop {
 			match rpc_send_rx.poll() {
 				Ok(Async::Ready(Some(message))) => {

bin/node/cli/src/cli.rs

@@ -185,23 +185,34 @@ where
 	T: AbstractService,
 	E: IntoExit,
 {
-	let (exit_send, exit) = exit_future::signal();
+	use futures::{FutureExt, TryFutureExt, channel::oneshot, future::select, compat::Future01CompatExt};
+
+	let (exit_send, exit) = oneshot::channel();
 
 	let informant = substrate_cli::informant::build(&service);
-	runtime.executor().spawn(exit.until(informant).map(|_| ()));
+
+	let future = select(informant, exit)
+		.map(|_| Ok(()))
+		.compat();
+
+	runtime.executor().spawn(future);
 
 	// we eagerly drop the service so that the internal exit future is fired,
 	// but we need to keep holding a reference to the global telemetry guard
 	let _telemetry = service.telemetry();
 
 	let service_res = {
-		let exit = e.into_exit().map_err(|_| error::Error::Other("Exit future failed.".into()));
-		let service = service.map_err(|err| error::Error::Service(err));
-		let select = service.select(exit).map(|_| ()).map_err(|(err, _)| err);
+		let exit = e.into_exit();
+		let service = service
+			.map_err(|err| error::Error::Service(err))
+			.compat();
+		let select = select(service, exit)
+			.map(|_| Ok(()))
+			.compat();
 		runtime.block_on(select)
 	};
 
-	exit_send.fire();
+	let _ = exit_send.send(());
 
 	// TODO [andre]: timeout this future #1318
 	let _ = runtime.shutdown_on_idle().wait();

bin/node/cli/src/service.rs

@@ -110,9 +110,9 @@ macro_rules! new_full_start {
 /// concrete types instead.
 macro_rules! new_full {
 	($config:expr, $with_startup_data: expr) => {{
-		use futures::sync::mpsc;
+		use futures01::sync::mpsc;
 		use network::DhtEvent;
-		use futures03::{
+		use futures::{
 			compat::Stream01CompatExt,
 			stream::StreamExt,
 			future::{FutureExt, TryFutureExt},
@@ -515,7 +515,7 @@ mod tests {
 				digest.push(<DigestItem as CompatibleDigestItem>::babe_pre_digest(babe_pre_digest));
 
 				let mut proposer = proposer_factory.init(&parent_header).unwrap();
-				let new_block = futures03::executor::block_on(proposer.propose(
+				let new_block = futures::executor::block_on(proposer.propose(
 					inherent_data,
 					digest,
 					std::time::Duration::from_secs(1),

client/Cargo.toml

@@ -13,8 +13,7 @@ kvdb = { git = "https://github.com/paritytech/parity-common", rev="b0317f649ab2c
 derive_more = { version = "0.15.0" }
 executor = { package = "substrate-executor", path = "executor" }
 fnv = { version = "1.0.6" }
-futures = { version = "0.1.29" }
-futures03 = { package = "futures-preview", version = "0.3.0-alpha.19", features = ["compat"] }
+futures = { version = "0.3.1", features = ["compat"] }
 hash-db = { version = "0.15.2" }
 header-metadata = { package = "substrate-header-metadata", path = "header-metadata" }
 hex-literal = { version = "0.2.1" }
@@ -38,4 +37,3 @@ client-db = { package = "substrate-client-db", path = "./db", features = ["kvdb-
 test-client = { package = "substrate-test-runtime-client", path = "../test/utils/runtime/client" }
 kvdb-memorydb = { git = "https://github.com/paritytech/parity-common", rev="b0317f649ab2c665b7987b8475878fc4d2e1f81d" }
 panic-handler = { package = "substrate-panic-handler", path = "../primitives/panic-handler" }
-

client/api/Cargo.toml

@@ -11,8 +11,7 @@ consensus = { package = "substrate-consensus-common", path = "../../primitives/c
 derive_more = { version = "0.15.0" }
 executor = { package = "substrate-executor", path = "../executor" }
 fnv = { version = "1.0.6" }
-futures = { version = "0.1.29" }
-futures03 = { package = "futures-preview", version = "0.3.0-alpha.19", features = ["compat"] }
+futures = { version = "0.3.1" }
 hash-db = { version = "0.15.2", default-features = false }
 header-metadata = { package = "substrate-header-metadata", path = "../header-metadata" }
 hex-literal = { version = "0.2.1" }

client/api/src/client.rs

@@ -15,7 +15,7 @@
 // along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
 
 use std::collections::HashMap;
-use futures03::channel::mpsc;
+use futures::channel::mpsc;
 use primitives::storage::StorageKey;
 use state_machine::ExecutionStrategy;
 use sr_primitives::{

client/api/src/light.rs

@@ -286,7 +286,7 @@ pub trait RemoteBlockchain<Block: BlockT>: Send + Sync {
 
 #[cfg(test)]
 pub mod tests {
-	use futures03::future::Ready;
+	use futures::future::Ready;
 	use parking_lot::Mutex;
     use crate::error::Error as ClientError;
     use test_primitives::{Block, Header, Extrinsic};
@@ -298,7 +298,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 {
@@ -321,7 +321,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 {
@@ -332,4 +332,4 @@ pub mod tests {
 			not_implemented_in_tests()
 		}
 	}
-}
+}

client/api/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![];

client/authority-discovery/Cargo.toml

@@ -14,8 +14,8 @@ bytes = "0.4.12"
 client-api = { package = "substrate-client-api", path = "../api" }
 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-timer = "0.4"
+futures = "0.3.1"
+futures-timer = "2.0"
 keystore = { package = "substrate-keystore", path = "../keystore" }
 libp2p = { version = "0.13.0", default-features = false, features = ["secp256k1", "libp2p-websocket"] }
 log = "0.4.8"