Watch transactions pool

ethcore/Cargo.toml

@@ -30,6 +30,7 @@ ethereum-types = "0.4"
 ethjson = { path = "../json" }
 ethkey = { path = "../accounts/ethkey" }
 evm = { path = "evm" }
+futures = "0.1"
 hash-db = "0.11.0"
 heapsize = "0.4"
 itertools = "0.5"

ethcore/light/Cargo.toml

@@ -18,6 +18,7 @@ memory-db = "0.11.0"
 trie-db = "0.11.0"
 patricia-trie-ethereum = { path = "../../util/patricia-trie-ethereum" }
 ethcore-network = { path = "../../util/network" }
+ethcore-miner = { path = "../../miner" }
 ethcore-io = { path = "../../util/io" }
 hash-db = "0.11.0"
 heapsize = "0.4"

ethcore/light/src/lib.rs

@@ -61,6 +61,7 @@ extern crate ethcore_io as io;
 extern crate ethcore_network as network;
 extern crate parity_bytes as bytes;
 extern crate ethereum_types;
+extern crate ethcore_miner as miner;
 extern crate ethcore;
 extern crate hash_db;
 extern crate heapsize;

ethcore/light/src/transaction_queue.rs

@@ -24,12 +24,15 @@
 //! address-wise manner.
 
 use std::fmt;
+use std::sync::Arc;
 use std::collections::{BTreeMap, HashMap};
 use std::collections::hash_map::Entry;
 
 use common_types::transaction::{self, Condition, PendingTransaction, SignedTransaction};
 use ethereum_types::{H256, U256, Address};
 use fastmap::H256FastMap;
+use futures::sync::mpsc;
+use miner::pool::TxStatus;
 
 // Knowledge of an account's current nonce.
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -134,6 +137,7 @@ pub struct TransactionQueue {
 	by_account: HashMap<Address, AccountTransactions>,
 	by_hash: H256FastMap<PendingTransaction>,
 	listeners: Vec<Listener>,
+	tx_statuses_listeners: Vec<mpsc::UnboundedSender<Arc<Vec<(H256, TxStatus)>>>>,
 }
 
 impl fmt::Debug for TransactionQueue {
@@ -231,7 +235,7 @@ impl TransactionQueue {
 		};
 
 		self.by_hash.insert(hash, tx);
-		self.notify(&promoted);
+		self.notify(&promoted, TxStatus::Added);
 		Ok(res)
 	}
 
@@ -343,6 +347,8 @@ impl TransactionQueue {
 		trace!(target: "txqueue", "Culled {} old transactions from sender {} (nonce={})",
 			removed_hashes.len(), address, cur_nonce);
 
+		self.notify(&removed_hashes, TxStatus::Culled);
+
 		for hash in removed_hashes {
 			self.by_hash.remove(&hash);
 		}
@@ -358,11 +364,26 @@ impl TransactionQueue {
 		self.listeners.push(f);
 	}
 
+	/// Add a transaction queue listener.
+	pub fn tx_statuses_receiver(&mut self) -> mpsc::UnboundedReceiver<Arc<Vec<(H256, TxStatus)>>> {
+		let (sender, receiver) = mpsc::unbounded();
+		self.tx_statuses_listeners.push(sender);
+		receiver
+	}
+
 	/// Notifies all listeners about new pending transaction.
-	fn notify(&self, hashes: &[H256]) {
+	fn notify(&mut self, hashes: &[H256], status: TxStatus) {
 		for listener in &self.listeners {
 			listener(hashes)
 		}
+
+		let to_send: Arc<Vec<(H256, TxStatus)>> = Arc::new(
+			hashes
+				.into_iter()
+				.map(|hash| (hash.clone(), status)).collect()
+		);
+
+		self.tx_statuses_listeners.retain(| listener| listener.unbounded_send(to_send.clone()).is_ok());
 	}
 }
 

ethcore/src/lib.rs

@@ -74,6 +74,7 @@ extern crate ethcore_miner;
 extern crate ethereum_types;
 extern crate ethjson;
 extern crate ethkey;
+extern crate futures;
 extern crate hash_db;
 extern crate heapsize;
 extern crate itertools;

ethcore/src/miner/miner.rs

@@ -24,11 +24,12 @@ use bytes::Bytes;
 use call_contract::CallContract;
 use ethcore_miner::gas_pricer::GasPricer;
 use ethcore_miner::local_accounts::LocalAccounts;
-use ethcore_miner::pool::{self, TransactionQueue, VerifiedTransaction, QueueStatus, PrioritizationStrategy};
+use ethcore_miner::pool::{self, TransactionQueue, VerifiedTransaction, QueueStatus, PrioritizationStrategy, TxStatus};
 use ethcore_miner::service_transaction_checker::ServiceTransactionChecker;
 #[cfg(feature = "work-notify")]
 use ethcore_miner::work_notify::NotifyWork;
 use ethereum_types::{H256, U256, Address};
+use futures::sync::mpsc;
 use io::IoChannel;
 use miner::pool_client::{PoolClient, CachedNonceClient, NonceCache};
 use miner;
@@ -263,6 +264,13 @@ impl Miner {
 		self.transaction_queue.add_listener(f);
 	}
 
+	/// Set a callback to be notified
+	pub fn tx_pool_receiver(&self) -> mpsc::UnboundedReceiver<Arc<Vec<(H256, TxStatus)>>> {
+		let (sender, receiver) = mpsc::unbounded();
+		self.transaction_queue.add_tx_pool_listener(sender);
+		receiver
+	}
+
 	/// Creates new instance of miner Arc.
 	pub fn new<A: LocalAccounts + 'static>(
 		options: MinerOptions,

miner/Cargo.toml

@@ -31,6 +31,9 @@ parity-runtime = { path = "../util/runtime" }
 parking_lot = "0.7"
 price-info = { path = "./price-info", optional = true }
 rlp = { version = "0.3.0", features = ["ethereum"] }
+serde = "1.0"
+serde_derive = "1.0"
+serde_json = "1.0"
 trace-time = "0.1"
 transaction-pool = "2.0"
 

miner/src/lib.rs

@@ -34,6 +34,7 @@ extern crate parking_lot;
 extern crate price_info;
 extern crate rlp;
 extern crate transaction_pool as txpool;
+extern crate serde;
 
 #[macro_use]
 extern crate ethabi_contract;
@@ -44,6 +45,8 @@ extern crate error_chain;
 #[macro_use]
 extern crate log;
 #[macro_use]
+extern crate serde_derive;
+#[macro_use]
 extern crate trace_time;
 
 #[cfg(test)]

miner/src/pool/listener.rs

@@ -20,9 +20,11 @@ use std::fmt;
 use std::sync::Arc;
 
 use ethereum_types::H256;
+use futures::sync::mpsc;
 use txpool::{self, VerifiedTransaction};
 
 use pool::VerifiedTransaction as Transaction;
+use pool::TxStatus;
 
 type Listener = Box<Fn(&[H256]) + Send + Sync>;
 
@@ -116,6 +118,65 @@ impl txpool::Listener<Transaction> for Logger {
 	}
 }
 
+/// Transactions pool notifier
+#[derive(Default)]
+pub struct TransactionsPoolNotifier {
+	listeners: Vec<mpsc::UnboundedSender<Arc<Vec<(H256, TxStatus)>>>>,
+	tx_statuses: Vec<(H256, TxStatus)>,
+}
+
+impl TransactionsPoolNotifier {
+	/// Add new listener to receive notifications.
+	pub fn add(&mut self, f: mpsc::UnboundedSender<Arc<Vec<(H256, TxStatus)>>>) {
+		self.listeners.push(f);
+	}
+
+	/// Notify listeners about all currently transactions.
+	pub fn notify(&mut self) {
+		if self.tx_statuses.is_empty() {
+			return;
+		}
+
+		let to_send = Arc::new(std::mem::replace(&mut self.tx_statuses, Vec::new()));
+		self.listeners
+			.retain(|listener| listener.unbounded_send(to_send.clone()).is_ok());
+	}
+}
+
+impl fmt::Debug for TransactionsPoolNotifier {
+	fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
+		fmt.debug_struct("TransactionsPoolNotifier")
+			.field("listeners", &self.listeners.len())
+			.finish()
+	}
+}
+
+impl txpool::Listener<Transaction> for TransactionsPoolNotifier {
+	fn added(&mut self, tx: &Arc<Transaction>, _old: Option<&Arc<Transaction>>) {
+		self.tx_statuses.push((tx.hash.clone(), TxStatus::Added));
+	}
+
+	fn rejected<H: fmt::Debug + fmt::LowerHex>(&mut self, tx: &Arc<Transaction>, _reason: &txpool::Error<H>) {
+		self.tx_statuses.push((tx.hash.clone(), TxStatus::Rejected));
+	}
+
+	fn dropped(&mut self, tx: &Arc<Transaction>, _new: Option<&Transaction>) {
+		self.tx_statuses.push((tx.hash.clone(), TxStatus::Dropped));
+	}
+
+	fn invalid(&mut self, tx: &Arc<Transaction>) {
+		self.tx_statuses.push((tx.hash.clone(), TxStatus::Invalid));
+	}
+
+	fn canceled(&mut self, tx: &Arc<Transaction>) {
+		self.tx_statuses.push((tx.hash.clone(), TxStatus::Canceled));
+	}
+
+	fn culled(&mut self, tx: &Arc<Transaction>) {
+		self.tx_statuses.push((tx.hash.clone(), TxStatus::Culled));
+	}
+}
+
 #[cfg(test)]
 mod tests {
 	use super::*;

miner/src/pool/mod.rs

@@ -199,3 +199,21 @@ impl ScoredTransaction for VerifiedTransaction {
 		self.transaction.nonce
 	}
 }
+
+/// Pool transactions status
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub enum TxStatus {
+	/// Added transaction
+	Added,
+	/// Rejected transaction
+	Rejected,
+	/// Dropped transaction
+	Dropped,
+	/// Invalid transaction
+	Invalid,
+	/// Canceled transaction
+	Canceled,
+	/// Culled transaction
+	Culled,
+}

miner/src/pool/queue.rs

@@ -22,17 +22,18 @@ use std::sync::atomic::{self, AtomicUsize};
 use std::collections::{BTreeMap, BTreeSet, HashMap};
 
 use ethereum_types::{H256, U256, Address};
+use futures::sync::mpsc;
 use parking_lot::RwLock;
 use txpool::{self, Verifier};
 use types::transaction;
 
 use pool::{
 	self, replace, scoring, verifier, client, ready, listener,
-	PrioritizationStrategy, PendingOrdering, PendingSettings,
+	PrioritizationStrategy, PendingOrdering, PendingSettings, TxStatus
 };
 use pool::local_transactions::LocalTransactionsList;
 
-type Listener = (LocalTransactionsList, (listener::Notifier, listener::Logger));
+type Listener = (LocalTransactionsList, (listener::Notifier, (listener::Logger, listener::TransactionsPoolNotifier)));
 type Pool = txpool::Pool<pool::VerifiedTransaction, scoring::NonceAndGasPrice, Listener>;
 
 /// Max cache time in milliseconds for pending transactions.
@@ -304,6 +305,8 @@ impl TransactionQueue {
 		// Notify about imported transactions.
 		(self.pool.write().listener_mut().1).0.notify();
 
+		((self.pool.write().listener_mut().1).1).1.notify();
+
 		if results.iter().any(|r| r.is_ok()) {
 			self.cached_pending.write().clear();
 		}
@@ -574,6 +577,12 @@ impl TransactionQueue {
 		(pool.listener_mut().1).0.add(f);
 	}
 
+	/// Add a listener to be notified about all transactions the pool
+	pub fn add_tx_pool_listener(&self, f: mpsc::UnboundedSender<Arc<Vec<(H256, TxStatus)>>>) {
+		let mut pool = self.pool.write();
+		((pool.listener_mut().1).1).1.add(f);
+	}
+
 	/// Check if pending set is cached.
 	#[cfg(test)]
 	pub fn is_pending_cached(&self) -> bool {

parity/cli/mod.rs

@@ -498,7 +498,7 @@ usage! {
 			"--jsonrpc-interface=[IP]",
 			"Specify the hostname portion of the HTTP JSON-RPC API server, IP should be an interface's IP address, or all (all interfaces) or local.",
 
-			ARG arg_jsonrpc_apis: (String) = "web3,eth,pubsub,net,parity,private,parity_pubsub,traces,rpc,shh,shh_pubsub", or |c: &Config| c.rpc.as_ref()?.apis.as_ref().map(|vec| vec.join(",")),
+			ARG arg_jsonrpc_apis: (String) = "web3,eth,pubsub,net,parity,private,parity_pubsub,traces,rpc,shh,shh_pubsub,parity_transactions_pool", or |c: &Config| c.rpc.as_ref()?.apis.as_ref().map(|vec| vec.join(",")),
 			"--jsonrpc-apis=[APIS]",
 			"Specify the APIs available through the HTTP JSON-RPC interface using a comma-delimited list of API names. Possible names are: all, safe, debug, web3, net, eth, pubsub, personal, signer, parity, parity_pubsub, parity_accounts, parity_set, traces, rpc, secretstore, shh, shh_pubsub. You can also disable a specific API by putting '-' in the front, example: all,-personal. 'safe' enables the following APIs: web3, net, eth, pubsub, parity, parity_pubsub, traces, rpc, shh, shh_pubsub",
 
@@ -539,7 +539,7 @@ usage! {
 			"--ws-interface=[IP]",
 			"Specify the hostname portion of the WebSockets JSON-RPC server, IP should be an interface's IP address, or all (all interfaces) or local.",
 
-			ARG arg_ws_apis: (String) = "web3,eth,pubsub,net,parity,parity_pubsub,private,traces,rpc,shh,shh_pubsub", or |c: &Config| c.websockets.as_ref()?.apis.as_ref().map(|vec| vec.join(",")),
+			ARG arg_ws_apis: (String) = "web3,eth,pubsub,net,parity,parity_pubsub,private,traces,rpc,shh,shh_pubsub,parity_transactions_pool", or |c: &Config| c.websockets.as_ref()?.apis.as_ref().map(|vec| vec.join(",")),
 			"--ws-apis=[APIS]",
 			"Specify the JSON-RPC APIs available through the WebSockets interface using a comma-delimited list of API names. Possible names are: all, safe, web3, net, eth, pubsub, personal, signer, parity, parity_pubsub, parity_accounts, parity_set, traces, rpc, secretstore, shh, shh_pubsub. You can also disable a specific API by putting '-' in the front, example: all,-personal. 'safe' enables the following APIs: web3, net, eth, pubsub, parity, parity_pubsub, traces, rpc, shh, shh_pubsub",
 
@@ -564,7 +564,7 @@ usage! {
 			"--ipc-path=[PATH]",
 			"Specify custom path for JSON-RPC over IPC service.",
 
-			ARG arg_ipc_apis: (String) = "web3,eth,pubsub,net,parity,parity_pubsub,parity_accounts,private,traces,rpc,shh,shh_pubsub", or |c: &Config| c.ipc.as_ref()?.apis.as_ref().map(|vec| vec.join(",")),
+			ARG arg_ipc_apis: (String) = "web3,eth,pubsub,net,parity,parity_pubsub,parity_accounts,private,traces,rpc,shh,shh_pubsub,parity_transactions_pool", or |c: &Config| c.ipc.as_ref()?.apis.as_ref().map(|vec| vec.join(",")),
 			"--ipc-apis=[APIS]",
 			"Specify custom API set available via JSON-RPC over IPC using a comma-delimited list of API names. Possible names are: all, safe, web3, net, eth, pubsub, personal, signer, parity, parity_pubsub, parity_accounts, parity_set, traces, rpc, secretstore, shh, shh_pubsub. You can also disable a specific API by putting '-' in the front, example: all,-personal. 'safe' enables the following APIs: web3, net, eth, pubsub, parity, parity_pubsub, traces, rpc, shh, shh_pubsub",