[r2r] broadcast tx to txhlp / refactor tx errors

mm2src/coins/eth.rs

@@ -1119,6 +1119,17 @@ impl MarketCoinOps for EthCoin {
         )
     }
 
+    fn send_raw_tx_bytes(&self, tx: &[u8]) -> Box<dyn Future<Item = String, Error = String> + Send> {
+        let bytes = try_fus!(hex::decode(tx));
+        Box::new(
+            self.web3
+                .eth()
+                .send_raw_transaction(bytes.into())
+                .map(|res| format!("{:02x}", res))
+                .map_err(|e| ERRL!("{}", e)),
+        )
+    }
+
     fn wait_for_confirmations(
         &self,
         tx: &[u8],

mm2src/coins/lightning.rs

@@ -398,6 +398,15 @@ impl MarketCoinOps for LightningCoin {
         ))
     }
 
+    fn send_raw_tx_bytes(&self, _tx: &[u8]) -> Box<dyn Future<Item = String, Error = String> + Send> {
+        Box::new(futures01::future::err(
+            MmError::new(
+                "send_raw_tx is not supported for lightning, please use send_payment method instead.".to_string(),
+            )
+            .to_string(),
+        ))
+    }
+
     // Todo: Implement this when implementing swaps for lightning as it's is used mainly for swaps
     fn wait_for_confirmations(
         &self,

mm2src/coins/lp_coins.rs

@@ -422,6 +422,8 @@ pub trait MarketCoinOps {
     /// Receives raw transaction bytes in hexadecimal format as input and returns tx hash in hexadecimal format
     fn send_raw_tx(&self, tx: &str) -> Box<dyn Future<Item = String, Error = String> + Send>;
 
+    fn send_raw_tx_bytes(&self, tx: &[u8]) -> Box<dyn Future<Item = String, Error = String> + Send>;
+
     fn wait_for_confirmations(
         &self,
         tx: &[u8],

mm2src/coins/qrc20.rs

@@ -1053,10 +1053,16 @@ impl MarketCoinOps for Qrc20Coin {
 
     fn platform_ticker(&self) -> &str { &self.0.platform }
 
+    #[inline(always)]
     fn send_raw_tx(&self, tx: &str) -> Box<dyn Future<Item = String, Error = String> + Send> {
         utxo_common::send_raw_tx(&self.utxo, tx)
     }
 
+    #[inline(always)]
+    fn send_raw_tx_bytes(&self, tx: &[u8]) -> Box<dyn Future<Item = String, Error = String> + Send> {
+        utxo_common::send_raw_tx(&self.utxo, tx)
+    }
+
     fn wait_for_confirmations(
         &self,
         tx: &[u8],

mm2src/coins/solana.rs

@@ -373,6 +373,20 @@ impl MarketCoinOps for SolanaCoin {
         Box::new(fut.boxed().compat())
     }
 
+    fn send_raw_tx_bytes(&self, tx: &[u8]) -> Box<dyn Future<Item = String, Error = String> + Send> {
+        let coin = self.clone();
+        let tx = tx.to_owned();
+        let fut = async move {
+            let bytes = hex::decode(tx).map_to_mm(|e| e).map_err(|e| format!("{:?}", e))?;
+            let tx: Transaction = deserialize(bytes.as_slice())
+                .map_to_mm(|e| e)
+                .map_err(|e| format!("{:?}", e))?;
+            let signature = coin.rpc().send_transaction(&tx).await.map_err(|e| format!("{:?}", e))?;
+            Ok(signature.to_string())
+        };
+        Box::new(fut.boxed().compat())
+    }
+
     fn wait_for_confirmations(
         &self,
         _tx: &[u8],

mm2src/coins/solana/solana_tests.rs

@@ -257,9 +257,15 @@ mod tests {
             .await
             .unwrap();
         println!("{:?}", valid_tx_details);
+
         let tx_str = str::from_utf8(&*valid_tx_details.tx_hex.0).unwrap();
         let res = sol_coin.send_raw_tx(tx_str).compat().await;
         assert_eq!(res.is_err(), false);
+
+        let res2 = sol_coin.send_raw_tx_bytes(&*valid_tx_details.tx_hex).compat().await;
+        assert_eq!(res2.is_err(), false);
+        assert_eq!(res, res2);
+
         //println!("{:?}", res);
     }
 

mm2src/coins/solana/spl.rs

@@ -218,10 +218,16 @@ impl MarketCoinOps for SplToken {
 
     fn platform_ticker(&self) -> &str { self.platform_coin.ticker() }
 
+    #[inline(always)]
     fn send_raw_tx(&self, tx: &str) -> Box<dyn Future<Item = String, Error = String> + Send> {
         self.platform_coin.send_raw_tx(tx)
     }
 
+    #[inline(always)]
+    fn send_raw_tx_bytes(&self, tx: &[u8]) -> Box<dyn Future<Item = String, Error = String> + Send> {
+        self.platform_coin.send_raw_tx_bytes(tx)
+    }
+
     fn wait_for_confirmations(
         &self,
         _tx: &[u8],

mm2src/coins/solana/spl_tests.rs

@@ -82,9 +82,17 @@ mod tests {
         assert_eq!(valid_tx_details.my_balance_change, withdraw_amount.neg());
         assert_eq!(valid_tx_details.coin, "USDC".to_string());
         assert_ne!(valid_tx_details.timestamp, 0);
+
         let tx_str = from_utf8(&*valid_tx_details.tx_hex.0).unwrap();
         let res = usdc_sol_coin.send_raw_tx(tx_str).compat().await;
         assert_eq!(res.is_err(), false);
         println!("{:?}", res);
+
+        let res2 = usdc_sol_coin
+            .send_raw_tx_bytes(&*valid_tx_details.tx_hex.0)
+            .compat()
+            .await;
+        assert_eq!(res2.is_err(), false);
+        assert_eq!(res, res2);
     }
 }

mm2src/coins/test_coin.rs

@@ -44,6 +44,8 @@ impl MarketCoinOps for TestCoin {
     /// Receives raw transaction bytes in hexadecimal format as input and returns tx hash in hexadecimal format
     fn send_raw_tx(&self, tx: &str) -> Box<dyn Future<Item = String, Error = String> + Send> { unimplemented!() }
 
+    fn send_raw_tx_bytes(&self, tx: &[u8]) -> Box<dyn Future<Item = String, Error = String> + Send> { unimplemented!() }
+
     fn wait_for_confirmations(
         &self,
         tx: &[u8],

mm2src/coins/utxo/bch.rs

@@ -1053,10 +1053,16 @@ impl MarketCoinOps for BchCoin {
 
     fn platform_ticker(&self) -> &str { self.ticker() }
 
+    #[inline(always)]
     fn send_raw_tx(&self, tx: &str) -> Box<dyn Future<Item = String, Error = String> + Send> {
         utxo_common::send_raw_tx(&self.utxo_arc, tx)
     }
 
+    #[inline(always)]
+    fn send_raw_tx_bytes(&self, tx: &[u8]) -> Box<dyn Future<Item = String, Error = String> + Send> {
+        utxo_common::send_raw_tx(&self.utxo_arc, tx)
+    }
+
     fn wait_for_confirmations(
         &self,
         tx: &[u8],

mm2src/coins/utxo/qtum.rs

@@ -673,10 +673,16 @@ impl MarketCoinOps for QtumCoin {
 
     fn platform_ticker(&self) -> &str { self.ticker() }
 
+    #[inline(always)]
     fn send_raw_tx(&self, tx: &str) -> Box<dyn Future<Item = String, Error = String> + Send> {
         utxo_common::send_raw_tx(&self.utxo_arc, tx)
     }
 
+    #[inline(always)]
+    fn send_raw_tx_bytes(&self, tx: &[u8]) -> Box<dyn Future<Item = String, Error = String> + Send> {
+        utxo_common::send_raw_tx(&self.utxo_arc, tx)
+    }
+
     fn wait_for_confirmations(
         &self,
         tx: &[u8],

mm2src/coins/utxo/slp.rs

@@ -1083,16 +1083,31 @@ impl MarketCoinOps for SlpToken {
 
     /// Receives raw transaction bytes in hexadecimal format as input and returns tx hash in hexadecimal format
     fn send_raw_tx(&self, tx: &str) -> Box<dyn Future<Item = String, Error = String> + Send> {
-        let coin = self.clone();
+        let _self = self.clone();
         let tx = tx.to_owned();
         let fut = async move {
             let bytes = hex::decode(tx).map_to_mm(|e| e).map_err(|e| format!("{:?}", e))?;
             let tx = deserialize(bytes.as_slice())
                 .map_to_mm(|e| e)
                 .map_err(|e| format!("{:?}", e))?;
-            let hash = coin.broadcast_tx(&tx).await.map_err(|e| format!("{:?}", e))?;
+            let hash = _self.broadcast_tx(&tx).await.map_err(|e| format!("{:?}", e))?;
             Ok(format!("{:?}", hash))
         };
+
+        Box::new(fut.boxed().compat())
+    }
+
+    fn send_raw_tx_bytes(&self, tx: &[u8]) -> Box<dyn Future<Item = String, Error = String> + Send> {
+        let _self = self.clone();
+        let bytes = tx.to_owned();
+        let fut = async move {
+            let tx = deserialize(bytes.as_slice())
+                .map_to_mm(|e| e)
+                .map_err(|e| format!("{:?}", e))?;
+            let hash = _self.broadcast_tx(&tx).await.map_err(|e| format!("{:?}", e))?;
+            Ok(format!("{:?}", hash))
+        };
+
         Box::new(fut.boxed().compat())
     }
 
@@ -2018,12 +2033,17 @@ mod slp_tests {
             142, 135, 205, 228, 173, 0, 0, 0, 0, 0, 25, 118, 169, 20, 140, 255, 252, 36, 9, 208, 99, 67, 125, 106, 168,
             183, 90, 0, 155, 155, 165, 27, 113, 252, 136, 172, 216, 36, 92, 97,
         ];
-        let tx_bytes_str = hex::encode(tx_bytes);
 
+        let tx_bytes_str = hex::encode(tx_bytes);
         let err = fusd.send_raw_tx(&tx_bytes_str).wait().unwrap_err();
         println!("{:?}", err);
         assert!(err.contains("is not valid with reason outputs greater than inputs"));
 
+        let err2 = fusd.send_raw_tx_bytes(tx_bytes).wait().unwrap_err();
+        println!("{:?}", err2);
+        assert!(err2.contains("is not valid with reason outputs greater than inputs"));
+        assert_eq!(err, err2);
+
         let utxo_tx: UtxoTx = deserialize(tx_bytes).unwrap();
         let err = block_on(fusd.broadcast_tx(&utxo_tx)).unwrap_err();
         match err.into_inner() {

mm2src/coins/utxo/utxo_common.rs

@@ -1696,7 +1696,10 @@ where
     Box::new(fut.boxed().compat())
 }
 
-pub fn send_raw_tx(coin: &UtxoCoinFields, tx: &str) -> Box<dyn Future<Item = String, Error = String> + Send> {
+pub fn send_raw_tx<T: AsRef<[u8]>>(
+    coin: &UtxoCoinFields,
+    tx: T,
+) -> Box<dyn Future<Item = String, Error = String> + Send> {
     let bytes = try_fus!(hex::decode(tx));
     Box::new(
         coin.rpc_client

mm2src/coins/utxo/utxo_standard.rs

@@ -449,10 +449,16 @@ impl MarketCoinOps for UtxoStandardCoin {
 
     fn platform_ticker(&self) -> &str { self.ticker() }
 
+    #[inline(always)]
     fn send_raw_tx(&self, tx: &str) -> Box<dyn Future<Item = String, Error = String> + Send> {
         utxo_common::send_raw_tx(&self.utxo_arc, tx)
     }
 
+    #[inline(always)]
+    fn send_raw_tx_bytes(&self, tx: &[u8]) -> Box<dyn Future<Item = String, Error = String> + Send> {
+        utxo_common::send_raw_tx(&self.utxo_arc, tx)
+    }
+
     fn wait_for_confirmations(
         &self,
         tx: &[u8],

mm2src/coins/z_coin.rs

@@ -729,10 +729,16 @@ impl MarketCoinOps for ZCoin {
 
     fn platform_ticker(&self) -> &str { self.ticker() }
 
+    #[inline(always)]
     fn send_raw_tx(&self, tx: &str) -> Box<dyn Future<Item = String, Error = String> + Send> {
         utxo_common::send_raw_tx(self.as_ref(), tx)
     }
 
+    #[inline(always)]
+    fn send_raw_tx_bytes(&self, tx: &[u8]) -> Box<dyn Future<Item = String, Error = String> + Send> {
+        utxo_common::send_raw_tx(self.as_ref(), tx)
+    }
+
     fn wait_for_confirmations(
         &self,
         tx: &[u8],

mm2src/lp_network.rs

@@ -16,11 +16,12 @@
 //  lp_network.rs
 //  marketmaker
 //
+use coins::lp_coinfind;
 use common::executor::spawn;
-use common::log;
 use common::mm_ctx::{MmArc, MmWeak};
 use common::mm_error::prelude::*;
 use common::mm_metrics::{ClockOps, MetricsOps};
+use common::{log, Future01CompatExt};
 use derive_more::Display;
 use futures::{channel::oneshot, StreamExt};
 use mm2_libp2p::atomicdex_behaviour::{AdexBehaviourCmd, AdexBehaviourEvent, AdexCmdTx, AdexEventRx, AdexResponse,
@@ -137,6 +138,16 @@ async fn process_p2p_message(
                 lp_swap::process_msg(ctx.clone(), split.next().unwrap_or_default(), &message.data).await;
                 to_propagate = true;
             },
+            Some(lp_swap::TX_HELPER_PREFIX) => {
+                if let Some(pair) = split.next() {
+                    if let Ok(Some(coin)) = lp_coinfind(&ctx, pair).await {
+                        match coin.send_raw_tx_bytes(&message.data).compat().await {
+                            Ok(id) => log::info!("Transaction broadcasted successfully: {:?} ", id),
+                            Err(e) => log::error!("Broadcast transaction failed. {}", e),
+                        }
+                    }
+                }
+            },
             None | Some(_) => (),
         }
     }
@@ -149,6 +160,7 @@ async fn process_p2p_message(
             &message.data,
             i_am_relay,
         );
+
         if process_fut.await {
             to_propagate = true;
         }

mm2src/lp_swap.rs

@@ -114,6 +114,8 @@ pub use trade_preimage::trade_preimage_rpc;
 
 pub const SWAP_PREFIX: TopicPrefix = "swap";
 
+pub const TX_HELPER_PREFIX: TopicPrefix = "txhlp";
+
 cfg_wasm32! {
     use common::indexed_db::{ConstructibleDb, DbLocked};
     use swap_wasm_db::{InitDbResult, SwapDb};
@@ -188,6 +190,16 @@ pub fn broadcast_swap_message(ctx: &MmArc, topic: String, msg: SwapMsg, p2p_priv
     broadcast_p2p_msg(ctx, vec![topic], encoded_msg);
 }
 
+pub fn broadcast_transaction_message(ctx: &MmArc, topic: String, msg: Vec<u8>, p2p_privkey: &Option<H256Json>) {
+    let p2p_private = match p2p_privkey {
+        Some(privkey) => privkey.0,
+        None => ctx.secp256k1_key_pair.or(&&|| panic!()).private().secret.take(),
+    };
+
+    let encoded_msg = encode_and_sign(&msg, &p2p_private).unwrap();
+    broadcast_p2p_msg(ctx, vec![topic], encoded_msg);
+}
+
 pub async fn process_msg(ctx: MmArc, topic: &str, msg: &[u8]) {
     let uuid = match Uuid::from_str(topic) {
         Ok(u) => u,
@@ -235,6 +247,16 @@ pub async fn process_msg(ctx: MmArc, topic: &str, msg: &[u8]) {
 
 pub fn swap_topic(uuid: &Uuid) -> String { pub_sub_topic(SWAP_PREFIX, &uuid.to_string()) }
 
+/// Formats and returns a topic format for `txhlp`.
+///
+/// # Usage
+/// ```ignore
+/// let topic = tx_helper_topic("BTC");
+/// // Returns topic format `txhlp/BTC` as String type.
+/// ```
+#[inline(always)]
+pub fn tx_helper_topic(coin: &str) -> String { pub_sub_topic(TX_HELPER_PREFIX, coin) }
+
 async fn recv_swap_msg<T>(
     ctx: MmArc,
     mut getter: impl FnMut(&mut SwapMsgStore) -> Option<T>,