Merge branch 'lightrpc' into light-txq

rpc/src/v1/helpers/dispatch.rs

@@ -28,6 +28,7 @@ use util::{Address, H520, H256, U256, Uint, Bytes, RwLock};
 use util::sha3::Hashable;
 
 use ethkey::Signature;
+use ethsync::LightSync;
 use ethcore::miner::MinerService;
 use ethcore::client::MiningBlockChainClient;
 use ethcore::transaction::{Action, SignedTransaction, PendingTransaction, Transaction};
@@ -95,10 +96,9 @@ impl<C: MiningBlockChainClient, M: MinerService> Dispatcher for FullDispatcher<C
 	fn fill_optional_fields(&self, request: TransactionRequest, default_sender: Address)
 		-> BoxFuture<FilledTransactionRequest, Error>
 	{
-		let inner = move || {
+		let (client, miner) = (take_weakf!(self.client), take_weakf!(self.miner));
-			let (client, miner) = (take_weak!(self.client), take_weak!(self.miner));
 		let request = request;
-			Ok(FilledTransactionRequest {
+		future::ok(FilledTransactionRequest {
 			from: request.from.unwrap_or(default_sender),
 			used_default_from: request.from.is_none(),
 			to: request.to,
@@ -108,19 +108,16 @@ impl<C: MiningBlockChainClient, M: MinerService> Dispatcher for FullDispatcher<C
 			value: request.value.unwrap_or_else(|| 0.into()),
 			data: request.data.unwrap_or_else(Vec::new),
 			condition: request.condition,
-			})
+		}).boxed()
-		};
-		future::done(inner()).boxed()
 	}
 
 	fn sign(&self, accounts: &AccountProvider, filled: FilledTransactionRequest, password: SignWith)
 		-> BoxFuture<WithToken<SignedTransaction>, Error>
 	{
-		let inner = move || {
+		let (client, miner) = (take_weakf!(self.client), take_weakf!(self.miner));
-			let (client, miner) = (take_weak!(self.client), take_weak!(self.miner));
 		let network_id = client.signing_network_id();
 		let address = filled.from;
-			let signed_transaction = {
+		future::ok({
 			let t = Transaction {
 				nonce: filled.nonce
 					.or_else(|| miner
@@ -136,16 +133,13 @@ impl<C: MiningBlockChainClient, M: MinerService> Dispatcher for FullDispatcher<C
 			};
 
 			let hash = t.hash(network_id);
-				let signature = signature(accounts, address, hash, password)?;
+			let signature = try_bf!(signature(accounts, address, hash, password));
+
 			signature.map(|sig| {
 				SignedTransaction::new(t.with_signature(sig, network_id))
 					.expect("Transaction was signed by AccountsProvider; it never produces invalid signatures; qed")
 			})
-			};
+		}).boxed()
-			Ok(signed_transaction)
-		};
-
-		future::done(inner()).boxed()
 	}
 
 	fn dispatch_transaction(&self, signed_transaction: PendingTransaction) -> Result<H256, Error> {
@@ -181,7 +175,7 @@ impl LightDispatcher {
 			sync: sync,
 			client: client,
 			on_demand: on_demand,
-			transaction_queue,
+			transaction_queue: transaction_queue,
 		}
 	}
 }

rpc/src/v1/impls/eth.rs

@@ -359,133 +359,108 @@ impl<C, SN: ?Sized, S: ?Sized, M, EM> Eth for EthClient<C, SN, S, M, EM> where
 	fn balance(&self, address: RpcH160, num: Trailing<BlockNumber>) -> BoxFuture<RpcU256, Error> {
 		let address = address.into();
 
-		let inner = || {
+		let res = match num.0.clone() {
-			match num.0.clone() {
+			BlockNumber::Pending => Ok(take_weakf!(self.miner).balance(&*take_weakf!(self.client), &address).into()),
-				BlockNumber::Pending => Ok(take_weak!(self.miner).balance(&*take_weak!(self.client), &address).into()),
 			id => {
-					let client = take_weak!(self.client);
+				let client = take_weakf!(self.client);
 
-					check_known(&*client, id.clone())?;
+				try_bf!(check_known(&*client, id.clone()));
 				match client.balance(&address, id.into()) {
 					Some(balance) => Ok(balance.into()),
 					None => Err(errors::state_pruned()),
 				}
 			}
-			}
 		};
 
-		future::done(inner()).boxed()
+		future::done(res).boxed()
 	}
 
 	fn storage_at(&self, address: RpcH160, pos: RpcU256, num: Trailing<BlockNumber>) -> BoxFuture<RpcH256, Error> {
 		let address: Address = RpcH160::into(address);
 		let position: U256 = RpcU256::into(pos);
 
-		let inner = || {
+		let res = match num.0.clone() {
-			match num.0.clone() {
+			BlockNumber::Pending => Ok(take_weakf!(self.miner).storage_at(&*take_weakf!(self.client), &address, &H256::from(position)).into()),
-				BlockNumber::Pending => Ok(take_weak!(self.miner).storage_at(&*take_weak!(self.client), &address, &H256::from(position)).into()),
 			id => {
-					let client = take_weak!(self.client);
+				let client = take_weakf!(self.client);
 
-					check_known(&*client, id.clone())?;
+				try_bf!(check_known(&*client, id.clone()));
 				match client.storage_at(&address, &H256::from(position), id.into()) {
 					Some(s) => Ok(s.into()),
 					None => Err(errors::state_pruned()),
 				}
 			}
-			}
 		};
 
-		future::done(inner()).boxed()
+		future::done(res).boxed()
 	}
 
 	fn transaction_count(&self, address: RpcH160, num: Trailing<BlockNumber>) -> BoxFuture<RpcU256, Error> {
 		let address: Address = RpcH160::into(address);
-		let inner = move || {
+		let res = match num.0.clone() {
-			match num.0.clone() {
+			BlockNumber::Pending => Ok(take_weakf!(self.miner).nonce(&*take_weakf!(self.client), &address).into()),
-				BlockNumber::Pending => Ok(take_weak!(self.miner).nonce(&*take_weak!(self.client), &address).into()),
 			id => {
-					let client = take_weak!(self.client);
+				let client = take_weakf!(self.client);
 
-					check_known(&*client, id.clone())?;
+				try_bf!(check_known(&*client, id.clone()));
 				match client.nonce(&address, id.into()) {
 					Some(nonce) => Ok(nonce.into()),
 					None => Err(errors::state_pruned()),
 				}
 			}
-			}
 		};
 
-		future::done(inner()).boxed()
+		future::done(res).boxed()
 	}
 
 	fn block_transaction_count_by_hash(&self, hash: RpcH256) -> BoxFuture<Option<RpcU256>, Error> {
-		let inner = || {
+		future::ok(take_weakf!(self.client).block(BlockId::Hash(hash.into()))
-			Ok(take_weak!(self.client).block(BlockId::Hash(hash.into()))
+			.map(|block| block.transactions_count().into())).boxed()
-				.map(|block| block.transactions_count().into()))
-		};
-
-		future::done(inner()).boxed()
 	}
 
 	fn block_transaction_count_by_number(&self, num: BlockNumber) -> BoxFuture<Option<RpcU256>, Error> {
-		let inner = || {
+		future::ok(match num {
-			match num {
+			BlockNumber::Pending => Some(
-				BlockNumber::Pending => Ok(Some(
+				take_weakf!(self.miner).status().transactions_in_pending_block.into()
-					take_weak!(self.miner).status().transactions_in_pending_block.into()
+			),
-				)),
+			_ =>
-				_ => Ok(
+				take_weakf!(self.client).block(num.into())
-					take_weak!(self.client).block(num.into())
 					.map(|block| block.transactions_count().into())
-					)
+		}).boxed()
-			}
-		};
-
-		future::done(inner()).boxed()
 	}
 
 	fn block_uncles_count_by_hash(&self, hash: RpcH256) -> BoxFuture<Option<RpcU256>, Error> {
-		let inner = || {
+			future::ok(take_weakf!(self.client).block(BlockId::Hash(hash.into()))
-			Ok(take_weak!(self.client).block(BlockId::Hash(hash.into()))
 				.map(|block| block.uncles_count().into()))
-		};
+				.boxed()
-
-		future::done(inner()).boxed()
 	}
 
 	fn block_uncles_count_by_number(&self, num: BlockNumber) -> BoxFuture<Option<RpcU256>, Error> {
-		let inner = || {
+		future::ok(match num {
-			match num {
+			BlockNumber::Pending => Some(0.into()),
-				BlockNumber::Pending => Ok(Some(0.into())),
+			_ => take_weakf!(self.client).block(num.into())
-				_ => Ok(
+					.map(|block| block.uncles_count().into()
-					take_weak!(self.client).block(num.into())
-						.map(|block| block.uncles_count().into())
 			),
-			}
+		}).boxed()
-		};
-
-		future::done(inner()).boxed()
 	}
 
 	fn code_at(&self, address: RpcH160, num: Trailing<BlockNumber>) -> BoxFuture<Bytes, Error> {
 		let address: Address = RpcH160::into(address);
 
-		let inner = || {
+		let res = match num.0.clone() {
-			match num.0.clone() {
+			BlockNumber::Pending => Ok(take_weakf!(self.miner).code(&*take_weakf!(self.client), &address).map_or_else(Bytes::default, Bytes::new)),
-				BlockNumber::Pending => Ok(take_weak!(self.miner).code(&*take_weak!(self.client), &address).map_or_else(Bytes::default, Bytes::new)),
 			id => {
-					let client = take_weak!(self.client);
+				let client = take_weakf!(self.client);
 
-					check_known(&*client, id.clone())?;
+				try_bf!(check_known(&*client, id.clone()));
 				match client.code(&address, id.into()) {
 					Some(code) => Ok(code.map_or_else(Bytes::default, Bytes::new)),
 					None => Err(errors::state_pruned()),
 				}
 			}
-			}
 		};
 
-		future::done(inner()).boxed()
+		future::done(res).boxed()
 	}
 
 	fn block_by_hash(&self, hash: RpcH256, include_txs: bool) -> BoxFuture<Option<RichBlock>, Error> {

rpc/src/v1/impls/personal.rs

@@ -95,15 +95,9 @@ impl<D: Dispatcher + 'static> Personal for PersonalClient<D> {
 	}
 
 	fn send_transaction(&self, meta: Metadata, request: TransactionRequest, password: String) -> BoxFuture<RpcH256, Error> {
-		let setup = || {
 		let dispatcher = self.dispatcher.clone();
-			let accounts = take_weak!(self.accounts);
+		let accounts = take_weakf!(self.accounts);
 
-			Ok((accounts, dispatcher))
-		};
-
-		future::done(setup())
-			.and_then(move |(accounts, dispatcher)| {
 		let default = match request.from.as_ref() {
 			Some(account) => Ok(account.clone().into()),
 			None => accounts
@@ -111,19 +105,20 @@ impl<D: Dispatcher + 'static> Personal for PersonalClient<D> {
 				.map_err(|e| errors::account("Cannot find default account.", e)),
 		};
 
-				let dis = dispatcher.clone();
+		let default = match default {
-				future::done(default)
+			Ok(default) => default,
-					.and_then(move |default| dis.fill_optional_fields(request.into(), default))
+			Err(e) => return future::err(e).boxed(),
-					.map(move |tx| (tx, accounts, dispatcher))
+		};
-			})
+
-			.and_then(move |(filled, accounts, dispatcher)| {
+		dispatcher.fill_optional_fields(request.into(), default)
+			.and_then(move |filled| {
 				let condition = filled.condition.clone().map(Into::into);
 				dispatcher.sign(&accounts, filled, SignWith::Password(password))
 					.map(|tx| tx.into_value())
 					.map(move |tx| PendingTransaction::new(tx, condition))
 					.map(move |tx| (tx, dispatcher))
 			})
-			.and_then(move |(pending_tx, dispatcher)| {
+			.and_then(|(pending_tx, dispatcher)| {
 				let network_id = pending_tx.network_id();
 				trace!(target: "miner", "send_transaction: dispatching tx: {} for network ID {:?}",
 					::rlp::encode(&*pending_tx).to_vec().pretty(), network_id);

rpc/src/v1/impls/signing.rs

@@ -82,20 +82,14 @@ impl<D: Dispatcher + 'static> SigningQueueClient<D> {
 	}
 
 	fn dispatch(&self, payload: RpcConfirmationPayload, default_account: DefaultAccount) -> BoxFuture<DispatchResult, Error> {
-		let setup = move || {
+		let accounts = take_weakf!(self.accounts);
-			let accounts = take_weak!(self.accounts);
-			let default_account = default_account;
 		let default_account = match default_account {
 			DefaultAccount::Provided(acc) => acc,
 			DefaultAccount::ForDapp(dapp) => accounts.default_address(dapp).ok().unwrap_or_default(),
 		};
 
-			Ok((self.dispatcher.clone(), accounts, default_account))
+		let dispatcher = self.dispatcher.clone();
-		};
+		let signer = take_weakf!(self.signer);
-
-		let weak_signer = self.signer.clone();
-		future::done(setup())
-			.and_then(move |(dispatcher, accounts, default_account)| {
 		dispatch::from_rpc(payload, default_account, &dispatcher)
 			.and_then(move |payload| {
 				let sender = payload.sender();
@@ -105,14 +99,13 @@ impl<D: Dispatcher + 'static> SigningQueueClient<D> {
 						.map(DispatchResult::Value)
 						.boxed()
 				} else {
-							future::lazy(move ||
+					future::done(
-								take_weak!(weak_signer).add_request(payload)
+						signer.add_request(payload)
 							.map(DispatchResult::Promise)
 							.map_err(|_| errors::request_rejected_limit())
 					).boxed()
 				}
 			})
-			})
 			.boxed()
 	}
 }
@@ -180,7 +173,7 @@ impl<D: Dispatcher + 'static> ParitySigning for SigningQueueClient<D> {
 			});
 
 			// and wait for that to resolve.
-			p.then(|result| futures::done(result.expect("Ready is never dropped nor canceled."))).boxed()
+			p.then(|result| futures::done(result.expect("Ready is never dropped nor canceled.")))
 		}).boxed()
 	}
 }
@@ -202,7 +195,7 @@ impl<D: Dispatcher + 'static> EthSigning for SigningQueueClient<D> {
 				}
 			});
 
-			p.then(|result| futures::done(result.expect("Ready is never dropped nor canceled."))).boxed()
+			p.then(|result| futures::done(result.expect("Ready is never dropped nor canceled.")))
 		}).boxed()
 	}
 
@@ -220,7 +213,7 @@ impl<D: Dispatcher + 'static> EthSigning for SigningQueueClient<D> {
 				}
 			});
 
-			p.then(|result| futures::done(result.expect("Ready is never dropped nor canceled."))).boxed()
+			p.then(|result| futures::done(result.expect("Ready is never dropped nor canceled.")))
 		}).boxed()
 	}
 
@@ -238,7 +231,7 @@ impl<D: Dispatcher + 'static> EthSigning for SigningQueueClient<D> {
 				}
 			});
 
-			p.then(|result| futures::done(result.expect("Ready is never dropped nor canceled."))).boxed()
+			p.then(|result| futures::done(result.expect("Ready is never dropped nor canceled.")))
 		}).boxed()
 	}
 }

rpc/src/v1/impls/signing_unsafe.rs

@@ -52,26 +52,18 @@ impl<D: Dispatcher + 'static> SigningUnsafeClient<D> {
 	}
 
 	fn handle(&self, payload: RpcConfirmationPayload, account: DefaultAccount) -> BoxFuture<RpcConfirmationResponse, Error> {
-		let setup = move || {
+		let accounts = take_weakf!(self.accounts);
-			let accounts = take_weak!(self.accounts);
+		let default = match account {
-			let default_account = account;
-			let default_account = match default_account {
 			DefaultAccount::Provided(acc) => acc,
 			DefaultAccount::ForDapp(dapp) => accounts.default_address(dapp).ok().unwrap_or_default(),
 		};
 
-			Ok((accounts, default_account))
-		};
-
 		let dis = self.dispatcher.clone();
-		future::done(setup())
-			.and_then(move |(accounts, default)| {
 		dispatch::from_rpc(payload, default, &dis)
 			.and_then(move |payload| {
 				dispatch::execute(dis, &accounts, payload, dispatch::SignWith::Nothing)
 			})
 			.map(|v| v.into_value())
-			})
 			.boxed()
 	}
 }

rpc/src/v1/mod.rs

@@ -18,11 +18,33 @@
 //!
 //! Compliant with ethereum rpc.
 
+// Upgrade a weak pointer, returning an error on failure.
 macro_rules! take_weak {
 	($weak: expr) => {
 		match $weak.upgrade() {
 			Some(arc) => arc,
-			None => return Err(Error::internal_error())
+			None => return Err(Error::internal_error()),
+		}
+	}
+}
+
+// Upgrade a weak pointer, returning an error leaf-future on failure.
+macro_rules! take_weakf {
+	($weak: expr) => {
+		match $weak.upgrade() {
+			Some(arc) => arc,
+			None => return ::futures::future::err(Error::internal_error()).boxed(),
+		}
+	}
+}
+
+// short for "try_boxfuture"
+// unwrap a result, returning a BoxFuture<_, Err> on failure.
+macro_rules! try_bf {
+	($res: expr) => {
+		match $res {
+			Ok(val) => val,
+			Err(e) => return ::futures::future::err(e.into()).boxed(),
 		}
 	}
 }