feat(txpool): add transaction_event_listener function (#3493)

crates/transaction-pool/src/lib.rs

@@ -280,6 +280,10 @@ where
         Ok(transactions)
     }
 
+    fn transaction_event_listener(&self, tx_hash: TxHash) -> Option<TransactionEvents> {
+        self.pool.add_transaction_event_listener(tx_hash)
+    }
+
     fn pending_transactions_listener(&self) -> Receiver<TxHash> {
         self.pool.add_pending_listener()
     }

crates/transaction-pool/src/pool/listener.rs

@@ -9,7 +9,7 @@ use std::{
 };
 use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
 
-/// A Stream that receives [TransactionEvent] for the transactions
+/// A Stream that receives [TransactionEvent] for the transaction with the given hash.
 #[derive(Debug)]
 #[must_use = "streams do nothing unless polled"]
 pub struct TransactionEvents {

crates/transaction-pool/src/pool/mod.rs

@@ -204,6 +204,20 @@ where
         rx
     }
 
+    /// If the pool contains the transaction, this adds a new listener that gets notified about
+    /// transaction events.
+    pub(crate) fn add_transaction_event_listener(
+        &self,
+        tx_hash: TxHash,
+    ) -> Option<TransactionEvents> {
+        let pool = self.pool.read();
+        if pool.contains(&tx_hash) {
+            Some(self.event_listener.write().subscribe(tx_hash))
+        } else {
+            None
+        }
+    }
+
     /// Returns hashes of _all_ transactions in the pool.
     pub(crate) fn pooled_transactions_hashes(&self) -> Vec<TxHash> {
         let pool = self.pool.read();

crates/transaction-pool/src/traits.rs

@@ -89,6 +89,11 @@ pub trait TransactionPool: Send + Sync + Clone {
         transactions: Vec<Self::Transaction>,
     ) -> PoolResult<Vec<PoolResult<TxHash>>>;
 
+    /// Returns a new transaction change event stream for the given transaction.
+    ///
+    /// Returns `None` if the transaction is not in the pool.
+    fn transaction_event_listener(&self, tx_hash: TxHash) -> Option<TransactionEvents>;
+
     /// Returns a new Stream that yields transactions hashes for new ready transactions.
     ///
     /// Consumer: RPC