Switch offchain workers to new futures (#3285)

* Switch offchain workers to new futures

* Fix tests

core/offchain/Cargo.toml

@@ -8,7 +8,7 @@ edition = "2018"
 
 [dependencies]
 client = { package = "substrate-client", path = "../../core/client" }
-futures = "0.1.25"
+futures-preview = "0.3.0-alpha.17"
 log = "0.4"
 offchain-primitives = { package = "substrate-offchain-primitives", path = "./primitives" }
 codec = { package = "parity-scale-codec", version = "1.0.0", features = ["derive"] }
@@ -23,7 +23,6 @@ keystore = { package = "substrate-keystore", path = "../keystore" }
 env_logger = "0.6"
 client-db = { package = "substrate-client-db", path = "../../core/client/db/", default-features = true }
 test-client = { package = "substrate-test-runtime-client", path = "../../core/test-runtime/client" }
-tokio = "0.1.7"
 
 [features]
 default = []

core/offchain/src/api.rs

@@ -23,7 +23,7 @@ use std::{
 };
 
 use client::backend::OffchainStorage;
-use futures::{Stream, Future, sync::mpsc};
+use futures::{StreamExt as _, Future, future, channel::mpsc};
 use log::{info, debug, warn, error};
 use network::{PeerId, Multiaddr, NetworkStateInfo};
 use codec::{Encode, Decode};
@@ -297,14 +297,14 @@ impl<A: ChainApi> AsyncApi<A> {
 	}
 
 	/// Run a processing task for the API
-	pub fn process(mut self) -> impl Future<Item=(), Error=()> {
+	pub fn process(mut self) -> impl Future<Output = ()> {
 		let receiver = self.receiver.take().expect("Take invoked only once.");
 
 		receiver.for_each(move |msg| {
 			match msg {
 				ExtMessage::SubmitExtrinsic(ext) => self.submit_extrinsic(ext),
 			}
-			Ok(())
+			future::ready(())
 		})
 	}
 

core/offchain/src/lib.rs

@@ -98,7 +98,7 @@ impl<Client, Storage, Block> OffchainWorkers<
 		number: &<Block::Header as traits::Header>::Number,
 		pool: &Arc<Pool<A>>,
 		network_state: Arc<dyn NetworkStateInfo + Send + Sync>,
-	) -> impl Future<Item = (), Error = ()> where
+	) -> impl Future<Output = ()> where
 		A: ChainApi<Block=Block> + 'static,
 	{
 		let runtime = self.client.runtime_api();
@@ -129,9 +129,9 @@ impl<Client, Storage, Block> OffchainWorkers<
 					log::error!("Error running offchain workers at {:?}: {:?}", at, e);
 				}
 			});
-			futures::future::Either::A(runner.process())
+			futures::future::Either::Left(runner.process())
 		} else {
-			futures::future::Either::B(futures::future::ok(()))
+			futures::future::Either::Right(futures::future::ready(()))
 		}
 	}
 }
@@ -152,7 +152,6 @@ fn spawn_worker(f: impl FnOnce() -> () + Send + 'static) {
 #[cfg(test)]
 mod tests {
 	use super::*;
-	use futures::Future;
 	use network::{Multiaddr, PeerId};
 
 	struct MockNetworkStateInfo();
@@ -171,18 +170,16 @@ mod tests {
 	fn should_call_into_runtime_and_produce_extrinsic() {
 		// given
 		let _ = env_logger::try_init();
-		let runtime = tokio::runtime::Runtime::new().unwrap();
 		let client = Arc::new(test_client::new());
 		let pool = Arc::new(Pool::new(Default::default(), transaction_pool::ChainApi::new(client.clone())));
 		let db = client_db::offchain::LocalStorage::new_test();
 		let network_state = Arc::new(MockNetworkStateInfo());
 
 		// when
 		let offchain = OffchainWorkers::new(client, db);
-		runtime.executor().spawn(offchain.on_block_imported(&0u64, &pool, network_state.clone()));
+		futures::executor::block_on(offchain.on_block_imported(&0u64, &pool, network_state));
 
 		// then
-		runtime.shutdown_on_idle().wait().unwrap();
 		assert_eq!(pool.status().ready, 1);
 		assert_eq!(pool.ready().next().unwrap().is_propagateable(), false);
 	}

core/service/src/components.rs

@@ -36,7 +36,7 @@ use crate::config::Configuration;
 use primitives::{Blake2Hasher, H256, traits::BareCryptoStorePtr};
 use rpc::{self, apis::system::SystemInfo};
 use futures::{prelude::*, future::Executor};
-use futures03::channel::mpsc;
+use futures03::{FutureExt as _, channel::mpsc, compat::Compat};
 
 // Type aliases.
 // These exist mainly to avoid typing `<F as Factory>::Foo` all over the code.
@@ -279,7 +279,9 @@ impl<C: Components> OffchainWorker<Self> for C where
 		pool: &Arc<TransactionPool<C::TransactionPoolApi>>,
 		network_state: &Arc<dyn NetworkStateInfo + Send + Sync>,
 	) -> error::Result<Box<dyn Future<Item = (), Error = ()> + Send>> {
-		Ok(Box::new(offchain.on_block_imported(number, pool, network_state.clone())))
+		let future = offchain.on_block_imported(number, pool, network_state.clone())
+			.map(|()| Ok(()));
+		Ok(Box::new(Compat::new(future)))
 	}
 }