feat(UTXO): balance event streaming for Electrum clients

mm2src/coins/tendermint/tendermint_balance_events.rs

@@ -114,6 +114,7 @@ impl EventBehaviour for TendermintCoin {
                         })
                         .collect();
 
+                    let mut balance_updates = vec![];
                     for denom in denoms {
                         if let Some((ticker, decimals)) = self.active_ticker_and_decimals_from_denom(&denom) {
                             let balance_denom = match self.account_balance_for_denom(&self.account_id, denom).await {
@@ -139,17 +140,22 @@ impl EventBehaviour for TendermintCoin {
                             }
 
                             if broadcast {
-                                let payload = json!({
+                                balance_updates.push(json!({
                                     "ticker": ticker,
                                     "balance": { "spendable": balance_decimal, "unspendable": BigDecimal::default() }
-                                });
-
-                                ctx.stream_channel_controller
-                                    .broadcast(Event::new(Self::EVENT_NAME.to_string(), payload.to_string()))
-                                    .await;
+                                }));
                             }
                         }
                     }
+
+                    if !balance_updates.is_empty() {
+                        ctx.stream_channel_controller
+                            .broadcast(Event::new(
+                                Self::EVENT_NAME.to_string(),
+                                json!(balance_updates).to_string(),
+                            ))
+                            .await;
+                    }
                 }
             }
         }

mm2src/coins/utxo.rs

@@ -32,6 +32,7 @@ pub mod rpc_clients;
 pub mod slp;
 pub mod spv;
 pub mod swap_proto_v2_scripts;
+pub mod utxo_balance_events;
 pub mod utxo_block_header_storage;
 pub mod utxo_builder;
 pub mod utxo_common;
@@ -65,7 +66,7 @@ pub use keys::{Address, AddressFormat as UtxoAddressFormat, AddressHashEnum, Key
                Type as ScriptType};
 #[cfg(not(target_arch = "wasm32"))]
 use lightning_invoice::Currency as LightningCurrency;
-use mm2_core::mm_ctx::MmArc;
+use mm2_core::mm_ctx::{MmArc, MmWeak};
 use mm2_err_handle::prelude::*;
 use mm2_metrics::MetricsArc;
 use mm2_number::BigDecimal;
@@ -141,6 +142,9 @@ pub type HistoryUtxoTxMap = HashMap<H256Json, HistoryUtxoTx>;
 pub type MatureUnspentMap = HashMap<Address, MatureUnspentList>;
 pub type RecentlySpentOutPointsGuard<'a> = AsyncMutexGuard<'a, RecentlySpentOutPoints>;
 pub type UtxoHDAddress = HDAddress<Address, Public>;
+pub type ScripthashNotificationSender = Option<Arc<AsyncMutex<AsyncSender<()>>>>;
+
+type ScripthashNotificationReceiver = Option<Arc<AsyncMutex<AsyncReceiver<()>>>>;
 
 #[cfg(windows)]
 #[cfg(not(target_arch = "wasm32"))]
@@ -610,6 +614,11 @@ pub struct UtxoCoinFields {
     /// This abortable system is used to spawn coin's related futures that should be aborted on coin deactivation
     /// and on [`MmArc::stop`].
     pub abortable_system: AbortableQueue,
+    pub(crate) ctx: MmWeak,
+    /// This is used for balance event streaming implementation for UTXOs.
+    /// If balance event streaming isn't enabled, this value will always be `None`; otherwise,
+    /// it will be used for receiving scripthash notifications to re-fetch balances.
+    scripthash_notification_receiver: ScripthashNotificationReceiver,
 }
 
 #[derive(Debug, Display)]

mm2src/coins/utxo/rpc_clients.rs

@@ -13,7 +13,7 @@ use common::executor::{abortable_queue, abortable_queue::AbortableQueue, Abortab
 use common::jsonrpc_client::{JsonRpcBatchClient, JsonRpcBatchResponse, JsonRpcClient, JsonRpcError, JsonRpcErrorType,
                              JsonRpcId, JsonRpcMultiClient, JsonRpcRemoteAddr, JsonRpcRequest, JsonRpcRequestEnum,
                              JsonRpcResponse, JsonRpcResponseEnum, JsonRpcResponseFut, RpcRes};
-use common::log::LogOnError;
+use common::log::{debug, LogOnError};
 use common::log::{error, info, warn};
 use common::{median, now_float, now_ms, now_sec, OrdRange};
 use derive_more::Display;
@@ -52,6 +52,8 @@ use std::sync::atomic::{AtomicU64, Ordering as AtomicOrdering};
 use std::sync::Arc;
 use std::time::Duration;
 
+use super::ScripthashNotificationSender;
+
 cfg_native! {
     use futures::future::Either;
     use futures::io::Error;
@@ -113,6 +115,15 @@ pub enum UtxoRpcClientEnum {
     Electrum(ElectrumClient),
 }
 
+impl ToString for UtxoRpcClientEnum {
+    fn to_string(&self) -> String {
+        match self {
+            UtxoRpcClientEnum::Native(_) => "native".to_owned(),
+            UtxoRpcClientEnum::Electrum(_) => "electrum".to_owned(),
+        }
+    }
+}
+
 impl From<ElectrumClient> for UtxoRpcClientEnum {
     fn from(client: ElectrumClient) -> UtxoRpcClientEnum { UtxoRpcClientEnum::Electrum(client) }
 }
@@ -345,6 +356,8 @@ pub trait UtxoRpcClientOps: fmt::Debug + Send + Sync + 'static {
     /// Submits the raw `tx` transaction (serialized, hex-encoded) to blockchain network.
     fn send_raw_transaction(&self, tx: BytesJson) -> UtxoRpcFut<H256Json>;
 
+    fn blockchain_scripthash_subscribe(&self, scripthash: String) -> UtxoRpcFut<Json>;
+
     /// Returns raw transaction (serialized, hex-encoded) by the given `txid`.
     fn get_transaction_bytes(&self, txid: &H256Json) -> UtxoRpcFut<BytesJson>;
 
@@ -701,12 +714,12 @@ impl JsonRpcClient for NativeClientImpl {
             .body(Vec::from(request_body))
             .map_err(|e| JsonRpcErrorType::InvalidRequest(e.to_string())));
 
-        let event_handles = self.event_handlers.clone();
+        let event_handlers = self.event_handlers.clone();
         Box::new(slurp_req(http_request).boxed().compat().then(
             move |result| -> Result<(JsonRpcRemoteAddr, JsonRpcResponseEnum), JsonRpcErrorType> {
                 let res = result.map_err(|e| e.into_inner())?;
                 // measure now only body length, because the `hyper` crate doesn't allow to get total HTTP packet length
-                event_handles.on_incoming_response(&res.2);
+                event_handlers.on_incoming_response(&res.2);
 
                 let body =
                     std::str::from_utf8(&res.2).map_err(|e| JsonRpcErrorType::parse_error(&uri, e.to_string()))?;
@@ -806,6 +819,13 @@ impl UtxoRpcClientOps for NativeClient {
         Box::new(rpc_func!(self, "sendrawtransaction", tx).map_to_mm_fut(UtxoRpcError::from))
     }
 
+    fn blockchain_scripthash_subscribe(&self, _scripthash: String) -> UtxoRpcFut<Json> {
+        Box::new(futures01::future::err(
+            UtxoRpcError::Internal("blockchain_scripthash_subscribe` is not supported for Native Clients".to_owned())
+                .into(),
+        ))
+    }
+
     fn get_transaction_bytes(&self, txid: &H256Json) -> UtxoRpcFut<BytesJson> {
         Box::new(self.get_raw_transaction_bytes(txid).map_to_mm_fut(UtxoRpcError::from))
     }
@@ -1438,6 +1458,7 @@ fn addr_to_socket_addr(input: &str) -> Result<SocketAddr, String> {
 pub fn spawn_electrum(
     req: &ElectrumRpcRequest,
     event_handlers: Vec<RpcTransportEventHandlerShared>,
+    scripthash_notification_sender: &ScripthashNotificationSender,
     abortable_system: AbortableQueue,
 ) -> Result<ElectrumConnection, String> {
     let config = match req.protocol {
@@ -1465,6 +1486,7 @@ pub fn spawn_electrum(
         req.url.clone(),
         config,
         event_handlers,
+        scripthash_notification_sender,
         abortable_system,
     ))
 }
@@ -1475,6 +1497,7 @@ pub fn spawn_electrum(
 pub fn spawn_electrum(
     req: &ElectrumRpcRequest,
     event_handlers: Vec<RpcTransportEventHandlerShared>,
+    scripthash_notification_sender: &ScripthashNotificationSender,
     abortable_system: AbortableQueue,
 ) -> Result<ElectrumConnection, String> {
     let mut url = req.url.clone();
@@ -1503,7 +1526,13 @@ pub fn spawn_electrum(
         },
     };
 
-    Ok(electrum_connect(url, config, event_handlers, abortable_system))
+    Ok(electrum_connect(
+        url,
+        config,
+        event_handlers,
+        scripthash_notification_sender,
+        abortable_system,
+    ))
 }
 
 /// Represents the active Electrum connection to selected address
@@ -1611,6 +1640,10 @@ pub struct ElectrumClientImpl {
     /// Please also note that this abortable system is a subsystem of [`UtxoCoinFields::abortable_system`].
     abortable_system: AbortableQueue,
     negotiate_version: bool,
+    /// This is used for balance event streaming implementation for UTXOs.
+    /// If balance event streaming isn't enabled, this value will always be `None`; otherwise,
+    /// it will be used for sending scripthash notifications to trigger re-fetching the balances.
+    scripthash_notification_sender: ScripthashNotificationSender,
 }
 
 async fn electrum_request_multi(
@@ -1700,7 +1733,12 @@ impl ElectrumClientImpl {
     /// Create an Electrum connection and spawn a green thread actor to handle it.
     pub async fn add_server(&self, req: &ElectrumRpcRequest) -> Result<(), String> {
         let subsystem = try_s!(self.abortable_system.create_subsystem());
-        let connection = try_s!(spawn_electrum(req, self.event_handlers.clone(), subsystem));
+        let connection = try_s!(spawn_electrum(
+            req,
+            self.event_handlers.clone(),
+            &self.scripthash_notification_sender,
+            subsystem,
+        ));
         self.connections.lock().await.push(connection);
         Ok(())
     }
@@ -1787,6 +1825,8 @@ impl Deref for ElectrumClient {
 
 const BLOCKCHAIN_HEADERS_SUB_ID: &str = "blockchain.headers.subscribe";
 
+const BLOCKCHAIN_SCRIPTHASH_SUB_ID: &str = "blockchain.scripthash.subscribe";
+
 impl UtxoJsonRpcClientInfo for ElectrumClient {
     fn coin_name(&self) -> &str { self.coin_ticker.as_str() }
 }
@@ -2236,6 +2276,10 @@ impl UtxoRpcClientOps for ElectrumClient {
         )
     }
 
+    fn blockchain_scripthash_subscribe(&self, scripthash: String) -> UtxoRpcFut<Json> {
+        Box::new(rpc_func!(self, BLOCKCHAIN_SCRIPTHASH_SUB_ID, scripthash).map_to_mm_fut(UtxoRpcError::from))
+    }
+
     /// https://electrumx.readthedocs.io/en/latest/protocol-methods.html#blockchain-transaction-get
     /// returns transaction bytes by default
     fn get_transaction_bytes(&self, txid: &H256Json) -> UtxoRpcFut<BytesJson> {
@@ -2396,6 +2440,7 @@ impl ElectrumClientImpl {
         block_headers_storage: BlockHeaderStorage,
         abortable_system: AbortableQueue,
         negotiate_version: bool,
+        scripthash_notification_sender: ScripthashNotificationSender,
     ) -> ElectrumClientImpl {
         let protocol_version = OrdRange::new(1.2, 1.4).unwrap();
         ElectrumClientImpl {
@@ -2409,6 +2454,7 @@ impl ElectrumClientImpl {
             block_headers_storage,
             abortable_system,
             negotiate_version,
+            scripthash_notification_sender,
         }
     }
 
@@ -2419,6 +2465,7 @@ impl ElectrumClientImpl {
         protocol_version: OrdRange<f32>,
         block_headers_storage: BlockHeaderStorage,
         abortable_system: AbortableQueue,
+        scripthash_notification_sender: ScripthashNotificationSender,
     ) -> ElectrumClientImpl {
         ElectrumClientImpl {
             protocol_version,
@@ -2428,6 +2475,7 @@ impl ElectrumClientImpl {
                 block_headers_storage,
                 abortable_system,
                 false,
+                scripthash_notification_sender,
             )
         }
     }
@@ -2438,17 +2486,25 @@ fn rx_to_stream(rx: mpsc::Receiver<Vec<u8>>) -> impl Stream<Item = Vec<u8>, Erro
     rx.map_err(|_| panic!("errors not possible on rx"))
 }
 
-async fn electrum_process_json(raw_json: Json, arc: &JsonRpcPendingRequestsShared) {
+async fn electrum_process_json(
+    raw_json: Json,
+    arc: &JsonRpcPendingRequestsShared,
+    scripthash_notification_sender: &ScripthashNotificationSender,
+) {
     // detect if we got standard JSONRPC response or subscription response as JSONRPC request
     #[derive(Deserialize)]
     #[serde(untagged)]
     enum ElectrumRpcResponseEnum {
+        /// The subscription response as JSONRPC request.
+        ///
+        /// NOTE Because JsonRpcResponse uses default values for each of its field,
+        /// this variant has to stay at top in this enumeration to be properly deserialized
+        /// from serde.
+        SubscriptionNotification(JsonRpcRequest),
         /// The standard JSONRPC single response.
         SingleResponse(JsonRpcResponse),
         /// The batch of standard JSONRPC responses.
         BatchResponses(JsonRpcBatchResponse),
-        /// The subscription response as JSONRPC request.
-        SubscriptionNotification(JsonRpcRequest),
     }
 
     let response: ElectrumRpcResponseEnum = match json::from_value(raw_json) {
@@ -2465,6 +2521,16 @@ async fn electrum_process_json(raw_json: Json, arc: &JsonRpcPendingRequestsShare
         ElectrumRpcResponseEnum::SubscriptionNotification(req) => {
             let id = match req.method.as_ref() {
                 BLOCKCHAIN_HEADERS_SUB_ID => BLOCKCHAIN_HEADERS_SUB_ID,
+                BLOCKCHAIN_SCRIPTHASH_SUB_ID => {
+                    if let Some(sender) = scripthash_notification_sender {
+                        debug!("Sending scripthash notification");
+                        if sender.lock().await.try_send(()).is_err() {
+                            error!("Failed sending scripthash notification");
+                            return;
+                        };
+                    };
+                    BLOCKCHAIN_SCRIPTHASH_SUB_ID
+                },
                 _ => {
                     error!("Couldn't get id of request {:?}", req);
                     return;
@@ -2487,7 +2553,11 @@ async fn electrum_process_json(raw_json: Json, arc: &JsonRpcPendingRequestsShare
     }
 }
 
-async fn electrum_process_chunk(chunk: &[u8], arc: &JsonRpcPendingRequestsShared) {
+async fn electrum_process_chunk(
+    chunk: &[u8],
+    arc: &JsonRpcPendingRequestsShared,
+    scripthash_notification_sender: &ScripthashNotificationSender,
+) {
     // we should split the received chunk because we can get several responses in 1 chunk.
     let split = chunk.split(|item| *item == b'\n');
     for chunk in split {
@@ -2500,7 +2570,7 @@ async fn electrum_process_chunk(chunk: &[u8], arc: &JsonRpcPendingRequestsShared
                     return;
                 },
             };
-            electrum_process_json(raw_json, arc).await
+            electrum_process_json(raw_json, arc, scripthash_notification_sender).await
         }
     }
 }
@@ -2629,6 +2699,7 @@ async fn connect_loop<Spawner: SpawnFuture>(
     responses: JsonRpcPendingRequestsShared,
     connection_tx: Arc<AsyncMutex<Option<mpsc::Sender<Vec<u8>>>>>,
     event_handlers: Vec<RpcTransportEventHandlerShared>,
+    scripthash_notification_sender: ScripthashNotificationSender,
     _spawner: Spawner,
 ) -> Result<(), ()> {
     let delay = Arc::new(AtomicU64::new(0));
@@ -2682,6 +2753,7 @@ async fn connect_loop<Spawner: SpawnFuture>(
             let delay = delay.clone();
             let addr = addr.clone();
             let responses = responses.clone();
+            let scripthash_notification_sender = scripthash_notification_sender.clone();
             let event_handlers = event_handlers.clone();
             async move {
                 let mut buffer = String::with_capacity(1024);
@@ -2705,7 +2777,7 @@ async fn connect_loop<Spawner: SpawnFuture>(
                     event_handlers.on_incoming_response(buffer.as_bytes());
                     last_chunk.store(now_ms(), AtomicOrdering::Relaxed);
 
-                    electrum_process_chunk(buffer.as_bytes(), &responses).await;
+                    electrum_process_chunk(buffer.as_bytes(), &responses, &scripthash_notification_sender).await;
                     buffer.clear();
                 }
             }
@@ -2749,6 +2821,7 @@ async fn connect_loop<Spawner: SpawnFuture>(
     responses: JsonRpcPendingRequestsShared,
     connection_tx: Arc<AsyncMutex<Option<mpsc::Sender<Vec<u8>>>>>,
     event_handlers: Vec<RpcTransportEventHandlerShared>,
+    scripthash_notification_sender: ScripthashNotificationSender,
     spawner: Spawner,
 ) -> Result<(), ()> {
     use std::sync::atomic::AtomicUsize;
@@ -2783,6 +2856,7 @@ async fn connect_loop<Spawner: SpawnFuture>(
             let delay = delay.clone();
             let addr = addr.clone();
             let responses = responses.clone();
+            let scripthash_notification_sender = scripthash_notification_sender.clone();
             let event_handlers = event_handlers.clone();
             async move {
                 while let Some(incoming_res) = transport_rx.next().await {
@@ -2795,7 +2869,7 @@ async fn connect_loop<Spawner: SpawnFuture>(
                             let incoming_str = incoming_json.to_string();
                             event_handlers.on_incoming_response(incoming_str.as_bytes());
 
-                            electrum_process_json(incoming_json, &responses).await;
+                            electrum_process_json(incoming_json, &responses, &scripthash_notification_sender).await;
                         },
                         Err(e) => {
                             error!("{} error: {:?}", addr, e);
@@ -2855,6 +2929,7 @@ fn electrum_connect(
     addr: String,
     config: ElectrumConfig,
     event_handlers: Vec<RpcTransportEventHandlerShared>,
+    scripthash_notification_sender: &ScripthashNotificationSender,
     abortable_system: AbortableQueue,
 ) -> ElectrumConnection {
     let responses = Arc::new(AsyncMutex::new(JsonRpcPendingRequests::default()));
@@ -2867,6 +2942,7 @@ fn electrum_connect(
         responses.clone(),
         tx.clone(),
         event_handlers,
+        scripthash_notification_sender.clone(),
         spawner.clone(),
     )
     .then(|_| futures::future::ready(()));