script/research/validatord: minor cleanup

script/research/validatord/src/lib.rs

@@ -0,0 +1 @@
+pub mod protocols;

script/research/validatord/src/main.rs

@@ -29,12 +29,9 @@ use darkfi::{
     Result,
 };
 
-mod protocol_tx_pool;
-mod tx_pool;
-
-use crate::{
+use validatord::protocols::{
     protocol_tx_pool::ProtocolTxPool,
-    tx_pool::{SeenTxHashes, SeenTxHashesPtr, Tx},
+    tx_pool::{Tx, TxPool, TxPoolPtr},
 };
 
 const CONFIG_FILE: &str = r"validatord_config.toml";
@@ -108,22 +105,22 @@ async fn start(executor: Arc<Executor<'_>>, opts: &Opt) -> Result<()> {
         ..Default::default()
     };
 
-    let seen_tx_hashes = SeenTxHashes::new();
+    let tx_pool = TxPool::new();
 
     // P2P registry setup
     let p2p = net::P2p::new(network_settings).await;
     let registry = p2p.protocol_registry();
 
     let (sender, _) = async_channel::unbounded();
-    let seen_tx_hashes2 = seen_tx_hashes.clone();
+    let tx_pool2 = tx_pool.clone();
     let sender2 = sender.clone();
 
     // Adding ProtocolTxPool to the registry
     registry
         .register(!net::SESSION_SEED, move |channel, p2p| {
             let sender = sender2.clone();
-            let seen_tx_hashes = seen_tx_hashes2.clone();
-            async move { ProtocolTxPool::init(channel, sender, seen_tx_hashes, p2p).await }
+            let tx_pool = tx_pool2.clone();
+            async move { ProtocolTxPool::init(channel, sender, tx_pool, p2p).await }
         })
         .await;
     // TODO: Add protocols for rest message types (block, vote)
@@ -145,7 +142,7 @@ async fn start(executor: Arc<Executor<'_>>, opts: &Opt) -> Result<()> {
     let ex2 = executor.clone();
     let ex3 = ex2.clone();
     let rpc_interface = Arc::new(JsonRpcInterface {
-        seen_tx_hashes: seen_tx_hashes.clone(),
+        tx_pool: tx_pool.clone(),
         p2p: p2p.clone(),
         _rpc_listen_addr: opts.rpc,
     });
@@ -166,7 +163,7 @@ async fn start(executor: Arc<Executor<'_>>, opts: &Opt) -> Result<()> {
 }
 
 struct JsonRpcInterface {
-    seen_tx_hashes: SeenTxHashesPtr,
+    tx_pool: TxPoolPtr,
     p2p: net::P2pPtr,
     _rpc_listen_addr: SocketAddr,
 }
@@ -205,9 +202,9 @@ impl JsonRpcInterface {
     }
 
     // --> {"jsonrpc": "2.0", "method": "get_tx_pool", "params": [], "id": 42}
-    // <-- {"jsonrpc": "2.0", "result": {"nodeID": [], "nodeinfo" [], "id": 42}
+    // <-- {"jsonrpc": "2.0", "result": {"tx_pool", "id": 42}
     async fn get_tx_pool(&self, id: Value, _params: Value) -> JsonResult {
-        let pool = format!("{:?}", self.seen_tx_hashes);
+        let pool = format!("{:?}", self.tx_pool);
         JsonResult::Resp(jsonresp(json!(pool), id))
     }
 
@@ -220,8 +217,12 @@ impl JsonRpcInterface {
             return jsonrpc::error(InvalidParams, None, id).into()
         }
 
+        // TODO: add proper tx hash here and check if its already in the pool
         let random_id = OsRng.next_u32();
-        self.seen_tx_hashes.add_seen(random_id).await;
+        let payload = String::from(args[0].as_str().unwrap());
+        let tx = Tx { hash: random_id, payload };
+
+        self.tx_pool.add_tx(tx).await;
         let protocol_tx = Tx { hash: random_id, payload: args[0].to_string() };
         let result = self.p2p.broadcast(protocol_tx).await;
 

script/research/validatord/src/protocols/mod.rs

@@ -0,0 +1,5 @@
+pub mod protocol_tx_pool;
+pub mod tx_pool;
+
+pub use protocol_tx_pool::ProtocolTxPool;
+pub use tx_pool::{Tx, TxPool, TxPoolPtr};

script/research/validatord/src/protocol_tx_pool.rs → script/research/validatord/src/protocols/protocol_tx_pool.rs

@@ -5,13 +5,13 @@ use darkfi::{net, Result};
 use log::debug;
 use std::sync::Arc;
 
-use crate::tx_pool::{SeenTxHashesPtr, Tx};
+use super::tx_pool::{Tx, TxPoolPtr};
 
 pub struct ProtocolTxPool {
     notify_queue_sender: async_channel::Sender<Arc<Tx>>,
     tx_pool_sub: net::MessageSubscription<Tx>,
     jobsman: net::ProtocolJobsManagerPtr,
-    seen_tx_hashes: SeenTxHashesPtr,
+    tx_pool: TxPoolPtr,
     p2p: net::P2pPtr,
 }
 
@@ -19,7 +19,7 @@ impl ProtocolTxPool {
     pub async fn init(
         channel: net::ChannelPtr,
         notify_queue_sender: async_channel::Sender<Arc<Tx>>,
-        seen_tx_hashes: SeenTxHashesPtr,
+        tx_pool: TxPoolPtr,
         p2p: net::P2pPtr,
     ) -> net::ProtocolBasePtr {
         let message_subsytem = channel.get_message_subsystem();
@@ -31,7 +31,7 @@ impl ProtocolTxPool {
             notify_queue_sender,
             tx_pool_sub: tx_sub,
             jobsman: net::ProtocolJobsManager::new("TxPoolProtocol", channel),
-            seen_tx_hashes,
+            tx_pool,
             p2p,
         })
     }
@@ -46,16 +46,16 @@ impl ProtocolTxPool {
                 "ProtocolTxPool::handle_receive_tx() received {:?}",
                 tx
             );
+            let tx_copy = (*tx).clone();
 
             // Do we already have this tx?
-            if self.seen_tx_hashes.is_seen(tx.hash).await {
+            if self.tx_pool.tx_exists(&tx_copy).await {
                 continue
             }
 
-            self.seen_tx_hashes.add_seen(tx.hash).await;
+            self.tx_pool.add_tx(tx_copy.clone()).await;
 
             // If not then broadcast to everybody else
-            let tx_copy = (*tx).clone();
             self.p2p.broadcast(tx_copy).await?;
 
             self.notify_queue_sender.send(tx).await.expect("notify_queue_sender send failed!");

script/research/validatord/src/tx_pool.rs → script/research/validatord/src/protocols/tx_pool.rs

@@ -11,7 +11,7 @@ use darkfi::{
 
 pub type TxHash = u32; // Change this to a proper hash type
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Eq, Hash, PartialEq)]
 pub struct Tx {
     pub hash: TxHash,
     pub payload: String,
@@ -39,22 +39,22 @@ impl Decodable for Tx {
 }
 
 #[derive(Debug)]
-pub struct SeenTxHashes {
-    seen_tx_hashes: Mutex<FxHashSet<TxHash>>,
+pub struct TxPool {
+    tx_pool: Mutex<FxHashSet<Tx>>,
 }
 
-pub type SeenTxHashesPtr = Arc<SeenTxHashes>;
+pub type TxPoolPtr = Arc<TxPool>;
 
-impl SeenTxHashes {
+impl TxPool {
     pub fn new() -> Arc<Self> {
-        Arc::new(Self { seen_tx_hashes: Mutex::new(FxHashSet::default()) })
+        Arc::new(Self { tx_pool: Mutex::new(FxHashSet::default()) })
     }
 
-    pub async fn add_seen(&self, hash: u32) {
-        self.seen_tx_hashes.lock().await.insert(hash);
+    pub async fn add_tx(&self, tx: Tx) {
+        self.tx_pool.lock().await.insert(tx);
     }
 
-    pub async fn is_seen(&self, hash: u32) -> bool {
-        self.seen_tx_hashes.lock().await.contains(&hash)
+    pub async fn tx_exists(&self, tx: &Tx) -> bool {
+        self.tx_pool.lock().await.contains(tx)
     }
 }