[WIP] concurrent balance loop controller

Signed-off-by: onur-ozkan <work@onurozkan.dev>

mm2src/coins/Cargo.toml

@@ -63,6 +63,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/events.rs

@@ -0,0 +1,7 @@
+use crate::MmCoin;
+use async_trait::async_trait;
+
+#[async_trait]
+pub trait BalanceEvent: MmCoin {
+    async fn stream(&self);
+}

mm2src/coins/lp_coins.rs

@@ -45,6 +45,7 @@ use async_trait::async_trait;
 use base58::FromBase58Error;
 use bip32::ExtendedPrivateKey;
 use common::custom_futures::timeout::TimeoutError;
+use common::executor::Timer;
 use common::executor::{abortable_queue::{AbortableQueue, WeakSpawner},
                        AbortSettings, AbortedError, SpawnAbortable, SpawnFuture};
 use common::log::{warn, LogOnError};
@@ -250,6 +251,8 @@ pub use test_coin::TestCoin;
 
 pub mod tx_history_storage;
 
+pub mod events;
+
 #[doc(hidden)]
 #[allow(unused_variables)]
 #[cfg(all(
@@ -3411,6 +3414,68 @@ fn lp_spawn_tx_history(ctx: MmArc, coin: MmCoinEnum) -> Result<(), String> {
     Ok(())
 }
 
+/// Concurrent balance event controller loop for active coins
+pub async fn balance_event_loop(ctx: MmArc) {
+    let cctx = CoinsContext::from_ctx(&ctx).unwrap();
+
+    // 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| {
+                if coin.is_available.load(AtomicOrdering::Relaxed) && coin.inner.is_platform_coin() {
+                    Some(coin.inner.clone())
+                } else {
+                    None
+                }
+            })
+            .collect();
+
+        drop(coins_mutex);
+
+        for coin in coins {
+            let ticker = coin.ticker().to_owned();
+
+            if event_pool.contains(&ticker) {
+                continue;
+            }
+
+            match coin {
+                MmCoinEnum::Tendermint(_) => {
+                    println!("COIN IS HERE {:?}", ticker);
+                },
+                MmCoinEnum::TendermintToken(_) => {
+                    unimplemented!();
+                },
+                MmCoinEnum::UtxoCoin(_) => todo!(),
+                MmCoinEnum::QtumCoin(_) => todo!(),
+                MmCoinEnum::Qrc20Coin(_) => todo!(),
+                MmCoinEnum::EthCoin(_) => todo!(),
+                MmCoinEnum::ZCoin(_) => todo!(),
+                MmCoinEnum::Bch(_) => todo!(),
+                MmCoinEnum::SlpToken(_) => todo!(),
+                MmCoinEnum::LightningCoin(_) => todo!(),
+                MmCoinEnum::Test(_) => todo!(),
+                #[cfg(all(
+                    feature = "enable-solana",
+                    not(target_os = "ios"),
+                    not(target_os = "android"),
+                    not(target_arch = "wasm32")
+                ))]
+                MmCoinEnum::SolanaCoin(_) | MmCoinEnum::SplToken(_) => todo!(),
+            }
+
+            event_pool.push(ticker);
+        }
+
+        Timer::sleep(5.).await;
+    }
+}
+
 /// NB: Returns only the enabled (aka active) coins.
 pub async fn lp_coinfind(ctx: &MmArc, ticker: &str) -> Result<Option<MmCoinEnum>, String> {
     let cctx = try_s!(CoinsContext::from_ctx(ctx));

mm2src/coins/tendermint/balance_event.rs

@@ -0,0 +1,9 @@
+use async_trait::async_trait;
+
+use super::TendermintCoin;
+use crate::events::BalanceEvent;
+
+#[async_trait]
+impl BalanceEvent for TendermintCoin {
+    async fn stream(&self) {}
+}

mm2src/coins/tendermint/mod.rs

@@ -2,6 +2,7 @@
 // Useful resources
 // https://docs.cosmos.network/
 
+pub mod balance_event;
 mod ibc;
 mod iris;
 mod rpc;

mm2src/mm2_main/src/lp_native_dex.rs

@@ -19,6 +19,7 @@
 //
 
 use bitcrypto::sha256;
+use coins::balance_event_loop;
 use coins::register_balance_update_handler;
 use common::executor::{SpawnFuture, Timer};
 use common::log::{info, warn};
@@ -382,10 +383,13 @@ fn migration_1(_ctx: &MmArc) {}
 #[cfg(not(target_arch = "wasm32"))]
 fn init_event_streaming(ctx: &MmArc) {
     // 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);
     }
+
+    ctx.spawner().spawn(balance_event_loop(ctx.clone()));
 }
 
 pub async fn lp_init_continue(ctx: MmArc) -> MmInitResult<()> {