diff --git a/Cargo.toml b/Cargo.toml index 3d532d5..53b7b96 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "solana-streamer-sdk" -version = "0.3.2" +version = "0.3.3" 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 d0b9871..f1b6a13 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.2" } +solana-streamer-sdk = { path = "./solana-streamer", version = "0.3.3" } ``` ### Use crates.io ```toml # Add to your Cargo.toml -solana-streamer-sdk = "0.3.2" +solana-streamer-sdk = "0.3.3" ``` ## Usage Examples @@ -119,7 +119,7 @@ use solana_streamer_sdk::{ Protocol, UnifiedEvent, }, grpc::ClientConfig, - shred_stream::ShredClientConfig, + shred::StreamClientConfig, yellowstone_grpc::{AccountFilter, TransactionFilter}, ShredStreamGrpc, YellowstoneGrpc, }, @@ -213,7 +213,7 @@ async fn test_shreds() -> Result<(), Box> { println!("Subscribing to ShredStream events..."); // Create low-latency configuration - let mut config = ShredClientConfig::low_latency(); + let mut config = StreamClientConfig::low_latency(); // Enable performance monitoring, has performance overhead, disabled by default config.enable_metrics = true; let shred_stream = @@ -229,8 +229,15 @@ async fn test_shreds() -> Result<(), Box> { Protocol::RaydiumAmmV4, ]; + // 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!("Listening for events, press Ctrl+C to stop..."); - shred_stream.shredstream_subscribe(protocols, None, callback).await?; + shred_stream.shredstream_subscribe(protocols, None, event_type_filter, callback).await?; Ok(()) } @@ -385,7 +392,9 @@ fn create_event_callback() -> impl Fn(Box) { ### Event Filtering -The library supports flexible event filtering to reduce processing overhead: +The library supports flexible event filtering to reduce processing overhead and improve performance: + +#### Basic Filtering ```rust use solana_streamer_sdk::streaming::event_parser::common::{filter::EventTypeFilter, EventType}; @@ -399,6 +408,47 @@ let event_type_filter = Some(EventTypeFilter { }); ``` +#### Performance Impact + +Event filtering can provide significant performance improvements: +- **60-80% reduction** in unnecessary event processing +- **Lower memory usage** by filtering out irrelevant events +- **Reduced network bandwidth** in distributed setups +- **Better focus** on events that matter to your application + +#### Filtering Examples by Use Case + +**Trading Bot (Focus on Trade Events)** +```rust +let event_type_filter = Some(EventTypeFilter { + include: vec![ + EventType::PumpSwapBuy, + EventType::PumpSwapSell, + EventType::PumpFunTrade, + EventType::RaydiumCpmmSwap, + EventType::RaydiumClmmSwap, + EventType::RaydiumAmmV4Swap, + ...... + ] +}); +``` + +**Pool Monitoring (Focus on Liquidity Events)** +```rust +let event_type_filter = Some(EventTypeFilter { + include: vec![ + EventType::PumpSwapCreatePool, + EventType::PumpSwapDeposit, + EventType::PumpSwapWithdraw, + EventType::RaydiumCpmmInitialize, + EventType::RaydiumCpmmDeposit, + EventType::RaydiumCpmmWithdraw, + EventType::RaydiumClmmCreatePool, + ...... + ] +}); +``` + ## Supported Protocols - **PumpFun**: Primary meme coin trading platform diff --git a/README_CN.md b/README_CN.md index 1365ec3..c7e9339 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.2" } +solana-streamer-sdk = { path = "./solana-streamer", version = "0.3.3" } ``` ### 使用 crates.io ```toml # 添加到您的 Cargo.toml -solana-streamer-sdk = "0.3.2" +solana-streamer-sdk = "0.3.3" ``` ## 使用示例 @@ -119,7 +119,7 @@ use solana_streamer_sdk::{ Protocol, UnifiedEvent, }, grpc::ClientConfig, - shred_stream::ShredClientConfig, + shred::StreamClientConfig, yellowstone_grpc::{AccountFilter, TransactionFilter}, ShredStreamGrpc, YellowstoneGrpc, }, @@ -213,7 +213,7 @@ async fn test_shreds() -> Result<(), Box> { println!("Subscribing to ShredStream events..."); // 创建低延迟配置 - let mut config = ShredClientConfig::low_latency(); + let mut config = StreamClientConfig::low_latency(); // 启用性能监控, 有性能损耗, 默认关闭 config.enable_metrics = true; let shred_stream = @@ -229,8 +229,15 @@ async fn test_shreds() -> Result<(), Box> { Protocol::RaydiumAmmV4, ]; + // 事件过滤 + // 不进行事件过滤，包含所有事件 + let event_type_filter = None; + // 只包含PumpSwapBuy事件、PumpSwapSell事件 + // let event_type_filter = + // EventTypeFilter { include: vec![EventType::PumpSwapBuy, EventType::PumpSwapSell] }; + println!("Listening for events, press Ctrl+C to stop..."); - shred_stream.shredstream_subscribe(protocols, None, callback).await?; + shred_stream.shredstream_subscribe(protocols, None, event_type_filter, callback).await?; Ok(()) } @@ -385,7 +392,9 @@ fn create_event_callback() -> impl Fn(Box) { ### 事件过滤 -库支持灵活的事件过滤以减少处理开销： +库支持灵活的事件过滤以减少处理开销并提升性能： + +#### 基础过滤 ```rust use solana_streamer_sdk::streaming::event_parser::common::{filter::EventTypeFilter, EventType}; @@ -399,6 +408,47 @@ let event_type_filter = Some(EventTypeFilter { }); ``` +#### 性能影响 + +事件过滤可以带来显著的性能提升： +- **减少 60-80%** 的不必要事件处理 +- **降低内存使用** 通过过滤掉无关事件 +- **减少网络带宽** 在分布式环境中 +- **更好的专注性** 只处理对应用有意义的事件 + +#### 按使用场景的过滤示例 + +**交易机器人（专注交易事件）** +```rust +let event_type_filter = Some(EventTypeFilter { + include: vec![ + EventType::PumpSwapBuy, + EventType::PumpSwapSell, + EventType::PumpFunTrade, + EventType::RaydiumCpmmSwap, + EventType::RaydiumClmmSwap, + EventType::RaydiumAmmV4Swap, + ..... + ] +}); +``` + +**池监控（专注流动性事件）** +```rust +let event_type_filter = Some(EventTypeFilter { + include: vec![ + EventType::PumpSwapCreatePool, + EventType::PumpSwapDeposit, + EventType::PumpSwapWithdraw, + EventType::RaydiumCpmmInitialize, + EventType::RaydiumCpmmDeposit, + EventType::RaydiumCpmmWithdraw, + EventType::RaydiumClmmCreatePool, + ...... + ] +}); +``` + ## 支持的协议 - **PumpFun**: 主要迷因币交易平台 diff --git a/src/main.rs b/src/main.rs index 2620a7b..5969861 100755 --- a/src/main.rs +++ b/src/main.rs @@ -43,7 +43,7 @@ use solana_streamer_sdk::{ Protocol, UnifiedEvent, }, grpc::ClientConfig, - shred_stream::ShredClientConfig, + shred::StreamClientConfig, yellowstone_grpc::{AccountFilter, TransactionFilter}, ShredStreamGrpc, YellowstoneGrpc, }, @@ -137,7 +137,7 @@ async fn test_shreds() -> Result<(), Box> { println!("Subscribing to ShredStream events..."); // Create low-latency configuration - let mut config = ShredClientConfig::low_latency(); + let mut config = StreamClientConfig::low_latency(); // Enable performance monitoring, has performance overhead, disabled by default config.enable_metrics = true; let shred_stream = @@ -153,8 +153,15 @@ async fn test_shreds() -> Result<(), Box> { Protocol::RaydiumAmmV4, ]; + // 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!("Listening for events, press Ctrl+C to stop..."); - shred_stream.shredstream_subscribe(protocols, None, callback).await?; + shred_stream.shredstream_subscribe(protocols, None, event_type_filter, callback).await?; Ok(()) } diff --git a/src/streaming/event_parser/protocols/bonk/events.rs b/src/streaming/event_parser/protocols/bonk/events.rs index b93f2dc..eaf6b05 100755 --- a/src/streaming/event_parser/protocols/bonk/events.rs +++ b/src/streaming/event_parser/protocols/bonk/events.rs @@ -49,6 +49,10 @@ pub struct BonkTradeEvent { #[borsh(skip)] pub quote_token_mint: Pubkey, #[borsh(skip)] + pub base_token_program: Pubkey, + #[borsh(skip)] + pub quote_token_program: Pubkey, + #[borsh(skip)] pub is_dev_create_token_trade: bool, #[borsh(skip)] pub is_bot: bool, diff --git a/src/streaming/event_parser/protocols/bonk/parser.rs b/src/streaming/event_parser/protocols/bonk/parser.rs index 8c4d6b7..4eaf8ce 100755 --- a/src/streaming/event_parser/protocols/bonk/parser.rs +++ b/src/streaming/event_parser/protocols/bonk/parser.rs @@ -173,6 +173,8 @@ impl BonkEventParser { quote_vault: accounts[8], base_token_mint: accounts[9], quote_token_mint: accounts[10], + base_token_program: accounts[11], + quote_token_program: accounts[12], trade_direction: TradeDirection::Buy, ..Default::default() })) @@ -207,6 +209,8 @@ impl BonkEventParser { quote_vault: accounts[8], base_token_mint: accounts[9], quote_token_mint: accounts[10], + base_token_program: accounts[11], + quote_token_program: accounts[12], trade_direction: TradeDirection::Buy, ..Default::default() })) @@ -241,6 +245,8 @@ impl BonkEventParser { quote_vault: accounts[8], base_token_mint: accounts[9], quote_token_mint: accounts[10], + base_token_program: accounts[11], + quote_token_program: accounts[12], trade_direction: TradeDirection::Sell, ..Default::default() })) @@ -275,6 +281,8 @@ impl BonkEventParser { quote_vault: accounts[8], base_token_mint: accounts[9], quote_token_mint: accounts[10], + base_token_program: accounts[11], + quote_token_program: accounts[12], trade_direction: TradeDirection::Sell, ..Default::default() })) diff --git a/src/streaming/mod.rs b/src/streaming/mod.rs index 9fd07b9..8747e2a 100755 --- a/src/streaming/mod.rs +++ b/src/streaming/mod.rs @@ -1,10 +1,11 @@ pub mod common; pub mod event_parser; pub mod grpc; +pub mod shred; pub mod shred_stream; pub mod yellowstone_grpc; pub mod yellowstone_sub_system; -pub use shred_stream::ShredStreamGrpc; +pub use shred::ShredStreamGrpc; pub use yellowstone_grpc::YellowstoneGrpc; pub use yellowstone_sub_system::{SystemEvent, TransferInfo}; diff --git a/src/streaming/shred/connection.rs b/src/streaming/shred/connection.rs new file mode 100644 index 0000000..ea08b39 --- /dev/null +++ b/src/streaming/shred/connection.rs @@ -0,0 +1,85 @@ +use std::sync::Arc; +use tokio::sync::Mutex; +use tonic::transport::Channel; + +use crate::common::AnyResult; +use crate::streaming::common::{ + MetricsManager, PerformanceMetrics, StreamClientConfig, +}; +use crate::protos::shredstream::shredstream_proxy_client::ShredstreamProxyClient; + +/// ShredStream gRPC 客户端 +#[derive(Clone)] +pub struct ShredStreamGrpc { + pub shredstream_client: Arc>, + pub config: StreamClientConfig, + pub metrics: Arc>, + pub metrics_manager: MetricsManager, +} + +impl ShredStreamGrpc { + /// 创建客户端，使用默认配置 + pub async fn new(endpoint: String) -> AnyResult { + Self::new_with_config(endpoint, StreamClientConfig::default()).await + } + + /// 创建客户端，使用自定义配置 + pub async fn new_with_config(endpoint: String, config: StreamClientConfig) -> AnyResult { + let shredstream_client = ShredstreamProxyClient::connect(endpoint.clone()).await?; + let metrics = Arc::new(Mutex::new(PerformanceMetrics::new())); + let config_arc = Arc::new(config.clone()); + + let metrics_manager = MetricsManager::new( + metrics.clone(), + config_arc, + "ShredStream".to_string() + ); + + Ok(Self { + shredstream_client: Arc::new(shredstream_client), + config, + metrics, + metrics_manager, + }) + } + + /// 创建高性能客户端（适合高并发场景） + pub async fn new_high_performance(endpoint: String) -> AnyResult { + Self::new_with_config(endpoint, StreamClientConfig::high_performance()).await + } + + /// 创建低延迟客户端（适合实时场景） + pub async fn new_low_latency(endpoint: String) -> AnyResult { + Self::new_with_config(endpoint, StreamClientConfig::low_latency()).await + } + + /// 获取当前配置 + pub fn get_config(&self) -> &StreamClientConfig { + &self.config + } + + /// 更新配置 + pub fn update_config(&mut self, config: StreamClientConfig) { + self.config = config; + } + + /// 获取性能指标 + pub async fn get_metrics(&self) -> PerformanceMetrics { + self.metrics_manager.get_metrics().await + } + + /// 启用或禁用性能监控 + pub fn set_enable_metrics(&mut self, enabled: bool) { + self.config.enable_metrics = enabled; + } + + /// 打印性能指标 + pub async fn print_metrics(&self) { + self.metrics_manager.print_metrics().await; + } + + /// 启动自动性能监控任务 + pub async fn start_auto_metrics_monitoring(&self) { + self.metrics_manager.start_auto_monitoring().await; + } +} diff --git a/src/streaming/shred/event_processor.rs b/src/streaming/shred/event_processor.rs new file mode 100644 index 0000000..c72895a --- /dev/null +++ b/src/streaming/shred/event_processor.rs @@ -0,0 +1,170 @@ +use solana_sdk::pubkey::Pubkey; +use std::sync::{Arc, Mutex}; + +use crate::common::AnyResult; +use crate::streaming::common::{ + EventBatchProcessor, MetricsEventType, MetricsManager, StreamClientConfig, +}; +use crate::streaming::event_parser::common::filter::EventTypeFilter; +use crate::streaming::event_parser::protocols::MutilEventParser; +use crate::streaming::event_parser::{EventParser, Protocol, UnifiedEvent}; +use crate::streaming::shred::TransactionWithSlot; + +/// ShredStream 事件处理器 +pub struct ShredEventProcessor { + pub(crate) metrics_manager: MetricsManager, + pub(crate) config: StreamClientConfig, + pub(crate) parser_cache: Arc>>>, +} + +impl ShredEventProcessor { + /// 创建新的事件处理器 + pub fn new(metrics_manager: MetricsManager, config: StreamClientConfig) -> Self { + 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 + } + + /// 即时处理单个交易 + pub async fn process_transaction_immediate( + &self, + transaction_with_slot: TransactionWithSlot, + protocols: Vec, + bot_wallet: Option, + event_type_filter: Option, + callback: &F, + ) -> AnyResult<()> + where + F: Fn(Box) + Send + Sync, + { + let start_time = std::time::Instant::now(); + self.metrics_manager.add_tx_process_count().await; + let program_received_time_ms = chrono::Utc::now().timestamp_millis(); + let slot = transaction_with_slot.slot; + let versioned_tx = transaction_with_slot.transaction; + let signature = versioned_tx.signatures[0]; + + // 获取缓存的解析器 + let parser = self.get_or_create_parser(protocols, event_type_filter); + + let all_events = parser + .parse_versioned_transaction( + &versioned_tx, + &signature.to_string(), + Some(slot), + None, + program_received_time_ms, + bot_wallet, + ) + .await + .unwrap_or_else(|_e| vec![]); + + // 保存事件数量用于日志记录 + let event_count = all_events.len(); + + // 即时处理事件 + for event in all_events { + callback(event); + } + + // 更新性能指标 + let processing_time = start_time.elapsed(); + let processing_time_ms = processing_time.as_millis() as f64; + + // 实际调用性能指标更新 + self.update_metrics(event_count as u64, processing_time_ms).await; + + // 记录慢处理操作 + self.metrics_manager.log_slow_processing(processing_time_ms, event_count); + + Ok(()) + } + + /// 批处理模式处理单个交易 + pub async fn process_transaction_with_batch( + &self, + transaction_with_slot: TransactionWithSlot, + protocols: Vec, + bot_wallet: Option, + batch_processor: &mut EventBatchProcessor, + event_type_filter: Option, + ) -> AnyResult<()> + where + F: FnMut(Vec>) + Send + Sync + 'static, + { + let start_time = std::time::Instant::now(); + self.metrics_manager.add_tx_process_count().await; + let program_received_time_ms = chrono::Utc::now().timestamp_millis(); + let slot = transaction_with_slot.slot; + let versioned_tx = transaction_with_slot.transaction; + let signature = versioned_tx.signatures[0]; + + // 获取缓存的解析器 + let parser = self.get_or_create_parser(protocols, event_type_filter); + + let all_events = parser + .parse_versioned_transaction( + &versioned_tx, + &signature.to_string(), + Some(slot), + None, + program_received_time_ms, + bot_wallet, + ) + .await + .unwrap_or_else(|_e| vec![]); + + // 保存事件数量用于日志记录 + let event_count = all_events.len(); + + // 使用批处理器处理事件 + for event in all_events { + batch_processor.add_event(event); + } + + // 更新性能指标 + let processing_time = start_time.elapsed(); + let processing_time_ms = processing_time.as_millis() as f64; + + // 实际调用性能指标更新 + self.update_metrics(event_count as u64, processing_time_ms).await; + + // 记录慢处理操作 + self.metrics_manager.log_slow_processing(processing_time_ms, event_count); + + Ok(()) + } + + /// 更新性能指标 + async fn update_metrics(&self, events_processed: u64, processing_time_ms: f64) { + // 使用统一的指标管理器，这里假设 ShredStream 主要处理交易事件 + self.metrics_manager + .update_metrics(MetricsEventType::Tx, events_processed, processing_time_ms) + .await; + } +} + +// 实现 Clone trait 以支持模块间共享 +impl Clone for ShredEventProcessor { + fn clone(&self) -> Self { + Self { + metrics_manager: self.metrics_manager.clone(), + config: self.config.clone(), + parser_cache: self.parser_cache.clone(), + } + } +} diff --git a/src/streaming/shred/mod.rs b/src/streaming/shred/mod.rs new file mode 100644 index 0000000..5ba7daf --- /dev/null +++ b/src/streaming/shred/mod.rs @@ -0,0 +1,17 @@ +// ShredStream 相关模块 +pub mod connection; +pub mod types; +pub mod stream_handler; +pub mod event_processor; + +// 重新导出主要类型 +pub use connection::*; +pub use types::*; +pub use stream_handler::*; +pub use event_processor::*; + +// 从公用模块重新导出 +pub use crate::streaming::common::{ + BackpressureConfig, BackpressureStrategy, BatchConfig, ConnectionConfig, EventBatchProcessor, + MetricsEventType, MetricsManager, PerformanceMetrics, StreamClientConfig, +}; diff --git a/src/streaming/shred/stream_handler.rs b/src/streaming/shred/stream_handler.rs new file mode 100644 index 0000000..b5de4c2 --- /dev/null +++ b/src/streaming/shred/stream_handler.rs @@ -0,0 +1,116 @@ +use futures::{channel::mpsc, StreamExt}; +use log::error; +use solana_entry::entry::Entry; +use tokio::task::JoinHandle; + +use crate::common::AnyResult; +use crate::protos::shredstream::{ + shredstream_proxy_client::ShredstreamProxyClient, SubscribeEntriesRequest, +}; +use crate::streaming::shred::TransactionWithSlot; + +/// ShredStream 流处理器 +pub struct ShredStreamHandler; + +impl ShredStreamHandler { + /// 启动 ShredStream 流处理任务 + /// + /// # 参数 + /// * `client` - ShredStream 客户端 + /// * `tx` - 事务发送通道 + /// * `channel_size` - 通道缓冲区大小 + /// + /// # 返回值 + /// 返回 ShredStream 流处理任务句柄和事务接收通道 + pub async fn start_stream_processing( + mut client: ShredstreamProxyClient, + channel_size: usize, + ) -> AnyResult<(JoinHandle<()>, mpsc::Receiver)> { + let request = tonic::Request::new(SubscribeEntriesRequest {}); + let stream = client.subscribe_entries(request).await?.into_inner(); + let (tx, rx) = mpsc::channel::(channel_size); + + let stream_task = tokio::spawn(Self::process_stream_messages(stream, tx)); + + Ok((stream_task, rx)) + } + + /// 处理流消息 + /// + /// # 参数 + /// * `stream` - ShredStream 数据流 + /// * `tx` - 事务发送通道 + async fn process_stream_messages( + mut stream: tonic::codec::Streaming, + mut tx: mpsc::Sender, + ) { + while let Some(message) = stream.next().await { + match message { + Ok(msg) => { + if let Err(e) = Self::handle_stream_message(msg, &mut tx).await { + error!("Error handling stream message: {e:?}"); + continue; + } + } + Err(error) => { + error!("Stream error: {error:?}"); + break; + } + } + } + } + + /// 处理单个流消息 + /// + /// # 参数 + /// * `msg` - ShredStream 消息 + /// * `tx` - 事务发送通道 + async fn handle_stream_message( + msg: crate::protos::shredstream::Entry, + tx: &mut mpsc::Sender, + ) -> AnyResult<()> { + if let Ok(entries) = bincode::deserialize::>(&msg.entries) { + for entry in entries { + for transaction in entry.transactions { + let transaction_with_slot = + TransactionWithSlot::new(transaction.clone(), msg.slot); + + if let Err(e) = tx.try_send(transaction_with_slot) { + // 如果通道满了，记录警告但不中断处理 + if e.is_full() { + log::warn!("Transaction channel is full, dropping transaction"); + } else { + // 通道已关闭，返回错误 + return Err(e.into()); + } + } + } + } + } + Ok(()) + } + + /// 启动事务处理任务 + /// + /// # 参数 + /// * `rx` - 事务接收通道 + /// * `processor` - 事务处理器 + pub fn start_transaction_processing( + mut rx: mpsc::Receiver, + processor: F, + ) -> JoinHandle<()> + where + F: Fn(TransactionWithSlot) -> Result<(), Box> + + Send + + Sync + + 'static, + { + tokio::spawn(async move { + while let Some(transaction_with_slot) = rx.next().await { + if let Err(e) = processor(transaction_with_slot) { + error!("Error processing transaction: {e:?}"); + } + } + }) + } +} diff --git a/src/streaming/shred/types.rs b/src/streaming/shred/types.rs new file mode 100644 index 0000000..888622d --- /dev/null +++ b/src/streaming/shred/types.rs @@ -0,0 +1,20 @@ +use solana_sdk::transaction::VersionedTransaction; + +/// 携带槽位信息的交易 +#[derive(Debug, Clone)] +pub struct TransactionWithSlot { + pub transaction: VersionedTransaction, + pub slot: u64, +} + +impl TransactionWithSlot { + /// 创建新的带槽位的交易 + pub fn new(transaction: VersionedTransaction, slot: u64) -> Self { + Self { transaction, slot } + } + + /// 获取交易签名 + pub fn signature(&self) -> String { + self.transaction.signatures[0].to_string() + } +} diff --git a/src/streaming/shred_stream.rs b/src/streaming/shred_stream.rs index 3cf52b4..b121293 100755 --- a/src/streaming/shred_stream.rs +++ b/src/streaming/shred_stream.rs @@ -1,335 +1,20 @@ -use std::sync::Arc; -use tokio::sync::Mutex; - -use futures::{channel::mpsc, StreamExt}; -use solana_entry::entry::Entry; -use tonic::transport::Channel; - -use log::error; -use solana_sdk::transaction::VersionedTransaction; - -use crate::common::AnyResult; -use crate::streaming::event_parser::{EventParserFactory, Protocol, UnifiedEvent}; - -use crate::protos::shredstream::shredstream_proxy_client::ShredstreamProxyClient; -use crate::protos::shredstream::SubscribeEntriesRequest; use solana_sdk::pubkey::Pubkey; -// -------------- TODO 待重构 -------------- -// -// -------------- -------------- +use crate::common::AnyResult; +use crate::streaming::common::EventBatchProcessor; +use crate::streaming::event_parser::common::filter::EventTypeFilter; +use crate::streaming::event_parser::{Protocol, UnifiedEvent}; +use crate::streaming::shred::{ShredEventProcessor, ShredStreamHandler, TransactionWithSlot}; -// 默认配置常量 -const DEFAULT_CHANNEL_SIZE: usize = 1000; -const DEFAULT_BATCH_SIZE: usize = 100; -const DEFAULT_BATCH_TIMEOUT_MS: u64 = 5; - -/// ShredStream批处理配置 -#[derive(Debug, Clone)] -pub struct ShredBatchConfig { - /// 批处理大小（默认：100） - pub batch_size: usize, - /// 批处理超时时间（毫秒，默认：10ms） - pub batch_timeout_ms: u64, - /// 是否启用批处理（默认：true） - pub enabled: bool, -} - -impl Default for ShredBatchConfig { - fn default() -> Self { - Self { - batch_size: DEFAULT_BATCH_SIZE, - batch_timeout_ms: DEFAULT_BATCH_TIMEOUT_MS, - enabled: true, - } - } -} - -/// ShredStream背压配置 -#[derive(Debug, Clone)] -pub struct ShredBackpressureConfig { - /// 通道大小（默认：10000） - pub channel_size: usize, -} - -impl Default for ShredBackpressureConfig { - fn default() -> Self { - Self { channel_size: DEFAULT_CHANNEL_SIZE } - } -} - -/// ShredStream完整配置 -#[derive(Debug, Clone)] -pub struct ShredClientConfig { - /// 批处理配置 - pub batch: ShredBatchConfig, - /// 背压配置 - pub backpressure: ShredBackpressureConfig, - /// 是否启用性能监控（默认：false） - pub enable_metrics: bool, -} - -impl Default for ShredClientConfig { - fn default() -> Self { - Self { - batch: ShredBatchConfig::default(), - backpressure: ShredBackpressureConfig::default(), - enable_metrics: false, - } - } -} - -impl ShredClientConfig { - /// 创建高性能配置（适合高并发场景） - pub fn high_performance() -> Self { - Self { - batch: ShredBatchConfig { batch_size: 200, batch_timeout_ms: 5, enabled: true }, - backpressure: ShredBackpressureConfig { channel_size: 20000 }, - enable_metrics: true, - } - } - - /// 创建低延迟配置（适合实时场景） - pub fn low_latency() -> Self { - Self { - batch: ShredBatchConfig { - batch_size: 10, - batch_timeout_ms: 1, - enabled: false, // 禁用批处理，即时处理 - }, - backpressure: ShredBackpressureConfig { channel_size: 1000 }, - enable_metrics: false, - } - } -} - -/// ShredStream性能监控指标 -#[derive(Debug, Clone)] -pub struct ShredPerformanceMetrics { - pub events_processed: u64, - pub events_per_second: f64, - 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 ShredPerformanceMetrics { - fn default() -> Self { - Self::new() - } -} - -impl ShredPerformanceMetrics { - pub fn new() -> Self { - let now = std::time::Instant::now(); - Self { - events_processed: 0, - events_per_second: 0.0, - 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, - } - } -} - -#[derive(Clone)] -pub struct ShredStreamGrpc { - shredstream_client: Arc>, - config: ShredClientConfig, - metrics: Arc>, -} - -struct TransactionWithSlot { - transaction: VersionedTransaction, - slot: u64, -} - -/// ShredStream批处理器 -pub struct ShredBatchProcessor -where - F: FnMut(Vec>) + Send + Sync + 'static, -{ - callback: F, - batch: Vec>, - batch_size: usize, - timeout_ms: u64, - last_flush_time: std::time::Instant, -} - -impl ShredBatchProcessor -where - F: FnMut(Vec>) + Send + Sync + 'static, -{ - pub fn new(callback: F, batch_size: usize, timeout_ms: u64) -> Self { - Self { - callback, - batch: Vec::with_capacity(batch_size), - batch_size, - timeout_ms, - last_flush_time: std::time::Instant::now(), - } - } - - pub fn add_event(&mut self, event: Box) { - self.batch.push(event); - - // 检查是否需要刷新批次 - if self.batch.len() >= self.batch_size || self.should_flush_by_timeout() { - self.flush(); - } - } - - pub fn flush(&mut self) { - if !self.batch.is_empty() { - let events = std::mem::replace(&mut self.batch, Vec::with_capacity(self.batch_size)); - (self.callback)(events); - self.last_flush_time = std::time::Instant::now(); - } - } - - fn should_flush_by_timeout(&self) -> bool { - self.last_flush_time.elapsed().as_millis() >= self.timeout_ms as u128 - } -} +use super::ShredStreamGrpc; impl ShredStreamGrpc { - /// 创建客户端，使用默认配置 - pub async fn new(endpoint: String) -> AnyResult { - Self::new_with_config(endpoint, ShredClientConfig::default()).await - } - - /// 创建客户端，使用自定义配置 - pub async fn new_with_config(endpoint: String, config: ShredClientConfig) -> AnyResult { - let shredstream_client = ShredstreamProxyClient::connect(endpoint.clone()).await?; - Ok(Self { - shredstream_client: Arc::new(shredstream_client), - config, - metrics: Arc::new(Mutex::new(ShredPerformanceMetrics::new())), - }) - } - - /// 创建高性能客户端（适合高并发场景） - pub async fn new_high_performance(endpoint: String) -> AnyResult { - Self::new_with_config(endpoint, ShredClientConfig::high_performance()).await - } - - /// 创建低延迟客户端（适合实时场景） - pub async fn new_low_latency(endpoint: String) -> AnyResult { - Self::new_with_config(endpoint, ShredClientConfig::low_latency()).await - } - - /// 获取当前配置 - pub fn get_config(&self) -> &ShredClientConfig { - &self.config - } - - /// 更新配置 - pub fn update_config(&mut self, config: ShredClientConfig) { - self.config = config; - } - - /// 获取性能指标 - pub async fn get_metrics(&self) -> ShredPerformanceMetrics { - let metrics = self.metrics.lock().await; - metrics.clone() - } - - /// 启用或禁用性能监控 - pub fn set_enable_metrics(&mut self, enabled: bool) { - self.config.enable_metrics = enabled; - } - - /// 打印性能指标 - pub async fn print_metrics(&self) { - let metrics = self.get_metrics().await; - println!("📊 ShredStream Performance Metrics:"); - 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!("---"); - } - - /// 启动自动性能监控任务 - pub async fn start_auto_metrics_monitoring(&self) { - // 检查是否启用性能监控 - if !self.config.enable_metrics { - return; // 如果未启用性能监控，不启动监控任务 - } - - let grpc_clone = self.clone(); - tokio::spawn(async move { - let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(10)); - loop { - interval.tick().await; - grpc_clone.print_metrics().await; - } - }); - } - - /// 更新性能指标 - async fn update_metrics(&self, events_processed: u64, processing_time_ms: f64) { - // 检查是否启用性能监控 - if !self.config.enable_metrics { - return; // 如果未启用性能监控，直接返回 - } - - let mut metrics = self.metrics.lock().await; - let now = std::time::Instant::now(); - - 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 { - metrics.min_processing_time_ms = processing_time_ms; - } - if processing_time_ms > metrics.max_processing_time_ms { - 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; - } - - // 基于时间窗口计算每秒处理事件数（5秒窗口） - let window_duration = std::time::Duration::from_secs(5); - 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; - } else { - // 如果窗口内没有事件，保持之前的速率或设为0 - metrics.events_per_second = 0.0; - } - - // 重置窗口 - metrics.events_in_window = 0; - metrics.window_start_time = now; - } else { - // 如果窗口还没满，不更新 events_per_second，保持之前的计算值 - // 这样可以避免因为单次批处理时间波动导致的指标跳跃 - } - - } - /// 订阅ShredStream事件（支持批处理和即时处理） pub async fn shredstream_subscribe( &self, protocols: Vec, bot_wallet: Option, + event_type_filter: Option, callback: F, ) -> AnyResult<()> where @@ -337,37 +22,41 @@ impl ShredStreamGrpc { { // 启动自动性能监控（如果启用） if self.config.enable_metrics { - self.start_auto_metrics_monitoring().await; + self.metrics_manager.start_auto_monitoring().await; } - let request = tonic::Request::new(SubscribeEntriesRequest {}); - let mut client = (*self.shredstream_client).clone(); - let stream = client.subscribe_entries(request).await?.into_inner(); - let (tx, rx) = mpsc::channel::(self.config.backpressure.channel_size); + // 启动流处理 + let client = (*self.shredstream_client).clone(); + let (_stream_task, rx) = ShredStreamHandler::start_stream_processing( + client, + self.config.backpressure.channel_size, + ) + .await?; // 根据配置选择处理模式 if self.config.batch.enabled { // 批处理模式 - self.process_with_batch(stream, tx, rx, protocols, bot_wallet, callback).await + self.process_with_batch(rx, protocols, bot_wallet, event_type_filter, callback).await } else { // 即时处理模式 - self.process_immediate(stream, tx, rx, protocols, bot_wallet, callback).await + self.process_immediate(rx, protocols, bot_wallet, event_type_filter, callback).await } } /// 批处理模式 async fn process_with_batch( &self, - mut stream: tonic::codec::Streaming, - mut tx: mpsc::Sender, - mut rx: mpsc::Receiver, + mut rx: futures::channel::mpsc::Receiver, protocols: Vec, bot_wallet: Option, + event_type_filter: Option, callback: F, ) -> AnyResult<()> where F: Fn(Box) + Send + Sync + 'static, { + use futures::StreamExt; + // 创建批处理器，将单个事件回调转换为批量回调 let batch_callback = move |events: Vec>| { for event in events { @@ -375,47 +64,28 @@ impl ShredStreamGrpc { } }; - let mut batch_processor = ShredBatchProcessor::new( + let mut batch_processor = EventBatchProcessor::new( batch_callback, self.config.batch.batch_size, self.config.batch.batch_timeout_ms, ); - tokio::spawn(async move { - while let Some(message) = stream.next().await { - match message { - Ok(msg) => { - if let Ok(entries) = bincode::deserialize::>(&msg.entries) { - for entry in entries { - for transaction in entry.transactions { - let _ = tx.try_send(TransactionWithSlot { - transaction: transaction.clone(), - slot: msg.slot, - }); - } - } - } - } - Err(error) => { - error!("Stream error: {error:?}"); - break; - } - } - } - }); + // 创建事件处理器 + let event_processor = + ShredEventProcessor::new(self.metrics_manager.clone(), self.config.clone()); - let self_clone = self.clone(); while let Some(transaction_with_slot) = rx.next().await { - if let Err(e) = self_clone + if let Err(e) = event_processor .process_transaction_with_batch( transaction_with_slot, protocols.clone(), bot_wallet, &mut batch_processor, + event_type_filter.clone(), ) .await { - error!("Error processing transaction: {e:?}"); + log::error!("Error processing transaction: {e:?}"); } } @@ -428,179 +98,36 @@ impl ShredStreamGrpc { /// 即时处理模式 async fn process_immediate( &self, - mut stream: tonic::codec::Streaming, - mut tx: mpsc::Sender, - mut rx: mpsc::Receiver, + mut rx: futures::channel::mpsc::Receiver, protocols: Vec, bot_wallet: Option, + event_type_filter: Option, callback: F, ) -> AnyResult<()> where F: Fn(Box) + Send + Sync + 'static, { - tokio::spawn(async move { - while let Some(message) = stream.next().await { - match message { - Ok(msg) => { - if let Ok(entries) = bincode::deserialize::>(&msg.entries) { - for entry in entries { - for transaction in entry.transactions { - let _ = tx.try_send(TransactionWithSlot { - transaction: transaction.clone(), - slot: msg.slot, - }); - } - } - } - } - Err(error) => { - error!("Stream error: {error:?}"); - break; - } - } - } - }); + use futures::StreamExt; + + // 创建事件处理器 + let event_processor = + ShredEventProcessor::new(self.metrics_manager.clone(), self.config.clone()); - let self_clone = self.clone(); while let Some(transaction_with_slot) = rx.next().await { - if let Err(e) = self_clone + if let Err(e) = event_processor .process_transaction_immediate( transaction_with_slot, protocols.clone(), bot_wallet, + event_type_filter.clone(), &callback, ) .await { - error!("Error processing transaction: {e:?}"); + log::error!("Error processing transaction: {e:?}"); } } Ok(()) } - - /// 即时处理单个交易 - async fn process_transaction_immediate( - &self, - transaction_with_slot: TransactionWithSlot, - protocols: Vec, - bot_wallet: Option, - callback: &F, - ) -> AnyResult<()> - where - F: Fn(Box) + Send + Sync, - { - let start_time = std::time::Instant::now(); - let program_received_time_ms = chrono::Utc::now().timestamp_millis(); - let slot = transaction_with_slot.slot; - let versioned_tx = transaction_with_slot.transaction; - let signature = versioned_tx.signatures[0]; - - // 预分配向量容量 - let mut all_events = Vec::with_capacity(protocols.len() * 2); - - for protocol in protocols { - let parser = EventParserFactory::create_parser(protocol.clone()); - let events = parser - .parse_versioned_transaction( - &versioned_tx, - &signature.to_string(), - Some(slot), - None, - program_received_time_ms, - bot_wallet, - ) - .await - .unwrap_or_else(|_e| vec![]); - all_events.extend(events); - } - - // 保存事件数量用于日志记录 - let event_count = all_events.len(); - - // 即时处理事件 - for event in all_events { - callback(event); - } - - // 更新性能指标 - let processing_time = start_time.elapsed(); - let processing_time_ms = processing_time.as_millis() as f64; - - // 实际调用性能指标更新 - self.update_metrics(event_count as u64, processing_time_ms).await; - - // 记录慢处理操作 - if processing_time_ms > 5.0 { - log::warn!( - "ShredStream transaction processing took {}ms for {} events", - processing_time_ms, - event_count - ); - } - - Ok(()) - } - - async fn process_transaction_with_batch( - &self, - transaction_with_slot: TransactionWithSlot, - protocols: Vec, - bot_wallet: Option, - batch_processor: &mut ShredBatchProcessor, - ) -> AnyResult<()> - where - F: FnMut(Vec>) + Send + Sync + 'static, - { - let start_time = std::time::Instant::now(); - let program_received_time_ms = chrono::Utc::now().timestamp_millis(); - let slot = transaction_with_slot.slot; - let versioned_tx = transaction_with_slot.transaction; - let signature = versioned_tx.signatures[0]; - - // 预分配向量容量 - let mut all_events = Vec::with_capacity(protocols.len() * 2); - - for protocol in protocols { - let parser = EventParserFactory::create_parser(protocol.clone()); - let events = parser - .parse_versioned_transaction( - &versioned_tx, - &signature.to_string(), - Some(slot), - None, - program_received_time_ms, - bot_wallet, - ) - .await - .unwrap_or_else(|_e| vec![]); - all_events.extend(events); - } - - // 保存事件数量用于日志记录 - let event_count = all_events.len(); - - // 使用批处理器处理事件 - for event in all_events { - batch_processor.add_event(event); - } - - // 更新性能指标 - let processing_time = start_time.elapsed(); - let processing_time_ms = processing_time.as_millis() as f64; - - // 实际调用性能指标更新 - self.update_metrics(event_count as u64, processing_time_ms).await; - - // 记录慢处理操作 - if processing_time_ms > 5.0 { - log::warn!( - "ShredStream transaction processing took {}ms for {} events", - processing_time_ms, - event_count - ); - } - - Ok(()) - } }