Problem: no functionality exists to dynamically fetch gas-prices

Currently, gas-prices are set upon bridge startup via the
Users's config TOML file; this value remains constant for the
life of the Bridge.

Solution: create a mechanism that asynchronously queries
gas-prices from an "Oracle" service on a timed interval. This
mechanism should be a stream of gas-prices that can be polled
from the Bridge.

Cargo.lock

@@ -157,6 +157,8 @@ dependencies = [
  "ethcore-transaction 0.1.0 (git+http://github.com/paritytech/parity?rev=991f0ca)",
  "ethereum-types 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "futures 0.1.21 (registry+https://github.com/rust-lang/crates.io-index)",
+ "hyper 0.11.27 (registry+https://github.com/rust-lang/crates.io-index)",
+ "hyper-tls 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)",
  "itertools 0.7.8 (registry+https://github.com/rust-lang/crates.io-index)",
  "jsonrpc-core 8.0.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "keccak-hash 0.1.0 (git+http://github.com/paritytech/parity?rev=991f0ca)",
@@ -792,7 +794,7 @@ source = "git+http://github.com/paritytech/parity?rev=991f0ca#991f0cac6ebb75b270
 dependencies = [
  "futures 0.1.21 (registry+https://github.com/rust-lang/crates.io-index)",
  "futures-timer 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
- "hyper 0.11.25 (registry+https://github.com/rust-lang/crates.io-index)",
+ "hyper 0.11.27 (registry+https://github.com/rust-lang/crates.io-index)",
  "hyper-rustls 0.11.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "log 0.4.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "tokio-core 0.1.17 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -997,7 +999,7 @@ dependencies = [
 
 [[package]]
 name = "hyper"
-version = "0.11.25"
+version = "0.11.27"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 dependencies = [
  "base64 0.9.1 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -1009,6 +1011,7 @@ dependencies = [
  "language-tags 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)",
  "log 0.4.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "mime 0.3.5 (registry+https://github.com/rust-lang/crates.io-index)",
+ "net2 0.2.32 (registry+https://github.com/rust-lang/crates.io-index)",
  "percent-encoding 1.0.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "relay 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "time 0.1.39 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -1017,6 +1020,7 @@ dependencies = [
  "tokio-proto 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "tokio-service 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "unicase 2.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
+ "want 0.0.4 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
 [[package]]
@@ -1026,7 +1030,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 dependencies = [
  "ct-logs 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "futures 0.1.21 (registry+https://github.com/rust-lang/crates.io-index)",
- "hyper 0.11.25 (registry+https://github.com/rust-lang/crates.io-index)",
+ "hyper 0.11.27 (registry+https://github.com/rust-lang/crates.io-index)",
  "rustls 0.11.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "tokio-core 0.1.17 (registry+https://github.com/rust-lang/crates.io-index)",
  "tokio-io 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -1042,7 +1046,7 @@ version = "0.1.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 dependencies = [
  "futures 0.1.21 (registry+https://github.com/rust-lang/crates.io-index)",
- "hyper 0.11.25 (registry+https://github.com/rust-lang/crates.io-index)",
+ "hyper 0.11.27 (registry+https://github.com/rust-lang/crates.io-index)",
  "native-tls 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)",
  "tokio-core 0.1.17 (registry+https://github.com/rust-lang/crates.io-index)",
  "tokio-io 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -2332,6 +2336,7 @@ version = "0.1.0"
 dependencies = [
  "bridge 0.2.0",
  "ethabi 5.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "ethcore 1.11.0 (git+http://github.com/paritytech/parity?rev=991f0ca)",
  "ethereum-types 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "futures 0.1.21 (registry+https://github.com/rust-lang/crates.io-index)",
  "jsonrpc-core 8.0.1 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -2606,6 +2611,11 @@ dependencies = [
  "rlp 0.2.1 (git+http://github.com/paritytech/parity?rev=991f0ca)",
 ]
 
+[[package]]
+name = "try-lock"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+
 [[package]]
 name = "typeable"
 version = "0.1.2"
@@ -2766,6 +2776,16 @@ name = "void"
 version = "1.0.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 
+[[package]]
+name = "want"
+version = "0.0.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "futures 0.1.21 (registry+https://github.com/rust-lang/crates.io-index)",
+ "log 0.4.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "try-lock 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
 [[package]]
 name = "wasm"
 version = "0.1.0"
@@ -2802,7 +2822,7 @@ dependencies = [
  "ethabi 5.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "ethereum-types 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "futures 0.1.21 (registry+https://github.com/rust-lang/crates.io-index)",
- "hyper 0.11.25 (registry+https://github.com/rust-lang/crates.io-index)",
+ "hyper 0.11.27 (registry+https://github.com/rust-lang/crates.io-index)",
  "hyper-tls 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)",
  "jsonrpc-core 8.0.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "log 0.4.1 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -3003,7 +3023,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 "checksum humantime 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)" = "0484fda3e7007f2a4a0d9c3a703ca38c71c54c55602ce4660c419fd32e188c9e"
 "checksum hyper 0.10.0-a.0 (git+https://github.com/paritytech/hyper)" = "<none>"
 "checksum hyper 0.10.13 (registry+https://github.com/rust-lang/crates.io-index)" = "368cb56b2740ebf4230520e2b90ebb0461e69034d85d1945febd9b3971426db2"
-"checksum hyper 0.11.25 (registry+https://github.com/rust-lang/crates.io-index)" = "549dbb86397490ce69d908425b9beebc85bbaad25157d67479d4995bb56fdf9a"
+"checksum hyper 0.11.27 (registry+https://github.com/rust-lang/crates.io-index)" = "34a590ca09d341e94cddf8e5af0bbccde205d5fbc2fa3c09dd67c7f85cea59d7"
 "checksum hyper-rustls 0.11.0 (registry+https://github.com/rust-lang/crates.io-index)" = "1d6cdc1751771a14b8175764394f025e309a28c825ed9eaf97fa62bb831dc8c5"
 "checksum hyper-tls 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)" = "a5aa51f6ae9842239b0fac14af5f22123b8432b4cc774a44ff059fcba0f675ca"
 "checksum idna 0.1.4 (registry+https://github.com/rust-lang/crates.io-index)" = "014b298351066f1512874135335d62a789ffe78a9974f94b43ed5621951eaf7d"
@@ -3179,6 +3199,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 "checksum transient-hashmap 0.4.1 (registry+https://github.com/rust-lang/crates.io-index)" = "aeb4b191d033a35edfce392a38cdcf9790b6cebcb30fa690c312c29da4dc433e"
 "checksum trezor-sys 1.0.0 (git+https://github.com/paritytech/trezor-sys)" = "<none>"
 "checksum triehash 0.1.0 (git+http://github.com/paritytech/parity?rev=991f0ca)" = "<none>"
+"checksum try-lock 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "ee2aa4715743892880f70885373966c83d73ef1b0838a664ef0c76fffd35e7c2"
 "checksum typeable 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)" = "1410f6f91f21d1612654e7cc69193b0334f909dcf2c790c4826254fbb86f8887"
 "checksum ucd-util 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)" = "fd2be2d6639d0f8fe6cdda291ad456e23629558d466e2789d2c3e9892bda285d"
 "checksum uint 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)" = "6477b2716357758c176c36719023e1f9726974d762150e4fc0a9c8c75488c343"
@@ -3202,6 +3223,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 "checksum version_check 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)" = "6b772017e347561807c1aa192438c5fd74242a670a6cffacc40f2defd1dc069d"
 "checksum vm 0.1.0 (git+http://github.com/paritytech/parity?rev=991f0ca)" = "<none>"
 "checksum void 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)" = "6a02e4885ed3bc0f2de90ea6dd45ebcbb66dacffe03547fadbb0eeae2770887d"
+"checksum want 0.0.4 (registry+https://github.com/rust-lang/crates.io-index)" = "a05d9d966753fa4b5c8db73fcab5eed4549cfe0e1e4e66911e5564a0085c35d1"
 "checksum wasm 0.1.0 (git+http://github.com/paritytech/parity?rev=991f0ca)" = "<none>"
 "checksum wasmi 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)" = "d19da510b59247935ad5f598357b3cc739912666d75d3d28318026478d95bbdb"
 "checksum web3 0.3.0 (registry+https://github.com/rust-lang/crates.io-index)" = "a806fc4a6fe10166a083806e3c06b21c7aa5e6f2cf8094416835f1070844e13a"

README.md

@@ -117,6 +117,10 @@ withdraw_confirm = { gas = 3000000, gas_price = 1000000000 }
 - `home/foreign.poll_interval` - specify how often home node should be polled for changes (in seconds, default: **1**)
 - `home/foreign.request_timeout` - specify request timeout (in seconds, default: **3600**)
 - `home/foreign.password` - path to the file containing a password for the validator's account (to decrypt the key from the keystore)
+- `home/foreign.gas_price_oracle_url` - the URL used to query the current gas-price for the home and foreign nodes, this service is known as the gas-price Oracle. This config option defaults to `None` if not supplied in the User's config TOML file. If this config value is `None`, no Oracle gas-price querying will occur, resulting in the config value for `home/foreign.default_gas_price` being used for all gas-prices.
+- `home/foreign.gas_price_timeout` - the number of seconds to wait for an HTTP response from the gas price oracle before using the default gas price. Defaults to `10 seconds`.
+- `home/foreign.gas_price_speed_type` - retrieve the gas-price corresponding to this speed when querying from an Oracle. Defaults to `fast`. The available values are: "instant", "fast", "standard", and "slow".
+- `home/foreign.default_gas_price` - the default gas price (in WEI) used in transactions with the home or foreign nodes. The `default_gas_price` is used when the Oracle cannot be reached. The default value is `15_000_000_000` WEI (ie. 15 GWEI).
 
 #### authorities options
 

bridge/Cargo.toml

@@ -25,6 +25,8 @@ keccak-hash = { git = "http://github.com/paritytech/parity", rev = "991f0ca" }
 ethcore-transaction = { git = "http://github.com/paritytech/parity", rev = "991f0ca" }
 itertools = "0.7"
 jsonrpc-core = "8.0"
+hyper = "0.11.27"
+hyper-tls = "0.1.3"
 
 [dev-dependencies]
 tempdir = "0.3"

bridge/src/bridge/deposit_relay.rs

@@ -42,7 +42,7 @@ enum DepositRelayState<T: Transport> {
 	Yield(Option<u64>),
 }
 
-pub fn create_deposit_relay<T: Transport + Clone>(app: Arc<App<T>>, init: &Database, foreign_balance: Arc<RwLock<Option<U256>>>, foreign_chain_id: u64) -> DepositRelay<T> {
+pub fn create_deposit_relay<T: Transport + Clone>(app: Arc<App<T>>, init: &Database, foreign_balance: Arc<RwLock<Option<U256>>>, foreign_chain_id: u64, foreign_gas_price: Arc<RwLock<u64>>) -> DepositRelay<T> {
 	let logs_init = api::LogStreamInit {
 		after: init.checked_deposit_relay,
 		request_timeout: app.config.home.request_timeout,
@@ -57,6 +57,7 @@ pub fn create_deposit_relay<T: Transport + Clone>(app: Arc<App<T>>, init: &Datab
 		app,
 		foreign_balance,
 		foreign_chain_id,
+		foreign_gas_price,
 	}
 }
 
@@ -67,6 +68,7 @@ pub struct DepositRelay<T: Transport> {
 	foreign_contract: Address,
 	foreign_balance: Arc<RwLock<Option<U256>>>,
 	foreign_chain_id: u64,
+	foreign_gas_price: Arc<RwLock<u64>>,
 }
 
 impl<T: Transport> Stream for DepositRelay<T> {
@@ -85,7 +87,11 @@ impl<T: Transport> Stream for DepositRelay<T> {
 					let item = try_stream!(self.logs.poll().map_err(|e| ErrorKind::ContextualizedError(Box::new(e), "polling home for deposits")));
 					let len = item.logs.len();
 					info!("got {} new deposits to relay", len);
-					let balance_required = U256::from(self.app.config.txs.deposit_relay.gas) * U256::from(self.app.config.txs.deposit_relay.gas_price) * U256::from(item.logs.len());
+
+					let gas = U256::from(self.app.config.txs.deposit_relay.gas);
+					let gas_price = U256::from(*self.foreign_gas_price.read().unwrap());
+					let balance_required = gas * gas_price * U256::from(item.logs.len());
+					
 					if balance_required > *foreign_balance.as_ref().unwrap() {
 						return Err(ErrorKind::InsufficientFunds.into())
 					}
@@ -96,8 +102,8 @@ impl<T: Transport> Stream for DepositRelay<T> {
 						.into_iter()
 						.map(|payload| {
 							let tx = Transaction {
-								gas: self.app.config.txs.deposit_relay.gas.into(),
+								gas,
-								gas_price: self.app.config.txs.deposit_relay.gas_price.into(),
+								gas_price,
 								value: U256::zero(),
 								data: payload.0,
 								nonce: U256::zero(),

bridge/src/bridge/gas_price.rs

@@ -0,0 +1,98 @@
+use std::collections::HashMap;
+use std::time::{Duration, Instant};
+
+use futures::{Async, Future, Poll, Stream};
+use hyper::{Chunk, client::HttpConnector, Client, Uri};
+use hyper_tls::HttpsConnector;
+use serde_json as json;
+use tokio_core::reactor::Handle;
+use tokio_timer::{Interval, Timer, Timeout};
+
+use config::{GasPriceSpeed, Node};
+use error::Error;
+
+const CACHE_TIMEOUT_DURATION: Duration = Duration::from_secs(5 * 60);
+const REQUEST_TIMEOUT_DURATION: Duration = Duration::from_secs(30);
+
+enum State {
+    Initial,
+    WaitingForResponse(Timeout<Box<Future<Item = Chunk, Error = Error>>>),
+    Yield(Option<u64>),
+}
+
+pub struct GasPriceStream {
+    state: State,
+    client: Client<HttpsConnector<HttpConnector>>,
+    uri: Uri,
+    speed: GasPriceSpeed,
+    request_timer: Timer,
+    interval: Interval,
+}
+
+impl GasPriceStream {
+    pub fn new(node: &Node, handle: &Handle, timer: &Timer) -> Self {
+        let client = Client::configure()
+            .connector(HttpsConnector::new(4, handle).unwrap())
+            .build(handle);
+
+        let uri: Uri = node.gas_price_oracle_url.clone().unwrap().parse().unwrap();
+
+        GasPriceStream {
+            state: State::Initial,
+            client,
+            uri,
+            speed: node.gas_price_speed,
+            request_timer: timer.clone(),
+            interval: timer.interval_at(Instant::now(), CACHE_TIMEOUT_DURATION),
+        }
+    }
+}
+
+impl Stream for GasPriceStream {
+    type Item = u64;
+    type Error = Error;
+
+    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
+        loop {
+            let next_state = match self.state {
+                State::Initial => {
+                    let _ = try_stream!(self.interval.poll());
+
+                    let request: Box<Future<Item = Chunk, Error = Error>> =
+                        Box::new(
+                            self.client.get(self.uri.clone())
+                                .and_then(|resp| resp.body().concat2())
+                                .map_err(|e| e.into())
+                        );
+
+                    let request_future = self.request_timer
+                        .timeout(request, REQUEST_TIMEOUT_DURATION);
+
+                    State::WaitingForResponse(request_future)
+                },
+                State::WaitingForResponse(ref mut request_future) => {
+                    match request_future.poll() {
+                        Ok(Async::NotReady) => return Ok(Async::NotReady),
+                        Ok(Async::Ready(chunk)) => {
+                            let json_obj: HashMap<String, json::Value> = json::from_slice(&chunk)?;
+
+                            let gas_price = match json_obj.get(self.speed.as_str()) {
+                                Some(json::Value::Number(price)) => (price.as_f64().unwrap() * 1_000_000_000.0).trunc() as u64,
+                                _ => unreachable!(),
+                            };
+
+                            State::Yield(Some(gas_price))
+                        },
+                        Err(e) => panic!(e), 
+                    }
+                },
+                State::Yield(ref mut opt) => match opt.take() {
+					None => State::Initial,
+					price => return Ok(Async::Ready(price)),
+				}
+            };
+
+            self.state = next_state;
+        }
+    }
+}

bridge/src/bridge/mod.rs

@@ -5,6 +5,7 @@ pub mod nonce;
 mod deposit_relay;
 mod withdraw_confirm;
 mod withdraw_relay;
+mod gas_price;
 
 use std::fs;
 use std::sync::{Arc, RwLock};
@@ -15,6 +16,7 @@ use web3::types::U256;
 use app::App;
 use database::Database;
 use error::{Error, ErrorKind, Result};
+use tokio_core::reactor::Handle;
 
 pub use self::deploy::{Deploy, Deployed, create_deploy};
 pub use self::balance::{BalanceCheck, create_balance_check};
@@ -22,6 +24,7 @@ pub use self::chain_id::{ChainIdRetrieval, create_chain_id_retrieval};
 pub use self::deposit_relay::{DepositRelay, create_deposit_relay};
 pub use self::withdraw_relay::{WithdrawRelay, create_withdraw_relay};
 pub use self::withdraw_confirm::{WithdrawConfirm, create_withdraw_confirm};
+pub use self::gas_price::GasPriceStream;
 
 /// Last block checked by the bridge components.
 #[derive(Clone, Copy)]
@@ -71,30 +74,52 @@ enum BridgeStatus {
 }
 
 /// Creates new bridge.
-pub fn create_bridge<T: Transport + Clone>(app: Arc<App<T>>, init: &Database, home_chain_id: u64, foreign_chain_id: u64) -> Bridge<T, FileBackend> {
+pub fn create_bridge<T: Transport + Clone>(app: Arc<App<T>>, init: &Database, handle: &Handle, home_chain_id: u64, foreign_chain_id: u64) -> Bridge<T, FileBackend> {
 	let backend = FileBackend {
 		path: app.database_path.clone(),
 		database: init.clone(),
 	};
 
-	create_bridge_backed_by(app, init, backend, home_chain_id, foreign_chain_id)
+	create_bridge_backed_by(app, init, backend, handle, home_chain_id, foreign_chain_id)
 }
 
 /// Creates new bridge writing to custom backend.
-pub fn create_bridge_backed_by<T: Transport + Clone, F: BridgeBackend>(app: Arc<App<T>>, init: &Database, backend: F, home_chain_id: u64, foreign_chain_id: u64) -> Bridge<T, F> {
+pub fn create_bridge_backed_by<T: Transport + Clone, F: BridgeBackend>(app: Arc<App<T>>, init: &Database, backend: F, handle: &Handle, home_chain_id: u64, foreign_chain_id: u64) -> Bridge<T, F> {
 	let home_balance = Arc::new(RwLock::new(None));
 	let foreign_balance = Arc::new(RwLock::new(None));
+
+	let home_gas_stream = if app.config.home.gas_price_oracle_url.is_some() {
+		let stream = GasPriceStream::new(&app.config.home, handle, &app.timer);
+		Some(stream)
+	} else {
+		None
+	};
+
+	let foreign_gas_stream = if app.config.foreign.gas_price_oracle_url.is_some() {
+		let stream = GasPriceStream::new(&app.config.foreign, handle, &app.timer);
+		Some(stream)
+	} else {
+		None
+	};
+
+	let home_gas_price = Arc::new(RwLock::new(app.config.home.default_gas_price));
+	let foreign_gas_price = Arc::new(RwLock::new(app.config.foreign.default_gas_price));
+
 	Bridge {
 		foreign_balance_check: create_balance_check(app.clone(), app.connections.foreign.clone(), app.config.foreign.clone()),
 		home_balance_check: create_balance_check(app.clone(), app.connections.home.clone(), app.config.home.clone()),
 		foreign_balance: foreign_balance.clone(),
 		home_balance: home_balance.clone(),
-		deposit_relay: create_deposit_relay(app.clone(), init, foreign_balance.clone(), foreign_chain_id),
+		deposit_relay: create_deposit_relay(app.clone(), init, foreign_balance.clone(), foreign_chain_id, foreign_gas_price.clone()),
-		withdraw_relay: create_withdraw_relay(app.clone(), init, home_balance.clone(), home_chain_id),
+		withdraw_relay: create_withdraw_relay(app.clone(), init, home_balance.clone(), home_chain_id, home_gas_price.clone()),
-		withdraw_confirm: create_withdraw_confirm(app.clone(), init, foreign_balance.clone(), foreign_chain_id),
+		withdraw_confirm: create_withdraw_confirm(app.clone(), init, foreign_balance.clone(), foreign_chain_id, foreign_gas_price.clone()),
 		state: BridgeStatus::Wait,
 		backend,
 		running: app.running.clone(),
+		home_gas_stream,
+		foreign_gas_stream,
+		home_gas_price,
+		foreign_gas_price,
 	}
 }
 
@@ -109,6 +134,10 @@ pub struct Bridge<T: Transport, F> {
 	state: BridgeStatus,
 	backend: F,
 	running: Arc<AtomicBool>,
+	home_gas_stream: Option<GasPriceStream>,
+	foreign_gas_stream: Option<GasPriceStream>,
+	home_gas_price: Arc<RwLock<u64>>,
+	foreign_gas_price: Arc<RwLock<u64>>,
 }
 
 use std::sync::atomic::{AtomicBool, Ordering};
@@ -134,6 +163,19 @@ impl<T: Transport, F: BridgeBackend> Bridge<T, F> {
 		}
 	}
 
+	fn get_gas_prices(&mut self) -> Poll<Option<()>, Error> {
+		if let Some(ref mut home_gas_stream) = self.home_gas_stream {
+			let mut home_price = self.home_gas_price.write().unwrap();
+			*home_price = try_bridge!(home_gas_stream.poll()).unwrap_or(*home_price);
+		}
+
+		if let Some(ref mut foreign_gas_stream) = self.foreign_gas_stream {
+			let mut foreign_price = self.foreign_gas_price.write().unwrap();
+			*foreign_price = try_bridge!(foreign_gas_stream.poll()).unwrap_or(*foreign_price);
+		}
+
+		Ok(Async::Ready(None))
+	}
 }
 
 impl<T: Transport, F: BridgeBackend> Stream for Bridge<T, F> {
@@ -161,6 +203,8 @@ impl<T: Transport, F: BridgeBackend> Stream for Bridge<T, F> {
 						}
 					}
 
+					let _ = self.get_gas_prices();
+
 					let d_relay = try_bridge!(self.deposit_relay.poll().map_err(|e| ErrorKind::ContextualizedError(Box::new(e), "deposit_relay")))
 						.map(BridgeChecked::DepositRelay);
 

bridge/src/bridge/withdraw_confirm.rs

@@ -37,7 +37,7 @@ enum WithdrawConfirmState<T: Transport> {
 	Yield(Option<u64>),
 }
 
-pub fn create_withdraw_confirm<T: Transport + Clone>(app: Arc<App<T>>, init: &Database, foreign_balance: Arc<RwLock<Option<U256>>>, foreign_chain_id: u64) -> WithdrawConfirm<T> {
+pub fn create_withdraw_confirm<T: Transport + Clone>(app: Arc<App<T>>, init: &Database, foreign_balance: Arc<RwLock<Option<U256>>>, foreign_chain_id: u64, foreign_gas_price: Arc<RwLock<u64>>) -> WithdrawConfirm<T> {
 	let logs_init = api::LogStreamInit {
 		after: init.checked_withdraw_confirm,
 		request_timeout: app.config.foreign.request_timeout,
@@ -53,6 +53,7 @@ pub fn create_withdraw_confirm<T: Transport + Clone>(app: Arc<App<T>>, init: &Da
 		app,
 		foreign_balance,
 		foreign_chain_id,
+		foreign_gas_price,
 	}
 }
 
@@ -63,6 +64,7 @@ pub struct WithdrawConfirm<T: Transport> {
 	foreign_contract: Address,
 	foreign_balance: Arc<RwLock<Option<U256>>>,
 	foreign_chain_id: u64,
+	foreign_gas_price: Arc<RwLock<u64>>,
 }
 
 impl<T: Transport> Stream for WithdrawConfirm<T> {
@@ -73,7 +75,7 @@ impl<T: Transport> Stream for WithdrawConfirm<T> {
 		// borrow checker...
 		let app = &self.app;
 		let gas = self.app.config.txs.withdraw_confirm.gas.into();
-		let gas_price = self.app.config.txs.withdraw_confirm.gas_price.into();
+		let gas_price = U256::from(*self.foreign_gas_price.read().unwrap());
 		let contract = self.foreign_contract.clone();
 		loop {
 			let next_state = match self.state {
@@ -110,7 +112,7 @@ impl<T: Transport> Stream for WithdrawConfirm<T> {
 
 					let block = item.to;
 
-					let balance_required = U256::from(self.app.config.txs.withdraw_confirm.gas) * U256::from(self.app.config.txs.withdraw_confirm.gas_price) * U256::from(signatures.len());
+					let balance_required = gas * gas_price * U256::from(signatures.len());
 					if balance_required > *foreign_balance.as_ref().unwrap() {
 						return Err(ErrorKind::InsufficientFunds.into())
 					}
@@ -124,7 +126,8 @@ impl<T: Transport> Stream for WithdrawConfirm<T> {
 						})
 						.map(|payload| {
 							let tx = Transaction {
-								gas, gas_price,
+								gas,
+								gas_price,
 								value: U256::zero(),
 								data: payload.0,
 								nonce: U256::zero(),

bridge/src/bridge/withdraw_relay.rs

@@ -76,7 +76,7 @@ pub enum WithdrawRelayState<T: Transport> {
 	Yield(Option<u64>),
 }
 
-pub fn create_withdraw_relay<T: Transport + Clone>(app: Arc<App<T>>, init: &Database, home_balance: Arc<RwLock<Option<U256>>>, home_chain_id: u64) -> WithdrawRelay<T> {
+pub fn create_withdraw_relay<T: Transport + Clone>(app: Arc<App<T>>, init: &Database, home_balance: Arc<RwLock<Option<U256>>>, home_chain_id: u64, home_gas_price: Arc<RwLock<u64>>) -> WithdrawRelay<T> {
 	let logs_init = api::LogStreamInit {
 		after: init.checked_withdraw_relay,
 		request_timeout: app.config.foreign.request_timeout,
@@ -93,6 +93,7 @@ pub fn create_withdraw_relay<T: Transport + Clone>(app: Arc<App<T>>, init: &Data
 		app,
 		home_balance,
 		home_chain_id,
+		home_gas_price,
 	}
 }
 
@@ -104,6 +105,7 @@ pub struct WithdrawRelay<T: Transport> {
 	home_contract: Address,
 	home_balance: Arc<RwLock<Option<U256>>>,
 	home_chain_id: u64,
+	home_gas_price: Arc<RwLock<u64>>,
 }
 
 impl<T: Transport> Stream for WithdrawRelay<T> {
@@ -113,6 +115,7 @@ impl<T: Transport> Stream for WithdrawRelay<T> {
 	fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
 		let app = &self.app;
 		let gas = self.app.config.txs.withdraw_relay.gas.into();
+		let gas_price = U256::from(*self.home_gas_price.read().unwrap());
 		let contract = self.home_contract.clone();
 		let home = &self.app.config.home;
 		let t = &self.app.connections.home;
@@ -178,7 +181,7 @@ impl<T: Transport> Stream for WithdrawRelay<T> {
 					info!("fetching messages and signatures complete");
 					assert_eq!(messages_raw.len(), signatures_raw.len());
 
-					let balance_required = U256::from(self.app.config.txs.withdraw_relay.gas) * U256::from(self.app.config.txs.withdraw_relay.gas_price) * U256::from(messages_raw.len());
+					let balance_required = gas * gas_price * U256::from(messages_raw.len());
 					if balance_required > *home_balance.as_ref().unwrap() {
 						return Err(ErrorKind::InsufficientFunds.into())
 					}
@@ -221,7 +224,8 @@ impl<T: Transport> Stream for WithdrawRelay<T> {
 								message.clone().0).into();
 							let gas_price = MessageToMainnet::from_bytes(message.0.as_slice()).mainnet_gas_price;
 							let tx = Transaction {
-									gas, gas_price,
+									gas,
+									gas_price,
 									value: U256::zero(),
 									data: payload.0,
 									nonce: U256::zero(),

bridge/src/config.rs

@@ -1,6 +1,7 @@
 use std::path::{Path, PathBuf};
 use std::fs;
 use std::io::Read;
+use std::str::FromStr;
 use std::time::Duration;
 #[cfg(feature = "deploy")]
 use rustc_hex::FromHex;
@@ -15,6 +16,9 @@ const DEFAULT_CONFIRMATIONS: usize = 12;
 const DEFAULT_TIMEOUT: u64 = 3600;
 const DEFAULT_RPC_PORT: u16 = 8545;
 const DEFAULT_CONCURRENCY: usize = 100;
+const DEFAULT_GAS_PRICE_SPEED: GasPriceSpeed = GasPriceSpeed::Fast;
+const DEFAULT_GAS_PRICE_TIMEOUT_SECS: u64 = 10;
+const DEFAULT_GAS_PRICE_WEI: u64 = 15_000_000_000;
 
 /// Application config.
 #[derive(Debug, PartialEq, Clone)]
@@ -71,6 +75,10 @@ pub struct Node {
 	pub rpc_port: u16,
 	pub password: PathBuf,
 	pub info: NodeInfo,
+	pub gas_price_oracle_url: Option<String>,
+	pub gas_price_speed: GasPriceSpeed,
+	pub gas_price_timeout: Duration,
+	pub default_gas_price: u64,
 }
 
 use std::sync::{Arc, RwLock};
@@ -97,6 +105,20 @@ impl PartialEq for NodeInfo {
 
 impl Node {
 	fn from_load_struct(node: load::Node) -> Result<Node, Error> {
+		let gas_price_oracle_url = node.gas_price_oracle_url.clone();
+		
+		let gas_price_speed = match node.gas_price_speed {
+			Some(ref s) => GasPriceSpeed::from_str(s).unwrap(),
+			None => DEFAULT_GAS_PRICE_SPEED
+		};
+
+		let gas_price_timeout = {
+			let n_secs = node.gas_price_timeout.unwrap_or(DEFAULT_GAS_PRICE_TIMEOUT_SECS);
+			Duration::from_secs(n_secs)
+		};
+
+		let default_gas_price = node.default_gas_price.unwrap_or(DEFAULT_GAS_PRICE_WEI);
+
 		let result = Node {
 			account: node.account,
 			#[cfg(feature = "deploy")]
@@ -115,6 +137,10 @@ impl Node {
 			rpc_port: node.rpc_port.unwrap_or(DEFAULT_RPC_PORT),
 			password: node.password,
 			info: Default::default(),
+			gas_price_oracle_url,
+			gas_price_speed,
+			gas_price_timeout,
+			default_gas_price,
 		};
 
 		Ok(result)
@@ -184,6 +210,40 @@ pub struct Authorities {
 	pub required_signatures: u32,
 }
 
+#[derive(Clone, Copy, Debug, PartialEq)]
+pub enum GasPriceSpeed {
+    Instant,
+    Fast,
+    Standard,
+    Slow,
+}
+
+impl FromStr for GasPriceSpeed {
+	type Err = ();
+	
+	fn from_str(s: &str) -> Result<Self, Self::Err> {
+		let speed = match s {
+			"instant" => GasPriceSpeed::Instant,
+			"fast" => GasPriceSpeed::Fast,
+			"standard" => GasPriceSpeed::Standard,
+			"slow" => GasPriceSpeed::Slow,
+			_ => return Err(()),
+		};
+		Ok(speed)
+	}
+}
+
+impl GasPriceSpeed {
+	pub fn as_str(&self) -> &str {
+		match *self {
+			GasPriceSpeed::Instant => "instant",
+			GasPriceSpeed::Fast => "fast",
+			GasPriceSpeed::Standard => "standard",
+			GasPriceSpeed::Slow => "slow",
+		}
+	}
+}
+
 /// Some config values may not be defined in `toml` file, but they should be specified at runtime.
 /// `load` module separates `Config` representation in file with optional from the one used
 /// in application.
@@ -213,6 +273,10 @@ mod load {
 		pub rpc_host: Option<String>,
 		pub rpc_port: Option<u16>,
 		pub password: PathBuf,
+		pub gas_price_oracle_url: Option<String>,
+		pub gas_price_speed: Option<String>,
+		pub gas_price_timeout: Option<u64>,
+		pub default_gas_price: Option<u64>,
 	}
 
 	#[derive(Deserialize)]
@@ -258,7 +322,7 @@ mod tests {
 	use super::ContractConfig;
 	#[cfg(feature = "deploy")]
     use super::TransactionConfig;
-	use super::DEFAULT_TIMEOUT;
+	use super::{DEFAULT_TIMEOUT, DEFAULT_CONCURRENCY, DEFAULT_GAS_PRICE_SPEED, DEFAULT_GAS_PRICE_TIMEOUT_SECS, DEFAULT_GAS_PRICE_WEI};
 
 	#[test]
 	fn load_full_setup_from_str() {
@@ -314,6 +378,10 @@ home_deploy = { gas = 20 }
 				rpc_port: 8545,
 				password: "password".into(),
 				info: Default::default(),
+				gas_price_oracle_url: None,
+				gas_price_speed: DEFAULT_GAS_PRICE_SPEED,
+				gas_price_timeout: Duration::from_secs(DEFAULT_GAS_PRICE_TIMEOUT_SECS),
+				default_gas_price: DEFAULT_GAS_PRICE_WEI,
 			},
 			foreign: Node {
 				account: "0000000000000000000000000000000000000001".into(),
@@ -328,6 +396,10 @@ home_deploy = { gas = 20 }
 				rpc_port: 8545,
 				password: "password".into(),
 				info: Default::default(),
+				gas_price_oracle_url: None,
+				gas_price_speed: DEFAULT_GAS_PRICE_SPEED,
+				gas_price_timeout: Duration::from_secs(DEFAULT_GAS_PRICE_TIMEOUT_SECS),
+				default_gas_price: DEFAULT_GAS_PRICE_WEI,
 			},
 			authorities: Authorities {
 				accounts: vec![
@@ -346,6 +418,7 @@ home_deploy = { gas = 20 }
 			expected.txs.home_deploy = TransactionConfig {
 				gas: 20,
 				gas_price: 0,
+				concurrency: DEFAULT_CONCURRENCY,
 			};
 		}
 
@@ -398,6 +471,10 @@ required_signatures = 2
 				rpc_port: 8545,
 				password: "password".into(),
 				info: Default::default(),
+				gas_price_oracle_url: None,
+				gas_price_speed: DEFAULT_GAS_PRICE_SPEED,
+				gas_price_timeout: Duration::from_secs(DEFAULT_GAS_PRICE_TIMEOUT_SECS),
+				default_gas_price: DEFAULT_GAS_PRICE_WEI,
 			},
 			foreign: Node {
 				account: "0000000000000000000000000000000000000001".into(),
@@ -412,6 +489,10 @@ required_signatures = 2
 				rpc_port: 8545,
 				password: "password".into(),
 				info: Default::default(),
+				gas_price_oracle_url: None,
+				gas_price_speed: DEFAULT_GAS_PRICE_SPEED,
+				gas_price_timeout: Duration::from_secs(DEFAULT_GAS_PRICE_TIMEOUT_SECS),
+				default_gas_price: DEFAULT_GAS_PRICE_WEI,
 			},
 			authorities: Authorities {
 				accounts: vec![

bridge/src/error.rs

@@ -5,6 +5,8 @@ use tokio_timer::{TimerError, TimeoutError};
 use {web3, toml, ethabi, rustc_hex};
 use ethcore::ethstore;
 use ethcore::account_provider::{SignError, Error as AccountError};
+use serde_json;
+use hyper;
 
 error_chain! {
 	types {
@@ -17,6 +19,8 @@ error_chain! {
 		Ethabi(ethabi::Error);
 		Timer(TimerError);
 		Hex(rustc_hex::FromHexError);
+		Json(serde_json::Error);
+		Hyper(hyper::Error);
 	}
 
 	errors {

bridge/src/lib.rs

@@ -30,6 +30,8 @@ extern crate keccak_hash;
 extern crate jsonrpc_core as rpc;
 
 extern crate itertools;
+extern crate hyper;
+extern crate hyper_tls;
 
 #[cfg(test)]
 #[macro_use]

cli/src/main.rs

@@ -123,6 +123,7 @@ fn execute<S, I>(command: I, running: Arc<AtomicBool>) -> Result<String, UserFac
 
 	info!(target: "bridge", "Starting event loop");
 	let mut event_loop = Core::new().unwrap();
+	let handle = event_loop.handle();
 
 	info!(target: "bridge", "Home rpc host {}", config.clone().home.rpc_host);
 	info!(target: "bridge", "Foreign rpc host {}", config.clone().foreign.rpc_host);
@@ -130,7 +131,7 @@ fn execute<S, I>(command: I, running: Arc<AtomicBool>) -> Result<String, UserFac
 	info!(target: "bridge", "Establishing connection:");
 
 	info!(target:"bridge", "  using RPC connection");
-	let app = match App::new_http(config.clone(), &args.arg_database, &event_loop.handle(), running.clone()) {
+	let app = match App::new_http(config.clone(), &args.arg_database, &handle, running.clone()) {
 		Ok(app) => app,
 		Err(e) => {
 			warn!("Can't establish an RPC connection: {:?}", e);
@@ -176,7 +177,7 @@ fn execute<S, I>(command: I, running: Arc<AtomicBool>) -> Result<String, UserFac
 	};
 
 	info!(target: "bridge", "Starting listening to events");
-	let bridge = create_bridge(app.clone(), &database, home_chain_id, foreign_chain_id).and_then(|_| future::ok(true)).collect();
+	let bridge = create_bridge(app.clone(), &database, &handle, home_chain_id, foreign_chain_id).and_then(|_| future::ok(true)).collect();
 	let mut result = event_loop.run(bridge);
 	loop {
 		match result {

integration-tests/bridge_config.toml

@@ -7,6 +7,7 @@ required_confirmations = 0
 rpc_host = "http://127.0.0.1"
 rpc_port = 8550
 password = "password.txt"
+default_gas_price = 0
 
 [home.contract]
 bin = "../compiled_contracts/HomeBridge.bin"
@@ -17,6 +18,7 @@ required_confirmations = 0
 rpc_host = "http://127.0.0.1"
 rpc_port = 8551
 password = "password.txt"
+default_gas_price = 0
 
 [foreign.contract]
 bin = "../compiled_contracts/ForeignBridge.bin"

tests/Cargo.toml

@@ -11,5 +11,6 @@ web3 = "0.3"
 serde_json = "1.0"
 pretty_assertions = "0.2.1"
 ethabi = "5.0"
+ethcore = { git = "http://github.com/paritytech/parity", rev = "991f0ca" }
 ethereum-types = "0.3"
 rustc-hex = "1.0"

tests/src/lib.rs

@@ -5,6 +5,7 @@ extern crate web3;
 extern crate bridge;
 #[macro_use]
 extern crate pretty_assertions;
+extern crate ethcore;
 
 use std::cell::Cell;
 use web3::Transport;
@@ -35,9 +36,47 @@ impl Transport for MockedTransport {
 	type Out = web3::Result<rpc::Value>;
 
 	fn prepare(&self, method: &str, params: Vec<rpc::Value>) -> (usize, rpc::Call) {
-		let n = self.requests.get();
+		let n = self.requests.get();		
 		assert_eq!(&self.expected_requests[n].method as &str, method, "invalid method called");
-		assert_eq!(self.expected_requests[n].params, params, "invalid method params");
+
+		for (expected_params, params) in self.expected_requests[n].params.iter().zip(params.iter()) {
+			assert_eq!(expected_params.get("address"), params.get("address"), "invalid method params, addresses do not match");
+			assert_eq!(expected_params.get("fromBlock"), params.get("fromBlock"), "invalid method params, from-blocks do not match");
+			assert_eq!(expected_params.get("limit"), params.get("limit"), "invalid method params, limits do not match");
+			assert_eq!(expected_params.get("toBlock"), params.get("toBlock"), "invalid method params, to-blocks do not match");
+						
+			let expected_topics: Vec<rpc::Value> = if let Some(ref topics) = expected_params.get("topics") {
+				topics.as_array().unwrap().clone()
+					.iter()
+					.filter_map(|topic|
+						if topic != &rpc::Value::Null {
+							Some(topic.clone())
+						} else {
+							None
+						}
+					)
+					.collect()
+			} else {
+				vec![]
+			};
+
+			let topics: Vec<rpc::Value> = if let Some(ref topics) = params.get("topics") {
+				topics.as_array().unwrap().clone()
+					.iter()
+					.filter_map(|topic|
+						if topic != &rpc::Value::Null {
+							Some(topic.clone())
+						} else {
+							None
+						}
+					)
+					.collect()
+			} else {
+				vec![]
+			};
+
+			assert_eq!(expected_topics, topics, "invalid method params, topics do not match");
+		}
 		self.requests.set(n + 1);
 
 		let request = web3::helpers::build_request(1, method, params);
@@ -98,15 +137,16 @@ macro_rules! test_app_stream {
 			use self::futures::{Future, Stream};
 			use self::bridge::app::{App, Connections};
 			use self::bridge::contracts::{foreign, home};
-			use self::bridge::config::{Config, Authorities, Node, ContractConfig, Transactions, TransactionConfig};
+			use self::bridge::config::{Config, Authorities, Node, NodeInfo, ContractConfig, Transactions, TransactionConfig, GasPriceSpeed};
 			use self::bridge::database::Database;
-
+			use ethcore::account_provider::AccountProvider;
+			
 			let home = $crate::MockedTransport {
 				requests: Default::default(),
 				expected_requests: vec![$($home_method),*].into_iter().zip(vec![$($home_req),*].into_iter()).map(Into::into).collect(),
 				mocked_responses: vec![$($home_res),*],
 			};
-
+			
 			let foreign = $crate::MockedTransport {
 				requests: Default::default(),
 				expected_requests: vec![$($foreign_method),*].into_iter().zip(vec![$($foreign_req),*].into_iter()).map(Into::into).collect(),
@@ -125,6 +165,12 @@ macro_rules! test_app_stream {
 					required_confirmations: $home_conf,
 					rpc_host: "".into(),
 					rpc_port: 8545,
+					password: "password.txt".into(),
+					info: NodeInfo::default(),
+					gas_price_oracle_url: None,
+					gas_price_speed: GasPriceSpeed::Fast,
+					gas_price_timeout: Duration::from_secs(5),
+					default_gas_price: 0,
 				},
 				foreign: Node {
 					account: $foreign_acc.parse().unwrap(),
@@ -136,6 +182,12 @@ macro_rules! test_app_stream {
 					required_confirmations: $foreign_conf,
 					rpc_host: "".into(),
 					rpc_port: 8545,
+					password: "password.txt".into(),
+					info: NodeInfo::default(),
+					gas_price_oracle_url: None,
+					gas_price_speed: GasPriceSpeed::Fast,
+					gas_price_timeout: Duration::from_secs(5),
+					default_gas_price: 0,
 				},
 				authorities: Authorities {
 					accounts: $authorities_accs.iter().map(|a: &&str| a.parse().unwrap()).collect(),
@@ -156,9 +208,10 @@ macro_rules! test_app_stream {
 				foreign_bridge: foreign::ForeignBridge::default(),
 				timer: Default::default(),
 				running: Arc::new(AtomicBool::new(true)),
+				keystore: AccountProvider::transient_provider(),
 			};
 
-			let app = Arc::new(app);
+			let app = Arc::new(app);			
 			let stream = $init_stream(app, &$db);
 			let res = stream.collect().wait();
 

tests/tests/deposit_relay.rs

@@ -4,6 +4,7 @@ extern crate serde_json;
 extern crate bridge;
 #[macro_use]
 extern crate tests;
+extern crate ethcore;
 
 use bridge::bridge::create_deposit_relay;
 
@@ -27,7 +28,7 @@ test_app_stream! {
 		],
 		signatures => 1;
 	txs => Transactions::default(),
-	init => |app, db| create_deposit_relay(app, db, Arc::new(RwLock::new(Some(99999999999u64.into())))).take(2),
+	init => |app, db| create_deposit_relay(app, db, Arc::new(RwLock::new(Some(99999999999u64.into()))), 17, Arc::new(RwLock::new(1))).take(2),
 	expected => vec![0x1005, 0x1006],
 	home_transport => [
 		"eth_blockNumber" =>
@@ -77,7 +78,7 @@ test_app_stream! {
 		],
 		signatures => 1;
 	txs => Transactions::default(),
-	init => |app, db| create_deposit_relay(app, db, Arc::new(RwLock::new(Some(99999999999u64.into())))).take(2),
+	init => |app, db| create_deposit_relay(app, db, Arc::new(RwLock::new(Some(99999999999u64.into()))), 17, Arc::new(RwLock::new(1))).take(2),
 	expected => vec![0x1005, 0x1006],
 	home_transport => [
 		"eth_blockNumber" =>
@@ -146,10 +147,11 @@ test_app_stream! {
 		deposit_relay: TransactionConfig {
 			gas: 0xfd,
 			gas_price: 0xa0,
+			concurrency: 100,
 		},
 		..Default::default()
 	},
-	init => |app, db| create_deposit_relay(app, db, Arc::new(RwLock::new(Some(99999999999u64.into())))).take(1),
+	init => |app, db| create_deposit_relay(app, db, Arc::new(RwLock::new(Some(99999999999u64.into()))), 17, Arc::new(RwLock::new(1))).take(1),
 	expected => vec![0x1005],
 	home_transport => [
 		"eth_blockNumber" =>
@@ -202,7 +204,7 @@ test_app_stream! {
 		],
 		signatures => 1;
 	txs => Transactions::default(),
-	init => |app, db| create_deposit_relay(app, db, Arc::new(RwLock::new(Some(99999999999u64.into())))).take(1),
+	init => |app, db| create_deposit_relay(app, db, Arc::new(RwLock::new(Some(99999999999u64.into()))), 17, Arc::new(RwLock::new(1))).take(1),
 	expected => vec![0x1005],
 	home_transport => [
 		"eth_blockNumber" =>
@@ -257,7 +259,7 @@ test_app_stream! {
 		],
 		signatures => 1;
 	txs => Transactions::default(),
-	init => |app, db| create_deposit_relay(app, db, Arc::new(RwLock::new(Some(99999999999u64.into())))).take(1),
+	init => |app, db| create_deposit_relay(app, db, Arc::new(RwLock::new(Some(99999999999u64.into()))), 17, Arc::new(RwLock::new(1))).take(1),
 	expected => vec![0x1005],
 	home_transport => [
 		"eth_blockNumber" =>
@@ -308,7 +310,7 @@ test_app_stream! {
 		],
 		signatures => 1;
 	txs => Transactions::default(),
-	init => |app, db| create_deposit_relay(app, db, Arc::new(RwLock::new(Some(99999999999u64.into())))).take(1),
+	init => |app, db| create_deposit_relay(app, db, Arc::new(RwLock::new(Some(99999999999u64.into()))), 17, Arc::new(RwLock::new(1))).take(1),
 	expected => vec![0x1005],
 	home_transport => [
 		"eth_blockNumber" =>

tests/tests/log_stream.rs

@@ -5,6 +5,7 @@ extern crate web3;
 extern crate bridge;
 #[macro_use]
 extern crate tests;
+extern crate ethcore;
 
 use std::time::Duration;
 use web3::types::{FilterBuilder, H160, H256, Log};
@@ -336,8 +337,14 @@ test_transport_stream! {
 			address: "0000000000000000000000000000000000000001".into(),
 			topics: vec![],
 			data: vec![0x10].into(),
-			log_type: "".into(),
+			block_hash: None,
-			..Default::default()
+			block_number: None, 
+			transaction_hash: None,
+			transaction_index: None,
+			log_index: None,
+			transaction_log_index: None,
+			log_type: None,
+			removed: None,
 		}],
 	}],
 	"eth_blockNumber" =>
@@ -379,8 +386,14 @@ test_transport_stream! {
 			address: "0000000000000000000000000000000000000001".into(),
 			topics: vec![],
 			data: vec![0x10].into(),
-			log_type: "".into(),
+			block_hash: None,
-			..Default::default()
+			block_number: None, 
+			transaction_hash: None,
+			transaction_index: None,
+			log_index: None,
+			transaction_log_index: None,
+			log_type: None,
+			removed: None,
 		}],
 	}, LogStreamItem {
 		from: 0x1007,
@@ -393,14 +406,26 @@ test_transport_stream! {
 			address: "0000000000000000000000000000000000000002".into(),
 			topics: vec![],
 			data: vec![0x20].into(),
-			log_type: "".into(),
+			block_hash: None,
-			..Default::default()
+			block_number: None, 
+			transaction_hash: None,
+			transaction_index: None,
+			log_index: None,
+			transaction_log_index: None,
+			log_type: None,
+			removed: None,
 		}, Log {
 			address: "0000000000000000000000000000000000000002".into(),
 			topics: vec![],
 			data: vec![0x30].into(),
-			log_type: "".into(),
+			block_hash: None,
-			..Default::default()
+			block_number: None, 
+			transaction_hash: None,
+			transaction_index: None,
+			log_index: None,
+			transaction_log_index: None,
+			log_type: None,
+			removed: None,
 		}],
 	}],
 	"eth_blockNumber" =>

tests/tests/withdraw_confirm.rs

@@ -9,6 +9,7 @@ extern crate tests;
 extern crate ethabi;
 extern crate rustc_hex;
 extern crate ethereum_types;
+extern crate ethcore;
 
 use rustc_hex::{ToHex, FromHex};
 use bridge::bridge::create_withdraw_confirm;
@@ -36,7 +37,7 @@ test_app_stream! {
 		],
 		signatures => 1;
 	txs => Transactions::default(),
-	init => |app, db| create_withdraw_confirm(app, db, Arc::new(RwLock::new(Some(99999999999u64.into())))).take(2),
+	init => |app, db| create_withdraw_confirm(app, db, Arc::new(RwLock::new(Some(99999999999u64.into()))), 17, Arc::new(RwLock::new(1))).take(2),
 	expected => vec![0x1005, 0x1006],
 	home_transport => [],
 	foreign_transport => [
@@ -86,7 +87,7 @@ test_app_stream! {
 		],
 		signatures => 1;
 	txs => Transactions::default(),
-	init => |app, db| create_withdraw_confirm(app, db, Arc::new(RwLock::new(Some(99999999999u64.into())))).take(2),
+	init => |app, db| create_withdraw_confirm(app, db, Arc::new(RwLock::new(Some(99999999999u64.into()))), 17, Arc::new(RwLock::new(1))).take(2),
 	expected => vec![0x1005, 0x1006],
 	home_transport => [],
 	foreign_transport => [
@@ -140,7 +141,7 @@ test_app_stream! {
 		],
 		signatures => 1;
 	txs => Transactions::default(),
-	init => |app, db| create_withdraw_confirm(app, db, Arc::new(RwLock::new(Some(99999999999u64.into())))).take(2),
+	init => |app, db| create_withdraw_confirm(app, db, Arc::new(RwLock::new(Some(99999999999u64.into()))), 17, Arc::new(RwLock::new(1))).take(2),
 	expected => vec![0x1005, 0x1006],
 	home_transport => [],
 	foreign_transport => [
@@ -198,10 +199,11 @@ test_app_stream! {
 		withdraw_confirm: TransactionConfig {
 			gas: 0xfe,
 			gas_price: 0xa1,
+			concurrency: 100,
 		},
 		..Default::default()
 	},
-	init => |app, db| create_withdraw_confirm(app, db, Arc::new(RwLock::new(Some(99999999999u64.into())))).take(2),
+	init => |app, db| create_withdraw_confirm(app, db, Arc::new(RwLock::new(Some(99999999999u64.into()))), 17, Arc::new(RwLock::new(1))).take(2),
 	expected => vec![0x1005, 0x1006],
 	home_transport => [],
 	foreign_transport => [
@@ -303,10 +305,11 @@ test_app_stream! {
 		withdraw_confirm: TransactionConfig {
 			gas: 0xff,
 			gas_price: 0xaa,
+			concurrency: 100,
 		},
 		..Default::default()
 	},
-	init => |app, db| create_withdraw_confirm(app, db, Arc::new(RwLock::new(Some(99999999999u64.into())))).take(2),
+	init => |app, db| create_withdraw_confirm(app, db, Arc::new(RwLock::new(Some(99999999999u64.into()))), 17, Arc::new(RwLock::new(1))).take(2),
 	expected => vec![0x2, 0x1006],
 	home_transport => [],
 	foreign_transport => [

tests/tests/withdraw_relay.rs

@@ -9,6 +9,7 @@ extern crate tests;
 extern crate ethabi;
 extern crate ethereum_types;
 extern crate rustc_hex;
+extern crate ethcore;
 
 use ethereum_types::{U256, H256};
 use rustc_hex::ToHex;
@@ -41,7 +42,7 @@ test_app_stream! {
 		],
 		signatures => 1;
 	txs => Transactions::default(),
-	init => |app, db| create_withdraw_relay(app, db, Arc::new(RwLock::new(Some(99999999999u64.into())))).take(2),
+	init => |app, db| create_withdraw_relay(app, db, Arc::new(RwLock::new(Some(99999999999u64.into()))), 17, Arc::new(RwLock::new(1))).take(2),
 	expected => vec![0x1005, 0x1006],
 	home_transport => [],
 	foreign_transport => [
@@ -92,7 +93,7 @@ test_app_stream! {
 		],
 		signatures => 1;
 	txs => Transactions::default(),
-	init => |app, db| create_withdraw_relay(app, db, Arc::new(RwLock::new(Some(99999999999u64.into())))).take(1),
+	init => |app, db| create_withdraw_relay(app, db, Arc::new(RwLock::new(Some(99999999999u64.into()))), 17, Arc::new(RwLock::new(1))).take(1),
 	expected => vec![0x1005],
 	home_transport => [],
 	foreign_transport => [
@@ -140,7 +141,7 @@ test_app_stream! {
 		],
 		signatures => 2;
 	txs => Transactions::default(),
-	init => |app, db| create_withdraw_relay(app, db, Arc::new(RwLock::new(Some(99999999999u64.into())))).take(1),
+	init => |app, db| create_withdraw_relay(app, db, Arc::new(RwLock::new(Some(99999999999u64.into()))), 17, Arc::new(RwLock::new(1))).take(1),
 	expected => vec![0x1005],
 	home_transport => [
 		// `HomeBridge.withdraw`