diff --git a/Cargo.lock b/Cargo.lock index c7d9b6952f..1911e05d20 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1020,6 +1020,7 @@ dependencies = [ "ethkey", "futures 0.1.29", "futures 0.3.28", + "futures-util", "group 0.8.0", "gstuff", "hex 0.4.3", @@ -1039,6 +1040,7 @@ dependencies = [ "mm2_core", "mm2_db", "mm2_err_handle", + "mm2_event_stream", "mm2_git", "mm2_io", "mm2_metamask", @@ -1086,6 +1088,7 @@ dependencies = [ "tiny-bip39", "tokio", "tokio-rustls", + "tokio-tungstenite-wasm", "tonic", "tonic-build", "url", @@ -1122,6 +1125,7 @@ dependencies = [ "lightning-invoice", "mm2_core", "mm2_err_handle", + "mm2_event_stream", "mm2_metamask", "mm2_metrics", "mm2_number", @@ -4333,6 +4337,7 @@ dependencies = [ "async-trait", "cfg-if 1.0.0", "common", + "futures 0.3.28", "parking_lot 0.12.0", "serde", "tokio", @@ -4989,6 +4994,12 @@ version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "624a8340c38c1b80fd549087862da4ba43e08858af025b236e509b6649fc13d5" +[[package]] +name = "openssl-probe" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff011a302c396a5197692431fc1948019154afc178baf7d8e37367442a4601cf" + [[package]] name = "ordered-float" version = "3.7.0" @@ -6316,6 +6327,18 @@ dependencies = [ "webpki 0.22.0", ] +[[package]] +name = "rustls-native-certs" +version = "0.6.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a9aace74cb666635c918e9c12bc0d348266037aa8eb599b5cba565709a8dff00" +dependencies = [ + "openssl-probe", + "rustls-pemfile 1.0.2", + "schannel", + "security-framework", +] + [[package]] name = "rustls-pemfile" version = "0.2.1" @@ -6399,6 +6422,15 @@ dependencies = [ "syn 1.0.95", ] +[[package]] +name = "schannel" +version = "0.1.22" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c3733bf4cf7ea0880754e19cb5a462007c4a8c1914bff372ccc95b464f1df88" +dependencies = [ + "windows-sys 0.48.0", +] + [[package]] name = "scoped-tls" version = "1.0.0" @@ -6511,6 +6543,29 @@ dependencies = [ "zeroize", ] +[[package]] +name = "security-framework" +version = "2.9.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "05b64fb303737d99b81884b2c63433e9ae28abebe5eb5045dcdd175dc2ecf4de" +dependencies = [ + "bitflags", + "core-foundation", + "core-foundation-sys", + "libc", + "security-framework-sys", +] + +[[package]] +name = "security-framework-sys" +version = "2.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e932934257d3b408ed8f30db49d85ea163bfe74961f017f405b025af298f0c7a" +dependencies = [ + "core-foundation-sys", + "libc", +] + [[package]] name = "semver" version = "0.9.0" @@ -8285,6 +8340,39 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-tungstenite" +version = "0.16.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e80b39df6afcc12cdf752398ade96a6b9e99c903dfdc36e53ad10b9c366bca72" +dependencies = [ + "futures-util", + "log", + "rustls 0.20.4", + "rustls-native-certs", + "tokio", + "tokio-rustls", + "tungstenite", + "webpki 0.22.0", +] + +[[package]] +name = "tokio-tungstenite-wasm" +version = "0.1.1-alpha.0" +source = "git+https://github.com/KomodoPlatform/tokio-tungstenite-wasm?rev=d20abdb#d20abdbbb2f03e302e3a8d11a1736ec8b50d0f58" +dependencies = [ + "futures-channel", + "futures-util", + "http 0.2.7", + "httparse", + "js-sys", + "thiserror", + "tokio", + "tokio-tungstenite", + "wasm-bindgen", + "web-sys", +] + [[package]] name = "tokio-util" version = "0.7.2" diff --git a/mm2src/adex_cli/Cargo.lock b/mm2src/adex_cli/Cargo.lock index ab1f4548d9..7d513ab8d2 100644 --- a/mm2src/adex_cli/Cargo.lock +++ b/mm2src/adex_cli/Cargo.lock @@ -1781,6 +1781,7 @@ dependencies = [ "async-trait", "cfg-if 1.0.0", "common", + "futures 0.3.28", "parking_lot", "serde", "tokio", diff --git a/mm2src/coins/Cargo.toml b/mm2src/coins/Cargo.toml index 1e45499ae6..b2ccc8c227 100644 --- a/mm2src/coins/Cargo.toml +++ b/mm2src/coins/Cargo.toml @@ -49,7 +49,9 @@ ethereum-types = { version = "0.13", default-features = false, features = ["std" ethkey = { git = "https://github.com/KomodoPlatform/mm2-parity-ethereum.git" } # Waiting for https://github.com/rust-lang/rust/issues/54725 to use on Stable. #enum_dispatch = "0.1" +tokio-tungstenite-wasm = { git = "https://github.com/KomodoPlatform/tokio-tungstenite-wasm", rev = "d20abdb", features = ["rustls-tls-native-roots"]} futures01 = { version = "0.1", package = "futures" } +futures-util = { version = "0.3", default-features = false, features = ["sink", "std"] } # using select macro requires the crate to be named futures, compilation failed with futures03 name futures = { version = "0.3", package = "futures", features = ["compat", "async-await"] } group = "0.8.0" @@ -63,6 +65,7 @@ lazy_static = "1.4" libc = "0.2" mm2_core = { path = "../mm2_core" } mm2_err_handle = { path = "../mm2_err_handle" } +mm2_event_stream = { path = "../mm2_event_stream" } mm2_git = { path = "../mm2_git" } mm2_io = { path = "../mm2_io" } mm2_metrics = { path = "../mm2_metrics" } diff --git a/mm2src/coins/tendermint/mod.rs b/mm2src/coins/tendermint/mod.rs index d480a4964e..60a4c61ec1 100644 --- a/mm2src/coins/tendermint/mod.rs +++ b/mm2src/coins/tendermint/mod.rs @@ -5,6 +5,7 @@ mod ibc; mod iris; mod rpc; +mod tendermint_balance_events; mod tendermint_coin; mod tendermint_token; pub mod tendermint_tx_history_v2; diff --git a/mm2src/coins/tendermint/rpc/tendermint_native_rpc.rs b/mm2src/coins/tendermint/rpc/tendermint_native_rpc.rs index dde181b3e3..4904a2ed30 100644 --- a/mm2src/coins/tendermint/rpc/tendermint_native_rpc.rs +++ b/mm2src/coins/tendermint/rpc/tendermint_native_rpc.rs @@ -309,6 +309,9 @@ impl HttpClient { }, }) } + + #[inline] + pub fn uri(&self) -> http::Uri { self.inner.uri() } } #[async_trait] @@ -481,6 +484,13 @@ mod sealed { HttpClient::Https(c) => c.perform(request).await, } } + + pub fn uri(&self) -> Uri { + match self { + HttpClient::Http(client) => client.uri.clone(), + HttpClient::Https(client) => client.uri.clone(), + } + } } async fn response_to_string(response: hyper::Response) -> Result { diff --git a/mm2src/coins/tendermint/rpc/tendermint_wasm_rpc.rs b/mm2src/coins/tendermint/rpc/tendermint_wasm_rpc.rs index 036815a25f..bcbc07c874 100644 --- a/mm2src/coins/tendermint/rpc/tendermint_wasm_rpc.rs +++ b/mm2src/coins/tendermint/rpc/tendermint_wasm_rpc.rs @@ -58,6 +58,9 @@ impl HttpClient { Ok(HttpClient { uri: url.to_owned() }) } + #[inline] + pub fn uri(&self) -> http::Uri { Uri::from_str(&self.uri).expect("This should never happen.") } + pub(crate) async fn perform(&self, request: R) -> Result where R: SimpleRequest, diff --git a/mm2src/coins/tendermint/tendermint_balance_events.rs b/mm2src/coins/tendermint/tendermint_balance_events.rs new file mode 100644 index 0000000000..122262eb51 --- /dev/null +++ b/mm2src/coins/tendermint/tendermint_balance_events.rs @@ -0,0 +1,180 @@ +use async_trait::async_trait; +use common::{executor::{AbortSettings, SpawnAbortable}, + http_uri_to_ws_address, log}; +use futures::channel::oneshot::{self, Receiver, Sender}; +use futures_util::{SinkExt, StreamExt}; +use jsonrpc_core::MethodCall; +use jsonrpc_core::{Id as RpcId, Params as RpcParams, Value as RpcValue, Version as RpcVersion}; +use mm2_core::mm_ctx::MmArc; +use mm2_event_stream::{behaviour::{EventBehaviour, EventInitStatus}, + Event, EventStreamConfiguration}; +use mm2_number::BigDecimal; +use std::collections::{HashMap, HashSet}; + +use super::TendermintCoin; +use crate::{tendermint::TendermintCommons, utxo::utxo_common::big_decimal_from_sat_unsigned, MarketCoinOps, MmCoin}; + +#[async_trait] +impl EventBehaviour for TendermintCoin { + const EVENT_NAME: &'static str = "COIN_BALANCE"; + + async fn handle(self, _interval: f64, tx: oneshot::Sender) { + fn generate_subscription_query(query_filter: String) -> String { + let mut params = serde_json::Map::with_capacity(1); + params.insert("query".to_owned(), RpcValue::String(query_filter)); + + let q = MethodCall { + id: RpcId::Num(0), + jsonrpc: Some(RpcVersion::V2), + method: "subscribe".to_owned(), + params: RpcParams::Map(params), + }; + + serde_json::to_string(&q).expect("This should never happen") + } + + let ctx = match MmArc::from_weak(&self.ctx) { + Some(ctx) => ctx, + None => { + let msg = "MM context must have been initialized already."; + tx.send(EventInitStatus::Failed(msg.to_owned())) + .expect("Receiver is dropped, which should never happen."); + panic!("{}", msg); + }, + }; + + let account_id = self.account_id.to_string(); + let mut current_balances: HashMap = HashMap::new(); + + let receiver_q = generate_subscription_query(format!("coin_received.receiver = '{}'", account_id)); + let receiver_q = tokio_tungstenite_wasm::Message::Text(receiver_q); + + let spender_q = generate_subscription_query(format!("coin_spent.spender = '{}'", account_id)); + let spender_q = tokio_tungstenite_wasm::Message::Text(spender_q); + + tx.send(EventInitStatus::Success) + .expect("Receiver is dropped, which should never happen."); + + loop { + let node_uri = match self.rpc_client().await { + Ok(client) => client.uri(), + Err(e) => { + log::error!("{e}"); + continue; + }, + }; + + let socket_address = format!("{}/{}", http_uri_to_ws_address(node_uri), "websocket"); + + let mut wsocket = match tokio_tungstenite_wasm::connect(socket_address).await { + Ok(ws) => ws, + Err(e) => { + log::error!("{e}"); + continue; + }, + }; + + // Filter received TX events + if let Err(e) = wsocket.send(receiver_q.clone()).await { + log::error!("{e}"); + continue; + } + + // Filter spent TX events + if let Err(e) = wsocket.send(spender_q.clone()).await { + log::error!("{e}"); + continue; + } + + while let Some(message) = wsocket.next().await { + let msg = match message { + Ok(tokio_tungstenite_wasm::Message::Text(data)) => data.clone(), + Ok(tokio_tungstenite_wasm::Message::Close(_)) => break, + Err(err) => { + log::error!("Server returned an unknown message type - {err}"); + break; + }, + _ => continue, + }; + + // Here, we receive raw data from the socket. + // To examine this data, you can use tools like wscat/websocat or visit + // https://pastebin.pl/view/499cbf2c for sample data. + if let Ok(json_val) = serde_json::from_str::(&msg) { + let transfers: Vec = + serde_json::from_value(json_val["result"]["events"]["transfer.amount"].clone()) + .unwrap_or_default(); + + let denoms: HashSet = transfers + .iter() + .map(|t| { + let amount: String = t.chars().take_while(|c| c.is_numeric()).collect(); + let denom = &t[amount.len()..]; + denom.to_owned() + }) + .collect(); + + 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 { + Ok(balance_denom) => balance_denom, + Err(e) => { + log::error!("{e}"); + continue; + }, + }; + + let balance_decimal = big_decimal_from_sat_unsigned(balance_denom, decimals); + + // Only broadcast when balance is changed + let mut broadcast = false; + if let Some(balance) = current_balances.get_mut(&ticker) { + if *balance != balance_decimal { + *balance = balance_decimal.clone(); + broadcast = true; + } + } else { + current_balances.insert(ticker.clone(), balance_decimal.clone()); + broadcast = true; + } + + if broadcast { + let payload = 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; + } + } + } + } + } + } + } + + async fn spawn_if_active(self, config: &EventStreamConfiguration) -> EventInitStatus { + if let Some(event) = config.get_event(Self::EVENT_NAME) { + log::info!( + "{} event is activated for {}. `stream_interval_seconds`({}) has no effect on this.", + Self::EVENT_NAME, + self.ticker(), + event.stream_interval_seconds + ); + + let (tx, rx): (Sender, Receiver) = oneshot::channel(); + let fut = self.clone().handle(event.stream_interval_seconds, tx); + let settings = + AbortSettings::info_on_abort(format!("{} event is stopped for {}.", Self::EVENT_NAME, self.ticker())); + self.spawner().spawn_with_settings(fut, settings); + + rx.await.unwrap_or_else(|e| { + EventInitStatus::Failed(format!("Event initialization status must be received: {}", e)) + }) + } else { + EventInitStatus::Inactive + } + } +} diff --git a/mm2src/coins/tendermint/tendermint_coin.rs b/mm2src/coins/tendermint/tendermint_coin.rs index cc569cb1d4..6d0432d703 100644 --- a/mm2src/coins/tendermint/tendermint_coin.rs +++ b/mm2src/coins/tendermint/tendermint_coin.rs @@ -60,7 +60,7 @@ use futures01::Future; use hex::FromHexError; use itertools::Itertools; use keys::KeyPair; -use mm2_core::mm_ctx::MmArc; +use mm2_core::mm_ctx::{MmArc, MmWeak}; use mm2_err_handle::prelude::*; use mm2_git::{FileMetadata, GitController, GithubClient, RepositoryOperations, GITHUB_API_URI}; use mm2_number::MmNumber; @@ -142,7 +142,7 @@ pub struct TendermintProtocolInfo { #[derive(Clone)] pub struct ActivatedTokenInfo { pub(crate) decimals: u8, - pub(crate) denom: Denom, + pub ticker: String, } pub struct TendermintConf { @@ -237,6 +237,7 @@ pub struct TendermintCoinImpl { pub(crate) history_sync_state: Mutex, client: TendermintRpcClient, chain_registry_name: Option, + pub(crate) ctx: MmWeak, } #[derive(Clone)] @@ -279,6 +280,7 @@ pub enum TendermintInitErrorKind { AvgBlockTimeMissing, #[display(fmt = "avg_blocktime must be in-between '0' and '255'.")] AvgBlockTimeInvalid, + BalanceStreamInitError(String), } #[derive(Display, Debug)] @@ -441,14 +443,14 @@ impl TendermintCommons for TendermintCoin { let ibc_assets_info = self.tokens_info.lock().clone(); let mut requests = Vec::new(); - for (ticker, info) in ibc_assets_info { + for (denom, info) in ibc_assets_info { let fut = async move { let balance_denom = self - .account_balance_for_denom(&self.account_id, info.denom.to_string()) + .account_balance_for_denom(&self.account_id, denom) .await .map_err(|e| e.into_inner())?; let balance_decimal = big_decimal_from_sat_unsigned(balance_denom, info.decimals); - Ok::<_, TendermintCoinRpcError>((ticker.clone(), balance_decimal)) + Ok::<_, TendermintCoinRpcError>((info.ticker, balance_decimal)) }; requests.push(fut); } @@ -544,6 +546,7 @@ impl TendermintCoin { history_sync_state: Mutex::new(history_sync_state), client: TendermintRpcClient(AsyncMutex::new(client_impl)), chain_registry_name: protocol_info.chain_registry_name, + ctx: ctx.weak(), }))) } @@ -1178,7 +1181,7 @@ impl TendermintCoin { pub fn add_activated_token_info(&self, ticker: String, decimals: u8, denom: Denom) { self.tokens_info .lock() - .insert(ticker, ActivatedTokenInfo { decimals, denom }); + .insert(denom.to_string(), ActivatedTokenInfo { decimals, ticker }); } fn estimate_blocks_from_duration(&self, duration: u64) -> i64 { @@ -1821,6 +1824,20 @@ impl TendermintCoin { _ => (self.gas_price(), fallback_gas_limit), } } + + pub(crate) fn active_ticker_and_decimals_from_denom(&self, denom: &str) -> Option<(String, u8)> { + if self.denom.as_ref() == denom { + return Some((self.ticker.clone(), self.decimals)); + } + + let tokens = self.tokens_info.lock(); + + if let Some(token_info) = tokens.get(denom) { + return Some((token_info.ticker.to_owned(), token_info.decimals)); + } + + None + } } fn clients_from_urls(rpc_urls: &[String]) -> MmResult, TendermintInitErrorKind> { diff --git a/mm2src/coins_activation/Cargo.toml b/mm2src/coins_activation/Cargo.toml index e6ae6401a0..09fd4adf8e 100644 --- a/mm2src/coins_activation/Cargo.toml +++ b/mm2src/coins_activation/Cargo.toml @@ -21,6 +21,7 @@ futures = { version = "0.3", package = "futures", features = ["compat", "async-a hex = "0.4.2" mm2_core = { path = "../mm2_core" } mm2_err_handle = { path = "../mm2_err_handle" } +mm2_event_stream = { path = "../mm2_event_stream" } mm2_metrics = { path = "../mm2_metrics" } mm2_number = { path = "../mm2_number" } parking_lot = { version = "0.12.0", features = ["nightly"] } diff --git a/mm2src/coins_activation/src/bch_with_tokens_activation.rs b/mm2src/coins_activation/src/bch_with_tokens_activation.rs index a6d4c54df8..b99b49235a 100644 --- a/mm2src/coins_activation/src/bch_with_tokens_activation.rs +++ b/mm2src/coins_activation/src/bch_with_tokens_activation.rs @@ -17,6 +17,7 @@ use common::{drop_mutability, true_f}; use crypto::CryptoCtxError; use mm2_core::mm_ctx::MmArc; use mm2_err_handle::prelude::*; +use mm2_event_stream::EventStreamConfiguration; use mm2_number::BigDecimal; use serde_derive::{Deserialize, Serialize}; use serde_json::Value as Json; @@ -330,4 +331,11 @@ impl PlatformWithTokensActivationOps for BchCoin { let settings = AbortSettings::info_on_abort(format!("bch_and_slp_history_loop stopped for {}", self.ticker())); self.spawner().spawn_with_settings(fut, settings); } + + async fn handle_balance_streaming( + &self, + _config: &EventStreamConfiguration, + ) -> Result<(), MmError> { + Ok(()) + } } diff --git a/mm2src/coins_activation/src/eth_with_token_activation.rs b/mm2src/coins_activation/src/eth_with_token_activation.rs index 0f7c8d8455..3a93f3ad07 100644 --- a/mm2src/coins_activation/src/eth_with_token_activation.rs +++ b/mm2src/coins_activation/src/eth_with_token_activation.rs @@ -15,6 +15,7 @@ use common::Future01CompatExt; use common::{drop_mutability, true_f}; use mm2_core::mm_ctx::MmArc; use mm2_err_handle::prelude::*; +use mm2_event_stream::EventStreamConfiguration; #[cfg(target_arch = "wasm32")] use mm2_metamask::MetamaskRpcError; use mm2_number::BigDecimal; @@ -277,6 +278,13 @@ impl PlatformWithTokensActivationOps for EthCoin { _initial_balance: Option, ) { } + + async fn handle_balance_streaming( + &self, + _config: &EventStreamConfiguration, + ) -> Result<(), MmError> { + Ok(()) + } } fn eth_priv_key_build_policy( diff --git a/mm2src/coins_activation/src/platform_coin_with_tokens.rs b/mm2src/coins_activation/src/platform_coin_with_tokens.rs index 94b9a16fc9..bd12c99a22 100644 --- a/mm2src/coins_activation/src/platform_coin_with_tokens.rs +++ b/mm2src/coins_activation/src/platform_coin_with_tokens.rs @@ -8,6 +8,7 @@ use crypto::CryptoCtxError; use derive_more::Display; use mm2_core::mm_ctx::MmArc; use mm2_err_handle::prelude::*; +use mm2_event_stream::EventStreamConfiguration; use mm2_number::BigDecimal; use ser_error_derive::SerializeErrorType; use serde_derive::{Deserialize, Serialize}; @@ -165,6 +166,11 @@ pub trait PlatformWithTokensActivationOps: Into { storage: impl TxHistoryStorage, initial_balance: Option, ); + + async fn handle_balance_streaming( + &self, + config: &EventStreamConfiguration, + ) -> Result<(), MmError>; } #[derive(Debug, Deserialize)] @@ -364,6 +370,10 @@ where ); } + if let Some(config) = &ctx.event_stream_configuration { + platform_coin.handle_balance_streaming(config).await?; + } + let coins_ctx = CoinsContext::from_ctx(&ctx).unwrap(); coins_ctx .add_platform_with_tokens(platform_coin.into(), mm_tokens) diff --git a/mm2src/coins_activation/src/solana_with_tokens_activation.rs b/mm2src/coins_activation/src/solana_with_tokens_activation.rs index b6bd7b123d..aa049d0867 100644 --- a/mm2src/coins_activation/src/solana_with_tokens_activation.rs +++ b/mm2src/coins_activation/src/solana_with_tokens_activation.rs @@ -18,6 +18,7 @@ use crypto::CryptoCtxError; use futures::future::try_join_all; use mm2_core::mm_ctx::MmArc; use mm2_err_handle::prelude::*; +use mm2_event_stream::EventStreamConfiguration; use mm2_number::BigDecimal; use serde_derive::{Deserialize, Serialize}; use serde_json::Value as Json; @@ -288,4 +289,11 @@ impl PlatformWithTokensActivationOps for SolanaCoin { _initial_balance: Option, ) { } + + async fn handle_balance_streaming( + &self, + _config: &EventStreamConfiguration, + ) -> Result<(), MmError> { + Ok(()) + } } diff --git a/mm2src/coins_activation/src/tendermint_with_assets_activation.rs b/mm2src/coins_activation/src/tendermint_with_assets_activation.rs index 006f7993d3..e2e8fda8f0 100644 --- a/mm2src/coins_activation/src/tendermint_with_assets_activation.rs +++ b/mm2src/coins_activation/src/tendermint_with_assets_activation.rs @@ -15,6 +15,8 @@ use common::{true_f, Future01CompatExt}; use crypto::StandardHDCoinAddress; use mm2_core::mm_ctx::MmArc; use mm2_err_handle::prelude::*; +use mm2_event_stream::behaviour::{EventBehaviour, EventInitStatus}; +use mm2_event_stream::EventStreamConfiguration; use mm2_number::BigDecimal; use serde::{Deserialize, Serialize}; use serde_json::Value as Json; @@ -231,7 +233,14 @@ impl PlatformWithTokensActivationOps for TendermintCoin { current_block, balance: None, tokens_balances: None, - tokens_tickers: Some(self.tokens_info.lock().clone().into_keys().collect()), + tokens_tickers: Some( + self.tokens_info + .lock() + .clone() + .into_values() + .map(|t| t.ticker) + .collect(), + ), }); } @@ -275,4 +284,17 @@ impl PlatformWithTokensActivationOps for TendermintCoin { let settings = AbortSettings::info_on_abort(format!("tendermint_history_loop stopped for {}", self.ticker())); self.spawner().spawn_with_settings(fut, settings); } + + async fn handle_balance_streaming( + &self, + config: &EventStreamConfiguration, + ) -> Result<(), MmError> { + if let EventInitStatus::Failed(err) = EventBehaviour::spawn_if_active(self.clone(), config).await { + return MmError::err(TendermintInitError { + ticker: self.ticker().to_owned(), + kind: TendermintInitErrorKind::BalanceStreamInitError(err), + }); + } + Ok(()) + } } diff --git a/mm2src/common/common.rs b/mm2src/common/common.rs index 058661f42c..6af8724e64 100644 --- a/mm2src/common/common.rs +++ b/mm2src/common/common.rs @@ -1035,3 +1035,40 @@ pub fn parse_rfc3339_to_timestamp(date_str: &str) -> Result bool { old_version == 0 && new_version == 1 } + +/// Takes `http:Uri` and converts it into `String` of websocket address +/// +/// Panics if the given URI doesn't contain a host value. +pub fn http_uri_to_ws_address(uri: http::Uri) -> String { + let address_prefix = match uri.scheme_str() { + Some("https") => "wss://", + _ => "ws://", + }; + + let host_address = uri.host().expect("Host can't be empty."); + let port = uri.port_u16().map(|p| format!(":{}", p)).unwrap_or_default(); + + format!("{}{}{}", address_prefix, host_address, port) +} + +#[test] +fn test_http_uri_to_ws_address() { + let uri = "https://cosmos-rpc.polkachu.com".parse::().unwrap(); + let ws_connection = http_uri_to_ws_address(uri); + assert_eq!(ws_connection, "wss://cosmos-rpc.polkachu.com"); + + let uri = "http://cosmos-rpc.polkachu.com".parse::().unwrap(); + let ws_connection = http_uri_to_ws_address(uri); + assert_eq!(ws_connection, "ws://cosmos-rpc.polkachu.com"); + + let uri = "http://34.82.96.8:26657".parse::().unwrap(); + let ws_connection = http_uri_to_ws_address(uri); + assert_eq!(ws_connection, "ws://34.82.96.8:26657"); +} + +#[test] +#[should_panic(expected = "Host can't be empty.")] +fn test_http_uri_to_ws_address_panic() { + let uri = "/demo/value".parse::().unwrap(); + http_uri_to_ws_address(uri); +} diff --git a/mm2src/mm2_event_stream/Cargo.toml b/mm2src/mm2_event_stream/Cargo.toml index 2865e0a01f..adf20e7ee2 100644 --- a/mm2src/mm2_event_stream/Cargo.toml +++ b/mm2src/mm2_event_stream/Cargo.toml @@ -7,6 +7,7 @@ edition = "2021" async-trait = "0.1" cfg-if = "1.0" common = { path = "../common" } +futures = { version = "0.3", default-features = false } parking_lot = "0.12" serde = { version = "1", features = ["derive", "rc"] } tokio = { version = "1", features = ["sync"] } diff --git a/mm2src/mm2_event_stream/src/behaviour.rs b/mm2src/mm2_event_stream/src/behaviour.rs index bb905af3fc..8539754061 100644 --- a/mm2src/mm2_event_stream/src/behaviour.rs +++ b/mm2src/mm2_event_stream/src/behaviour.rs @@ -1,5 +1,13 @@ use crate::EventStreamConfiguration; use async_trait::async_trait; +use futures::channel::oneshot; + +#[derive(Clone, Debug)] +pub enum EventInitStatus { + Inactive, + Success, + Failed(String), +} #[async_trait] pub trait EventBehaviour { @@ -7,9 +15,9 @@ pub trait EventBehaviour { const EVENT_NAME: &'static str; /// Event handler that is responsible for broadcasting event data to the streaming channels. - async fn handle(self, interval: f64); + async fn handle(self, interval: f64, tx: oneshot::Sender); /// Spawns the `Self::handle` in a separate thread if the event is active according to the mm2 configuration. /// Does nothing if the event is not active. - fn spawn_if_active(self, config: &EventStreamConfiguration); + async fn spawn_if_active(self, config: &EventStreamConfiguration) -> EventInitStatus; } diff --git a/mm2src/mm2_event_stream/src/lib.rs b/mm2src/mm2_event_stream/src/lib.rs index afa5c7e1be..cc3b86f7d5 100644 --- a/mm2src/mm2_event_stream/src/lib.rs +++ b/mm2src/mm2_event_stream/src/lib.rs @@ -40,9 +40,12 @@ pub struct EventStreamConfiguration { #[derive(Clone, Default, Deserialize)] pub struct EventConfig { /// The interval in seconds at which the event should be streamed. + #[serde(default = "default_stream_interval")] pub stream_interval_seconds: f64, } +const fn default_stream_interval() -> f64 { 5. } + impl Default for EventStreamConfiguration { fn default() -> Self { Self { diff --git a/mm2src/mm2_libp2p/src/atomicdex_behaviour.rs b/mm2src/mm2_libp2p/src/atomicdex_behaviour.rs index 8b1930c9e0..482251fb93 100644 --- a/mm2src/mm2_libp2p/src/atomicdex_behaviour.rs +++ b/mm2src/mm2_libp2p/src/atomicdex_behaviour.rs @@ -26,7 +26,7 @@ use libp2p_floodsub::{Floodsub, FloodsubEvent, Topic as FloodsubTopic}; use log::{debug, error, info}; use rand::seq::SliceRandom; use rand::Rng; -use std::{collections::hash_map::{DefaultHasher, HashMap}, +use std::{collections::{hash_map::DefaultHasher, BTreeMap}, hash::{Hash, Hasher}, iter, net::IpAddr, @@ -47,7 +47,7 @@ const ANNOUNCE_INITIAL_DELAY: Duration = Duration::from_secs(60); const CHANNEL_BUF_SIZE: usize = 1024 * 8; /// Returns info about connected peers -pub async fn get_peers_info(mut cmd_tx: AdexCmdTx) -> HashMap> { +pub async fn get_peers_info(mut cmd_tx: AdexCmdTx) -> BTreeMap> { let (result_tx, rx) = oneshot::channel(); let cmd = AdexBehaviourCmd::GetPeersInfo { result_tx }; cmd_tx.send(cmd).await.expect("Rx should be present"); @@ -55,21 +55,21 @@ pub async fn get_peers_info(mut cmd_tx: AdexCmdTx) -> HashMap HashMap> { +pub async fn get_gossip_mesh(mut cmd_tx: AdexCmdTx) -> BTreeMap> { let (result_tx, rx) = oneshot::channel(); let cmd = AdexBehaviourCmd::GetGossipMesh { result_tx }; cmd_tx.send(cmd).await.expect("Rx should be present"); rx.await.expect("Tx should be present") } -pub async fn get_gossip_peer_topics(mut cmd_tx: AdexCmdTx) -> HashMap> { +pub async fn get_gossip_peer_topics(mut cmd_tx: AdexCmdTx) -> BTreeMap> { let (result_tx, rx) = oneshot::channel(); let cmd = AdexBehaviourCmd::GetGossipPeerTopics { result_tx }; cmd_tx.send(cmd).await.expect("Rx should be present"); rx.await.expect("Tx should be present") } -pub async fn get_gossip_topic_peers(mut cmd_tx: AdexCmdTx) -> HashMap> { +pub async fn get_gossip_topic_peers(mut cmd_tx: AdexCmdTx) -> BTreeMap> { let (result_tx, rx) = oneshot::channel(); let cmd = AdexBehaviourCmd::GetGossipTopicPeers { result_tx }; cmd_tx.send(cmd).await.expect("Rx should be present"); @@ -133,16 +133,16 @@ pub enum AdexBehaviourCmd { response_channel: AdexResponseChannel, }, GetPeersInfo { - result_tx: oneshot::Sender>>, + result_tx: oneshot::Sender>>, }, GetGossipMesh { - result_tx: oneshot::Sender>>, + result_tx: oneshot::Sender>>, }, GetGossipPeerTopics { - result_tx: oneshot::Sender>>, + result_tx: oneshot::Sender>>, }, GetGossipTopicPeers { - result_tx: oneshot::Sender>>, + result_tx: oneshot::Sender>>, }, GetRelayMesh { result_tx: oneshot::Sender>, diff --git a/mm2src/mm2_main/src/lp_native_dex.rs b/mm2src/mm2_main/src/lp_native_dex.rs index db0600c633..cfe63525da 100644 --- a/mm2src/mm2_main/src/lp_native_dex.rs +++ b/mm2src/mm2_main/src/lp_native_dex.rs @@ -28,9 +28,11 @@ use enum_from::EnumFromTrait; use mm2_core::mm_ctx::{MmArc, MmCtx}; use mm2_err_handle::common_errors::InternalError; use mm2_err_handle::prelude::*; +use mm2_event_stream::behaviour::{EventBehaviour, EventInitStatus}; use mm2_libp2p::{spawn_gossipsub, AdexBehaviourError, NodeType, RelayAddress, RelayAddressError, SwarmRuntime, WssCerts}; use mm2_metrics::mm_gauge; +use mm2_net::network_event::NetworkEvent; use mm2_net::p2p::P2PContext; use rpc_task::RpcTaskError; use serde_json::{self as json}; @@ -52,17 +54,19 @@ use crate::mm2::rpc::spawn_rpc; cfg_native! { use db_common::sqlite::rusqlite::Error as SqlError; - use mm2_event_stream::behaviour::EventBehaviour; use mm2_io::fs::{ensure_dir_is_writable, ensure_file_is_writable}; use mm2_net::ip_addr::myipaddr; - use mm2_net::network_event::NetworkEvent; } #[path = "lp_init/init_context.rs"] mod init_context; #[path = "lp_init/init_hw.rs"] pub mod init_hw; -#[cfg(target_arch = "wasm32")] -#[path = "lp_init/init_metamask.rs"] -pub mod init_metamask; + +cfg_wasm32! { + use mm2_net::wasm_event_stream::handle_worker_stream; + + #[path = "lp_init/init_metamask.rs"] + pub mod init_metamask; +} const NETID_8762_SEEDNODES: [&str; 3] = [ "streamseed1.komodo.earth", @@ -161,6 +165,8 @@ pub enum MmInitError { EmptyPassphrase, #[display(fmt = "Invalid passphrase: {}", _0)] InvalidPassphrase(String), + #[display(fmt = "NETWORK event initialization failed: {}", _0)] + NetworkEventInitFailed(String), #[from_trait(WithHwRpcError::hw_rpc_error)] #[display(fmt = "{}", _0)] HwError(HwRpcError), @@ -384,12 +390,21 @@ fn migrate_db(ctx: &MmArc) -> MmInitResult<()> { #[cfg(not(target_arch = "wasm32"))] fn migration_1(_ctx: &MmArc) {} -#[cfg(not(target_arch = "wasm32"))] -fn init_event_streaming(ctx: &MmArc) { +async fn init_event_streaming(ctx: &MmArc) -> MmInitResult<()> { // This condition only executed if events were enabled in mm2 configuration. if let Some(config) = &ctx.event_stream_configuration { - // Network event handling - NetworkEvent::new(ctx.clone()).spawn_if_active(config); + if let EventInitStatus::Failed(err) = NetworkEvent::new(ctx.clone()).spawn_if_active(config).await { + return MmError::err(MmInitError::NetworkEventInitFailed(err)); + } + } + + Ok(()) +} + +#[cfg(target_arch = "wasm32")] +fn init_wasm_event_streaming(ctx: &MmArc) { + if ctx.event_stream_configuration.is_some() { + ctx.spawner().spawn(handle_worker_stream(ctx.clone())); } } @@ -423,13 +438,15 @@ pub async fn lp_init_continue(ctx: MmArc) -> MmInitResult<()> { // an order and start new swap that might get started 2 times because of kick-start kick_start(ctx.clone()).await?; - #[cfg(not(target_arch = "wasm32"))] - init_event_streaming(&ctx); + init_event_streaming(&ctx).await?; ctx.spawner().spawn(lp_ordermatch_loop(ctx.clone())); ctx.spawner().spawn(broadcast_maker_orders_keep_alive_loop(ctx.clone())); + #[cfg(target_arch = "wasm32")] + init_wasm_event_streaming(&ctx); + ctx.spawner().spawn(clean_memory_loop(ctx.weak())); Ok(()) diff --git a/mm2src/mm2_net/Cargo.toml b/mm2src/mm2_net/Cargo.toml index f9b67767e5..9c42cd3d38 100644 --- a/mm2src/mm2_net/Cargo.toml +++ b/mm2src/mm2_net/Cargo.toml @@ -37,7 +37,7 @@ mm2_state_machine = { path = "../mm2_state_machine"} wasm-bindgen = "0.2.86" wasm-bindgen-test = { version = "0.3.2" } wasm-bindgen-futures = "0.4.21" -web-sys = { version = "0.3.55", features = ["console", "CloseEvent", "DomException", "ErrorEvent", "IdbDatabase", "IdbCursor", "IdbCursorWithValue", "IdbFactory", "IdbIndex", "IdbIndexParameters", "IdbObjectStore", "IdbObjectStoreParameters", "IdbOpenDbRequest", "IdbKeyRange", "IdbTransaction", "IdbTransactionMode", "IdbVersionChangeEvent", "MessageEvent", "WebSocket"] } +web-sys = { version = "0.3.55", features = ["console", "CloseEvent", "DomException", "ErrorEvent", "IdbDatabase", "IdbCursor", "IdbCursorWithValue", "IdbFactory", "IdbIndex", "IdbIndexParameters", "IdbObjectStore", "IdbObjectStoreParameters", "IdbOpenDbRequest", "IdbKeyRange", "IdbTransaction", "IdbTransactionMode", "IdbVersionChangeEvent", "MessageEvent", "WebSocket", "Worker"] } js-sys = "0.3.27" [target.'cfg(not(target_arch = "wasm32"))'.dependencies] diff --git a/mm2src/mm2_net/src/lib.rs b/mm2src/mm2_net/src/lib.rs index 379f0f928a..edd13738b9 100644 --- a/mm2src/mm2_net/src/lib.rs +++ b/mm2src/mm2_net/src/lib.rs @@ -1,4 +1,5 @@ pub mod grpc_web; +#[cfg(feature = "event-stream")] pub mod network_event; #[cfg(feature = "p2p")] pub mod p2p; pub mod transport; @@ -6,8 +7,8 @@ pub mod transport; #[cfg(not(target_arch = "wasm32"))] pub mod native_http; #[cfg(not(target_arch = "wasm32"))] pub mod native_tls; #[cfg(all(feature = "event-stream", not(target_arch = "wasm32")))] -pub mod network_event; -#[cfg(all(feature = "event-stream", not(target_arch = "wasm32")))] pub mod sse_handler; +#[cfg(all(feature = "event-stream", target_arch = "wasm32"))] +pub mod wasm_event_stream; #[cfg(target_arch = "wasm32")] pub mod wasm_http; #[cfg(target_arch = "wasm32")] pub mod wasm_ws; diff --git a/mm2src/mm2_net/src/network_event.rs b/mm2src/mm2_net/src/network_event.rs index 35ce5de40d..beee72e36f 100644 --- a/mm2src/mm2_net/src/network_event.rs +++ b/mm2src/mm2_net/src/network_event.rs @@ -2,9 +2,10 @@ use crate::p2p::P2PContext; use async_trait::async_trait; use common::{executor::{SpawnFuture, Timer}, log::info}; +use futures::channel::oneshot::{self, Receiver, Sender}; use mm2_core::mm_ctx::MmArc; pub use mm2_event_stream::behaviour::EventBehaviour; -use mm2_event_stream::{Event, EventStreamConfiguration}; +use mm2_event_stream::{behaviour::EventInitStatus, Event, EventStreamConfiguration}; use mm2_libp2p::behaviours::atomicdex; use serde_json::json; @@ -20,8 +21,11 @@ impl NetworkEvent { impl EventBehaviour for NetworkEvent { const EVENT_NAME: &'static str = "NETWORK"; - async fn handle(self, interval: f64) { + async fn handle(self, interval: f64, tx: oneshot::Sender) { let p2p_ctx = P2PContext::fetch_from_mm_arc(&self.ctx); + let mut previously_sent = json!({}); + + tx.send(EventInitStatus::Success).unwrap(); loop { let p2p_cmd_tx = p2p_ctx.cmd_tx.lock().clone(); @@ -40,22 +44,34 @@ impl EventBehaviour for NetworkEvent { "relay_mesh": relay_mesh, }); - self.ctx - .stream_channel_controller - .broadcast(Event::new(Self::EVENT_NAME.to_string(), event_data.to_string())) - .await; + if previously_sent != event_data { + self.ctx + .stream_channel_controller + .broadcast(Event::new(Self::EVENT_NAME.to_string(), event_data.to_string())) + .await; + + previously_sent = event_data; + } Timer::sleep(interval).await; } } - fn spawn_if_active(self, config: &EventStreamConfiguration) { + async fn spawn_if_active(self, config: &EventStreamConfiguration) -> EventInitStatus { if let Some(event) = config.get_event(Self::EVENT_NAME) { info!( "NETWORK event is activated with {} seconds interval.", event.stream_interval_seconds ); - self.ctx.spawner().spawn(self.handle(event.stream_interval_seconds)); + + let (tx, rx): (Sender, Receiver) = oneshot::channel(); + self.ctx.spawner().spawn(self.handle(event.stream_interval_seconds, tx)); + + rx.await.unwrap_or_else(|e| { + EventInitStatus::Failed(format!("Event initialization status must be received: {}", e)) + }) + } else { + EventInitStatus::Inactive } } } diff --git a/mm2src/mm2_net/src/wasm_event_stream.rs b/mm2src/mm2_net/src/wasm_event_stream.rs new file mode 100644 index 0000000000..1c6f8182fc --- /dev/null +++ b/mm2src/mm2_net/src/wasm_event_stream.rs @@ -0,0 +1,26 @@ +use mm2_core::mm_ctx::MmArc; +use serde_json::json; + +/// Handles broadcasted messages from `mm2_event_stream` continuously for WASM. +pub async fn handle_worker_stream(ctx: MmArc) { + let config = ctx + .event_stream_configuration + .as_ref() + .expect("Event stream configuration couldn't be found. This should never happen."); + + let mut channel_controller = ctx.stream_channel_controller.clone(); + let mut rx = channel_controller.create_channel(config.total_active_events()); + + while let Some(event) = rx.recv().await { + let data = json!({ + "_type": event.event_type(), + "message": event.message(), + }); + + let worker = web_sys::Worker::new("worker.js").expect("Missing worker.js"); + let message_js = wasm_bindgen::JsValue::from_str(&data.to_string()); + + worker.post_message(&message_js) + .expect("Incompatible browser!\nSee https://developer.mozilla.org/en-US/docs/Web/API/Worker/postMessage#browser_compatibility for details."); + } +}