[r2r] broadcast tx to txhlp / refactor tx errors

mm2src/coins/eth.rs

@@ -1119,6 +1119,20 @@ impl MarketCoinOps for EthCoin {
         )
     }
 
+    fn send_raw_tx_bytes(&self, mut tx: &[u8]) -> Box<dyn Future<Item = String, Error = String> + Send> {
+        if tx.starts_with(b"0x") {
+            tx = &tx[2..];
+        }
+
+        Box::new(
+            self.web3
+                .eth()
+                .send_raw_transaction(tx.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

@@ -1057,6 +1057,10 @@ impl MarketCoinOps for Qrc20Coin {
         utxo_common::send_raw_tx(&self.utxo, tx)
     }
 
+    fn send_raw_tx_bytes(&self, tx: &[u8]) -> Box<dyn Future<Item = String, Error = String> + Send> {
+        utxo_common::send_raw_tx_bytes(&self.utxo, tx)
+    }
+
     fn wait_for_confirmations(
         &self,
         tx: &[u8],

mm2src/coins/solana.rs

@@ -373,6 +373,19 @@ 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 tx: Transaction = deserialize(tx.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/spl.rs

@@ -222,6 +222,10 @@ impl MarketCoinOps for SplToken {
         self.platform_coin.send_raw_tx(tx)
     }
 
+    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/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

@@ -1057,6 +1057,10 @@ impl MarketCoinOps for BchCoin {
         utxo_common::send_raw_tx(&self.utxo_arc, tx)
     }
 
+    fn send_raw_tx_bytes(&self, tx: &[u8]) -> Box<dyn Future<Item = String, Error = String> + Send> {
+        utxo_common::send_raw_tx_bytes(&self.utxo_arc, tx)
+    }
+
     fn wait_for_confirmations(
         &self,
         tx: &[u8],

mm2src/coins/utxo/qtum.rs

@@ -677,6 +677,10 @@ impl MarketCoinOps for QtumCoin {
         utxo_common::send_raw_tx(&self.utxo_arc, tx)
     }
 
+    fn send_raw_tx_bytes(&self, tx: &[u8]) -> Box<dyn Future<Item = String, Error = String> + Send> {
+        utxo_common::send_raw_tx_bytes(&self.utxo_arc, tx)
+    }
+
     fn wait_for_confirmations(
         &self,
         tx: &[u8],

mm2src/coins/utxo/slp.rs

@@ -1096,6 +1096,19 @@ impl MarketCoinOps for SlpToken {
         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 tx = deserialize(tx.as_slice())
+                .map_to_mm(|e| e)
+                .map_err(|e| format!("{:?}", e))?;
+            let hash = coin.broadcast_tx(&tx).await.map_err(|e| format!("{:?}", e))?;
+            Ok(format!("{:?}", hash))
+        };
+        Box::new(fut.boxed().compat())
+    }
+
     fn wait_for_confirmations(
         &self,
         tx: &[u8],

mm2src/coins/utxo/utxo_common.rs

@@ -1706,6 +1706,15 @@ pub fn send_raw_tx(coin: &UtxoCoinFields, tx: &str) -> Box<dyn Future<Item = Str
     )
 }
 
+pub fn send_raw_tx_bytes(coin: &UtxoCoinFields, tx: &[u8]) -> Box<dyn Future<Item = String, Error = String> + Send> {
+    Box::new(
+        coin.rpc_client
+            .send_raw_transaction(tx.into())
+            .map_err(|e| ERRL!("{}", e))
+            .map(|hash| format!("{:?}", hash)),
+    )
+}
+
 pub fn wait_for_confirmations(
     coin: &UtxoCoinFields,
     tx: &[u8],

mm2src/coins/utxo/utxo_standard.rs

@@ -453,6 +453,10 @@ impl MarketCoinOps for UtxoStandardCoin {
         utxo_common::send_raw_tx(&self.utxo_arc, tx)
     }
 
+    fn send_raw_tx_bytes(&self, tx: &[u8]) -> Box<dyn Future<Item = String, Error = String> + Send> {
+        utxo_common::send_raw_tx_bytes(&self.utxo_arc, tx)
+    }
+
     fn wait_for_confirmations(
         &self,
         tx: &[u8],

mm2src/coins/z_coin.rs

@@ -733,6 +733,10 @@ impl MarketCoinOps for ZCoin {
         utxo_common::send_raw_tx(self.as_ref(), tx)
     }
 
+    fn send_raw_tx_bytes(&self, tx: &[u8]) -> Box<dyn Future<Item = String, Error = String> + Send> {
+        utxo_common::send_raw_tx_bytes(self.as_ref(), tx)
+    }
+
     fn wait_for_confirmations(
         &self,
         tx: &[u8],

mm2src/lp_network.rs

@@ -1,3 +1,4 @@
+use coins::lp_coinfind;
 /******************************************************************************
  * Copyright © 2014-2019 The SuperNET Developers.                             *
  *                                                                            *
@@ -17,10 +18,10 @@
 //  marketmaker
 //
 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>,

mm2src/lp_swap/maker_swap.rs

@@ -11,6 +11,7 @@ use super::{broadcast_my_swap_status, broadcast_swap_message_every, check_other_
 use crate::mm2::lp_dispatcher::{DispatcherContext, LpEvents};
 use crate::mm2::lp_network::subscribe_to_topic;
 use crate::mm2::lp_ordermatch::{MakerOrderBuilder, OrderConfirmationsSettings};
+use crate::mm2::lp_swap::{broadcast_transaction_message, tx_helper_topic};
 use crate::mm2::MM_VERSION;
 use bigdecimal::BigDecimal;
 use bitcrypto::dhash160;
@@ -853,6 +854,13 @@ impl MakerSwap {
             },
         };
 
+        broadcast_transaction_message(
+            &self.ctx,
+            tx_helper_topic(self.taker_coin.ticker()),
+            transaction.tx_hex(),
+            &self.p2p_privkey,
+        );
+
         let tx_hash = transaction.tx_hash();
         log!({ "Taker payment spend tx {:02x}", tx_hash });
         let tx_ident = TransactionIdentifier {
@@ -925,6 +933,14 @@ impl MakerSwap {
                 ]))
             },
         };
+
+        broadcast_transaction_message(
+            &self.ctx,
+            tx_helper_topic(self.maker_coin.ticker()),
+            transaction.tx_hex(),
+            &self.p2p_privkey,
+        );
+
         let tx_hash = transaction.tx_hash();
         log!({ "Maker payment refund tx {:02x}", tx_hash });
         let tx_ident = TransactionIdentifier {

mm2src/lp_swap/taker_swap.rs

@@ -10,6 +10,7 @@ use super::{broadcast_my_swap_status, broadcast_swap_message_every, check_other_
             SwapConfirmationsSettings, SwapError, SwapMsg, SwapsContext, TransactionIdentifier, WAIT_CONFIRM_INTERVAL};
 use crate::mm2::lp_network::subscribe_to_topic;
 use crate::mm2::lp_ordermatch::{MatchBy, OrderConfirmationsSettings, TakerAction, TakerOrderBuilder};
+use crate::mm2::lp_swap::{broadcast_transaction_message, tx_helper_topic};
 use crate::mm2::MM_VERSION;
 use bigdecimal::BigDecimal;
 use coins::{lp_coinfind, CanRefundHtlc, FeeApproxStage, FoundSwapTxSpend, MmCoinEnum, TradeFee, TradePreimageValue,
@@ -1287,6 +1288,13 @@ impl TakerSwap {
             },
         };
 
+        broadcast_transaction_message(
+            &self.ctx,
+            tx_helper_topic(self.maker_coin.ticker()),
+            transaction.tx_hex(),
+            &self.p2p_privkey,
+        );
+
         let tx_hash = transaction.tx_hash();
         log!({"Maker payment spend tx {:02x}", tx_hash });
         let tx_ident = TransactionIdentifier {
@@ -1330,6 +1338,13 @@ impl TakerSwap {
             },
         };
 
+        broadcast_transaction_message(
+            &self.ctx,
+            tx_helper_topic(self.taker_coin.ticker()),
+            transaction.tx_hex(),
+            &self.p2p_privkey,
+        );
+
         let tx_hash = transaction.tx_hash();
         log!({"Taker refund tx hash {:02x}", tx_hash });
         let tx_ident = TransactionIdentifier {