diff --git a/Cargo.toml b/Cargo.toml index c78ac55..3d532d5 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "solana-streamer-sdk" -version = "0.3.1" +version = "0.3.2" edition = "2021" authors = ["William ", "sgxiang ", "wei <1415121722@qq.com>"] repository = "https://github.com/0xfnzero/solana-streamer" diff --git a/README.md b/README.md index d541891..d0b9871 100755 --- a/README.md +++ b/README.md @@ -44,14 +44,14 @@ Add the dependency to your `Cargo.toml`: ```toml # Add to your Cargo.toml -solana-streamer-sdk = { path = "./solana-streamer", version = "0.3.1" } +solana-streamer-sdk = { path = "./solana-streamer", version = "0.3.2" } ``` ### Use crates.io ```toml # Add to your Cargo.toml -solana-streamer-sdk = "0.3.1" +solana-streamer-sdk = "0.3.2" ``` ## Usage Examples @@ -71,13 +71,14 @@ This example demonstrates: The example uses a predefined transaction signature and shows how to extract protocol-specific events from the transaction data. -### Advanced Usage with Batch Processing and Backpressure +### Advanced Usage - Complete Example ```rust use solana_streamer_sdk::{ match_event, streaming::{ event_parser::{ + common::{filter::EventTypeFilter, EventType}, protocols::{ bonk::{ parser::BONK_PROGRAM_ID, BonkGlobalConfigAccountEvent, BonkMigrateToAmmEvent, @@ -127,13 +128,8 @@ use solana_streamer_sdk::{ #[tokio::main] async fn main() -> Result<(), Box> { println!("Starting Solana Streamer..."); - - // Test Yellowstone gRPC with performance monitoring test_grpc().await?; - - // Test ShredStream with performance monitoring test_shreds().await?; - Ok(()) } @@ -178,26 +174,33 @@ async fn test_grpc() -> Result<(), Box> { let account_exclude = vec![]; let account_required = vec![]; - println!("Starting to listen for events, press Ctrl+C to stop..."); - println!("Monitoring programs: {:?}", account_include); - - println!("Starting subscription..."); - - // 监听交易数据 + // Transaction filter for monitoring transaction events let transaction_filter = TransactionFilter { account_include: account_include.clone(), account_exclude, account_required, }; - // 监听属于owner程序的账号数据 -> 账号事件监听 + // Account filter for monitoring account state changes let account_filter = AccountFilter { account: vec![], owner: account_include.clone() }; + // Event type filtering - optional + // No event filtering, includes all events + let event_type_filter = None; + // Only include PumpSwapBuy and PumpSwapSell events + // let event_type_filter = Some(EventTypeFilter { include: vec![EventType::PumpSwapBuy, EventType::PumpSwapSell] }); + + println!("Starting to listen for events, press Ctrl+C to stop..."); + println!("Monitoring programs: {:?}", account_include); + + println!("Starting subscription..."); + grpc.subscribe_events_immediate( protocols, None, transaction_filter, account_filter, + event_type_filter, None, callback, ) @@ -236,11 +239,11 @@ fn create_event_callback() -> impl Fn(Box) { |event: Box| { println!("🎉 Event received! Type: {:?}, ID: {}", event.event_type(), event.id()); match_event!(event, { - // block meta + // -------------------------- block meta ----------------------- BlockMetaEvent => |e: BlockMetaEvent| { println!("BlockMetaEvent: {e:?}"); }, - // bonk + // -------------------------- bonk ----------------------- BonkPoolCreateEvent => |e: BonkPoolCreateEvent| { // When using grpc, you can get block_time from each event println!("block_time: {:?}, block_time_ms: {:?}", e.metadata.block_time, e.metadata.block_time_ms); @@ -255,7 +258,7 @@ fn create_event_callback() -> impl Fn(Box) { BonkMigrateToCpswapEvent => |e: BonkMigrateToCpswapEvent| { println!("BonkMigrateToCpswapEvent: {e:?}"); }, - // pumpfun + // -------------------------- pumpfun ----------------------- PumpFunTradeEvent => |e: PumpFunTradeEvent| { println!("PumpFunTradeEvent: {e:?}"); }, @@ -265,7 +268,7 @@ fn create_event_callback() -> impl Fn(Box) { PumpFunCreateTokenEvent => |e: PumpFunCreateTokenEvent| { println!("PumpFunCreateTokenEvent: {e:?}"); }, - // pumpswap + // -------------------------- pumpswap ----------------------- PumpSwapBuyEvent => |e: PumpSwapBuyEvent| { println!("Buy event: {e:?}"); }, @@ -281,7 +284,7 @@ fn create_event_callback() -> impl Fn(Box) { PumpSwapWithdrawEvent => |e: PumpSwapWithdrawEvent| { println!("Withdraw event: {e:?}"); }, - // raydium_cpmm + // -------------------------- raydium_cpmm ----------------------- RaydiumCpmmSwapEvent => |e: RaydiumCpmmSwapEvent| { println!("RaydiumCpmmSwapEvent: {e:?}"); }, @@ -294,7 +297,7 @@ fn create_event_callback() -> impl Fn(Box) { RaydiumCpmmWithdrawEvent => |e: RaydiumCpmmWithdrawEvent| { println!("RaydiumCpmmWithdrawEvent: {e:?}"); }, - // raydium_clmm + // -------------------------- raydium_clmm ----------------------- RaydiumClmmSwapEvent => |e: RaydiumClmmSwapEvent| { println!("RaydiumClmmSwapEvent: {e:?}"); }, @@ -319,7 +322,7 @@ fn create_event_callback() -> impl Fn(Box) { RaydiumClmmOpenPositionV2Event => |e: RaydiumClmmOpenPositionV2Event| { println!("RaydiumClmmOpenPositionV2Event: {e:?}"); }, - // raydium_amm_v4 + // -------------------------- raydium_amm_v4 ----------------------- RaydiumAmmV4SwapEvent => |e: RaydiumAmmV4SwapEvent| { println!("RaydiumAmmV4SwapEvent: {e:?}"); }, @@ -380,6 +383,22 @@ fn create_event_callback() -> impl Fn(Box) { } ``` +### Event Filtering + +The library supports flexible event filtering to reduce processing overhead: + +```rust +use solana_streamer_sdk::streaming::event_parser::common::{filter::EventTypeFilter, EventType}; + +// No filtering - receive all events +let event_type_filter = None; + +// Filter specific event types - only receive PumpSwap buy/sell events +let event_type_filter = Some(EventTypeFilter { + include: vec![EventType::PumpSwapBuy, EventType::PumpSwapSell] +}); +``` + ## Supported Protocols - **PumpFun**: Primary meme coin trading platform diff --git a/README_CN.md b/README_CN.md index c5be0de..1365ec3 100644 --- a/README_CN.md +++ b/README_CN.md @@ -44,14 +44,14 @@ git clone https://github.com/0xfnzero/solana-streamer ```toml # 添加到您的 Cargo.toml -solana-streamer-sdk = { path = "./solana-streamer", version = "0.3.1" } +solana-streamer-sdk = { path = "./solana-streamer", version = "0.3.2" } ``` ### 使用 crates.io ```toml # 添加到您的 Cargo.toml -solana-streamer-sdk = "0.3.1" +solana-streamer-sdk = "0.3.2" ``` ## 使用示例 @@ -71,13 +71,14 @@ cargo run --example parse_tx_events 该示例使用预定义的交易签名，展示如何从交易数据中提取协议特定的事件。 -### 高级用法示例 +### 高级用法 - 完整示例 ```rust use solana_streamer_sdk::{ match_event, streaming::{ event_parser::{ + common::{filter::EventTypeFilter, EventType}, protocols::{ bonk::{ parser::BONK_PROGRAM_ID, BonkGlobalConfigAccountEvent, BonkMigrateToAmmEvent, @@ -173,11 +174,6 @@ async fn test_grpc() -> Result<(), Box> { let account_exclude = vec![]; let account_required = vec![]; - println!("Starting to listen for events, press Ctrl+C to stop..."); - println!("Monitoring programs: {:?}", account_include); - - println!("Starting subscription..."); - // 监听交易数据 let transaction_filter = TransactionFilter { account_include: account_include.clone(), @@ -188,11 +184,23 @@ async fn test_grpc() -> Result<(), Box> { // 监听属于owner程序的账号数据 -> 账号事件监听 let account_filter = AccountFilter { account: vec![], owner: account_include.clone() }; + // 事件过滤 - 可选 + // 不进行事件过滤，包含所有事件 + let event_type_filter = None; + // 只包含PumpSwapBuy事件、PumpSwapSell事件 + // let event_type_filter = Some(EventTypeFilter { include: vec![EventType::PumpSwapBuy, EventType::PumpSwapSell] }); + + println!("Starting to listen for events, press Ctrl+C to stop..."); + println!("Monitoring programs: {:?}", account_include); + + println!("Starting subscription..."); + grpc.subscribe_events_immediate( protocols, None, transaction_filter, account_filter, + event_type_filter, None, callback, ) @@ -231,11 +239,11 @@ fn create_event_callback() -> impl Fn(Box) { |event: Box| { println!("🎉 Event received! Type: {:?}, ID: {}", event.event_type(), event.id()); match_event!(event, { - // block meta + // -------------------------- block meta ----------------------- BlockMetaEvent => |e: BlockMetaEvent| { println!("BlockMetaEvent: {e:?}"); }, - // bonk + // -------------------------- bonk ----------------------- BonkPoolCreateEvent => |e: BonkPoolCreateEvent| { // 使用grpc的时候，可以从每个事件中获取到block_time println!("block_time: {:?}, block_time_ms: {:?}", e.metadata.block_time, e.metadata.block_time_ms); @@ -250,7 +258,7 @@ fn create_event_callback() -> impl Fn(Box) { BonkMigrateToCpswapEvent => |e: BonkMigrateToCpswapEvent| { println!("BonkMigrateToCpswapEvent: {e:?}"); }, - // pumpfun + // -------------------------- pumpfun ----------------------- PumpFunTradeEvent => |e: PumpFunTradeEvent| { println!("PumpFunTradeEvent: {e:?}"); }, @@ -260,7 +268,7 @@ fn create_event_callback() -> impl Fn(Box) { PumpFunCreateTokenEvent => |e: PumpFunCreateTokenEvent| { println!("PumpFunCreateTokenEvent: {e:?}"); }, - // pumpswap + // -------------------------- pumpswap ----------------------- PumpSwapBuyEvent => |e: PumpSwapBuyEvent| { println!("Buy event: {e:?}"); }, @@ -276,7 +284,7 @@ fn create_event_callback() -> impl Fn(Box) { PumpSwapWithdrawEvent => |e: PumpSwapWithdrawEvent| { println!("Withdraw event: {e:?}"); }, - // raydium_cpmm + // -------------------------- raydium_cpmm ----------------------- RaydiumCpmmSwapEvent => |e: RaydiumCpmmSwapEvent| { println!("RaydiumCpmmSwapEvent: {e:?}"); }, @@ -289,7 +297,7 @@ fn create_event_callback() -> impl Fn(Box) { RaydiumCpmmWithdrawEvent => |e: RaydiumCpmmWithdrawEvent| { println!("RaydiumCpmmWithdrawEvent: {e:?}"); }, - // raydium_clmm + // -------------------------- raydium_clmm ----------------------- RaydiumClmmSwapEvent => |e: RaydiumClmmSwapEvent| { println!("RaydiumClmmSwapEvent: {e:?}"); }, @@ -314,7 +322,7 @@ fn create_event_callback() -> impl Fn(Box) { RaydiumClmmOpenPositionV2Event => |e: RaydiumClmmOpenPositionV2Event| { println!("RaydiumClmmOpenPositionV2Event: {e:?}"); }, - // raydium_amm_v4 + // -------------------------- raydium_amm_v4 ----------------------- RaydiumAmmV4SwapEvent => |e: RaydiumAmmV4SwapEvent| { println!("RaydiumAmmV4SwapEvent: {e:?}"); }, @@ -375,6 +383,22 @@ fn create_event_callback() -> impl Fn(Box) { } ``` +### 事件过滤 + +库支持灵活的事件过滤以减少处理开销： + +```rust +use solana_streamer_sdk::streaming::event_parser::common::{filter::EventTypeFilter, EventType}; + +// 无过滤 - 接收所有事件 +let event_type_filter = None; + +// 过滤特定事件类型 - 只接收 PumpSwap 买入/卖出事件 +let event_type_filter = Some(EventTypeFilter { + include: vec![EventType::PumpSwapBuy, EventType::PumpSwapSell] +}); +``` + ## 支持的协议 - **PumpFun**: 主要迷因币交易平台 diff --git a/examples/parse_tx_events.rs b/examples/parse_tx_events.rs index c023e5d..cbb5bb6 100644 --- a/examples/parse_tx_events.rs +++ b/examples/parse_tx_events.rs @@ -97,7 +97,7 @@ async fn get_single_transaction_details(signature_str: &str) -> Result<()> { Protocol::RaydiumCpmm, Protocol::RaydiumAmmV4, ]; - let parser: Arc = Arc::new(MutilEventParser::new(protocols)); + let parser: Arc = Arc::new(MutilEventParser::new(protocols, None)); let start_time = std::time::Instant::now(); let events = parser .parse_transaction( diff --git a/src/main.rs b/src/main.rs index ef347ab..2620a7b 100755 --- a/src/main.rs +++ b/src/main.rs @@ -2,6 +2,7 @@ use solana_streamer_sdk::{ match_event, streaming::{ event_parser::{ + common::{filter::EventTypeFilter, EventType}, protocols::{ bonk::{ parser::BONK_PROGRAM_ID, BonkGlobalConfigAccountEvent, BonkMigrateToAmmEvent, @@ -97,16 +98,22 @@ async fn test_grpc() -> Result<(), Box> { let account_exclude = vec![]; let account_required = vec![]; - // 监听交易数据 + // Listen to transaction data let transaction_filter = TransactionFilter { account_include: account_include.clone(), account_exclude, account_required, }; - // 监听属于owner程序的账号数据 -> 账号事件监听 + // Listen to account data belonging to owner programs -> account event monitoring let account_filter = AccountFilter { account: vec![], owner: account_include.clone() }; + // Event filtering + // No event filtering, includes all events + let event_type_filter = None; + // Only include PumpSwapBuy events and PumpSwapSell events + // let event_type_filter = EventTypeFilter { include: vec![EventType::PumpSwapBuy, EventType::PumpSwapSell] }; + println!("Starting to listen for events, press Ctrl+C to stop..."); println!("Monitoring programs: {:?}", account_include); @@ -117,6 +124,7 @@ async fn test_grpc() -> Result<(), Box> { None, transaction_filter, account_filter, + event_type_filter, None, callback, ) diff --git a/src/streaming/common/metrics.rs b/src/streaming/common/metrics.rs index 69540d4..26126df 100644 --- a/src/streaming/common/metrics.rs +++ b/src/streaming/common/metrics.rs @@ -4,19 +4,37 @@ use tokio::sync::Mutex; use super::config::StreamClientConfig; use super::constants::*; +/// 单个事件类型的指标 +#[derive(Debug, Clone)] +pub struct EventMetrics { + pub process_count: u64, + pub events_processed: u64, + pub events_per_second: f64, + pub events_in_window: u64, + pub window_start_time: std::time::Instant, +} + +impl EventMetrics { + fn new(now: std::time::Instant) -> Self { + Self { + process_count: 0, + events_processed: 0, + events_per_second: 0.0, + events_in_window: 0, + window_start_time: now, + } + } +} + /// 通用性能监控指标 #[derive(Debug, Clone)] pub struct PerformanceMetrics { pub start_time: std::time::Instant, - pub process_count: u64, - pub events_processed: u64, - pub events_per_second: f64, + pub event_metrics: [EventMetrics; 3], // [Tx, Account, BlockMeta] pub average_processing_time_ms: f64, pub min_processing_time_ms: f64, pub max_processing_time_ms: f64, pub last_update_time: std::time::Instant, - pub events_in_window: u64, - pub window_start_time: std::time::Instant, } impl Default for PerformanceMetrics { @@ -25,20 +43,89 @@ impl Default for PerformanceMetrics { } } +pub enum MetricsEventType { + Tx, + Account, + BlockMeta, +} + +impl MetricsEventType { + fn as_index(&self) -> usize { + match self { + MetricsEventType::Tx => 0, + MetricsEventType::Account => 1, + MetricsEventType::BlockMeta => 2, + } + } +} + impl PerformanceMetrics { pub fn new() -> Self { let now = std::time::Instant::now(); Self { - start_time: std::time::Instant::now(), - process_count: 0, - events_processed: 0, - events_per_second: 0.0, + start_time: now, + event_metrics: [EventMetrics::new(now), EventMetrics::new(now), EventMetrics::new(now)], average_processing_time_ms: 0.0, min_processing_time_ms: 0.0, max_processing_time_ms: 0.0, last_update_time: now, - events_in_window: 0, - window_start_time: now, + } + } + + /// 更新时间窗口指标 + fn update_window_metrics( + &mut self, + event_type: &MetricsEventType, + now: std::time::Instant, + window_duration: std::time::Duration, + ) { + let index = event_type.as_index(); + let event_metric = &mut self.event_metrics[index]; + + if now.duration_since(event_metric.window_start_time) >= window_duration { + let window_seconds = now.duration_since(event_metric.window_start_time).as_secs_f64(); + // 修复：正确计算每秒事件数，避免除零错误 + event_metric.events_per_second = if window_seconds > 0.001 { + // 避免极小的时间差 + event_metric.events_in_window as f64 / window_seconds + } else { + 0.0 // 时间太短时设为0，而不是事件总数 + }; + + // 重置窗口 + event_metric.events_in_window = 0; + event_metric.window_start_time = now; + } + } + + /// 计算实时每秒事件数（用于显示） + fn calculate_real_time_events_per_second( + &self, + event_type: &MetricsEventType, + now: std::time::Instant, + ) -> f64 { + let index = event_type.as_index(); + let event_metric = &self.event_metrics[index]; + + let current_window_duration = + now.duration_since(event_metric.window_start_time).as_secs_f64(); + + // 如果当前窗口有足够的时间和事件，使用当前窗口的数据 + if current_window_duration > 1.0 && event_metric.events_in_window > 0 { + event_metric.events_in_window as f64 / current_window_duration + } + // 如果当前窗口时间太短或没有事件，使用上一个完整窗口的值 + else if event_metric.events_per_second > 0.0 { + event_metric.events_per_second + } + // 如果都没有，计算总体平均值 + else { + let total_duration = now.duration_since(self.start_time).as_secs_f64(); + if total_duration > 1.0 && event_metric.events_processed > 0 { + event_metric.events_processed as f64 / total_duration + } else { + 0.0 + } } } } @@ -69,15 +156,46 @@ impl MetricsManager { /// 打印性能指标 pub async fn print_metrics(&self) { let metrics = self.get_metrics().await; - println!("📊 {} Performance Metrics:", self.stream_name); + let event_names = ["TX", "Account", "Block Meta"]; + let event_types = + [MetricsEventType::Tx, MetricsEventType::Account, MetricsEventType::BlockMeta]; + let now = std::time::Instant::now(); + + println!("\n📊 {} Performance Metrics", self.stream_name); println!(" Run Time: {:?}", metrics.start_time.elapsed()); - println!(" Process Count: {}", metrics.process_count); - println!(" Events Processed: {}", metrics.events_processed); - println!(" Events/Second: {:.2}", metrics.events_per_second); - println!(" Avg Processing Time: {:.2}ms", metrics.average_processing_time_ms); - println!(" Min Processing Time: {:.2}ms", metrics.min_processing_time_ms); - println!(" Max Processing Time: {:.2}ms", metrics.max_processing_time_ms); - println!("---"); + + // 打印表格头部 + println!("┌─────────────┬──────────────┬──────────────────┬─────────────────┐"); + println!("│ Event Type │ Process Count│ Events Processed │ Events/Second │"); + println!("├─────────────┼──────────────┼──────────────────┼─────────────────┤"); + + // 打印每种事件类型的数据 + for (i, name) in event_names.iter().enumerate() { + let event_metric = &metrics.event_metrics[i]; + // 使用实时计算的每秒事件数，而不是窗口更新的值 + let real_time_eps = metrics.calculate_real_time_events_per_second(&event_types[i], now); + + println!( + "│ {:11} │ {:12} │ {:16} │ {:13.2} │", + name, + event_metric.process_count, + event_metric.events_processed, + real_time_eps + ); + } + + println!("└─────────────┴──────────────┴──────────────────┴─────────────────┘"); + + // 打印处理时间统计表格 + println!("\n⏱️ Processing Time Statistics"); + println!("┌─────────────────────┬─────────────┐"); + println!("│ Metric │ Value (ms) │"); + println!("├─────────────────────┼─────────────┤"); + println!("│ Average │ {:9.2} │", metrics.average_processing_time_ms); + println!("│ Minimum │ {:9.2} │", metrics.min_processing_time_ms); + println!("│ Maximum │ {:9.2} │", metrics.max_processing_time_ms); + println!("└─────────────────────┴─────────────┘"); + println!(); } /// 启动自动性能监控任务 @@ -100,29 +218,50 @@ impl MetricsManager { } /// 更新处理次数 - pub async fn add_process_count(&self) { + pub async fn add_process_count(&self, event_type: MetricsEventType) { if !self.config.enable_metrics { return; } let mut metrics = self.metrics.lock().await; - metrics.process_count += 1; + metrics.event_metrics[event_type.as_index()].process_count += 1; + } + + // 保持向后兼容的方法 + pub async fn add_tx_process_count(&self) { + self.add_process_count(MetricsEventType::Tx).await; + } + + pub async fn add_account_process_count(&self) { + self.add_process_count(MetricsEventType::Account).await; + } + + pub async fn add_block_meta_process_count(&self) { + self.add_process_count(MetricsEventType::BlockMeta).await; } /// 更新性能指标 - pub async fn update_metrics(&self, events_processed: u64, processing_time_ms: f64) { + pub async fn update_metrics( + &self, + event_type: MetricsEventType, + events_processed: u64, + processing_time_ms: f64, + ) { // 检查是否启用性能监控 if !self.config.enable_metrics { - return; // 如果未启用性能监控，直接返回 + return; } let mut metrics = self.metrics.lock().await; let now = std::time::Instant::now(); + let index = event_type.as_index(); + + // 更新事件计数 + metrics.event_metrics[index].events_processed += events_processed; + metrics.event_metrics[index].events_in_window += events_processed; - metrics.events_processed += events_processed; - metrics.events_in_window += events_processed; metrics.last_update_time = now; - // 更新最快和最慢处理时间 + // 更新处理时间统计 if processing_time_ms < metrics.min_processing_time_ms || metrics.min_processing_time_ms == 0.0 { @@ -132,29 +271,24 @@ impl MetricsManager { metrics.max_processing_time_ms = processing_time_ms; } - // 计算平均处理时间 - if metrics.events_processed > 0 { - metrics.average_processing_time_ms = (metrics.average_processing_time_ms - * (metrics.events_processed - events_processed) as f64 - + processing_time_ms) - / metrics.events_processed as f64; - } + // 计算平均处理时间 - 使用增量更新避免重复计算 + let total_events = metrics.event_metrics[index].events_processed; + if total_events > 0 { + let total_events_f64 = total_events as f64; + let old_total = (total_events_f64 - events_processed as f64).max(0.0); - // 基于时间窗口计算每秒处理事件数 - let window_duration = std::time::Duration::from_secs(DEFAULT_METRICS_WINDOW_SECONDS); - if now.duration_since(metrics.window_start_time) >= window_duration { - let window_seconds = now.duration_since(metrics.window_start_time).as_secs_f64(); - if window_seconds > 0.0 && metrics.events_in_window > 0 { - metrics.events_per_second = metrics.events_in_window as f64 / window_seconds; + metrics.average_processing_time_ms = if old_total > 0.0 { + (metrics.average_processing_time_ms * old_total + + processing_time_ms * events_processed as f64) + / total_events_f64 } else { - // 如果窗口内没有事件，保持之前的速率或设为0 - metrics.events_per_second = 0.0; - } - - // 重置窗口 - metrics.events_in_window = 0; - metrics.window_start_time = now; + processing_time_ms + }; } + + // 更新时间窗口指标 + let window_duration = std::time::Duration::from_secs(DEFAULT_METRICS_WINDOW_SECONDS); + metrics.update_window_metrics(&event_type, now, window_duration); } /// 记录慢处理操作 diff --git a/src/streaming/event_parser/common/filter.rs b/src/streaming/event_parser/common/filter.rs new file mode 100644 index 0000000..8fa3296 --- /dev/null +++ b/src/streaming/event_parser/common/filter.rs @@ -0,0 +1,24 @@ +use crate::streaming::event_parser::common::{ + types::EventType, ACCOUNT_EVENT_TYPES, BLOCK_EVENT_TYPES, +}; + +#[derive(Debug, Clone)] +pub struct EventTypeFilter { + pub include: Vec, +} + +impl EventTypeFilter { + pub fn include_transaction_event(&self) -> bool { + self.include + .iter() + .any(|event| !ACCOUNT_EVENT_TYPES.contains(event) && !BLOCK_EVENT_TYPES.contains(event)) + } + + pub fn include_account_event(&self) -> bool { + self.include.iter().any(|event| ACCOUNT_EVENT_TYPES.contains(event)) + } + + pub fn include_block_event(&self) -> bool { + self.include.iter().any(|event| BLOCK_EVENT_TYPES.contains(event)) + } +} diff --git a/src/streaming/event_parser/common/mod.rs b/src/streaming/event_parser/common/mod.rs index aa905b6..3ecc161 100755 --- a/src/streaming/event_parser/common/mod.rs +++ b/src/streaming/event_parser/common/mod.rs @@ -1,5 +1,6 @@ pub mod types; pub mod utils; +pub mod filter; /// 自动生成UnifiedEvent trait实现的宏 #[macro_export] diff --git a/src/streaming/event_parser/common/types.rs b/src/streaming/event_parser/common/types.rs index c1cbc0c..03098b4 100755 --- a/src/streaming/event_parser/common/types.rs +++ b/src/streaming/event_parser/common/types.rs @@ -180,6 +180,24 @@ pub enum EventType { Unknown, } +pub const ACCOUNT_EVENT_TYPES: &[EventType] = &[ + EventType::AccountRaydiumAmmV4AmmInfo, + EventType::AccountPumpSwapGlobalConfig, + EventType::AccountPumpSwapPool, + EventType::AccountBonkPoolState, + EventType::AccountBonkGlobalConfig, + EventType::AccountBonkPlatformConfig, + EventType::AccountBonkVestingRecord, + EventType::AccountPumpFunBondingCurve, + EventType::AccountPumpFunGlobal, + EventType::AccountRaydiumClmmAmmConfig, + EventType::AccountRaydiumClmmPoolState, + EventType::AccountRaydiumClmmTickArrayState, + EventType::AccountRaydiumCpmmAmmConfig, + EventType::AccountRaydiumCpmmPoolState, +]; +pub const BLOCK_EVENT_TYPES: &[EventType] = &[EventType::BlockMeta]; + impl EventType { #[allow(clippy::inherent_to_string)] pub fn to_string(&self) -> String { diff --git a/src/streaming/event_parser/core/account_event_parser.rs b/src/streaming/event_parser/core/account_event_parser.rs index 855a394..7e95698 100644 --- a/src/streaming/event_parser/core/account_event_parser.rs +++ b/src/streaming/event_parser/core/account_event_parser.rs @@ -3,6 +3,7 @@ use std::sync::OnceLock; use solana_sdk::pubkey::Pubkey; +use crate::streaming::event_parser::common::filter::EventTypeFilter; use crate::streaming::event_parser::common::{EventMetadata, EventType, ProtocolType}; use crate::streaming::event_parser::core::traits::UnifiedEvent; use crate::streaming::event_parser::protocols::bonk::parser::BONK_PROGRAM_ID; @@ -34,7 +35,7 @@ static PROTOCOL_CONFIGS_CACHE: OnceLock) -> Vec { + pub fn configs(protocols: Vec, event_type_filter: Option) -> Vec { let protocols_map = PROTOCOL_CONFIGS_CACHE.get_or_init(|| { let mut map: HashMap> = HashMap::new(); map.insert(Protocol::PumpSwap, vec![ @@ -145,7 +146,11 @@ impl AccountEventParser { let mut configs = vec![]; for protocol in protocols { - configs.extend(protocols_map.get(&protocol).unwrap_or(&vec![]).clone()); + let protocol_configs = protocols_map.get(&protocol).unwrap_or(&vec![]).clone(); + let filtered_configs: Vec = protocol_configs.into_iter().filter(|config| { + event_type_filter.as_ref().map(|filter| filter.include.contains(&config.event_type)).unwrap_or(true) + }).collect(); + configs.extend(filtered_configs); } configs } @@ -154,8 +159,9 @@ impl AccountEventParser { protocols: Vec, account: AccountPretty, program_received_time_ms: i64, + event_type_filter: Option, ) -> Option> { - let configs = Self::configs(protocols); + let configs = Self::configs(protocols, event_type_filter); for config in configs { if account.owner == config.program_id.to_string() && account.data[..config.account_discriminator.len()] diff --git a/src/streaming/event_parser/protocols/mutil/parser.rs b/src/streaming/event_parser/protocols/mutil/parser.rs index 9b5d75b..af3902d 100755 --- a/src/streaming/event_parser/protocols/mutil/parser.rs +++ b/src/streaming/event_parser/protocols/mutil/parser.rs @@ -4,6 +4,7 @@ use prost_types::Timestamp; use solana_sdk::{instruction::CompiledInstruction, pubkey::Pubkey}; use solana_transaction_status::UiCompiledInstruction; +use crate::streaming::event_parser::common::filter::EventTypeFilter; use crate::streaming::event_parser::{ core::traits::{EventParser, GenericEventParseConfig, GenericEventParser, UnifiedEvent}, EventParserFactory, Protocol, @@ -14,7 +15,7 @@ pub struct MutilEventParser { } impl MutilEventParser { - pub fn new(protocols: Vec) -> Self { + pub fn new(protocols: Vec, event_type_filter: Option) -> Self { let mut inner = GenericEventParser::new(vec![], vec![]); // Configure all event types for protocol in protocols { @@ -22,12 +23,18 @@ impl MutilEventParser { // Merge inner_instruction_configs, append configurations to existing Vec for (key, configs) in parse.inner_instruction_configs() { - inner.inner_instruction_configs.entry(key).or_insert_with(Vec::new).extend(configs); + let filtered_configs: Vec = configs.into_iter().filter(|config| { + event_type_filter.as_ref().map(|filter| filter.include.contains(&config.event_type)).unwrap_or(true) + }).collect(); + inner.inner_instruction_configs.entry(key).or_insert_with(Vec::new).extend(filtered_configs); } // Merge instruction_configs, append configurations to existing Vec for (key, configs) in parse.instruction_configs() { - inner.instruction_configs.entry(key).or_insert_with(Vec::new).extend(configs); + let filtered_configs: Vec = configs.into_iter().filter(|config| { + event_type_filter.as_ref().map(|filter| filter.include.contains(&config.event_type)).unwrap_or(true) + }).collect(); + inner.instruction_configs.entry(key).or_insert_with(Vec::new).extend(filtered_configs); } // Append program_ids (this is already appending) diff --git a/src/streaming/grpc/event_processor.rs b/src/streaming/grpc/event_processor.rs index a315340..10454fe 100644 --- a/src/streaming/grpc/event_processor.rs +++ b/src/streaming/grpc/event_processor.rs @@ -1,12 +1,14 @@ -use std::sync::Arc; +use std::sync::{Arc, Mutex}; use solana_sdk::pubkey::Pubkey; use super::types::EventPretty; use crate::common::AnyResult; use crate::streaming::common::{ - EventBatchProcessor as EventBatchCollector, MetricsManager, StreamClientConfig as ClientConfig, + EventBatchProcessor as EventBatchCollector, MetricsEventType, MetricsManager, + StreamClientConfig as ClientConfig, }; +use crate::streaming::event_parser::common::filter::EventTypeFilter; use crate::streaming::event_parser::core::account_event_parser::AccountEventParser; use crate::streaming::event_parser::core::common_event_parser::CommonEventParser; use crate::streaming::event_parser::EventParser; @@ -18,12 +20,29 @@ use crate::streaming::event_parser::{ pub struct EventProcessor { pub(crate) metrics_manager: MetricsManager, pub(crate) config: ClientConfig, + pub(crate) parser_cache: Arc>>>, } impl EventProcessor { /// 创建新的事件处理器 pub fn new(metrics_manager: MetricsManager, config: ClientConfig) -> Self { - Self { metrics_manager, config } + Self { metrics_manager, config, parser_cache: Arc::new(Mutex::new(None)) } + } + + /// 获取或创建解析器，使用缓存机制避免重复创建 + fn get_or_create_parser( + &self, + protocols: Vec, + event_type_filter: Option, + ) -> Arc { + let mut cache = self.parser_cache.lock().unwrap(); + if let Some(cached_parser) = cache.clone() { + return cached_parser.clone(); + } + let parser: Arc = + Arc::new(MutilEventParser::new(protocols.clone(), event_type_filter.clone())); + *cache = Some(parser.clone()); + parser } /// 使用性能监控处理事件交易 @@ -33,19 +52,21 @@ impl EventProcessor { callback: &F, bot_wallet: Option, protocols: Vec, + event_type_filter: Option, ) -> AnyResult<()> where F: Fn(Box) + Send + Sync, { match event_pretty { EventPretty::Account(account_pretty) => { - self.metrics_manager.add_process_count().await; + self.metrics_manager.add_account_process_count().await; let start_time = std::time::Instant::now(); let program_received_time_ms = chrono::Utc::now().timestamp_millis(); let account_event = AccountEventParser::parse_account_event( protocols.clone(), account_pretty, program_received_time_ms, + event_type_filter, ); if let Some(event) = account_event { callback(event); @@ -53,21 +74,22 @@ impl EventProcessor { let processing_time = start_time.elapsed(); let processing_time_ms = processing_time.as_millis() as f64; // 更新性能指标（如果启用） - self.metrics_manager.update_metrics(1, processing_time_ms).await; + self.metrics_manager + .update_metrics(MetricsEventType::Account, 1, processing_time_ms) + .await; // 记录慢处理操作 self.metrics_manager.log_slow_processing(processing_time_ms, 1); } } EventPretty::Transaction(transaction_pretty) => { - self.metrics_manager.add_process_count().await; + self.metrics_manager.add_tx_process_count().await; let start_time = std::time::Instant::now(); let program_received_time_ms = chrono::Utc::now().timestamp_millis(); let slot = transaction_pretty.slot; let signature = transaction_pretty.signature.to_string(); - // 直接创建解析器并处理事务 - let parser: Arc = - Arc::new(MutilEventParser::new(protocols.clone())); + // 使用缓存获取解析器 + let parser = self.get_or_create_parser(protocols.clone(), event_type_filter); let all_events = parser .parse_transaction( transaction_pretty.tx.clone(), @@ -98,13 +120,15 @@ impl EventProcessor { let processing_time_ms = processing_time.as_millis() as f64; // 更新性能指标（如果启用） - self.metrics_manager.update_metrics(event_count as u64, processing_time_ms).await; + self.metrics_manager + .update_metrics(MetricsEventType::Tx, event_count as u64, processing_time_ms) + .await; // 记录慢处理操作 self.metrics_manager.log_slow_processing(processing_time_ms, event_count); } EventPretty::BlockMeta(block_meta_pretty) => { let start_time = std::time::Instant::now(); - self.metrics_manager.add_process_count().await; + self.metrics_manager.add_block_meta_process_count().await; let block_time_ms = block_meta_pretty .block_time .map(|ts| ts.seconds * 1000 + ts.nanos as i64 / 1_000_000) @@ -119,7 +143,9 @@ impl EventProcessor { let processing_time = start_time.elapsed(); let processing_time_ms = processing_time.as_millis() as f64; // 更新性能指标（如果启用） - self.metrics_manager.update_metrics(1, processing_time_ms).await; + self.metrics_manager + .update_metrics(MetricsEventType::BlockMeta, 1, processing_time_ms) + .await; // 记录慢处理操作 self.metrics_manager.log_slow_processing(processing_time_ms, 1); } @@ -135,19 +161,21 @@ impl EventProcessor { batch_processor: &mut EventBatchCollector, bot_wallet: Option, protocols: Vec, + event_type_filter: Option, ) -> AnyResult<()> where F: Fn(Vec>) + Send + Sync + 'static, { match event_pretty { EventPretty::Account(account_pretty) => { - self.metrics_manager.add_process_count().await; + self.metrics_manager.add_account_process_count().await; let start_time = std::time::Instant::now(); let program_received_time_ms = chrono::Utc::now().timestamp_millis(); let account_event = AccountEventParser::parse_account_event( protocols.clone(), account_pretty, program_received_time_ms, + event_type_filter, ); if let Some(event) = account_event { (batch_processor.callback)(vec![event]); @@ -155,21 +183,22 @@ impl EventProcessor { let processing_time = start_time.elapsed(); let processing_time_ms = processing_time.as_millis() as f64; // 实际调用性能指标更新 - self.metrics_manager.update_metrics(1, processing_time_ms).await; + self.metrics_manager + .update_metrics(MetricsEventType::Account, 1, processing_time_ms) + .await; // 记录慢处理操作 self.metrics_manager.log_slow_processing(processing_time_ms, 1); } } EventPretty::Transaction(transaction_pretty) => { - self.metrics_manager.add_process_count().await; + self.metrics_manager.add_tx_process_count().await; let start_time = std::time::Instant::now(); let program_received_time_ms = chrono::Utc::now().timestamp_millis(); let slot = transaction_pretty.slot; let signature = transaction_pretty.signature.to_string(); - // 直接创建解析器并处理事务 - let parser: Arc = - Arc::new(MutilEventParser::new(protocols.clone())); + // 使用缓存获取解析器 + let parser = self.get_or_create_parser(protocols.clone(), event_type_filter); let result = parser .parse_transaction( transaction_pretty.tx.clone(), @@ -224,13 +253,16 @@ impl EventProcessor { let processing_time_ms = processing_time.as_millis() as f64; // 实际调用性能指标更新 - self.metrics_manager.update_metrics(total_events as u64, processing_time_ms).await; + self.metrics_manager + .update_metrics(MetricsEventType::Tx, total_events as u64, processing_time_ms) + .await; // 记录慢处理操作 self.metrics_manager.log_slow_processing(processing_time_ms, total_events); } EventPretty::BlockMeta(block_meta_pretty) => { let start_time = std::time::Instant::now(); + self.metrics_manager.add_block_meta_process_count().await; let block_time_ms = block_meta_pretty .block_time .map(|ts| ts.seconds * 1000 + ts.nanos as i64 / 1_000_000) @@ -245,7 +277,9 @@ impl EventProcessor { let processing_time = start_time.elapsed(); let processing_time_ms = processing_time.as_millis() as f64; // 更新性能指标（如果启用） - self.metrics_manager.update_metrics(1, processing_time_ms).await; + self.metrics_manager + .update_metrics(MetricsEventType::BlockMeta, 1, processing_time_ms) + .await; // 记录慢处理操作 self.metrics_manager.log_slow_processing(processing_time_ms, 1); } diff --git a/src/streaming/grpc/subscription.rs b/src/streaming/grpc/subscription.rs index 9315ab2..2efdddd 100644 --- a/src/streaming/grpc/subscription.rs +++ b/src/streaming/grpc/subscription.rs @@ -12,6 +12,7 @@ use super::types::AccountsFilterMap; use super::types::TransactionsFilterMap; use crate::common::AnyResult; use crate::streaming::common::StreamClientConfig as ClientConfig; +use crate::streaming::event_parser::common::filter::EventTypeFilter; /// 订阅管理器 #[derive(Clone)] @@ -41,17 +42,27 @@ impl SubscriptionManager { /// 创建订阅请求并返回流 pub async fn subscribe_with_request( &self, - transactions: TransactionsFilterMap, + transactions: Option, accounts: Option, commitment: Option, + event_type_filter: Option, ) -> AnyResult<( impl Sink, impl Stream>, )> { + let blocks_meta = if event_type_filter.is_some() + && event_type_filter.as_ref().unwrap().include_block_event() + { + hashmap! { "".to_owned() => SubscribeRequestFilterBlocksMeta {} } + } else if event_type_filter.is_none() { + hashmap! { "".to_owned() => SubscribeRequestFilterBlocksMeta {} } + } else { + hashmap! {} + }; let subscribe_request = SubscribeRequest { accounts: accounts.unwrap_or_default(), - transactions, - blocks_meta: hashmap! { "".to_owned() => SubscribeRequestFilterBlocksMeta {} }, + transactions: transactions.unwrap_or_default(), + blocks_meta, commitment: if let Some(commitment) = commitment { Some(commitment as i32) } else { @@ -69,10 +80,16 @@ impl SubscriptionManager { &self, account: Vec, owner: Vec, + event_type_filter: Option, ) -> Option { if account.len() == 0 && owner.len() == 0 { return None; } + if event_type_filter.is_some() + && !event_type_filter.as_ref().unwrap().include_account_event() + { + return None; + } let mut accounts = HashMap::new(); accounts.insert( "".to_owned(), @@ -92,7 +109,13 @@ impl SubscriptionManager { account_include: Vec, account_exclude: Vec, account_required: Vec, - ) -> TransactionsFilterMap { + event_type_filter: Option, + ) -> Option { + if event_type_filter.is_some() + && !event_type_filter.as_ref().unwrap().include_transaction_event() + { + return None; + } let mut transactions = HashMap::new(); transactions.insert( "client".to_string(), @@ -105,7 +128,7 @@ impl SubscriptionManager { account_required, }, ); - transactions + Some(transactions) } /// 获取配置 diff --git a/src/streaming/yellowstone_grpc.rs b/src/streaming/yellowstone_grpc.rs index 4e09fc0..0ceff0c 100644 --- a/src/streaming/yellowstone_grpc.rs +++ b/src/streaming/yellowstone_grpc.rs @@ -9,6 +9,7 @@ use crate::common::AnyResult; use crate::streaming::common::{ EventBatchProcessor, MetricsManager, PerformanceMetrics, StreamClientConfig, }; +use crate::streaming::event_parser::common::filter::EventTypeFilter; use crate::streaming::event_parser::{Protocol, UnifiedEvent}; use crate::streaming::grpc::{EventPretty, EventProcessor, StreamHandler, SubscriptionManager}; @@ -111,13 +112,26 @@ impl YellowstoneGrpc { self.config.enable_metrics = enabled; } - /// 简化的即时事件订阅（推荐用于简单场景） + /// Simplified immediate event subscription (recommended for simple scenarios) + /// + /// # Parameters + /// * `protocols` - List of protocols to monitor + /// * `bot_wallet` - Optional bot wallet address for filtering related transactions + /// * `transaction_filter` - Transaction filter specifying accounts to include/exclude + /// * `account_filter` - Account filter specifying accounts and owners to monitor + /// * `event_filter` - Optional event filter for further event filtering, no filtering if None + /// * `commitment` - Optional commitment level, defaults to Confirmed + /// * `callback` - Event callback function that receives parsed unified events + /// + /// # Returns + /// Returns `AnyResult<()>`, `Ok(())` on success, error information on failure pub async fn subscribe_events_immediate( &self, protocols: Vec, bot_wallet: Option, transaction_filter: TransactionFilter, account_filter: AccountFilter, + event_type_filter: Option, commitment: Option, callback: F, ) -> AnyResult<()> @@ -133,15 +147,18 @@ impl YellowstoneGrpc { transaction_filter.account_include, transaction_filter.account_exclude, transaction_filter.account_required, + event_type_filter.clone(), + ); + let accounts = self.subscription_manager.subscribe_with_account_request( + account_filter.account, + account_filter.owner, + event_type_filter.clone(), ); - let accounts = self - .subscription_manager - .subscribe_with_account_request(account_filter.account, account_filter.owner); // 订阅事件 let (mut subscribe_tx, mut stream) = self .subscription_manager - .subscribe_with_request(transactions, accounts, commitment) + .subscribe_with_request(transactions, accounts, commitment, event_type_filter.clone()) .await?; // 创建通道，使用配置中的通道大小 @@ -183,6 +200,7 @@ impl YellowstoneGrpc { &callback, bot_wallet, protocols.clone(), + event_type_filter.clone(), ) .await { @@ -195,13 +213,32 @@ impl YellowstoneGrpc { Ok(()) } - /// 高级模式订阅（包含批处理和背压处理） + /// Advanced event subscription with batch processing and backpressure handling + /// + /// # Parameters + /// * `protocols` - List of protocols to monitor + /// * `bot_wallet` - Optional bot wallet address for filtering related transactions + /// * `transaction_filter` - Transaction filter specifying accounts to include/exclude + /// * `account_filter` - Account filter specifying accounts and owners to monitor + /// * `event_filter` - Optional event filter for further event filtering, no filtering if None + /// * `commitment` - Optional commitment level, defaults to Confirmed + /// * `callback` - Event callback function that receives parsed unified events + /// + /// # Features + /// * Batch processing for improved throughput + /// * Backpressure handling to prevent memory overflow + /// * Automatic performance monitoring (if enabled) + /// * Configurable batch size and timeout + /// + /// # Returns + /// Returns `AnyResult<()>`, `Ok(())` on success, error information on failure pub async fn subscribe_events_advanced( &self, protocols: Vec, bot_wallet: Option, transaction_filter: TransactionFilter, account_filter: AccountFilter, + event_type_filter: Option, commitment: Option, callback: F, ) -> AnyResult<()> @@ -217,15 +254,18 @@ impl YellowstoneGrpc { transaction_filter.account_include, transaction_filter.account_exclude, transaction_filter.account_required, + event_type_filter.clone(), + ); + let accounts = self.subscription_manager.subscribe_with_account_request( + account_filter.account, + account_filter.owner, + event_type_filter.clone(), ); - let accounts = self - .subscription_manager - .subscribe_with_account_request(account_filter.account, account_filter.owner); // Subscribe to events let (mut subscribe_tx, mut stream) = self .subscription_manager - .subscribe_with_request(transactions, accounts, commitment) + .subscribe_with_request(transactions, accounts, commitment, event_type_filter.clone()) .await?; // Create channel @@ -280,6 +320,7 @@ impl YellowstoneGrpc { &mut batch_processor, bot_wallet, protocols.clone(), + event_type_filter.clone(), ) .await { @@ -299,6 +340,10 @@ impl YellowstoneGrpc { // 实现 Clone trait 以支持模块间共享 impl Clone for EventProcessor { fn clone(&self) -> Self { - Self { metrics_manager: self.metrics_manager.clone(), config: self.config.clone() } + Self { + metrics_manager: self.metrics_manager.clone(), + config: self.config.clone(), + parser_cache: self.parser_cache.clone(), + } } } diff --git a/src/streaming/yellowstone_sub_system.rs b/src/streaming/yellowstone_sub_system.rs index 2bdbc97..ebb4aa9 100755 --- a/src/streaming/yellowstone_sub_system.rs +++ b/src/streaming/yellowstone_sub_system.rs @@ -45,9 +45,12 @@ impl YellowstoneGrpc { account_include, account_exclude, addrs, + None, ); - let (mut subscribe_tx, mut stream) = - self.subscription_manager.subscribe_with_request(transactions, None, None).await?; + let (mut subscribe_tx, mut stream) = self + .subscription_manager + .subscribe_with_request(transactions, None, None, None) + .await?; let (mut tx, mut rx) = mpsc::channel::(CHANNEL_SIZE); let callback = Box::new(callback);