[dependencies]: upgrade tokio to v0.3

Cargo.toml

@@ -24,11 +24,11 @@ serde_json = "1.0.58"
 smallvec = { version = "1.4.2", default-features = false }
 thiserror = "1.0.20"
 
-# used for tests and HTTP/hyper
-# the features are used only for tests but enabled because `dev-depedencies` are leaked into dependencies.
-tokio = { version = "0.2.22", features = ["stream", "rt-threaded", "macros"], optional = true }
-
 # HTTP-related dependencies
+# the features `rt-multi-thread` and `macros` are used only for tests but enabled because `dev-dependencies` are leaked into dependencies.
+tokio = { version = "0.3.0", features = ["stream", "rt-multi-thread", "macros"], optional = true }
+# hyper is using tokio v0.2, remove when hyper upgrades.
+tokio-compat-02 = "0.1.0"
 hyper = { version = "0.13.8", features = ["stream"], optional = true }
 unicase = { version = "2.6.0", optional = true }
 

src/client/http/transport.rs

@@ -17,9 +17,9 @@ use std::pin::Pin;
 use std::{fmt, io, thread};
 
 use crate::common;
-
 use futures::{channel::mpsc, prelude::*};
 use thiserror::Error;
+use tokio_compat_02::FutureExt;
 
 /// Implementation of a raw client for HTTP requests.
 pub struct HttpTransportClient {
@@ -52,12 +52,14 @@ impl HttpTransportClient {
 			.name("jsonrpsee-hyper-client".to_string())
 			.spawn(move || {
 				let client = hyper::Client::new();
+				// hyper v0.13 requires tokio v0.2 that's why compat() is used.
 				background_thread(requests_rx, move |rq| {
 					// cloning Hyper client = cloning references
 					let client = client.clone();
 					async move {
 						let _ = rq.send_back.unbounded_send(client.request(rq.request).await);
 					}
+					.compat()
 				})
 			})
 			.unwrap();
@@ -173,7 +175,7 @@ fn background_thread<T, ProcessRequest: Future<Output = ()>>(
 	mut requests_rx: mpsc::Receiver<T>,
 	process_request: impl Fn(T) -> ProcessRequest,
 ) {
-	let mut runtime = match tokio::runtime::Builder::new().basic_scheduler().enable_all().build() {
+	let runtime = match tokio::runtime::Builder::new_current_thread().enable_all().build() {
 		Ok(r) => r,
 		Err(err) => {
 			// Ideally, we would try to initialize the tokio runtime in the main thread then move

src/http/transport/background.rs

@@ -31,6 +31,7 @@ use futures::{channel::mpsc, channel::oneshot, prelude::*};
 use hyper::service::{make_service_fn, service_fn};
 use hyper::Error;
 use std::{error, io, net::SocketAddr, thread};
+use tokio_compat_02::FutureExt;
 
 /// Background thread that serves HTTP requests.
 pub(super) struct BackgroundHttp {
@@ -79,29 +80,33 @@ impl BackgroundHttp {
 
 		// Because hyper can only be polled through tokio, we spawn it in a background thread.
 		thread::Builder::new().name("jsonrpsee-hyper-server".to_string()).spawn(move || {
-			let mut runtime = match tokio::runtime::Builder::new().basic_scheduler().enable_all().build() {
+			let runtime = match tokio::runtime::Builder::new_current_thread().enable_all().build() {
 				Ok(r) => r,
 				Err(err) => {
 					log::error!("Failed to initialize tokio runtime in HTTP JSON-RPC server: {}", err);
 					return;
 				}
 			};
 
-			runtime.block_on(async move {
-				match hyper::Server::try_bind(&addr) {
-					Ok(builder) => {
-						let server = builder.serve(make_service);
-						let _ = addr_tx.send(Ok(server.local_addr()));
-						if let Err(err) = server.await {
-							log::error!("HTTP JSON-RPC server closed with an error: {}", err);
+			// hyper v0.13 is using tokio v0.2 that's why compat() is used.
+			runtime.block_on(
+				async move {
+					match hyper::Server::try_bind(&addr) {
+						Ok(builder) => {
+							let server = builder.serve(make_service);
+							let _ = addr_tx.send(Ok(server.local_addr()));
+							if let Err(err) = server.await {
+								log::error!("HTTP JSON-RPC server closed with an error: {}", err);
+							}
 						}
-					}
-					Err(err) => {
-						log::error!("Failed to bind to address {}: {}", addr, err);
-						let _ = addr_tx.send(Err(err));
-					}
-				};
-			});
+						Err(err) => {
+							log::error!("Failed to bind to address {}: {}", addr, err);
+							let _ = addr_tx.send(Err(err));
+						}
+					};
+				}
+				.compat(),
+			);
 		})?;
 
 		let local_addr = addr_rx.await??;

test-utils/Cargo.toml

@@ -13,5 +13,6 @@ hyper = "0.13.8"
 serde = { version = "1.0.116", default-features = false, features = ["derive"] }
 serde_json = "1.0.58"
 soketto = "0.4.2"
-tokio = { version = "0.2.22", features = ["dns", "stream", "tcp", "rt-threaded", "macros"] }
-tokio-util = { version = "0.3", features = ["compat"] }
+tokio = { version = "0.3", features = ["net", "rt-multi-thread", "macros"] }
+tokio-util = { version = "0.4", features = ["compat"] }
+tokio-compat-02 = "0.1.0"

test-utils/src/helpers.rs

@@ -1,6 +1,7 @@
 use crate::types::{Body, HttpResponse, Id, Uri};
 use serde_json::Value;
 use std::net::SocketAddr;
+use tokio_compat_02::FutureExt;
 
 /// Converts a sockaddress to a WebSocket URI.
 pub fn to_ws_uri_string(addr: SocketAddr) -> String {
@@ -54,7 +55,7 @@ pub async fn http_request(body: Body, uri: Uri) -> Result<HttpResponse, String>
 		.header(hyper::header::CONTENT_TYPE, hyper::header::HeaderValue::from_static("application/json"))
 		.body(body)
 		.expect("uri and request headers are valid; qed");
-	let res = client.request(r).await.map_err(|e| format!("{:?}", e))?;
+	let res = client.request(r).compat().await.map_err(|e| format!("{:?}", e))?;
 
 	let (parts, body) = res.into_parts();
 	let bytes = hyper::body::to_bytes(body).await.unwrap();