feat: add common account event parser with SPL token support

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

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<dyn UnifiedEvent>) {
             RaydiumCpmmPoolStateAccountEvent => |e: RaydiumCpmmPoolStateAccountEvent| {
                 println!("RaydiumCpmmPoolStateAccountEvent: {e:?}");
             },
+            CommonAccountEvent => |e: CommonAccountEvent| {
+                println!("CommonAccountEvent: {e:?}");
+            },
         });
     }
 }

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

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<u64>,
+}
+impl_unified_event!(CommonAccountEvent,);
+
 /// 账户事件解析器
 pub type AccountEventParserFn =
     fn(account: &AccountPretty, metadata: EventMetadata) -> Option<Box<dyn UnifiedEvent>>;
@@ -33,6 +50,9 @@ pub type AccountEventParserFn =
 static PROTOCOL_CONFIGS_CACHE: OnceLock<HashMap<Protocol, Vec<AccountEventParseConfig>>> =
     OnceLock::new();
 
+// 通用账户解析配置的静态缓存
+static COMMON_CONFIG: OnceLock<AccountEventParseConfig> = 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<AccountEventParseConfig> = 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<Box<dyn UnifiedEvent>> {
         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<Box<dyn UnifiedEvent>> {
+        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));
+    }
 }