Slurp refactoring #914

mm2src/coins/eth.rs

@@ -24,7 +24,8 @@ use common::custom_futures::TimedAsyncMutex;
 use common::executor::Timer;
 use common::mm_ctx::{MmArc, MmWeak};
 use common::mm_error::prelude::*;
-use common::{now_ms, slurp_url, small_rng, DEX_FEE_ADDR_RAW_PUBKEY};
+use common::transport::{slurp_url, SlurpError};
+use common::{now_ms, small_rng, DEX_FEE_ADDR_RAW_PUBKEY};
 use derive_more::Display;
 use ethabi::{Contract, Token};
 use ethcore_transaction::{Action, Transaction as UnSignedEthTx, UnverifiedTransaction};
@@ -109,16 +110,33 @@ pub type GasStationResult = Result<GasStationData, MmError<GasStationReqErr>>;
 
 #[derive(Debug, Display)]
 pub enum GasStationReqErr {
-    #[display(fmt = "Transport: {}", _0)]
-    Transport(String),
+    #[display(fmt = "Transport '{}' error: {}", uri, error)]
+    Transport {
+        uri: String,
+        error: String,
+    },
     #[display(fmt = "Invalid response: {}", _0)]
     InvalidResponse(String),
+    Internal(String),
 }
 
 impl From<serde_json::Error> for GasStationReqErr {
     fn from(e: serde_json::Error) -> Self { GasStationReqErr::InvalidResponse(e.to_string()) }
 }
 
+impl From<SlurpError> for GasStationReqErr {
+    fn from(e: SlurpError) -> Self {
+        let error = e.to_string();
+        match e {
+            SlurpError::ErrorDeserializing { .. } => GasStationReqErr::InvalidResponse(error),
+            SlurpError::Transport { uri, .. } | SlurpError::Timeout { uri, .. } => {
+                GasStationReqErr::Transport { uri, error }
+            },
+            SlurpError::Internal(_) | SlurpError::InvalidRequest(_) => GasStationReqErr::Internal(error),
+        }
+    }
+}
+
 #[derive(Debug, Display)]
 pub enum Web3RpcError {
     #[display(fmt = "Transport: {}", _0)]
@@ -132,8 +150,9 @@ pub enum Web3RpcError {
 impl From<GasStationReqErr> for Web3RpcError {
     fn from(err: GasStationReqErr) -> Self {
         match err {
-            GasStationReqErr::Transport(err) => Web3RpcError::Transport(err),
+            GasStationReqErr::Transport { .. } => Web3RpcError::Transport(err.to_string()),
             GasStationReqErr::InvalidResponse(err) => Web3RpcError::InvalidResponse(err),
+            GasStationReqErr::Internal(err) => Web3RpcError::Internal(err),
         }
     }
 }
@@ -292,10 +311,13 @@ pub enum EthAddressFormat {
 
 #[cfg_attr(test, mockable)]
 async fn make_gas_station_request(url: &str) -> GasStationResult {
-    let resp = slurp_url(url).await.map_to_mm(GasStationReqErr::Transport)?;
+    let resp = slurp_url(url).await?;
     if resp.0 != StatusCode::OK {
         let error = format!("Gas price request failed with status code {}", resp.0);
-        return MmError::err(GasStationReqErr::Transport(error));
+        return MmError::err(GasStationReqErr::Transport {
+            uri: url.to_owned(),
+            error,
+        });
     }
     let result: GasStationData = json::from_slice(&resp.2)?;
     Ok(result)

mm2src/coins/eth/web3_transport.rs

@@ -1,4 +1,5 @@
 use super::{RpcTransportEventHandler, RpcTransportEventHandlerShared};
+use common::log::warn;
 #[cfg(not(target_arch = "wasm32"))] use futures::FutureExt;
 use futures::TryFutureExt;
 use futures01::{Future, Poll};
@@ -11,12 +12,14 @@ use web3::error::{Error, ErrorKind};
 use web3::helpers::{build_request, to_result_from_output, to_string};
 use web3::{RequestId, Transport};
 
+const REQUEST_TIMEOUT_S: f64 = 60.;
+
 /// Parse bytes RPC response into `Result`.
 /// Implementation copied from Web3 HTTP transport
 #[cfg(not(target_arch = "wasm32"))]
-fn single_response<T: Deref<Target = [u8]>>(response: T) -> Result<Json, Error> {
-    let response =
-        serde_json::from_slice(&*response).map_err(|e| Error::from(ErrorKind::InvalidResponse(format!("{}", e))))?;
+fn single_response<T: Deref<Target = [u8]>>(response: T, rpc_url: &str) -> Result<Json, Error> {
+    let response = serde_json::from_slice(&*response)
+        .map_err(|e| Error::from(ErrorKind::InvalidResponse(format!("{}: {}", rpc_url, e))))?;
 
     match response {
         Response::Single(output) => to_result_from_output(output),
@@ -107,7 +110,7 @@ async fn send_request(
     event_handlers: Vec<RpcTransportEventHandlerShared>,
 ) -> Result<Json, Error> {
     use common::executor::Timer;
-    use common::wio::slurp_reqʹ;
+    use common::transport::slurp_req;
     use futures::future::{select, Either};
     use gstuff::binprint;
     use http::header::HeaderValue;
@@ -122,33 +125,40 @@ async fn send_request(
         *req.uri_mut() = uri.clone();
         req.headers_mut()
             .insert(http::header::CONTENT_TYPE, HeaderValue::from_static("application/json"));
-        let timeout = Timer::sleep(60.);
-        let req = Box::pin(slurp_reqʹ(req));
+        let timeout = Timer::sleep(REQUEST_TIMEOUT_S);
+        let req = Box::pin(slurp_req(req));
         let rc = select(req, timeout).await;
         let res = match rc {
             Either::Left((r, _t)) => r,
             Either::Right((_t, _r)) => {
-                errors.push(ERRL!("timeout"));
+                let error = ERRL!("Error requesting '{}': {}s timeout expired", uri, REQUEST_TIMEOUT_S);
+                warn!("{}", error);
+                errors.push(error);
                 continue;
             },
         };
 
         let (status, _headers, body) = match res {
             Ok(r) => r,
             Err(err) => {
-                errors.push(err);
+                errors.push(err.to_string());
                 continue;
             },
         };
 
         event_handlers.on_incoming_response(&body);
 
         if !status.is_success() {
-            errors.push(ERRL!("!200: {}, {}", status, binprint(&body, b'.')));
+            errors.push(ERRL!(
+                "Server '{}' response !200: {}, {}",
+                uri,
+                status,
+                binprint(&body, b'.')
+            ));
             continue;
         }
 
-        return single_response(body);
+        return single_response(body, &uri.to_string());
     }
     Err(ErrorKind::Transport(fomat!(
         "request " [request] " failed: "
@@ -189,7 +199,7 @@ async fn send_request_once(
     uri: &http::Uri,
     event_handlers: &Vec<RpcTransportEventHandlerShared>,
 ) -> Result<Json, Error> {
-    use common::wasm_http::FetchRequest;
+    use common::transport::wasm_http::FetchRequest;
 
     macro_rules! try_or {
         ($exp:expr, $errkind:ident) => {

mm2src/coins/utxo/rpc_clients.rs

@@ -538,7 +538,7 @@ impl JsonRpcClient for NativeClientImpl {
 
     #[cfg(not(target_arch = "wasm32"))]
     fn transport(&self, request: JsonRpcRequest) -> JsonRpcResponseFut {
-        use common::wio::slurp_req;
+        use common::transport::slurp_req;
 
         let request_body = try_fus!(json::to_string(&request));
         // measure now only body length, because the `hyper` crate doesn't allow to get total HTTP packet length
@@ -2052,7 +2052,7 @@ async fn connect_loop(
         static ref CONN_IDX: Arc<AtomicUsize> = Arc::new(AtomicUsize::new(0));
     }
 
-    use common::wasm_ws::ws_transport;
+    use common::transport::wasm_ws::ws_transport;
 
     let mut delay: u64 = 0;
     loop {