diff --git a/Cargo.toml b/Cargo.toml index 7a7f89d..0e61ef4 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -68,6 +68,7 @@ crossbeam = "0.8.4" crossbeam-queue = "0.3.12" parking_lot = "0.12.1" wide = "0.7" +spl-token = "8.0.0" [dev-dependencies] criterion = { version = "0.5", features = ["html_reports"] } \ No newline at end of file diff --git a/src/main.rs b/src/main.rs index 5c3fd40..c28031d 100755 --- a/src/main.rs +++ b/src/main.rs @@ -3,6 +3,7 @@ use solana_streamer_sdk::{ streaming::{ event_parser::{ common::{filter::EventTypeFilter, EventType}, + core::account_event_parser::CommonAccountEvent, protocols::{ bonk::{ parser::BONK_PROGRAM_ID, BonkGlobalConfigAccountEvent, BonkMigrateToAmmEvent, @@ -333,6 +334,9 @@ fn create_event_callback() -> impl Fn(Box) { RaydiumCpmmPoolStateAccountEvent => |e: RaydiumCpmmPoolStateAccountEvent| { println!("RaydiumCpmmPoolStateAccountEvent: {e:?}"); }, + CommonAccountEvent => |e: CommonAccountEvent| { + println!("CommonAccountEvent: {e:?}"); + }, }); } } diff --git a/src/streaming/event_parser/common/types.rs b/src/streaming/event_parser/common/types.rs index 888a088..b935557 100755 --- a/src/streaming/event_parser/common/types.rs +++ b/src/streaming/event_parser/common/types.rs @@ -141,6 +141,8 @@ pub enum EventType { AccountRaydiumCpmmAmmConfig, AccountRaydiumCpmmPoolState, + AccountCommon, + // Common events BlockMeta, Unknown, @@ -225,6 +227,7 @@ impl fmt::Display for EventType { } EventType::AccountRaydiumCpmmAmmConfig => write!(f, "AccountRaydiumCpmmAmmConfig"), EventType::AccountRaydiumCpmmPoolState => write!(f, "AccountRaydiumCpmmPoolState"), + EventType::AccountCommon => write!(f, "AccountCommon"), EventType::BlockMeta => write!(f, "BlockMeta"), EventType::Unknown => write!(f, "Unknown"), } diff --git a/src/streaming/event_parser/core/account_event_parser.rs b/src/streaming/event_parser/core/account_event_parser.rs index a9926c5..4ac5c9a 100644 --- a/src/streaming/event_parser/core/account_event_parser.rs +++ b/src/streaming/event_parser/core/account_event_parser.rs @@ -1,8 +1,12 @@ use std::collections::HashMap; use std::sync::OnceLock; +use serde::{Deserialize, Serialize}; +use solana_sdk::program_pack::Pack; use solana_sdk::pubkey::Pubkey; +use spl_token::state::Account; +use crate::impl_unified_event; use crate::streaming::common::SimdUtils; use crate::streaming::event_parser::common::filter::EventTypeFilter; use crate::streaming::event_parser::common::{EventMetadata, EventType, ProtocolType}; @@ -26,6 +30,19 @@ pub struct AccountEventParseConfig { pub account_parser: AccountEventParserFn, } +/// 通用账户事件 +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct CommonAccountEvent { + pub metadata: EventMetadata, + pub pubkey: Pubkey, + pub executable: bool, + pub lamports: u64, + pub owner: Pubkey, + pub rent_epoch: u64, + pub amount: Option, +} +impl_unified_event!(CommonAccountEvent,); + /// 账户事件解析器 pub type AccountEventParserFn = fn(account: &AccountPretty, metadata: EventMetadata) -> Option>; @@ -33,6 +50,9 @@ pub type AccountEventParserFn = static PROTOCOL_CONFIGS_CACHE: OnceLock>> = OnceLock::new(); +// 通用账户解析配置的静态缓存 +static COMMON_CONFIG: OnceLock = OnceLock::new(); + pub struct AccountEventParser {} impl AccountEventParser { @@ -148,21 +168,39 @@ impl AccountEventParser { map }); - let mut configs = vec![]; - let empty_vec = vec![]; + let mut configs = Vec::new(); + let empty_vec = Vec::new(); + + // 预估容量以减少重新分配 + let estimated_capacity = protocols.len() * 3; // 大多数协议有2-3个配置 + configs.reserve(estimated_capacity); + for protocol in protocols { let protocol_configs = protocols_map.get(protocol).unwrap_or(&empty_vec); - let filtered_configs: Vec = protocol_configs - .iter() - .filter(|config| { - event_type_filter - .map(|filter| filter.include.contains(&config.event_type)) - .unwrap_or(true) - }) - .cloned() - .collect(); - configs.extend(filtered_configs); + // 如果没有过滤器,直接扩展所有配置 + if event_type_filter.is_none() { + configs.extend(protocol_configs.iter().cloned()); + } else { + // 有过滤器时才进行过滤 + let filter = event_type_filter.unwrap(); + configs.extend( + protocol_configs + .iter() + .filter(|config| filter.include.contains(&config.event_type)) + .cloned(), + ); + } } + + let common_config = COMMON_CONFIG.get_or_init(|| AccountEventParseConfig { + program_id: Pubkey::default(), + protocol_type: ProtocolType::Common, + event_type: EventType::AccountCommon, + account_discriminator: &[], + account_parser: Self::parse_token_account_event, + }); + configs.push(common_config.clone()); + configs } @@ -173,8 +211,12 @@ impl AccountEventParser { ) -> Option> { let configs = Self::configs(protocols, event_type_filter); for config in configs { - if account.owner == config.program_id - && SimdUtils::fast_discriminator_match(&account.data, config.account_discriminator) + if config.program_id == Pubkey::default() + || (account.owner == config.program_id + && SimdUtils::fast_discriminator_match( + &account.data, + config.account_discriminator, + )) { let event = (config.account_parser)( &account, @@ -189,13 +231,29 @@ impl AccountEventParser { }, ); if let Some(mut event) = event { - event.set_handle_us(elapsed_micros_since( - account.recv_us, - )); + event.set_handle_us(elapsed_micros_since(account.recv_us)); return Some(event); } } } None } + + pub fn parse_token_account_event( + account: &AccountPretty, + metadata: EventMetadata, + ) -> Option> { + let info = Account::unpack(&account.data); + let mut event = CommonAccountEvent { + metadata, + pubkey: account.pubkey, + executable: account.executable, + lamports: account.lamports, + owner: account.owner, + rent_epoch: account.rent_epoch, + amount: if let Ok(info) = info { Some(info.amount) } else { None }, + }; + event.set_handle_us(elapsed_micros_since(account.recv_us)); + return Some(Box::new(event)); + } }