test(txpool): listener it tests (#4019)

Cargo.lock

@@ -5927,6 +5927,7 @@ name = "reth-transaction-pool"
 version = "0.1.0-alpha.4"
 dependencies = [
  "aquamarine",
+ "assert_matches",
  "async-trait",
  "auto_impl",
  "bitflags 1.3.2",

crates/transaction-pool/Cargo.toml

@@ -51,6 +51,7 @@ paste = "1.0"
 rand = "0.8"
 proptest = "1.0"
 criterion = "0.5"
+assert_matches = "1.5"
 
 [features]
 default = ["serde"]

crates/transaction-pool/tests/it/listeners.rs

@@ -0,0 +1,39 @@
+use assert_matches::assert_matches;
+use reth_transaction_pool::{
+    test_utils::{testing_pool, MockTransactionFactory},
+    FullTransactionEvent, TransactionEvent, TransactionOrigin, TransactionPool,
+};
+use tokio_stream::StreamExt;
+
+#[tokio::test(flavor = "multi_thread")]
+async fn txpool_listener_by_hash() {
+    let txpool = testing_pool();
+    let mut mock_tx_factory = MockTransactionFactory::default();
+    let transaction = mock_tx_factory.create_eip1559();
+
+    let result = txpool
+        .add_transaction_and_subscribe(TransactionOrigin::External, transaction.transaction.clone())
+        .await;
+    assert_matches!(result, Ok(_));
+
+    let mut events = result.unwrap();
+    assert_matches!(events.next().await, Some(TransactionEvent::Pending));
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn txpool_listener_all() {
+    let txpool = testing_pool();
+    let mut mock_tx_factory = MockTransactionFactory::default();
+    let transaction = mock_tx_factory.create_eip1559();
+
+    let mut all_tx_events = txpool.all_transactions_event_listener();
+
+    let added_result =
+        txpool.add_transaction(TransactionOrigin::External, transaction.transaction.clone()).await;
+    assert_matches!(added_result, Ok(hash) if hash == transaction.transaction.get_hash());
+
+    assert_matches!(
+        all_tx_events.next().await,
+        Some(FullTransactionEvent::Pending(hash)) if hash == transaction.transaction.get_hash()
+    );
+}

crates/transaction-pool/tests/it/main.rs

@@ -1,3 +1,6 @@
 //! transaction-pool integration tests
 
+#[cfg(feature = "test-utils")]
+mod listeners;
+
 fn main() {}