add function to start the pool

binaries/cuprated/src/main.rs

@@ -12,6 +12,8 @@
     reason = "TODO: remove after v1.0.0"
 )]
 
+use crate::txpool::IncomingTxHandler;
+
 mod blockchain;
 mod config;
 mod constants;

binaries/cuprated/src/txpool.rs

@@ -2,6 +2,40 @@
 //!
 //! Will handle initiating the tx-pool, providing the preprocessor required for the dandelion pool.
 
+use cuprate_consensus::BlockChainContextService;
+use cuprate_p2p::NetworkInterface;
+use cuprate_p2p_core::ClearNet;
+use cuprate_txpool::service::{TxpoolReadHandle, TxpoolWriteHandle};
+
+use crate::blockchain::ConcreteTxVerifierService;
+
 mod dandelion;
 mod incoming_tx;
 mod txs_being_handled;
+
+pub use incoming_tx::{IncomingTxError, IncomingTxHandler, IncomingTxs};
+
+pub fn init_incoming_tx_handler(
+    clear_net: NetworkInterface<ClearNet>,
+    txpool_write_handle: TxpoolWriteHandle,
+    txpool_read_handle: TxpoolReadHandle,
+    blockchain_context_cache: BlockChainContextService,
+    tx_verifier_service: ConcreteTxVerifierService,
+) -> IncomingTxHandler {
+    let dandelion_router = dandelion::dandelion_router(clear_net);
+
+    let dandelion_pool_manager = dandelion::start_dandelion_pool_manager(
+        dandelion_router,
+        txpool_read_handle.clone(),
+        txpool_write_handle.clone(),
+    );
+
+    IncomingTxHandler {
+        txs_being_handled: txs_being_handled::TxsBeingHandled::new(),
+        blockchain_context_cache,
+        dandelion_pool_manager,
+        tx_verifier_service,
+        txpool_write_handle,
+        txpool_read_handle,
+    }
+}

binaries/cuprated/src/txpool/dandelion.rs

@@ -8,15 +8,12 @@ use cuprate_p2p_core::ClearNet;
 use cuprate_txpool::service::{TxpoolReadHandle, TxpoolWriteHandle};
 use cuprate_wire::NetworkAddress;
 
+use super::incoming_tx::{DandelionTx, TxId};
+
 mod diffuse_service;
 mod stem_service;
 mod tx_store;
 
-#[derive(Clone)]
-pub struct DandelionTx(Bytes);
-
-type TxId = [u8; 32];
-
 const DANDELION_CONFIG: DandelionConfig = DandelionConfig {
     time_between_hop: Duration::from_millis(175),
     epoch_duration: Duration::from_secs(10 * 60),
@@ -38,7 +35,7 @@ pub fn start_dandelion_pool_manager(
     txpool_write_handle: TxpoolWriteHandle,
 ) -> DandelionPoolService<DandelionTx, TxId, NetworkAddress> {
     cuprate_dandelion_tower::pool::start_dandelion_pool_manager(
-        12,
+        32,
         router,
         tx_store::TxStoreService {
             txpool_read_handle,

binaries/cuprated/src/txpool/incoming_tx.rs

@@ -51,27 +51,27 @@ pub struct IncomingTxs {
 
 ///  The transaction type used for dandelion++.
 #[derive(Clone)]
-struct DandelionTx(Bytes);
+pub struct DandelionTx(pub Bytes);
 
 /// A transaction ID/hash.
-type TxId = [u8; 32];
+pub(super) type TxId = [u8; 32];
 
 /// The service than handles incoming transaction pool transactions.
 ///
 /// This service handles everything including verifying the tx, adding it to the pool and routing it to other nodes.
 pub struct IncomingTxHandler {
     /// A store of txs currently being handled in incoming tx requests.
-    txs_being_handled: TxsBeingHandled,
+    pub(super) txs_being_handled: TxsBeingHandled,
     /// The blockchain context cache.
-    blockchain_context_cache: BlockChainContextService,
+    pub(super) blockchain_context_cache: BlockChainContextService,
     /// The dandelion txpool manager.
-    dandelion_pool_manager: DandelionPoolService<DandelionTx, TxId, NetworkAddress>,
+    pub(super) dandelion_pool_manager: DandelionPoolService<DandelionTx, TxId, NetworkAddress>,
     /// The transaction verifier service.
-    tx_verifier_service: ConcreteTxVerifierService,
+    pub(super) tx_verifier_service: ConcreteTxVerifierService,
     /// The txpool write handle.
-    txpool_write_handle: TxpoolWriteHandle,
+    pub(super) txpool_write_handle: TxpoolWriteHandle,
     /// The txpool read handle.
-    txpool_read_handle: TxpoolReadHandle,
+    pub(super) txpool_read_handle: TxpoolReadHandle,
 }
 
 impl Service<IncomingTxs> for IncomingTxHandler {

binaries/cuprated/src/txpool/txs_being_handled.rs

@@ -12,6 +12,10 @@ pub fn tx_blob_hash(tx_bytes: &[u8]) -> [u8; 32] {
 pub struct TxsBeingHandled(Arc<DashSet<[u8; 32]>>);
 
 impl TxsBeingHandled {
+    pub fn new() -> Self {
+        Self(Arc::new(DashSet::new()))
+    }
+
     pub fn local_tracker(&self) -> TxBeingHandledLocally {
         TxBeingHandledLocally {
             txs_being_handled: self.clone(),