Add execute_with_headers method to Transport trait to allow passing H…

…TTP headers

Cargo.toml

@@ -14,8 +14,8 @@ edition = "2018"
 [dependencies]
 arrayvec = "0.7.1"
 derive_more = "0.99.1"
-ethabi = "17.0.0"
-ethereum-types = "0.13.0"
+ethabi = "18.0.0"
+ethereum-types = "0.14.1"
 futures = "0.3.5"
 futures-timer = "3.0.2"
 hex = "0.4"

examples/bench.rs

@@ -30,7 +30,7 @@ where
     for _ in 0..max {
         let ticker = ticker.clone();
         ticker.start();
-        let accounts = web3.eth().block_number().then(move |res| {
+        let accounts = web3.eth().block_number(None).then(move |res| {
             if let Err(e) = res {
                 println!("Error: {:?}", e);
             }

examples/transport_batch.rs

@@ -5,7 +5,7 @@ async fn main() -> web3::Result {
     let web3 = web3::Web3::new(web3::transports::Batch::new(http));
 
     let accounts = web3.eth().accounts();
-    let block = web3.eth().block_number();
+    let block = web3.eth().block_number(None);
 
     let result = web3.transport().submit_batch().await?;
     println!("Result: {:?}", result);

src/api/eth.rs

@@ -1,5 +1,7 @@
 //! `Eth` namespace
 
+use headers::HeaderMap;
+
 use crate::{
     api::Namespace,
     helpers::{self, CallFuture},
@@ -37,8 +39,8 @@ impl<T: Transport> Eth<T> {
     }
 
     /// Get current block number
-    pub fn block_number(&self) -> CallFuture<U64, T::Out> {
-        CallFuture::new(self.transport.execute("eth_blockNumber", vec![]))
+    pub fn block_number(&self, headers: Option<HeaderMap>) -> CallFuture<U64, T::Out> {
+        CallFuture::new(self.transport.execute_with_headers("eth_blockNumber", vec![], headers))
     }
 
     /// Call a constant method of contract without changing the state of the blockchain.
@@ -116,23 +118,28 @@ impl<T: Transport> Eth<T> {
     }
 
     /// Get all logs matching a given filter object
-    pub fn logs(&self, filter: Filter) -> CallFuture<Vec<Log>, T::Out> {
+    pub fn logs(&self, filter: Filter, headers: Option<HeaderMap>) -> CallFuture<Vec<Log>, T::Out> {
         let filter = helpers::serialize(&filter);
-        CallFuture::new(self.transport.execute("eth_getLogs", vec![filter]))
+        CallFuture::new(
+            self.transport
+                .execute_with_headers("eth_getLogs", vec![filter], headers),
+        )
     }
 
     /// Get block details with transaction hashes.
-    pub fn block(&self, block: BlockId) -> CallFuture<Option<Block<H256>>, T::Out> {
+    pub fn block(&self, block: BlockId, headers: Option<HeaderMap>) -> CallFuture<Option<Block<H256>>, T::Out> {
         let include_txs = helpers::serialize(&false);
 
         let result = match block {
             BlockId::Hash(hash) => {
                 let hash = helpers::serialize(&hash);
-                self.transport.execute("eth_getBlockByHash", vec![hash, include_txs])
+                self.transport
+                    .execute_with_headers("eth_getBlockByHash", vec![hash, include_txs], headers)
             }
             BlockId::Number(num) => {
                 let num = helpers::serialize(&num);
-                self.transport.execute("eth_getBlockByNumber", vec![num, include_txs])
+                self.transport
+                    .execute_with_headers("eth_getBlockByNumber", vec![num, include_txs], headers)
             }
         };
 
@@ -558,7 +565,7 @@ mod tests {
     );
 
     rpc_test! (
-      Eth:block_number => "eth_blockNumber";
+      Eth:block_number, None => "eth_blockNumber", Vec::<String>::new();
       Value::String("0x123".into()) => 0x123
     );
 
@@ -653,21 +660,21 @@ mod tests {
     );
 
     rpc_test! (
-      Eth:logs, FilterBuilder::default().build() => "eth_getLogs", vec!["{}"];
+      Eth:logs, FilterBuilder::default().build(), None => "eth_getLogs", vec!["{}"];
       Value::Array(vec![::serde_json::from_str(EXAMPLE_LOG).unwrap()])
       => vec![::serde_json::from_str::<Log>(EXAMPLE_LOG).unwrap()]
     );
 
     rpc_test! (
-      Eth:block:block_by_hash, BlockId::Hash(H256::from_low_u64_be(0x123))
+      Eth:block:block_by_hash, BlockId::Hash(H256::from_low_u64_be(0x123)), None
       =>
       "eth_getBlockByHash", vec![r#""0x0000000000000000000000000000000000000000000000000000000000000123""#, r#"false"#];
       ::serde_json::from_str(EXAMPLE_BLOCK).unwrap()
       => Some(::serde_json::from_str::<Block<H256>>(EXAMPLE_BLOCK).unwrap())
     );
 
     rpc_test! (
-      Eth:block, BlockNumber::Pending
+      Eth:block, BlockNumber::Pending, None
       =>
       "eth_getBlockByNumber", vec![r#""pending""#, r#"false"#];
       ::serde_json::from_str(EXAMPLE_PENDING_BLOCK).unwrap()

src/confirm.rs

@@ -55,7 +55,7 @@ where
     loop {
         let _ = filter_stream.next().await;
         if let Some(confirmation_block_number) = check.check().await? {
-            let block_number = eth.block_number().await?;
+            let block_number = eth.block_number(None).await?;
             if confirmation_block_number.low_u64() + confirmations as u64 <= block_number.low_u64() {
                 return Ok(());
             }

src/contract/mod.rs

@@ -310,7 +310,7 @@ impl<T: Transport> Contract<T> {
 
         let logs = self
             .eth
-            .logs(FilterBuilder::default().topic_filter(filter).build())
+            .logs(FilterBuilder::default().topic_filter(filter).build(), None)
             .await?;
         logs.into_iter()
             .map(move |l| {

src/lib.rs

@@ -11,6 +11,7 @@
 // select! in WS transport
 #![recursion_limit = "256"]
 
+use headers::HeaderMap;
 use jsonrpc_core as rpc;
 
 /// Re-export of the `futures` crate.
@@ -52,12 +53,18 @@ 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) -> Self::Out;
+    fn send(&self, id: RequestId, request: rpc::Call, headers: Option<HeaderMap>) -> 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)
+        self.send(id, request, 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 {
+        let (id, request) = self.prepare(method, params);
+        self.send(id, request, headers)
     }
 }
 
@@ -97,8 +104,8 @@ where
         (**self).prepare(method, params)
     }
 
-    fn send(&self, id: RequestId, request: rpc::Call) -> Self::Out {
-        (**self).send(id, request)
+    fn send(&self, id: RequestId, request: rpc::Call, headers: Option<HeaderMap>) -> Self::Out {
+        (**self).send(id, request, headers)
     }
 }
 
@@ -143,6 +150,7 @@ mod tests {
 
     use crate::api::Web3;
     use futures::future::BoxFuture;
+    use headers::HeaderMap;
     use std::sync::Arc;
 
     #[derive(Debug, Clone)]
@@ -155,7 +163,7 @@ mod tests {
             unimplemented!()
         }
 
-        fn send(&self, _id: RequestId, _request: rpc::Call) -> Self::Out {
+        fn send(&self, _id: RequestId, _request: rpc::Call, _headers: Option<HeaderMap>) -> Self::Out {
             unimplemented!()
         }
     }

src/transports/batch.rs

@@ -9,6 +9,7 @@ use futures::{
     task::{Context, Poll},
     Future, FutureExt,
 };
+use headers::HeaderMap;
 use parking_lot::Mutex;
 use std::{collections::BTreeMap, pin::Pin, sync::Arc};
 
@@ -72,7 +73,7 @@ where
         self.transport.prepare(method, params)
     }
 
-    fn send(&self, id: RequestId, request: rpc::Call) -> Self::Out {
+    fn send(&self, id: RequestId, request: rpc::Call, _headers: Option<HeaderMap>) -> Self::Out {
         let (tx, rx) = oneshot::channel();
         self.pending.lock().insert(id, tx);
         self.batch.lock().push((id, request));

src/transports/either.rs

@@ -5,6 +5,7 @@ use futures::{
     future::{BoxFuture, FutureExt},
     stream::{BoxStream, StreamExt},
 };
+use headers::HeaderMap;
 
 /// A wrapper over two possible transports.
 ///
@@ -36,10 +37,10 @@ where
         }
     }
 
-    fn send(&self, id: RequestId, request: rpc::Call) -> Self::Out {
+    fn send(&self, id: RequestId, request: rpc::Call, headers: Option<HeaderMap>) -> Self::Out {
         match *self {
-            Self::Left(ref a) => a.send(id, request).boxed(),
-            Self::Right(ref b) => b.send(id, request).boxed(),
+            Self::Left(ref a) => a.send(id, request, headers).boxed(),
+            Self::Right(ref b) => b.send(id, request, headers).boxed(),
         }
     }
 }

src/transports/http.rs

@@ -8,6 +8,7 @@ use crate::{
 use futures::future::BoxFuture;
 #[cfg(feature = "wasm")]
 use futures::future::LocalBoxFuture as BoxFuture;
+use headers::HeaderMap;
 use jsonrpc_core::types::{Call, Output, Request, Value};
 use reqwest::{Client, Url};
 use serde::de::DeserializeOwned;
@@ -73,11 +74,21 @@ impl Http {
 }
 
 // Id is only used for logging.
-async fn execute_rpc<T: DeserializeOwned>(client: &Client, url: Url, request: &Request, id: RequestId) -> Result<T> {
+async fn execute_rpc<T: DeserializeOwned>(
+    client: &Client,
+    url: Url,
+    request: &Request,
+    id: RequestId,
+    headers: Option<HeaderMap>,
+) -> Result<T> {
     log::debug!("[id:{}] sending request: {:?}", id, serde_json::to_string(&request)?);
-    let response = client
-        .post(url)
-        .json(request)
+    let mut request_builder = client.post(url).json(request);
+
+    if let Some(headers) = headers {
+        request_builder = request_builder.headers(headers);
+    }
+
+    let response = request_builder
         .send()
         .await
         .map_err(|err| Error::Transport(TransportError::Message(format!("failed to send request: {}", err))))?;
@@ -116,10 +127,10 @@ impl Transport for Http {
         (id, request)
     }
 
-    fn send(&self, id: RequestId, call: Call) -> Self::Out {
+    fn send(&self, id: RequestId, call: Call, headers: Option<HeaderMap>) -> Self::Out {
         let (client, url) = self.new_request();
         Box::pin(async move {
-            let output: Output = execute_rpc(&client, url, &Request::Single(call), id).await?;
+            let output: Output = execute_rpc(&client, url, &Request::Single(call), id, headers).await?;
             helpers::to_result_from_output(output)
         })
     }
@@ -137,7 +148,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 outputs: Vec<Output> = execute_rpc(&client, url, &Request::Batch(calls), id).await?;
+            let outputs: Vec<Output> = execute_rpc(&client, url, &Request::Batch(calls), id, None).await?;
             handle_batch_response(&ids, outputs)
         })
     }

src/transports/ipc.rs

@@ -8,6 +8,7 @@ use futures::{
     future::{join_all, JoinAll},
     stream::StreamExt,
 };
+use headers::HeaderMap;
 use jsonrpc_core as rpc;
 use std::{
     collections::BTreeMap,
@@ -63,7 +64,7 @@ impl Transport for Ipc {
         (id, request)
     }
 
-    fn send(&self, id: RequestId, call: rpc::Call) -> Self::Out {
+    fn send(&self, id: RequestId, call: rpc::Call, _headers: Option<HeaderMap>) -> Self::Out {
         let (response_tx, response_rx) = oneshot::channel();
         let message = TransportMessage::Single((id, call, response_tx));
 
@@ -356,7 +357,7 @@ mod test {
                 "test": -1,
             })],
         );
-        let response = ipc.send(req_id, request).await;
+        let response = ipc.send(req_id, request, None).await;
         let expected_response_json: serde_json::Value = json!({
             "test": 1,
         });
@@ -368,7 +369,7 @@ mod test {
                 "test": 3,
             })],
         );
-        let response = ipc.send(req_id, request).await;
+        let response = ipc.send(req_id, request, None).await;
         let expected_response_json: serde_json::Value = json!({
             "test": "string1",
         });

src/transports/test.rs

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

src/transports/ws.rs

@@ -11,6 +11,7 @@ use futures::{
     task::{Context, Poll},
     AsyncRead, AsyncWrite, Future, FutureExt, Stream, StreamExt,
 };
+use headers::HeaderMap;
 use soketto::{
     connection,
     handshake::{Client, ServerResponse},
@@ -438,7 +439,7 @@ impl Transport for WebSocket {
         (id, request)
     }
 
-    fn send(&self, id: RequestId, request: rpc::Call) -> Self::Out {
+    fn send(&self, id: RequestId, request: rpc::Call, _headers: Option<HeaderMap>) -> Self::Out {
         let response = self.send_request(id, rpc::Request::Single(request));
         Response::new(response, batch_to_single)
     }