Replace custom hyper client with reqwest (#593)

* [DNM] Replace custom hyper client with reqwest

* Fix runtime error

impls/src/adapters/http.rs

@@ -13,7 +13,7 @@
 // limitations under the License.
 
 /// HTTP Wallet 'plugin' implementation
-use crate::client_utils::{Client, ClientError};
+use crate::client_utils::{Client, ClientError, ClientErrorKind};
 use crate::libwallet::slate_versions::{SlateVersion, VersionedSlate};
 use crate::libwallet::{Error, ErrorKind, Slate};
 use crate::SlateSender;
@@ -163,11 +163,15 @@ impl HttpSlateSender {
 	where
 		IN: Serialize,
 	{
-		let mut client = Client::new();
-		if self.use_socks {
-			client.use_socks = true;
-			client.socks_proxy_addr = self.socks_proxy_addr;
-		}
+		let client =
+			if self.use_socks {
+				Client::new()
+			} else {
+				Client::with_socks_proxy(self.socks_proxy_addr.ok_or_else(|| {
+					ClientErrorKind::Internal("No socks proxy address set".into())
+				})?)
+			}
+			.map_err(|_| ClientErrorKind::Internal("Unable to create http client".into()))?;
 		let req = client.create_post_request(url, api_secret, &input)?;
 		let res = client.send_request(req)?;
 		Ok(res)

impls/src/adapters/keybase.rs

@@ -345,7 +345,7 @@ impl SlateReceiver for KeybaseAllChannels {
 		account: &str,
 		node_api_secret: Option<String>,
 	) -> Result<(), Error> {
-		let node_client = HTTPNodeClient::new(&config.check_node_api_http_addr, node_api_secret);
+		let node_client = HTTPNodeClient::new(&config.check_node_api_http_addr, node_api_secret)?;
 		let mut wallet =
 			Box::new(DefaultWalletImpl::<'static, HTTPNodeClient>::new(node_client).unwrap())
 				as Box<

impls/src/client_utils/client.rs

@@ -16,12 +16,9 @@
 
 use crate::util::to_base64;
 use failure::{Backtrace, Context, Fail, ResultExt};
-use hyper::body;
-use hyper::header::{ACCEPT, AUTHORIZATION, CONTENT_TYPE, USER_AGENT};
-use hyper::{self, Body, Client as HyperClient, Request, Uri};
-use hyper_rustls;
-use hyper_timeout::TimeoutConnector;
 use lazy_static::lazy_static;
+use reqwest::header::{HeaderMap, HeaderValue, ACCEPT, AUTHORIZATION, CONTENT_TYPE, USER_AGENT};
+use reqwest::{ClientBuilder, Method, Proxy, RequestBuilder};
 use serde::{Deserialize, Serialize};
 use serde_json;
 use std::fmt::{self, Display};
@@ -103,19 +100,41 @@ impl From<Context<ErrorKind>> for Error {
 
 #[derive(Clone)]
 pub struct Client {
-	/// Whether to use socks proxy
-	pub use_socks: bool,
-	/// Proxy url/port
-	pub socks_proxy_addr: Option<SocketAddr>,
+	client: reqwest::Client,
 }
 
 impl Client {
 	/// New client
-	pub fn new() -> Self {
-		Client {
-			use_socks: false,
-			socks_proxy_addr: None,
+	pub fn new() -> Result<Self, Error> {
+		Self::build(None)
+	}
+
+	pub fn with_socks_proxy(socks_proxy_addr: SocketAddr) -> Result<Self, Error> {
+		Self::build(Some(socks_proxy_addr))
+	}
+
+	fn build(socks_proxy_addr: Option<SocketAddr>) -> Result<Self, Error> {
+		let mut headers = HeaderMap::new();
+		headers.insert(USER_AGENT, HeaderValue::from_static("grin-client"));
+		headers.insert(ACCEPT, HeaderValue::from_static("application/json"));
+		headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
+
+		let mut builder = ClientBuilder::new()
+			.timeout(Duration::from_secs(20))
+			.use_rustls_tls()
+			.default_headers(headers);
+
+		if let Some(s) = socks_proxy_addr {
+			let proxy = Proxy::all(&format!("socks5://{}:{}", s.ip(), s.port()))
+				.map_err(|e| ErrorKind::Internal(format!("Unable to create proxy: {}", e)))?;
+			builder = builder.proxy(proxy);
 		}
+
+		let client = builder
+			.build()
+			.map_err(|e| ErrorKind::Internal(format!("Unable to build client: {}", e)))?;
+
+		Ok(Client { client })
 	}
 
 	/// Helper function to easily issue a HTTP GET request against a given URL that
@@ -125,7 +144,7 @@ impl Client {
 	where
 		for<'de> T: Deserialize<'de>,
 	{
-		self.handle_request(self.build_request(url, "GET", api_secret, None)?)
+		self.handle_request(self.build_request(url, Method::GET, api_secret, None)?)
 	}
 
 	/// Helper function to easily issue an async HTTP GET request against a given
@@ -139,7 +158,7 @@ impl Client {
 	where
 		for<'de> T: Deserialize<'de> + Send + 'static,
 	{
-		self.handle_request_async(self.build_request(url, "GET", api_secret, None)?)
+		self.handle_request_async(self.build_request(url, Method::GET, api_secret, None)?)
 			.await
 	}
 
@@ -147,7 +166,7 @@ impl Client {
 	/// on a given URL that returns nothing. Handles request
 	/// building and response code checking.
 	pub fn _get_no_ret(&self, url: &str, api_secret: Option<String>) -> Result<(), Error> {
-		let req = self.build_request(url, "GET", api_secret, None)?;
+		let req = self.build_request(url, Method::GET, api_secret, None)?;
 		self.send_request(req)?;
 		Ok(())
 	}
@@ -228,32 +247,21 @@ impl Client {
 	fn build_request(
 		&self,
 		url: &str,
-		method: &str,
+		method: Method,
 		api_secret: Option<String>,
 		body: Option<String>,
-	) -> Result<Request<Body>, Error> {
-		let uri: Uri = url
-			.parse()
-			.map_err(|_| ErrorKind::RequestError(format!("Invalid url {}", url)))?;
-		let mut builder = Request::builder();
+	) -> Result<RequestBuilder, Error> {
+		let mut builder = self.client.request(method, url);
 		if let Some(api_secret) = api_secret {
 			let basic_auth = format!("Basic {}", to_base64(&format!("grin:{}", api_secret)));
 			builder = builder.header(AUTHORIZATION, basic_auth);
 		}
 
-		builder
-			.method(method)
-			.uri(uri)
-			.header(USER_AGENT, "grin-client")
-			.header(ACCEPT, "application/json")
-			.header(CONTENT_TYPE, "application/json")
-			.body(match body {
-				None => Body::empty(),
-				Some(json) => json.into(),
-			})
-			.map_err(|e| {
-				ErrorKind::RequestError(format!("Bad request {} {}: {}", method, url, e)).into()
-			})
+		if let Some(body) = body {
+			builder = builder.body(body);
+		}
+
+		Ok(builder)
 	}
 
 	pub fn create_post_request<IN>(
@@ -261,17 +269,17 @@ impl Client {
 		url: &str,
 		api_secret: Option<String>,
 		input: &IN,
-	) -> Result<Request<Body>, Error>
+	) -> Result<RequestBuilder, Error>
 	where
 		IN: Serialize,
 	{
 		let json = serde_json::to_string(input).context(ErrorKind::Internal(
 			"Could not serialize data to JSON".to_owned(),
 		))?;
-		self.build_request(url, "POST", api_secret, Some(json))
+		self.build_request(url, Method::POST, api_secret, Some(json))
 	}
 
-	fn handle_request<T>(&self, req: Request<Body>) -> Result<T, Error>
+	fn handle_request<T>(&self, req: RequestBuilder) -> Result<T, Error>
 	where
 		for<'de> T: Deserialize<'de>,
 	{
@@ -282,7 +290,7 @@ impl Client {
 		})
 	}
 
-	async fn handle_request_async<T>(&self, req: Request<Body>) -> Result<T, Error>
+	async fn handle_request_async<T>(&self, req: RequestBuilder) -> Result<T, Error>
 	where
 		for<'de> T: Deserialize<'de> + Send + 'static,
 	{
@@ -292,54 +300,19 @@ impl Client {
 		Ok(ser)
 	}
 
-	async fn send_request_async(&self, req: Request<Body>) -> Result<String, Error> {
-		let resp = if !self.use_socks {
-			let https = hyper_rustls::HttpsConnector::new();
-			let mut connector = TimeoutConnector::new(https);
-			connector.set_connect_timeout(Some(Duration::from_secs(20)));
-			connector.set_read_timeout(Some(Duration::from_secs(20)));
-			connector.set_write_timeout(Some(Duration::from_secs(20)));
-			let client = HyperClient::builder().build::<_, Body>(connector);
-
-			client.request(req).await
-		} else {
-			let addr = self.socks_proxy_addr.ok_or_else(|| {
-				ErrorKind::RequestError("Missing Socks proxy address".to_string())
-			})?;
-			let auth = format!("{}:{}", addr.ip(), addr.port());
-
-			let https = hyper_rustls::HttpsConnector::new();
-			let socks = hyper_socks2_mw::SocksConnector {
-				proxy_addr: hyper::Uri::builder()
-					.scheme("socks5")
-					.authority(auth.as_str())
-					.path_and_query("/")
-					.build()
-					.map_err(|_| {
-						ErrorKind::RequestError("Can't parse Socks proxy address".to_string())
-					})?,
-				auth: None,
-				connector: https,
-			};
-			let mut connector = TimeoutConnector::new(socks);
-			connector.set_connect_timeout(Some(Duration::from_secs(20)));
-			connector.set_read_timeout(Some(Duration::from_secs(20)));
-			connector.set_write_timeout(Some(Duration::from_secs(20)));
-			let client = HyperClient::builder().build::<_, Body>(connector);
-
-			client.request(req).await
-		};
-		let resp =
-			resp.map_err(|e| ErrorKind::RequestError(format!("Cannot make request: {}", e)))?;
-
-		let raw = body::to_bytes(resp)
+	async fn send_request_async(&self, req: RequestBuilder) -> Result<String, Error> {
+		let resp = req
+			.send()
 			.await
-			.map_err(|e| ErrorKind::RequestError(format!("Cannot read response body: {}", e)))?;
-
-		Ok(String::from_utf8_lossy(&raw).to_string())
+			.map_err(|e| ErrorKind::RequestError(format!("Cannot make request: {}", e)))?;
+		let text = resp
+			.text()
+			.await
+			.map_err(|e| ErrorKind::ResponseError(format!("Cannot parse response: {}", e)))?;
+		Ok(text)
 	}
 
-	pub fn send_request(&self, req: Request<Body>) -> Result<String, Error> {
+	pub fn send_request(&self, req: RequestBuilder) -> Result<String, Error> {
 		// This client is currently used both outside and inside of a tokio runtime
 		// context. In the latter case we are not allowed to do a blocking call to
 		// our global runtime, which unfortunately means we have to spawn a new thread

impls/src/client_utils/json_rpc.rs

@@ -16,7 +16,6 @@
 //! JSON RPC Client functionality
 use std::{error, fmt};
 
-use hyper;
 use serde_json;
 
 /// Builds a request
@@ -97,8 +96,6 @@ impl Response {
 pub enum Error {
 	/// Json error
 	Json(serde_json::Error),
-	/// Client error
-	Hyper(hyper::error::Error),
 	/// Error response
 	Rpc(RpcError),
 	/// Response to a request did not have the expected nonce
@@ -121,12 +118,6 @@ impl From<serde_json::Error> for Error {
 	}
 }
 
-impl From<hyper::error::Error> for Error {
-	fn from(e: hyper::error::Error) -> Error {
-		Error::Hyper(e)
-	}
-}
-
 impl From<RpcError> for Error {
 	fn from(e: RpcError) -> Error {
 		Error::Rpc(e)
@@ -137,7 +128,6 @@ impl fmt::Display for Error {
 	fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
 		match *self {
 			Error::Json(ref e) => write!(f, "JSON decode error: {}", e),
-			Error::Hyper(ref e) => write!(f, "Hyper error: {}", e),
 			Error::Rpc(ref r) => write!(f, "RPC error response: {:?}", r),
 			Error::_BatchDuplicateResponseId(ref v) => {
 				write!(f, "duplicate RPC batch response ID: {}", v)
@@ -152,7 +142,6 @@ impl std::error::Error for Error {
 	fn description(&self) -> &str {
 		match *self {
 			Error::Json(_) => "JSON decode error",
-			Error::Hyper(_) => "Hyper error",
 			Error::Rpc(_) => "RPC error response",
 			Error::_NonceMismatch => "Nonce of response did not match nonce of request",
 			Error::_VersionMismatch => "`jsonrpc` field set to non-\"2.0\"",
@@ -168,7 +157,6 @@ impl std::error::Error for Error {
 	fn cause(&self) -> Option<&dyn error::Error> {
 		match *self {
 			Error::Json(ref e) => Some(e),
-			Error::Hyper(ref e) => Some(e),
 			_ => None,
 		}
 	}

impls/src/client_utils/mod.rs

@@ -15,4 +15,4 @@
 mod client;
 pub mod json_rpc;
 
-pub use client::{Client, Error as ClientError, RUNTIME};
+pub use client::{Client, Error as ClientError, ErrorKind as ClientErrorKind, RUNTIME};

impls/src/node_clients/http.rs

@@ -21,7 +21,6 @@ use futures::stream::FuturesUnordered;
 use futures::TryStreamExt;
 use std::collections::HashMap;
 use std::env;
-use tokio::runtime::Handle;
 
 use crate::client_utils::{Client, RUNTIME};
 use crate::libwallet;
@@ -35,6 +34,7 @@ const ENDPOINT: &str = "/v2/foreign";
 
 #[derive(Clone)]
 pub struct HTTPNodeClient {
+	client: Client,
 	node_url: String,
 	node_api_secret: Option<String>,
 	node_version_info: Option<NodeVersionInfo>,
@@ -42,12 +42,16 @@ pub struct HTTPNodeClient {
 
 impl HTTPNodeClient {
 	/// Create a new client that will communicate with the given grin node
-	pub fn new(node_url: &str, node_api_secret: Option<String>) -> HTTPNodeClient {
-		HTTPNodeClient {
+	pub fn new(
+		node_url: &str,
+		node_api_secret: Option<String>,
+	) -> Result<HTTPNodeClient, libwallet::Error> {
+		Ok(HTTPNodeClient {
+			client: Client::new().map_err(|_| libwallet::ErrorKind::Node)?,
 			node_url: node_url.to_owned(),
 			node_api_secret: node_api_secret,
 			node_version_info: None,
-		}
+		})
 	}
 
 	/// Allow returning the chain height without needing a wallet instantiated
@@ -61,9 +65,10 @@ impl HTTPNodeClient {
 		params: &serde_json::Value,
 	) -> Result<D, libwallet::Error> {
 		let url = format!("{}{}", self.node_url(), ENDPOINT);
-		let client = Client::new();
 		let req = build_request(method, params);
-		let res = client.post::<Request, Response>(url.as_str(), self.node_api_secret(), &req);
+		let res = self
+			.client
+			.post::<Request, Response>(url.as_str(), self.node_api_secret(), &req);
 
 		match res {
 			Err(e) => {
@@ -156,9 +161,10 @@ impl NodeClient for HTTPNodeClient {
 		let params = json!([excess.0.as_ref().to_hex(), min_height, max_height]);
 		// have to handle this manually since the error needs to be parsed
 		let url = format!("{}{}", self.node_url(), ENDPOINT);
-		let client = Client::new();
 		let req = build_request(method, &params);
-		let res = client.post::<Request, Response>(url.as_str(), self.node_api_secret(), &req);
+		let res = self
+			.client
+			.post::<Request, Response>(url.as_str(), self.node_api_secret(), &req);
 
 		match res {
 			Err(e) => {
@@ -225,8 +231,6 @@ impl NodeClient for HTTPNodeClient {
 		let url = format!("{}{}", self.node_url(), ENDPOINT);
 		let api_secret = self.node_api_secret();
 		let task = async move {
-			let client = Client::new();
-
 			let params: Vec<_> = query_params
 				.chunks(chunk_size)
 				.map(|c| json!([c, null, null, false, false]))
@@ -239,7 +243,7 @@ impl NodeClient for HTTPNodeClient {
 
 			let mut tasks = Vec::with_capacity(params.len());
 			for req in &reqs {
-				tasks.push(client.post_async::<Request, Response>(
+				tasks.push(self.client.post_async::<Request, Response>(
 					url.as_str(),
 					req,
 					api_secret.clone(),
@@ -250,14 +254,7 @@ impl NodeClient for HTTPNodeClient {
 			task.try_collect().await
 		};
 
-		let res: Result<Vec<Response>, _> = if Handle::try_current().is_ok() {
-			let rt = RUNTIME.clone();
-			std::thread::spawn(move || rt.lock().unwrap().block_on(task))
-				.join()
-				.unwrap()
-		} else {
-			RUNTIME.lock().unwrap().block_on(task)
-		};
+		let res: Result<Vec<_>, _> = RUNTIME.lock().unwrap().block_on(task);
 
 		let results: Vec<OutputPrintable> = match res {
 			Ok(resps) => {