Pass optional timeout alongside headers

grandinetech · Jul 22, 2024 · 9d791a0 · 9d791a0
1 parent 7397650
commit 9d791a0
Show file tree

Hide file tree

Showing 8 changed files with 89 additions and 27 deletions.
diff --git a/src/api/eth.rs b/src/api/eth.rs
@@ -40,7 +40,10 @@ impl<T: Transport> Eth<T> {
 
     /// Get current block number
     pub fn block_number(&self, headers: Option<HeaderMap>) -> CallFuture<U64, T::Out> {
-        CallFuture::new(self.transport.execute_with_headers("eth_blockNumber", vec![], headers))
+        CallFuture::new(
+            self.transport
+                .execute_with_headers("eth_blockNumber", vec![], headers, None),
+        )
     }
 
     /// Call a constant method of contract without changing the state of the blockchain.
@@ -122,7 +125,7 @@ impl<T: Transport> Eth<T> {
         let filter = helpers::serialize(&filter);
         CallFuture::new(
             self.transport
-                .execute_with_headers("eth_getLogs", vec![filter], headers),
+                .execute_with_headers("eth_getLogs", vec![filter], headers, None),
         )
     }
 
@@ -134,12 +137,12 @@ impl<T: Transport> Eth<T> {
             BlockId::Hash(hash) => {
                 let hash = helpers::serialize(&hash);
                 self.transport
-                    .execute_with_headers("eth_getBlockByHash", vec![hash, include_txs], headers)
+                    .execute_with_headers("eth_getBlockByHash", vec![hash, include_txs], headers, None)
             }
             BlockId::Number(num) => {
                 let num = helpers::serialize(&num);
                 self.transport
-                    .execute_with_headers("eth_getBlockByNumber", vec![num, include_txs], headers)
+                    .execute_with_headers("eth_getBlockByNumber", vec![num, include_txs], headers, None)
             }
         };
 

diff --git a/src/lib.rs b/src/lib.rs
@@ -11,6 +11,8 @@
 // select! in WS transport
 #![recursion_limit = "256"]
 
+use std::time::Duration;
+
 use headers::HeaderMap;
 use jsonrpc_core as rpc;
 
@@ -53,18 +55,30 @@ pub trait Transport: std::fmt::Debug + Clone {
     fn prepare(&self, method: &str, params: Vec<rpc::Value>) -> (RequestId, rpc::Call);
 
     /// Execute prepared RPC call.
-    fn send(&self, id: RequestId, request: rpc::Call, headers: Option<HeaderMap>) -> Self::Out;
+    fn send(
+        &self,
+        id: RequestId,
+        request: rpc::Call,
+        headers: Option<HeaderMap>,
+        timeout: Option<Duration>,
+    ) -> Self::Out;
 
     /// Execute remote method with given parameters.
     fn execute(&self, method: &str, params: Vec<rpc::Value>) -> Self::Out {
         let (id, request) = self.prepare(method, params);
-        self.send(id, request, None)
+        self.send(id, request, None, None)
     }
 
     /// Execute remote method with given parameters and headers.
-    fn execute_with_headers(&self, method: &str, params: Vec<rpc::Value>, headers: Option<HeaderMap>) -> Self::Out {
+    fn execute_with_headers(
+        &self,
+        method: &str,
+        params: Vec<rpc::Value>,
+        headers: Option<HeaderMap>,
+        timeout: Option<Duration>,
+    ) -> Self::Out {
         let (id, request) = self.prepare(method, params);
-        self.send(id, request, headers)
+        self.send(id, request, headers, timeout)
     }
 }
 
@@ -104,8 +118,14 @@ where
         (**self).prepare(method, params)
     }
 
-    fn send(&self, id: RequestId, request: rpc::Call, headers: Option<HeaderMap>) -> Self::Out {
-        (**self).send(id, request, headers)
+    fn send(
+        &self,
+        id: RequestId,
+        request: rpc::Call,
+        headers: Option<HeaderMap>,
+        timeout: Option<Duration>,
+    ) -> Self::Out {
+        (**self).send(id, request, headers, timeout)
     }
 }
 
@@ -151,7 +171,7 @@ mod tests {
     use crate::api::Web3;
     use futures::future::BoxFuture;
     use headers::HeaderMap;
-    use std::sync::Arc;
+    use std::{sync::Arc, time::Duration};
 
     #[derive(Debug, Clone)]
     struct FakeTransport;
@@ -163,7 +183,13 @@ mod tests {
             unimplemented!()
         }
 
-        fn send(&self, _id: RequestId, _request: rpc::Call, _headers: Option<HeaderMap>) -> Self::Out {
+        fn send(
+            &self,
+            _id: RequestId,
+            _request: rpc::Call,
+            _headers: Option<HeaderMap>,
+            _timeout: Option<Duration>,
+        ) -> Self::Out {
             unimplemented!()
         }
     }

diff --git a/src/transports/batch.rs b/src/transports/batch.rs
@@ -11,7 +11,7 @@ use futures::{
 };
 use headers::HeaderMap;
 use parking_lot::Mutex;
-use std::{collections::BTreeMap, pin::Pin, sync::Arc};
+use std::{collections::BTreeMap, pin::Pin, sync::Arc, time::Duration};
 
 type Pending = oneshot::Sender<error::Result<rpc::Value>>;
 type PendingRequests = Arc<Mutex<BTreeMap<RequestId, Pending>>>;
@@ -76,7 +76,13 @@ where
         self.transport.prepare(method, params)
     }
 
-    fn send(&self, id: RequestId, request: rpc::Call, _headers: Option<HeaderMap>) -> Self::Out {
+    fn send(
+        &self,
+        id: RequestId,
+        request: rpc::Call,
+        _headers: Option<HeaderMap>,
+        _timeout: Option<Duration>,
+    ) -> Self::Out {
         let (tx, rx) = oneshot::channel();
         self.pending.lock().insert(id, tx);
         self.batch.lock().push((id, request));

diff --git a/src/transports/either.rs b/src/transports/either.rs
@@ -1,5 +1,7 @@
 //! A strongly-typed transport alternative.
 
+use std::time::Duration;
+
 use crate::{api, error, rpc, BatchTransport, DuplexTransport, RequestId, Transport};
 use futures::{
     future::{BoxFuture, FutureExt},
@@ -37,10 +39,16 @@ where
         }
     }
 
-    fn send(&self, id: RequestId, request: rpc::Call, headers: Option<HeaderMap>) -> Self::Out {
+    fn send(
+        &self,
+        id: RequestId,
+        request: rpc::Call,
+        headers: Option<HeaderMap>,
+        timeout: Option<Duration>,
+    ) -> Self::Out {
         match *self {
-            Self::Left(ref a) => a.send(id, request, headers).boxed(),
-            Self::Right(ref b) => b.send(id, request, headers).boxed(),
+            Self::Left(ref a) => a.send(id, request, headers, timeout).boxed(),
+            Self::Right(ref b) => b.send(id, request, headers, timeout).boxed(),
         }
     }
 }

diff --git a/src/transports/http.rs b/src/transports/http.rs
@@ -18,6 +18,7 @@ use std::{
         atomic::{AtomicUsize, Ordering},
         Arc,
     },
+    time::Duration,
 };
 
 /// HTTP Transport
@@ -80,6 +81,7 @@ async fn execute_rpc<T: DeserializeOwned>(
     request: &Request,
     id: RequestId,
     headers: Option<HeaderMap>,
+    timeout: Option<Duration>,
 ) -> Result<T> {
     log::debug!("[id:{}] sending request: {:?}", id, serde_json::to_string(&request)?);
     let mut request_builder = client.post(url).json(request);
@@ -88,6 +90,10 @@ async fn execute_rpc<T: DeserializeOwned>(
         request_builder = request_builder.headers(headers);
     }
 
+    if let Some(timeout) = timeout {
+        request_builder = request_builder.timeout(timeout);
+    }
+
     let response = request_builder
         .send()
         .await
@@ -127,10 +133,10 @@ impl Transport for Http {
         (id, request)
     }
 
-    fn send(&self, id: RequestId, call: Call, headers: Option<HeaderMap>) -> Self::Out {
+    fn send(&self, id: RequestId, call: Call, headers: Option<HeaderMap>, timeout: Option<Duration>) -> Self::Out {
         let (client, url) = self.new_request();
         Box::pin(async move {
-            let output: Output = execute_rpc(&client, url, &Request::Single(call), id, headers).await?;
+            let output: Output = execute_rpc(&client, url, &Request::Single(call), id, headers, timeout).await?;
             helpers::to_result_from_output(output)
         })
     }
@@ -148,7 +154,7 @@ impl BatchTransport for Http {
         let (client, url) = self.new_request();
         let (ids, calls): (Vec<_>, Vec<_>) = requests.into_iter().unzip();
         Box::pin(async move {
-            let value = execute_rpc(&client, url, &Request::Batch(calls), id, None).await?;
+            let value = execute_rpc(&client, url, &Request::Batch(calls), id, None, None).await?;
             let outputs = handle_possible_error_object_for_batched_request(value)?;
             handle_batch_response(&ids, outputs)
         })

diff --git a/src/transports/ipc.rs b/src/transports/ipc.rs
@@ -16,6 +16,7 @@ use std::{
     pin::Pin,
     sync::{atomic::AtomicUsize, Arc},
     task::{Context, Poll},
+    time::Duration,
 };
 use tokio::{
     io::AsyncWriteExt,
@@ -64,7 +65,13 @@ impl Transport for Ipc {
         (id, request)
     }
 
-    fn send(&self, id: RequestId, call: rpc::Call, _headers: Option<HeaderMap>) -> Self::Out {
+    fn send(
+        &self,
+        id: RequestId,
+        call: rpc::Call,
+        _headers: Option<HeaderMap>,
+        _timeout: Option<Duration>,
+    ) -> Self::Out {
         let (response_tx, response_rx) = oneshot::channel();
         let message = TransportMessage::Single((id, call, response_tx));
 
@@ -357,7 +364,7 @@ mod test {
                 "test": -1,
             })],
         );
-        let response = ipc.send(req_id, request, None).await;
+        let response = ipc.send(req_id, request, None, None).await;
         let expected_response_json: serde_json::Value = json!({
             "test": 1,
         });
@@ -369,7 +376,7 @@ mod test {
                 "test": 3,
             })],
         );
-        let response = ipc.send(req_id, request, None).await;
+        let response = ipc.send(req_id, request, None, None).await;
         let expected_response_json: serde_json::Value = json!({
             "test": "string1",
         });

diff --git a/src/transports/test.rs b/src/transports/test.rs
@@ -6,7 +6,7 @@ use crate::{
 };
 use futures::future::{self, BoxFuture, FutureExt};
 use headers::HeaderMap;
-use std::{cell::RefCell, collections::VecDeque, rc::Rc};
+use std::{cell::RefCell, collections::VecDeque, rc::Rc, time::Duration};
 
 type Result<T> = BoxFuture<'static, error::Result<T>>;
 
@@ -27,7 +27,13 @@ impl Transport for TestTransport {
         (self.requests.borrow().len(), request)
     }
 
-    fn send(&self, id: RequestId, request: rpc::Call, _headers: Option<HeaderMap>) -> Result<rpc::Value> {
+    fn send(
+        &self,
+        id: RequestId,
+        request: rpc::Call,
+        _headers: Option<HeaderMap>,
+        _timeout: Option<Duration>,
+    ) -> Result<rpc::Value> {
         future::ready(match self.responses.borrow_mut().pop_front() {
             Some(response) => Ok(response),
             None => {

diff --git a/src/transports/ws.rs b/src/transports/ws.rs
@@ -21,7 +21,7 @@ use std::{
     fmt,
     marker::Unpin,
     pin::Pin,
-    sync::{atomic, Arc},
+    sync::{atomic, Arc}, time::Duration,
 };
 use url::Url;
 
@@ -474,7 +474,7 @@ impl Transport for WebSocket {
         (id, request)
     }
 
-    fn send(&self, id: RequestId, request: rpc::Call, _headers: Option<HeaderMap>) -> Self::Out {
+    fn send(&self, id: RequestId, request: rpc::Call, _headers: Option<HeaderMap>, _timeout: Option<Duration>) -> Self::Out {
         let response = self.send_request(id, rpc::Request::Single(request));
         Response::new(response, batch_to_single)
     }