add start_newPendingTransactions_loop

client/src/rpc/impls/eth_pubsub.rs

@@ -50,9 +50,12 @@ pub struct PubSubClient {
  handler: Arc<ChainNotificationHandler>,
  heads_subscribers: Arc<RwLock<Subscribers<Client>>>,
  logs_subscribers: Arc<RwLock<Subscribers<(Client, LogFilter)>>>,
+ pending_transactions_subscribers: Arc<RwLock<Subscribers<Client>>>,
  epochs_ordered: Arc<Channel<(u64, Vec<H256>)>>,
  consensus: SharedConsensusGraph,
  heads_loop_started: Arc<RwLock<bool>>,
+ pending_transactions_loop_started: Arc<RwLock<bool>>,
+ new_pending_transactions: Arc<Channel<H256>>,
 }
 
 impl PubSubClient {
@@ -64,21 +67,26 @@ impl PubSubClient {
  {
  let heads_subscribers = Arc::new(RwLock::new(Subscribers::default()));
  let logs_subscribers = Arc::new(RwLock::new(Subscribers::default()));
+ let pending_transactions_subscribers = Arc::new(RwLock::new(Subscribers::default()));
 
  let handler = Arc::new(ChainNotificationHandler {
  executor,
  consensus: consensus.clone(),
  data_man: consensus.get_data_manager().clone(),
  heads_subscribers: heads_subscribers.clone(),
+ pending_transactions_subscribers: pending_transactions_subscribers.clone()
  });
 
  PubSubClient {
  handler,
  heads_subscribers,
  logs_subscribers,
+ pending_transactions_subscribers,
  epochs_ordered: notifications.epochs_ordered.clone(),
  consensus: consensus.clone(),
+ new_pending_transactions: notifications.new_pending_transactions.clone(),
  heads_loop_started: Arc::new(RwLock::new(false)),
+ pending_transactions_loop_started: Arc::new(RwLock::new(false)),
  }
  }
 
@@ -91,6 +99,31 @@ impl PubSubClient {
  self.epochs_ordered.clone()
  }
 
+ fn start_new_pending_transactions_loop(&self) {
+ let mut loop_started = self.pending_transactions_loop_started.write();
+ if *loop_started {
+ return;
+ }
+
+ debug!("start_pending_transactions_loop");
+ *loop_started = true;
+
+ let mut receiver = self.new_pending_transactions.subscribe();
+
+ let handler_clone = self.handler.clone();
+
+ let fut = async move {
+ while let Some(new_tx) = receiver.recv().await {
+ trace!("new_pending_transactions_loop: {:?}", new_tx);
+
+ handler_clone.notify_new_pending_transaction(new_tx);
+ }
+ };
+
+ let fut = fut.unit_error().boxed().compat();
+ self.handler.executor.spawn(fut);
+ }
+
  fn start_heads_loop(&self) {
  let mut loop_started = self.heads_loop_started.write();
  if *loop_started {
@@ -232,6 +265,7 @@ pub struct ChainNotificationHandler {
  consensus: SharedConsensusGraph,
  data_man: Arc<BlockDataManager>,
  heads_subscribers: Arc<RwLock<Subscribers<Client>>>,
+ pending_transactions_subscribers: Arc<RwLock<Subscribers<Client>>>,
 }
 
 impl ChainNotificationHandler {
@@ -254,6 +288,28 @@ impl ChainNotificationHandler {
  let _ = fut.compat().await;
  }
 
+ // notify each subscriber about new pending transaction `hash` concurrently
+ fn notify_new_pending_transaction(&self, hash: H256) {
+ trace!("notify_new_pending_transaction({:?})", hash);
+
+ let subscribers = self.pending_transactions_subscribers.read();
+
+ // do not retrieve anything unnecessarily
+ if subscribers.is_empty() {
+ trace!("No subscribers for pending transactions");
+ return;
+ }
+
+ debug!("Notify {}", hash);
+ for subscriber in subscribers.values() {
+ Self::notify(
+ &self.executor,
+ subscriber,
+ pubsub::Result::TransactionHash(hash),
+ );
+ }
+ }
+
  // notify each subscriber about header `hash` concurrently
  // NOTE: multiple calls to this method will result in concurrent
  // notifications, so the headers published might be reordered.
@@ -542,6 +598,13 @@ impl PubSub for PubSubClient {
  )
  {
  let error = match (kind, params) {
+ // --------- newPendingTransactions ---------
+ (pubsub::Kind::NewPendingTransactions, None) => {
+ info!("eth pubsub newPendingTransactions");
+ self.pending_transactions_subscribers.write().push(subscriber);
+ self.start_new_pending_transactions_loop();
+ return;
+ }
  // --------- newHeads ---------
  (pubsub::Kind::NewHeads, None) => {
  info!("eth pubsub newheads");

core/src/channel.rs

@@ -113,6 +113,7 @@ pub struct Notifications {
  pub new_block_hashes: Arc<Channel<H256>>,
  pub epochs_ordered: Arc<Channel<(u64, Vec<H256>)>>,
  pub blame_verification_results: Arc<Channel<(u64, Option<u64>)>>, /* <height, witness> */
+ pub new_pending_transactions: Arc<Channel<H256>>,
 }
 
 impl Notifications {
@@ -123,6 +124,7 @@ impl Notifications {
  blame_verification_results: Arc::new(Channel::new(
  "blame-verification-results",
  )),
+ new_pending_transactions: Arc::new(Channel::new("new-pending-transactions")),
  })
  }
 }