feat(channels): wasm, coin balance and socket impl

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" }

mm2src/coins/coin_balance_event.rs

@@ -0,0 +1,124 @@
+use crate::{CoinsContext, MmCoin, MmCoinEnum};
+use async_trait::async_trait;
+use common::{executor::{SpawnFuture, Timer},
+             log::info};
+use mm2_core::mm_ctx::MmArc;
+use mm2_event_stream::{behaviour::EventBehaviour, EventStreamConfiguration};
+use std::sync::atomic::Ordering;
+
+/// Event tag for broadcasting balance events
+pub(crate) const COIN_BALANCE_EVENT_TAG: &str = "COIN_BALANCE";
+
+pub struct CoinBalanceEvent {
+    ctx: MmArc,
+}
+
+impl CoinBalanceEvent {
+    pub fn new(ctx: MmArc) -> Self { Self { ctx } }
+}
+
+#[async_trait]
+impl EventBehaviour for CoinBalanceEvent {
+    const EVENT_NAME: &'static str = COIN_BALANCE_EVENT_TAG;
+
+    async fn handle(self, _interval: f64) {
+        let cctx = CoinsContext::from_ctx(&self.ctx).expect("Unexpected internal panic.");
+
+        // Events that are already fired
+        let mut event_pool: Vec<String> = vec![];
+
+        loop {
+            let coins_mutex = cctx.coins.lock().await;
+
+            let coins: Vec<MmCoinEnum> = coins_mutex
+                .values()
+                .filter_map(|coin| {
+                    // We loop this over and over, so it's not necessary to sequentially load the atomics all over
+                    // the threads, since the cost of it is way too higher than the `AtomicOrdering::Relaxed`
+                    if coin.is_available.load(Ordering::Relaxed) {
+                        Some(coin.inner.clone())
+                    } else {
+                        None
+                    }
+                })
+                .collect();
+
+            // Similar to above, we don't need to held the lock(which will block all other processes that depends
+            // on this lock(like coin activation)) since we loop this over continuously.
+            drop(coins_mutex);
+
+            // Handle balance streaming concurrently for each coin
+            for coin in coins {
+                let ticker = coin.ticker().to_owned();
+
+                if event_pool.contains(&ticker) {
+                    continue;
+                }
+
+                match coin {
+                    MmCoinEnum::UtxoCoin(inner) => {
+                        self.ctx.spawner().spawn(inner.handle_balance_stream(self.ctx.clone()))
+                    },
+                    MmCoinEnum::QtumCoin(inner) => {
+                        self.ctx.spawner().spawn(inner.handle_balance_stream(self.ctx.clone()))
+                    },
+                    MmCoinEnum::Qrc20Coin(inner) => {
+                        self.ctx.spawner().spawn(inner.handle_balance_stream(self.ctx.clone()))
+                    },
+                    MmCoinEnum::EthCoin(inner) => {
+                        self.ctx.spawner().spawn(inner.handle_balance_stream(self.ctx.clone()))
+                    },
+                    MmCoinEnum::ZCoin(inner) => self.ctx.spawner().spawn(inner.handle_balance_stream(self.ctx.clone())),
+                    MmCoinEnum::Bch(inner) => self.ctx.spawner().spawn(inner.handle_balance_stream(self.ctx.clone())),
+                    MmCoinEnum::SlpToken(inner) => {
+                        self.ctx.spawner().spawn(inner.handle_balance_stream(self.ctx.clone()))
+                    },
+                    MmCoinEnum::Tendermint(inner) => {
+                        self.ctx.spawner().spawn(inner.handle_balance_stream(self.ctx.clone()))
+                    },
+                    MmCoinEnum::TendermintToken(inner) => {
+                        self.ctx.spawner().spawn(inner.handle_balance_stream(self.ctx.clone()))
+                    },
+                    #[cfg(not(target_arch = "wasm32"))]
+                    MmCoinEnum::LightningCoin(inner) => {
+                        self.ctx.spawner().spawn(inner.handle_balance_stream(self.ctx.clone()))
+                    },
+                    MmCoinEnum::Test(inner) => self.ctx.spawner().spawn(inner.handle_balance_stream(self.ctx.clone())),
+                    #[cfg(all(
+                        feature = "enable-solana",
+                        not(target_os = "ios"),
+                        not(target_os = "android"),
+                        not(target_arch = "wasm32")
+                    ))]
+                    MmCoinEnum::SolanaCoin(inner) => {
+                        self.ctx.spawner().spawn(inner.handle_balance_stream(self.ctx.clone()))
+                    },
+                    #[cfg(all(
+                        feature = "enable-solana",
+                        not(target_os = "ios"),
+                        not(target_os = "android"),
+                        not(target_arch = "wasm32")
+                    ))]
+                    MmCoinEnum::SplToken(inner) => {
+                        self.ctx.spawner().spawn(inner.handle_balance_stream(self.ctx.clone()))
+                    },
+                }
+
+                event_pool.push(ticker);
+            }
+
+            Timer::sleep(5.).await;
+        }
+    }
+
+    fn spawn_if_active(self, config: &EventStreamConfiguration) {
+        if let Some(event) = config.get_event(Self::EVENT_NAME) {
+            info!(
+                "{} event is activated. `stream_interval_seconds`({}) has no effect for this event.",
+                Self::EVENT_NAME,
+                event.stream_interval_seconds
+            );
+            self.ctx.spawner().spawn(self.handle(event.stream_interval_seconds));
+        }
+    }
+}

mm2src/coins/eth.rs

@@ -4743,6 +4743,8 @@ impl MmCoin for EthCoin {
             tokens.remove(ticker);
         };
     }
+
+    async fn handle_balance_stream(self, _ctx: MmArc) { todo!() }
 }
 
 pub trait TryToAddress {

mm2src/coins/lightning.rs

@@ -1452,4 +1452,6 @@ impl MmCoin for LightningCoin {
     fn on_disabled(&self) -> Result<(), AbortedError> { AbortableSystem::abort_all(&self.platform.abortable_system) }
 
     fn on_token_deactivated(&self, _ticker: &str) {}
+
+    async fn handle_balance_stream(self, _ctx: MmArc) { todo!() }
 }

mm2src/coins/lp_coins.rs

@@ -250,6 +250,8 @@ pub use test_coin::TestCoin;
 
 pub mod tx_history_storage;
 
+pub mod coin_balance_event;
+
 #[doc(hidden)]
 #[allow(unused_variables)]
 #[cfg(all(
@@ -2576,6 +2578,9 @@ pub trait MmCoin:
 
     /// For Handling the removal/deactivation of token on platform coin deactivation.
     fn on_token_deactivated(&self, ticker: &str);
+
+    // Handler for coin balance streaming continuously to the thread/stream channels
+    async fn handle_balance_stream(self, ctx: MmArc);
 }
 
 /// The coin futures spawner. It's used to spawn futures that can be aborted immediately or after a timeout

mm2src/coins/qrc20.rs

@@ -1486,6 +1486,8 @@ impl MmCoin for Qrc20Coin {
     fn on_disabled(&self) -> Result<(), AbortedError> { AbortableSystem::abort_all(&self.as_ref().abortable_system) }
 
     fn on_token_deactivated(&self, _ticker: &str) {}
+
+    async fn handle_balance_stream(self, _ctx: MmArc) { todo!() }
 }
 
 pub fn qrc20_swap_id(time_lock: u32, secret_hash: &[u8]) -> Vec<u8> {

mm2src/coins/solana.rs

@@ -776,4 +776,6 @@ impl MmCoin for SolanaCoin {
     fn on_disabled(&self) -> Result<(), AbortedError> { AbortableSystem::abort_all(&self.abortable_system) }
 
     fn on_token_deactivated(&self, _ticker: &str) {}
+
+    async fn handle_balance_stream(self, _ctx: MmArc) { todo!() }
 }

mm2src/coins/solana/spl.rs

@@ -570,4 +570,6 @@ impl MmCoin for SplToken {
     fn on_disabled(&self) -> Result<(), AbortedError> { self.conf.abortable_system.abort_all() }
 
     fn on_token_deactivated(&self, _ticker: &str) {}
+
+    async fn handle_balance_stream(self, _ctx: MmArc) { todo!() }
 }

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<hyper::Body>) -> Result<String, Error> {

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<R>(&self, request: R) -> Result<R::Response, PerformError>
     where
         R: SimpleRequest,