kabu_defi_market/
logs_parser.rs

1use alloy_network::Network;
2use alloy_primitives::Log;
3use alloy_provider::Provider;
4use eyre::Result;
5use std::collections::HashMap;
6use tokio::sync::broadcast;
7use tracing::error;
8
9use kabu_types_events::LoomTask;
10use kabu_types_market::PoolLoaders;
11
12pub async fn process_log_entries<P, N>(
13    log_entries: Vec<Log>,
14    pool_loaders: &PoolLoaders<P, N>,
15    tasks_tx: broadcast::Sender<LoomTask>,
16) -> Result<()>
17where
18    N: Network,
19    P: Provider<N> + Send + Sync + Clone + 'static,
20{
21    let mut pool_to_fetch = Vec::new();
22    let mut processed_pools = HashMap::new();
23
24    for log_entry in log_entries.into_iter() {
25        if let Some((pool_id, pool_class)) = pool_loaders.determine_pool_class(&log_entry) {
26            // was this pool already processed?
27            if processed_pools.insert(log_entry.address, true).is_some() {
28                continue;
29            }
30
31            pool_to_fetch.push((pool_id, pool_class));
32        }
33    }
34
35    if let Err(e) = tasks_tx.send(LoomTask::FetchAndAddPools(pool_to_fetch)) {
36        error!("Failed to send task: {}", e);
37    }
38    Ok(())
39}
kabu_defi_market/logs_parser.rs

kabu_defi_market/
logs_parser.rs
