diff --git a/src/streaming/event_parser/factory.rs b/src/streaming/event_parser/factory.rs index b7830d5..cf6d918 100755 --- a/src/streaming/event_parser/factory.rs +++ b/src/streaming/event_parser/factory.rs @@ -1,6 +1,6 @@ use anyhow::{anyhow, Result}; use solana_sdk::pubkey::Pubkey; -use std::sync::Arc; +use std::{collections::HashMap, sync::{Arc, LazyLock}}; use crate::streaming::event_parser::protocols::{ bonk::parser::BONK_PROGRAM_ID, pumpfun::parser::PUMPFUN_PROGRAM_ID, @@ -63,19 +63,27 @@ impl std::str::FromStr for Protocol { } } +static EVENT_PARSERS: LazyLock>> = LazyLock::new(|| { + let mut parsers: HashMap> = HashMap::new(); + parsers.insert(Protocol::PumpSwap, Arc::new(PumpSwapEventParser::new())); + parsers.insert(Protocol::PumpFun, Arc::new(PumpFunEventParser::new())); + parsers.insert(Protocol::Bonk, Arc::new(BonkEventParser::new())); + parsers.insert(Protocol::RaydiumCpmm, Arc::new(RaydiumCpmmEventParser::new())); + parsers.insert(Protocol::RaydiumClmm, Arc::new(RaydiumClmmEventParser::new())); + parsers +}); + + /// 事件解析器工厂 - 用于创建不同协议的事件解析器 pub struct EventParserFactory; impl EventParserFactory { + /// 创建指定协议的事件解析器 pub fn create_parser(protocol: Protocol) -> Arc { - match protocol { - Protocol::PumpSwap => Arc::new(PumpSwapEventParser::new()), - Protocol::PumpFun => Arc::new(PumpFunEventParser::new()), - Protocol::Bonk => Arc::new(BonkEventParser::new()), - Protocol::RaydiumCpmm => Arc::new(RaydiumCpmmEventParser::new()), - Protocol::RaydiumClmm => Arc::new(RaydiumClmmEventParser::new()), - } + EVENT_PARSERS.get(&protocol).cloned().unwrap_or_else(|| { + panic!("Parser for protocol {} not found", protocol); + }) } /// 创建所有协议的事件解析器 diff --git a/src/streaming/yellowstone_grpc.rs b/src/streaming/yellowstone_grpc.rs index 403699d..5a44da0 100755 --- a/src/streaming/yellowstone_grpc.rs +++ b/src/streaming/yellowstone_grpc.rs @@ -261,7 +261,6 @@ impl YellowstoneGrpc { Ok(()) } - /// 处理事件交易 async fn process_event_transaction( transaction_pretty: TransactionPretty, callback: &F, @@ -271,24 +270,39 @@ impl YellowstoneGrpc { where F: Fn(Box) + Send + Sync, { + // let start_time = std::time::Instant::now(); let slot = transaction_pretty.slot; let signature = transaction_pretty.signature.to_string(); + let mut futures = Vec::new(); for protocol in protocols { let parser = EventParserFactory::create_parser(protocol); - let events = parser - .parse_transaction( - transaction_pretty.tx.clone(), - &signature, + let tx_clone = transaction_pretty.tx.clone(); + let signature_clone = signature.clone(); + let bot_wallet_clone = bot_wallet.clone(); + + futures.push(tokio::spawn(async move { + parser.parse_transaction( + tx_clone, + &signature_clone, Some(slot), transaction_pretty.block_time, - bot_wallet.clone(), + bot_wallet_clone, ) .await - .unwrap_or_else(|_e| vec![]); - for event in events { - callback(event); + .unwrap_or_else(|_e| vec![]) + })); + } + + let results = futures::future::join_all(futures).await; + for result in results { + if let Ok(events) = result { + for event in events { + callback(event); + } } } + // let elapsed = start_time.elapsed(); + // println!("处理交易耗时: {:?}", elapsed); Ok(()) }