Prefer using tokio_compat everywhere for now

Author: Xanewok

http/src/lib.rs

@@ -53,7 +53,7 @@ pub use crate::handler::ServerHandler;
 pub use crate::response::Response;
 pub use crate::server_utils::cors::{self, AccessControlAllowOrigin, AllowCors, Origin};
 pub use crate::server_utils::hosts::{DomainsValidation, Host};
-pub use crate::server_utils::{tokio, SuspendableStream};
+pub use crate::server_utils::{tokio, tokio_compat, SuspendableStream};
 pub use crate::utils::{cors_allow_headers, cors_allow_origin, is_host_allowed};
 
 /// Action undertaken by a middleware.
@@ -300,7 +300,7 @@ where
 	/// Utilize existing event loop executor to poll RPC results.
 	///
 	/// Applies only to 1 of the threads. Other threads will spawn their own Event Loops.
-	pub fn event_loop_executor(mut self, executor: tokio::runtime::TaskExecutor) -> Self {
+	pub fn event_loop_executor(mut self, executor: tokio_compat::runtime::TaskExecutor) -> Self {
 		self.executor = UninitializedExecutor::Shared(executor);
 		self
 	}
@@ -519,7 +519,7 @@ fn serve<M: jsonrpc::Metadata, S: jsonrpc::Middleware<M>>(
 		mpsc::Sender<io::Result<SocketAddr>>,
 		oneshot::Sender<()>,
 	),
-	executor: tokio::runtime::TaskExecutor,
+	executor: tokio_compat::runtime::TaskExecutor,
 	addr: SocketAddr,
 	cors_domains: CorsDomains,
 	cors_max_age: Option<u32>,

ipc/src/server.rs

@@ -11,8 +11,8 @@ use tokio_service::{self, Service as _};
 
 use crate::server_utils::{
 	codecs, reactor, session,
-	tokio::{reactor::Handle, runtime::TaskExecutor},
-	tokio_compat,
+	tokio::reactor::Handle,
+	tokio_compat::runtime::TaskExecutor,
 	tokio_util,
 };
 
@@ -107,12 +107,6 @@ where
 		self
 	}
 
-	/// Sets shared different event loop executor.
-	pub fn event_loop_executor_compat(mut self, executor: tokio_compat::runtime::TaskExecutor) -> Self {
-		self.executor = reactor::UninitializedExecutor::Compat(executor);
-		self
-	}
-
 	/// Sets different event loop I/O reactor.
 	pub fn event_loop_reactor(mut self, reactor: Handle) -> Self {
 		self.reactor = Some(reactor);
@@ -565,9 +559,7 @@ mod tests {
 		};
 
 		let io = MetaIoHandler::<Arc<SessionEndMeta>>::default();
-		let mut rt = tokio_compat::runtime::Builder::new().build().unwrap();
 		let builder = ServerBuilder::with_meta_extractor(io, session_metadata_extractor);
-		let builder = builder.event_loop_executor_compat(rt.executor());
 		let server = builder.start(path).expect("Server must run with no issues");
 		{
 			let _ = UnixStream::connect(path).wait().expect("Socket should connect");

server-utils/src/reactor.rs

@@ -6,17 +6,14 @@
 //! that `tokio::runtime` can be multi-threaded.
 
 use std::io;
-use tokio;
 
 use futures01::Future;
 
 /// Possibly uninitialized event loop executor.
 #[derive(Debug)]
 pub enum UninitializedExecutor {
 	/// Shared instance of executor.
-	Shared(tokio::runtime::TaskExecutor),
-	/// Shared instance of executor.
-	Compat(tokio_compat::runtime::TaskExecutor),
+	Shared(tokio_compat::runtime::TaskExecutor),
 	/// Event Loop should be spawned by the transport.
 	Unspawned,
 }
@@ -35,7 +32,6 @@ impl UninitializedExecutor {
 	pub fn init_with_name<T: Into<String>>(self, name: T) -> io::Result<Executor> {
 		match self {
 			UninitializedExecutor::Shared(executor) => Ok(Executor::Shared(executor)),
-			UninitializedExecutor::Compat(executor) => Ok(Executor::Compat(executor)),
 			UninitializedExecutor::Unspawned => RpcEventLoop::with_name(Some(name.into())).map(Executor::Spawned),
 		}
 	}
@@ -45,19 +41,16 @@ impl UninitializedExecutor {
 #[derive(Debug)]
 pub enum Executor {
 	/// Shared instance
-	Shared(tokio::runtime::TaskExecutor),
-	/// Shared instance
-	Compat(tokio_compat::runtime::TaskExecutor),
+	Shared(tokio_compat::runtime::TaskExecutor),
 	/// Spawned Event Loop
 	Spawned(RpcEventLoop),
 }
 
 impl Executor {
 	/// Get tokio executor associated with this event loop.
-	pub fn executor(&self) -> tokio::runtime::TaskExecutor {
-		match *self {
+	pub fn executor(&self) -> tokio_compat::runtime::TaskExecutor {
+		match self {
 			Executor::Shared(ref executor) => executor.clone(),
-			Executor::Compat(..) => panic!(),
 			Executor::Spawned(ref eloop) => eloop.executor(),
 		}
 	}
@@ -69,7 +62,6 @@ impl Executor {
 	{
 		match self {
 			Executor::Shared(exe) => exe.spawn(future),
-			Executor::Compat(exe) => exe.spawn(future),
 			Executor::Spawned(eloop) => eloop.executor().spawn(future),
 		}
 	}
@@ -92,9 +84,9 @@ impl Executor {
 /// A handle to running event loop. Dropping the handle will cause event loop to finish.
 #[derive(Debug)]
 pub struct RpcEventLoop {
-	executor: tokio::runtime::TaskExecutor,
+	executor: tokio_compat::runtime::TaskExecutor,
 	close: Option<futures01::Complete<()>>,
-	handle: Option<tokio::runtime::Shutdown>,
+	runtime: Option<tokio_compat::runtime::Runtime>,
 }
 
 impl Drop for RpcEventLoop {
@@ -113,34 +105,33 @@ impl RpcEventLoop {
 	pub fn with_name(name: Option<String>) -> io::Result<Self> {
 		let (stop, stopped) = futures01::oneshot();
 
-		let mut tb = tokio::runtime::Builder::new();
+		let mut tb = tokio_compat::runtime::Builder::new();
 		tb.core_threads(1);
 
 		if let Some(name) = name {
 			tb.name_prefix(name);
 		}
 
-		let mut runtime = tb.build()?;
+		let runtime = tb.build()?;
 		let executor = runtime.executor();
 		let terminate = futures01::empty().select(stopped).map(|_| ()).map_err(|_| ());
 		runtime.spawn(terminate);
-		let handle = runtime.shutdown_on_idle();
 
 		Ok(RpcEventLoop {
 			executor,
 			close: Some(stop),
-			handle: Some(handle),
+			runtime: Some(runtime),
 		})
 	}
 
 	/// Get executor for this event loop.
-	pub fn executor(&self) -> tokio::runtime::TaskExecutor {
+	pub fn executor(&self) -> tokio_compat::runtime::TaskExecutor {
 		self.executor.clone()
 	}
 
 	/// Blocks current thread and waits until the event loop is finished.
 	pub fn wait(mut self) -> Result<(), ()> {
-		self.handle.take().ok_or(())?.wait()
+		self.runtime.take().ok_or(())?.shutdown_on_idle().wait()
 	}
 
 	/// Finishes this event loop.

tcp/src/server.rs

@@ -8,7 +8,7 @@ use futures01::sync::oneshot;
 use futures01::{future, Future, Sink, Stream};
 
 use crate::jsonrpc::{middleware, MetaIoHandler, Metadata, Middleware};
-use crate::server_utils::{codecs, reactor, tokio, tokio_codec::Framed, SuspendableStream};
+use crate::server_utils::{codecs, reactor, tokio, tokio_compat, tokio_codec::Framed, SuspendableStream};
 
 use crate::dispatch::{Dispatcher, PeerMessageQueue, SenderChannels};
 use crate::meta::{MetaExtractor, NoopExtractor, RequestContext};
@@ -60,7 +60,7 @@ where
 	}
 
 	/// Utilize existing event loop executor.
-	pub fn event_loop_executor(mut self, handle: tokio::runtime::TaskExecutor) -> Self {
+	pub fn event_loop_executor(mut self, handle: tokio_compat::runtime::TaskExecutor) -> Self {
 		self.executor = reactor::UninitializedExecutor::Shared(handle);
 		self
 	}

ws/src/metadata.rs

@@ -3,7 +3,7 @@ use std::sync::{atomic, Arc};
 
 use crate::core;
 use crate::core::futures::channel::mpsc;
-use crate::server_utils::{session, tokio::runtime::TaskExecutor};
+use crate::server_utils::{session, tokio_compat::runtime::TaskExecutor};
 use crate::ws;
 
 use crate::error;

ws/src/server_builder.rs

@@ -69,7 +69,7 @@ where
 	}
 
 	/// Utilize existing event loop executor to poll RPC results.
-	pub fn event_loop_executor(mut self, executor: server_utils::tokio::runtime::TaskExecutor) -> Self {
+	pub fn event_loop_executor(mut self, executor: server_utils::tokio_compat::runtime::TaskExecutor) -> Self {
 		self.executor = UninitializedExecutor::Shared(executor);
 		self
 	}

ws/src/session.rs

@@ -11,7 +11,7 @@ use slab::Slab;
 use crate::server_utils::cors::Origin;
 use crate::server_utils::hosts::Host;
 use crate::server_utils::session::{SessionId, SessionStats};
-use crate::server_utils::tokio::runtime::TaskExecutor;
+use crate::server_utils::tokio_compat::runtime::TaskExecutor;
 use crate::server_utils::Pattern;
 use crate::ws;