gavindiaz/solana-streamer
1
0
Fork 0
mirror of https://github.com/0xfnzero/solana-streamer.git synced 2026-07-27 17:37:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: enhance event filtering and monitoring system

Browse Source 
- Add EventTypeFilter for precise event type filtering
- Refactor metrics to track events by type (tx/account/block)
- Add parser caching for improved performance
- Enhance gRPC subscription management
- Update examples and type definitions
This commit is contained in:
ysq
2025-08-16 14:48:51 +08:00
parent 89f4a85a11
commit ea7ecd1d0c
15 changed files with 477 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "solana-streamer-sdk"
version = "0.3.1"
version = "0.3.2"
edition = "2021"
authors = ["William <byteblock6@gmail.com>", "sgxiang <sgxiang@gmail.com>", "wei <1415121722@qq.com>"]
repository = "https://github.com/0xfnzero/solana-streamer"
README.md
+41 -22
View File
@@ -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<dyn std::error::Error>> {
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<dyn std::error::Error>> {
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<dyn UnifiedEvent>) {
|event: Box<dyn UnifiedEvent>| {
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<dyn UnifiedEvent>) {
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<dyn UnifiedEvent>) {
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<dyn UnifiedEvent>) {
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<dyn UnifiedEvent>) {
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<dyn UnifiedEvent>) {
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<dyn UnifiedEvent>) {
}
```
### 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
README_CN.md
+39 -15
View File
@@ -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<dyn std::error::Error>> {
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<dyn std::error::Error>> {
// 监听属于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<dyn UnifiedEvent>) {
|event: Box<dyn UnifiedEvent>| {
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<dyn UnifiedEvent>) {
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<dyn UnifiedEvent>) {
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<dyn UnifiedEvent>) {
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<dyn UnifiedEvent>) {
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<dyn UnifiedEvent>) {
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<dyn UnifiedEvent>) {
}
```
### 事件过滤
库支持灵活的事件过滤以减少处理开销：
```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**: 主要迷因币交易平台
examples/parse_tx_events.rs
+1 -1
View File
@@ -97,7 +97,7 @@ async fn get_single_transaction_details(signature_str: &str) -> Result<()> {
Protocol::RaydiumCpmm,
Protocol::RaydiumAmmV4,
];
let parser: Arc<dyn EventParser> = Arc::new(MutilEventParser::new(protocols));
let parser: Arc<dyn EventParser> = Arc::new(MutilEventParser::new(protocols, None));
let start_time = std::time::Instant::now();
let events = parser
.parse_transaction(
src/main.rs
+10 -2
View File
@@ -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<dyn std::error::Error>> {
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<dyn std::error::Error>> {
None,
transaction_filter,
account_filter,
event_type_filter,
None,
callback,
)
src/streaming/common/metrics.rs
+180 -46
View File
@@ -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);
}
/// 记录慢处理操作
src/streaming/event_parser/common/filter.rs
+24
View File
@@ -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<EventType>,
}
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))
}
}
src/streaming/event_parser/common/mod.rs
+1
View File
@@ -1,5 +1,6 @@
pub mod types;
pub mod utils;
pub mod filter;
/// 自动生成UnifiedEvent trait实现的宏
#[macro_export]
src/streaming/event_parser/common/types.rs
+18
View File
@@ -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 {
src/streaming/event_parser/core/account_event_parser.rs
+9 -3
View File
@@ -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<HashMap<Protocol, Vec<AccountEventParseC
pub struct AccountEventParser {}
impl AccountEventParser {
pub fn configs(protocols: Vec<Protocol>) -> Vec<AccountEventParseConfig> {
pub fn configs(protocols: Vec<Protocol>, event_type_filter: Option<EventTypeFilter>) -> Vec<AccountEventParseConfig> {
let protocols_map = PROTOCOL_CONFIGS_CACHE.get_or_init(|| {
let mut map: HashMap<Protocol, Vec<AccountEventParseConfig>> = 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<AccountEventParseConfig> = 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<Protocol>,
account: AccountPretty,
program_received_time_ms: i64,
event_type_filter: Option<EventTypeFilter>,
) -> Option<Box<dyn UnifiedEvent>> {
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()]
src/streaming/event_parser/protocols/mutil/parser.rs
+10 -3
View File
@@ -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<Protocol>) -> Self {
pub fn new(protocols: Vec<Protocol>, event_type_filter: Option<EventTypeFilter>) -> 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<GenericEventParseConfig> = 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<GenericEventParseConfig> = 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)
src/streaming/grpc/event_processor.rs
+54 -20
View File
@@ -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<Mutex<Option<Arc<dyn EventParser>>>>,
}
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<Protocol>,
event_type_filter: Option<EventTypeFilter>,
) -> Arc<dyn EventParser> {
let mut cache = self.parser_cache.lock().unwrap();
if let Some(cached_parser) = cache.clone() {
return cached_parser.clone();
}
let parser: Arc<dyn EventParser> =
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<Pubkey>,
protocols: Vec<Protocol>,
event_type_filter: Option<EventTypeFilter>,
) -> AnyResult<()>
where
F: Fn(Box<dyn UnifiedEvent>) + 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<dyn EventParser> =
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<F>,
bot_wallet: Option<Pubkey>,
protocols: Vec<Protocol>,
event_type_filter: Option<EventTypeFilter>,
) -> AnyResult<()>
where
F: Fn(Vec<Box<dyn UnifiedEvent>>) + 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<dyn EventParser> =
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);
}
src/streaming/grpc/subscription.rs
+28 -5
View File
@@ -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<TransactionsFilterMap>,
accounts: Option<AccountsFilterMap>,
commitment: Option<CommitmentLevel>,
event_type_filter: Option<EventTypeFilter>,
) -> AnyResult<(
impl Sink<SubscribeRequest, Error = mpsc::SendError>,
impl Stream<Item = Result<SubscribeUpdate, Status>>,
)> {
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<String>,
owner: Vec<String>,
event_type_filter: Option<EventTypeFilter>,
) -> Option<AccountsFilterMap> {
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<String>,
account_exclude: Vec<String>,
account_required: Vec<String>,
) -> TransactionsFilterMap {
event_type_filter: Option<EventTypeFilter>,
) -> Option<TransactionsFilterMap> {
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)
}
/// 获取配置
src/streaming/yellowstone_grpc.rs
+56 -11
View File
@@ -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<F>(
&self,
protocols: Vec<Protocol>,
bot_wallet: Option<Pubkey>,
transaction_filter: TransactionFilter,
account_filter: AccountFilter,
event_type_filter: Option<EventTypeFilter>,
commitment: Option<CommitmentLevel>,
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<F>(
&self,
protocols: Vec<Protocol>,
bot_wallet: Option<Pubkey>,
transaction_filter: TransactionFilter,
account_filter: AccountFilter,
event_type_filter: Option<EventTypeFilter>,
commitment: Option<CommitmentLevel>,
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(),
}
}
}
src/streaming/yellowstone_sub_system.rs
+5 -2
View File
@@ -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::<EventPretty>(CHANNEL_SIZE);
let callback = Box::new(callback);