RPC Server: pushTransaction and pushHighPriorityTransaction methods broadcast the transaction

lib/src/extras/rpc_server.rs

@@ -49,7 +49,7 @@ pub fn initialize_rpc_server(
     ));
     dispatcher.add(NetworkDispatcher::new(client.network()));
     if let Some(mempool) = client.mempool() {
-        dispatcher.add(MempoolDispatcher::new(mempool));
+        dispatcher.add(MempoolDispatcher::new(client.consensus_proxy(), mempool));
     }
     dispatcher.add(PolicyDispatcher {});
     if let Some(validator_proxy) = client.validator_proxy() {

mempool/src/mempool.rs

@@ -612,7 +612,7 @@ impl Mempool {
         (txs, size)
     }
 
-    /// Adds a transaction to the Mempool.
+    /// Adds a transaction to the local mempool without broadcasting it over the network.
     pub fn add_transaction(
         &self,
         transaction: Transaction,

rpc-interface/src/mempool.rs

@@ -9,11 +9,11 @@ use crate::types::{HashOrTx, MempoolInfo, RPCResult};
 pub trait MempoolInterface {
     type Error;
 
-    /// Pushes a raw transaction into the mempool, it will be assigned a default priority.
+    /// Pushes a raw transaction with a default priority assigned into the mempool and broadcast it to the network.
     async fn push_transaction(&mut self, raw_tx: String)
         -> RPCResult<Blake2bHash, (), Self::Error>;
 
-    /// Pushes a raw transaction into the mempool with high priority.
+    /// Pushes a raw transaction with a high priority assigned into the mempool and broadcast it to the network.
     async fn push_high_priority_transaction(
         &mut self,
         raw_tx: String,

rpc-server/src/dispatchers/mempool.rs

@@ -1,8 +1,10 @@
 use std::sync::Arc;
 
 use async_trait::async_trait;
+use nimiq_consensus::ConsensusProxy;
 use nimiq_hash::{Blake2bHash, Hash};
 use nimiq_mempool::{mempool::Mempool, mempool_transactions::TxPriority};
+use nimiq_network_libp2p::Network;
 use nimiq_rpc_interface::{
     mempool::MempoolInterface,
     types::{HashOrTx, MempoolInfo, RPCResult},
@@ -14,12 +16,13 @@ use crate::error::Error;
 
 #[allow(dead_code)]
 pub struct MempoolDispatcher {
+    consensus: ConsensusProxy<Network>,
     mempool: Arc<Mempool>,
 }
 
 impl MempoolDispatcher {
-    pub fn new(mempool: Arc<Mempool>) -> Self {
-        MempoolDispatcher { mempool }
+    pub fn new(consensus: ConsensusProxy<Network>, mempool: Arc<Mempool>) -> Self {
+        MempoolDispatcher { consensus, mempool }
     }
 }
 
@@ -35,8 +38,13 @@ impl MempoolInterface for MempoolDispatcher {
         let tx = Transaction::deserialize_from_vec(&hex::decode(&raw_tx)?)?;
         let txid = tx.hash::<Blake2bHash>();
 
-        match self.mempool.add_transaction(tx, None) {
-            Ok(_) => Ok(txid.into()),
+        match self.mempool.add_transaction(tx.clone(), None) {
+            Ok(_) => self
+                .consensus
+                .send_transaction(tx)
+                .await
+                .map(|_| txid.into())
+                .map_err(Error::NetworkError),
             Err(e) => Err(Error::MempoolError(e)),
         }
     }
@@ -48,8 +56,16 @@ impl MempoolInterface for MempoolDispatcher {
         let tx = Transaction::deserialize_from_vec(&hex::decode(&raw_tx)?)?;
         let txid = tx.hash::<Blake2bHash>();
 
-        match self.mempool.add_transaction(tx, Some(TxPriority::High)) {
-            Ok(_) => Ok(txid.into()),
+        match self
+            .mempool
+            .add_transaction(tx.clone(), Some(TxPriority::High))
+        {
+            Ok(_) => self
+                .consensus
+                .send_transaction(tx)
+                .await
+                .map(|_| txid.into())
+                .map_err(Error::NetworkError),
             Err(e) => Err(Error::MempoolError(e)),
         }
     }