diff --git a/Cargo.toml b/Cargo.toml index 64aa143..a03c355 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "solana-streamer-sdk" -version = "0.1.11" +version = "0.2.0" edition = "2021" authors = ["William ", "sgxiang ", "wei <1415121722@qq.com>"] repository = "https://github.com/0xfnzero/solana-streamer" @@ -40,7 +40,7 @@ yellowstone-grpc-client = { version = "8.0.0" } yellowstone-grpc-proto = { version = "8.0.0" } tokio = { version = "1.42.0", features = ["full", "rt-multi-thread"]} tonic = { version = "0.12.3", features = ["tls", "tls-roots", "tls-webpki-roots"] } -rustls = { version = "0.23.23", features = ["ring"] } +rustls = { version = "0.23.23", features = ["ring"], default-features = false } rustls-native-certs = "0.8.1" tokio-rustls = "0.26.1" log = "0.4.22" diff --git a/README.md b/README.md index 3b3e7e5..4cdf257 100755 --- a/README.md +++ b/README.md @@ -18,7 +18,7 @@ A lightweight Rust library for real-time event streaming from Solana DEX trading 6. **Event Parsing System**: Automatic parsing and categorization of protocol-specific events 7. **High Performance**: Optimized for low-latency event processing 8. **Batch Processing Optimization**: Batch processing events to reduce callback overhead -9. **Performance Monitoring**: Built-in performance metrics monitoring, including event processing speed, memory usage, etc. +9. **Performance Monitoring**: Built-in performance metrics monitoring, including event processing speed, etc. 10. **Memory Optimization**: Object pooling and caching mechanisms to reduce memory allocations 11. **Flexible Configuration System**: Support for custom batch sizes, backpressure strategies, channel sizes, and other parameters 12. **Preset Configurations**: Provides high-performance, low-latency, ordered processing, and other preset configurations @@ -41,18 +41,33 @@ Add the dependency to your `Cargo.toml`: ```toml # Add to your Cargo.toml -solana-streamer-sdk = { path = "./solana-streamer", version = "0.1.11" } +solana-streamer-sdk = { path = "./solana-streamer", version = "0.2.0" } ``` ### Use crates.io ```toml # Add to your Cargo.toml -solana-streamer-sdk = "0.1.11" +solana-streamer-sdk = "0.2.0" ``` ## Usage Examples +### Quick Start - Parse Transaction Events + +You can quickly test the library by running the built-in example that parses transaction events: + +```bash +cargo run --example parse_tx_events +``` + +This example demonstrates: +- How to parse transaction data from Solana mainnet using RPC +- Event parsing for multiple protocols (PumpFun, PumpSwap, Bonk, Raydium CPMM/CLMM) +- Transaction details extraction including fees, logs, and compute units + +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 ```rust @@ -61,7 +76,10 @@ use solana_streamer_sdk::{ streaming::{ event_parser::{ protocols::{ - bonk::{parser::BONK_PROGRAM_ID, BonkPoolCreateEvent, BonkTradeEvent}, + bonk::{ + parser::BONK_PROGRAM_ID, BonkMigrateToAmmEvent, BonkMigrateToCpswapEvent, + BonkPoolCreateEvent, BonkTradeEvent, + }, pumpfun::{parser::PUMPFUN_PROGRAM_ID, PumpFunCreateTokenEvent, PumpFunTradeEvent}, pumpswap::{ parser::PUMPSWAP_PROGRAM_ID, PumpSwapBuyEvent, PumpSwapCreatePoolEvent, @@ -70,11 +88,10 @@ use solana_streamer_sdk::{ raydium_clmm::{ parser::RAYDIUM_CLMM_PROGRAM_ID, RaydiumClmmSwapEvent, RaydiumClmmSwapV2Event, }, - raydium_cpmm::{parser::RAYDIUM_CPMM_PROGRAM_ID, RaydiumCpmmSwapEvent}, + raydium_cpmm::{parser::RAYDIUM_CPMM_PROGRAM_ID, RaydiumCpmmSwapEvent}, BlockMetaEvent, }, Protocol, UnifiedEvent, - }, - ShredStreamGrpc, YellowstoneGrpc, + }, grpc::ClientConfig, shred_stream::ShredClientConfig, ShredStreamGrpc, YellowstoneGrpc }, }; @@ -104,9 +121,11 @@ async fn test_grpc() -> Result<(), Box> { config, )?; + println!("GRPC client created successfully"); + let callback = create_event_callback(); - // Configure protocols to monitor + // Will try to parse corresponding protocol events from transactions let protocols = vec![ Protocol::PumpFun, Protocol::PumpSwap, @@ -115,7 +134,9 @@ async fn test_grpc() -> Result<(), Box> { Protocol::RaydiumClmm, ]; - // Configure account filtering + println!("Protocols to monitor: {:?}", protocols); + + // Filter accounts let account_include = vec![ PUMPFUN_PROGRAM_ID.to_string(), // Monitor pumpfun program ID PUMPSWAP_PROGRAM_ID.to_string(), // Monitor pumpswap program ID @@ -129,11 +150,10 @@ async fn test_grpc() -> Result<(), Box> { println!("Starting to listen for events, press Ctrl+C to stop..."); println!("Monitoring programs: {:?}", account_include); - + println!("Starting subscription..."); - - // Subscribe with automatic performance monitoring - grpc.subscribe_events_v2( + + grpc.subscribe_events_immediate( protocols, None, account_include, @@ -167,11 +187,7 @@ async fn test_shreds() -> Result<(), Box> { ]; println!("Listening for events, press Ctrl+C to stop..."); - - // Subscribe with automatic performance monitoring - shred_stream - .shredstream_subscribe(protocols, None, callback) - .await?; + shred_stream.shredstream_subscribe(protocols, None, callback).await?; Ok(()) } @@ -180,13 +196,16 @@ fn create_event_callback() -> impl Fn(Box) { |event: Box| { println!("🎉 Event received! Type: {:?}, ID: {}", event.event_type(), event.id()); match_event!(event, { + BlockMetaEvent => |e: BlockMetaEvent| { + println!("BlockMetaEvent: {e:?}"); + }, 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); println!("BonkPoolCreateEvent: {:?}", e.base_mint_param.symbol); }, BonkTradeEvent => |e: BonkTradeEvent| { - println!("BonkTradeEvent: {:?}", e); + println!("BonkTradeEvent: {e:?}"); }, BonkMigrateToAmmEvent => |e: BonkMigrateToAmmEvent| { println!("BonkMigrateToAmmEvent: {e:?}"); @@ -195,34 +214,34 @@ fn create_event_callback() -> impl Fn(Box) { println!("BonkMigrateToCpswapEvent: {e:?}"); }, PumpFunTradeEvent => |e: PumpFunTradeEvent| { - println!("PumpFunTradeEvent: {:?}", e); + println!("PumpFunTradeEvent: {e:?}"); }, PumpFunCreateTokenEvent => |e: PumpFunCreateTokenEvent| { - println!("PumpFunCreateTokenEvent: {:?}", e); + println!("PumpFunCreateTokenEvent: {e:?}"); }, PumpSwapBuyEvent => |e: PumpSwapBuyEvent| { - println!("Buy event: {:?}", e); + println!("Buy event: {e:?}"); }, PumpSwapSellEvent => |e: PumpSwapSellEvent| { - println!("Sell event: {:?}", e); + println!("Sell event: {e:?}"); }, PumpSwapCreatePoolEvent => |e: PumpSwapCreatePoolEvent| { - println!("CreatePool event: {:?}", e); + println!("CreatePool event: {e:?}"); }, PumpSwapDepositEvent => |e: PumpSwapDepositEvent| { - println!("Deposit event: {:?}", e); + println!("Deposit event: {e:?}"); }, PumpSwapWithdrawEvent => |e: PumpSwapWithdrawEvent| { - println!("Withdraw event: {:?}", e); + println!("Withdraw event: {e:?}"); }, RaydiumCpmmSwapEvent => |e: RaydiumCpmmSwapEvent| { - println!("RaydiumCpmmSwapEvent: {:?}", e); + println!("RaydiumCpmmSwapEvent: {e:?}"); }, RaydiumClmmSwapEvent => |e: RaydiumClmmSwapEvent| { - println!("RaydiumClmmSwapEvent: {:?}", e); + println!("RaydiumClmmSwapEvent: {e:?}"); }, RaydiumClmmSwapV2Event => |e: RaydiumClmmSwapV2Event| { - println!("RaydiumClmmSwapV2Event: {:?}", e); + println!("RaydiumClmmSwapV2Event: {e:?}"); } }); } @@ -309,7 +328,6 @@ MIT License 1. **Network Stability**: Ensure stable network connection for continuous event streaming 2. **Rate Limiting**: Be aware of rate limits on public gRPC endpoints 3. **Error Recovery**: Implement proper error handling and reconnection logic -4. **Resource Management**: Monitor memory and CPU usage for long-running streams 5. **Compliance**: Ensure compliance with relevant laws and regulations ## Language Versions diff --git a/README_CN.md b/README_CN.md index 8ad6d81..fe2ab0d 100644 --- a/README_CN.md +++ b/README_CN.md @@ -18,7 +18,7 @@ 6. **事件解析系统**: 自动解析和分类协议特定事件 7. **高性能**: 针对低延迟事件处理进行优化 8. **批处理优化**: 批量处理事件以减少回调开销 -9. **性能监控**: 内置性能指标监控,包括事件处理速度、内存使用等 +9. **性能监控**: 内置性能指标监控,包括事件处理速度等 10. **内存优化**: 对象池和缓存机制减少内存分配 11. **灵活配置系统**: 支持自定义批处理大小、背压策略、通道大小等参数 12. **预设配置**: 提供高性能、低延迟、有序处理等预设配置 @@ -41,25 +41,45 @@ git clone https://github.com/0xfnzero/solana-streamer ```toml # 添加到您的 Cargo.toml -solana-streamer-sdk = { path = "./solana-streamer", version = "0.1.11" } +solana-streamer-sdk = { path = "./solana-streamer", version = "0.2.0" } ``` ### 使用 crates.io ```toml # 添加到您的 Cargo.toml -solana-streamer-sdk = "0.1.11" +solana-streamer-sdk = "0.2.0" ``` ## 使用示例 +### 快速开始 - 解析交易事件 + +您可以通过运行内置示例来快速测试库的交易事件解析功能: + +```bash +cargo run --example parse_tx_events +``` + +该示例演示了: +- 如何使用 RPC 从 Solana 主网解析交易数据 +- 多协议事件解析(PumpFun、PumpSwap、Bonk、Raydium CPMM/CLMM) +- 交易详情提取,包括费用、日志和计算单元 + +该示例使用预定义的交易签名,展示如何从交易数据中提取协议特定的事件。 + +### 高级用法示例 + ```rust use solana_streamer_sdk::{ match_event, streaming::{ event_parser::{ protocols::{ - bonk::{parser::BONK_PROGRAM_ID, BonkPoolCreateEvent, BonkTradeEvent}, + bonk::{ + parser::BONK_PROGRAM_ID, BonkMigrateToAmmEvent, BonkMigrateToCpswapEvent, + BonkPoolCreateEvent, BonkTradeEvent, + }, pumpfun::{parser::PUMPFUN_PROGRAM_ID, PumpFunCreateTokenEvent, PumpFunTradeEvent}, pumpswap::{ parser::PUMPSWAP_PROGRAM_ID, PumpSwapBuyEvent, PumpSwapCreatePoolEvent, @@ -68,23 +88,23 @@ use solana_streamer_sdk::{ raydium_clmm::{ parser::RAYDIUM_CLMM_PROGRAM_ID, RaydiumClmmSwapEvent, RaydiumClmmSwapV2Event, }, - raydium_cpmm::{parser::RAYDIUM_CPMM_PROGRAM_ID, RaydiumCpmmSwapEvent}, + raydium_cpmm::{parser::RAYDIUM_CPMM_PROGRAM_ID, RaydiumCpmmSwapEvent}, BlockMetaEvent, }, Protocol, UnifiedEvent, - }, - ShredStreamGrpc, YellowstoneGrpc, + }, grpc::ClientConfig, shred_stream::ShredClientConfig, ShredStreamGrpc, YellowstoneGrpc }, }; #[tokio::main] async fn main() -> Result<(), Box> { + println!("Starting Solana Streamer..."); test_grpc().await?; test_shreds().await?; Ok(()) } async fn test_grpc() -> Result<(), Box> { - println!("正在订阅 Yellowstone gRPC 事件..."); + println!("Subscribing to Yellowstone gRPC events..."); // 创建低延迟配置 let mut config = ClientConfig::low_latency(); @@ -96,6 +116,8 @@ async fn test_grpc() -> Result<(), Box> { config, )?; + println!("GRPC client created successfully"); + let callback = create_event_callback(); // 将会从交易中尝试解析对应的协议事件 @@ -107,6 +129,8 @@ async fn test_grpc() -> Result<(), Box> { Protocol::RaydiumClmm, ]; + println!("Protocols to monitor: {:?}", protocols); + // 过滤账号 let account_include = vec![ PUMPFUN_PROGRAM_ID.to_string(), // 监听 pumpfun 程序ID @@ -119,8 +143,12 @@ async fn test_grpc() -> Result<(), Box> { let account_exclude = vec![]; let account_required = vec![]; - println!("开始监听事件,按 Ctrl+C 停止..."); - grpc.subscribe_events_v2( + 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, account_include, @@ -135,13 +163,15 @@ async fn test_grpc() -> Result<(), Box> { } async fn test_shreds() -> Result<(), Box> { - println!("正在订阅 ShredStream 事件..."); + println!("Subscribing to ShredStream events..."); + // 创建低延迟配置 let mut config = ShredClientConfig::low_latency(); // 启用性能监控, 有性能损耗, 默认关闭 config.enable_metrics = true; let shred_stream = ShredStreamGrpc::new_with_config("http://127.0.0.1:10800".to_string(), config).await?; + let callback = create_event_callback(); let protocols = vec![ Protocol::PumpFun, @@ -151,24 +181,26 @@ async fn test_shreds() -> Result<(), Box> { Protocol::RaydiumClmm, ]; - println!("开始监听事件,按 Ctrl+C 停止..."); - shred_stream - .shredstream_subscribe(protocols, None, callback) - .await?; + println!("Listening for events, press Ctrl+C to stop..."); + shred_stream.shredstream_subscribe(protocols, None, callback).await?; Ok(()) } fn create_event_callback() -> impl Fn(Box) { |event: Box| { + println!("🎉 Event received! Type: {:?}, ID: {}", event.event_type(), event.id()); match_event!(event, { + BlockMetaEvent => |e: BlockMetaEvent| { + println!("BlockMetaEvent: {e:?}"); + }, BonkPoolCreateEvent => |e: BonkPoolCreateEvent| { // 使用grpc的时候,可以从每个事件中获取到block_time println!("block_time: {:?}, block_time_ms: {:?}", e.metadata.block_time, e.metadata.block_time_ms); println!("BonkPoolCreateEvent: {:?}", e.base_mint_param.symbol); }, BonkTradeEvent => |e: BonkTradeEvent| { - println!("BonkTradeEvent: {:?}", e); + println!("BonkTradeEvent: {e:?}"); }, BonkMigrateToAmmEvent => |e: BonkMigrateToAmmEvent| { println!("BonkMigrateToAmmEvent: {e:?}"); @@ -177,34 +209,34 @@ fn create_event_callback() -> impl Fn(Box) { println!("BonkMigrateToCpswapEvent: {e:?}"); }, PumpFunTradeEvent => |e: PumpFunTradeEvent| { - println!("PumpFunTradeEvent: {:?}", e); + println!("PumpFunTradeEvent: {e:?}"); }, PumpFunCreateTokenEvent => |e: PumpFunCreateTokenEvent| { - println!("PumpFunCreateTokenEvent: {:?}", e); + println!("PumpFunCreateTokenEvent: {e:?}"); }, PumpSwapBuyEvent => |e: PumpSwapBuyEvent| { - println!("Buy event: {:?}", e); + println!("Buy event: {e:?}"); }, PumpSwapSellEvent => |e: PumpSwapSellEvent| { - println!("Sell event: {:?}", e); + println!("Sell event: {e:?}"); }, PumpSwapCreatePoolEvent => |e: PumpSwapCreatePoolEvent| { - println!("CreatePool event: {:?}", e); + println!("CreatePool event: {e:?}"); }, PumpSwapDepositEvent => |e: PumpSwapDepositEvent| { - println!("Deposit event: {:?}", e); + println!("Deposit event: {e:?}"); }, PumpSwapWithdrawEvent => |e: PumpSwapWithdrawEvent| { - println!("Withdraw event: {:?}", e); + println!("Withdraw event: {e:?}"); }, RaydiumCpmmSwapEvent => |e: RaydiumCpmmSwapEvent| { - println!("RaydiumCpmmSwapEvent: {:?}", e); + println!("RaydiumCpmmSwapEvent: {e:?}"); }, RaydiumClmmSwapEvent => |e: RaydiumClmmSwapEvent| { - println!("RaydiumClmmSwapEvent: {:?}", e); + println!("RaydiumClmmSwapEvent: {e:?}"); }, RaydiumClmmSwapV2Event => |e: RaydiumClmmSwapV2Event| { - println!("RaydiumClmmSwapV2Event: {:?}", e); + println!("RaydiumClmmSwapV2Event: {e:?}"); } }); } @@ -291,7 +323,6 @@ MIT 许可证 1. **网络稳定性**: 确保稳定的网络连接以进行连续的事件流传输 2. **速率限制**: 注意公共 gRPC 端点的速率限制 3. **错误恢复**: 实现适当的错误处理和重连逻辑 -4. **资源管理**: 监控长时间运行流的内存和 CPU 使用情况 5. **合规性**: 确保遵守相关法律法规 ## 语言版本 diff --git a/examples/parse_tx_events.rs b/examples/parse_tx_events.rs new file mode 100644 index 0000000..c1693f6 --- /dev/null +++ b/examples/parse_tx_events.rs @@ -0,0 +1,124 @@ +use anyhow::Result; +use solana_sdk::commitment_config::CommitmentConfig; +use solana_streamer_sdk::streaming::event_parser::{ + protocols::MutilEventParser, EventParser, Protocol, +}; +use std::str::FromStr; +use std::sync::Arc; + +/// Get transaction data based on transaction signature +#[tokio::main] +async fn main() -> Result<()> { + let signatures = vec![ + "tb6KxvMAZtctQw2yszKk6QGVuzbv1CAbS5mKK7JjVaeb2i5rCafNHfnqE1PhaWukx79sDYcsoZqbHXRJgPu93w4", + ]; + // Validate signature format + let mut valid_signatures = Vec::new(); + for sig_str in &signatures { + match solana_sdk::signature::Signature::from_str(sig_str) { + Ok(_) => valid_signatures.push(*sig_str), + Err(e) => println!("Invalid signature format: {}", e), + } + } + if valid_signatures.is_empty() { + println!("No valid transaction signatures"); + return Ok(()); + } + for signature in valid_signatures { + println!("Starting transaction parsing: {}", signature); + get_single_transaction_details(signature).await?; + println!("Transaction parsing completed: {}\n\n\n", signature); + } + + Ok(()) +} + +/// Get details of a single transaction +async fn get_single_transaction_details(signature_str: &str) -> Result<()> { + use solana_sdk::signature::Signature; + use solana_transaction_status::UiTransactionEncoding; + + let signature = Signature::from_str(signature_str)?; + + // Create Solana RPC client + let rpc_url = "https://api.mainnet-beta.solana.com"; + println!("Connecting to Solana RPC: {}", rpc_url); + + let client = solana_client::nonblocking::rpc_client::RpcClient::new(rpc_url.to_string()); + + match client + .get_transaction_with_config( + &signature, + solana_client::rpc_config::RpcTransactionConfig { + encoding: Some(UiTransactionEncoding::Binary), + commitment: Some(CommitmentConfig::confirmed()), + max_supported_transaction_version: Some(0), + }, + ) + .await + { + Ok(transaction) => { + println!("Transaction signature: {}", signature_str); + println!("Block slot: {}", transaction.slot); + + if let Some(block_time) = transaction.block_time { + println!("Block time: {}", block_time); + } + + if let Some(meta) = &transaction.transaction.meta { + println!("Transaction fee: {} lamports", meta.fee); + println!("Status: {}", if meta.err.is_none() { "Success" } else { "Failed" }); + if let Some(err) = &meta.err { + println!("Error details: {:?}", err); + } + // Compute units consumed + if let solana_transaction_status::option_serializer::OptionSerializer::Some(units) = + &meta.compute_units_consumed + { + println!("Compute units consumed: {}", units); + } + // Display logs (all) + if let solana_transaction_status::option_serializer::OptionSerializer::Some(logs) = + &meta.log_messages + { + println!("Transaction logs (all {} entries):", logs.len()); + for (i, log) in logs.iter().enumerate() { + println!(" [{}] {}", i + 1, log); + } + } + } + let protocols = vec![ + Protocol::Bonk, + Protocol::RaydiumClmm, + Protocol::PumpSwap, + Protocol::PumpFun, + Protocol::RaydiumCpmm, + ]; + let parser: Arc = Arc::new(MutilEventParser::new(protocols)); + let start_time = std::time::Instant::now(); + let events = parser + .parse_transaction( + transaction.transaction.clone(), + &signature.to_string(), + Some(transaction.slot), + None, + 0, + None, + ) + .await + .unwrap_or_else(|_e| vec![]); + + let end_time = std::time::Instant::now(); + let duration = end_time.duration_since(start_time); + println!("Parsing time: {:?}", duration); + for event in events { + println!("{:?}\n", event); + } + } + Err(e) => { + println!("Failed to get transaction: {}", e); + } + } + + Ok(()) +} diff --git a/src/main.rs b/src/main.rs index 78f3137..8daf70f 100755 --- a/src/main.rs +++ b/src/main.rs @@ -15,13 +15,10 @@ use solana_streamer_sdk::{ raydium_clmm::{ parser::RAYDIUM_CLMM_PROGRAM_ID, RaydiumClmmSwapEvent, RaydiumClmmSwapV2Event, }, - raydium_cpmm::{parser::RAYDIUM_CPMM_PROGRAM_ID, RaydiumCpmmSwapEvent}, + raydium_cpmm::{parser::RAYDIUM_CPMM_PROGRAM_ID, RaydiumCpmmSwapEvent}, BlockMetaEvent, }, Protocol, UnifiedEvent, - }, - ShredStreamGrpc, YellowstoneGrpc, - yellowstone_grpc::ClientConfig, - shred_stream::ShredClientConfig, + }, grpc::ClientConfig, shred_stream::ShredClientConfig, ShredStreamGrpc, YellowstoneGrpc }, }; @@ -77,7 +74,7 @@ async fn test_grpc() -> Result<(), Box> { println!("Starting subscription..."); - grpc.subscribe_events_v2( + grpc.subscribe_events_immediate( protocols, None, account_include, @@ -120,6 +117,9 @@ fn create_event_callback() -> impl Fn(Box) { |event: Box| { println!("🎉 Event received! Type: {:?}, ID: {}", event.event_type(), event.id()); match_event!(event, { + BlockMetaEvent => |e: BlockMetaEvent| { + println!("BlockMetaEvent: {e:?}"); + }, 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); diff --git a/src/streaming/common/batch.rs b/src/streaming/common/batch.rs new file mode 100644 index 0000000..f9c092c --- /dev/null +++ b/src/streaming/common/batch.rs @@ -0,0 +1,131 @@ +use crate::streaming::event_parser::UnifiedEvent; + +/// 通用批处理事件收集器 +pub struct EventBatchProcessor +where + F: FnMut(Vec>) + Send + Sync + 'static, +{ + pub(crate) callback: F, + batch: Vec>, + batch_size: usize, + timeout_ms: u64, + last_flush_time: std::time::Instant, +} + +impl EventBatchProcessor +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) { + log::debug!("Adding event to batch: {} (type: {:?})", event.id(), event.event_type()); + self.batch.push(event); + + // 检查是否需要刷新批次 + if self.batch.len() >= self.batch_size || self.should_flush_by_timeout() { + log::debug!("Flushing batch: size={}, timeout={}", self.batch.len(), 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)); + log::debug!("Flushing {} events from batch processor", events.len()); + + // 添加调试信息(仅在debug模式下) + if log::log_enabled!(log::Level::Debug) { + for (i, event) in events.iter().enumerate() { + log::debug!("Event {}: Type={:?}, ID={}", i, event.event_type(), event.id()); + } + } + + // 执行回调并捕获可能的错误 + log::debug!("Executing batch callback with {} events", events.len()); + match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + (self.callback)(events); + })) { + Ok(_) => { + log::debug!("Batch callback executed successfully"); + } + Err(e) => { + log::error!("Batch callback panicked: {:?}", e); + } + } + + self.last_flush_time = std::time::Instant::now(); + } else { + log::debug!("No events to flush"); + } + } + + /// 获取当前批次大小 + pub fn current_batch_size(&self) -> usize { + self.batch.len() + } + + /// 检查是否应该基于超时刷新 + fn should_flush_by_timeout(&self) -> bool { + self.last_flush_time.elapsed().as_millis() >= self.timeout_ms as u128 + } + + /// 检查批次是否已满 + pub fn is_batch_full(&self) -> bool { + self.batch.len() >= self.batch_size + } + + /// 检查是否需要刷新(大小或超时) + pub fn should_flush(&self) -> bool { + self.is_batch_full() || self.should_flush_by_timeout() + } +} + +/// 简单的事件批处理器,用于将单个事件回调转换为批量回调 +pub struct SimpleEventBatchProcessor +where + F: Fn(Box) + Send + Sync + 'static, +{ + callback: F, +} + +impl SimpleEventBatchProcessor +where + F: Fn(Box) + Send + Sync + 'static, +{ + pub fn new(callback: F) -> Self { + Self { callback } + } + + /// 将批量事件拆分为单个事件处理 + pub fn process_batch(&self, events: Vec>) { + for event in events { + (self.callback)(event); + } + } +} + +/// 批处理器包装器,用于将单个事件回调适配为批量处理 +pub fn create_batch_callback_adapter( + single_event_callback: F, +) -> impl FnMut(Vec>) + Send + Sync + 'static +where + F: Fn(Box) + Send + Sync + 'static, +{ + move |events: Vec>| { + for event in events { + single_event_callback(event); + } + } +} diff --git a/src/streaming/common/config.rs b/src/streaming/common/config.rs new file mode 100644 index 0000000..1ae1500 --- /dev/null +++ b/src/streaming/common/config.rs @@ -0,0 +1,131 @@ +use super::constants::*; + +/// 背压处理策略 +#[derive(Debug, Clone, Copy)] +pub enum BackpressureStrategy { + /// 阻塞等待(默认) + Block, + /// 丢弃消息 + Drop, + /// 重试有限次数后丢弃 + Retry { max_attempts: usize, wait_ms: u64 }, +} + +impl Default for BackpressureStrategy { + fn default() -> Self { + Self::Block + } +} + +/// 批处理配置 +#[derive(Debug, Clone)] +pub struct BatchConfig { + /// 批处理大小(默认:100) + pub batch_size: usize, + /// 批处理超时时间(毫秒,默认:5ms) + pub batch_timeout_ms: u64, + /// 是否启用批处理(默认:true) + pub enabled: bool, +} + +impl Default for BatchConfig { + fn default() -> Self { + Self { + batch_size: DEFAULT_BATCH_SIZE, + batch_timeout_ms: DEFAULT_BATCH_TIMEOUT_MS, + enabled: true, + } + } +} + +/// 背压配置 +#[derive(Debug, Clone)] +pub struct BackpressureConfig { + /// 通道大小(默认:1000) + pub channel_size: usize, + /// 背压处理策略(默认:Block) + pub strategy: BackpressureStrategy, +} + +impl Default for BackpressureConfig { + fn default() -> Self { + Self { channel_size: DEFAULT_CHANNEL_SIZE, strategy: BackpressureStrategy::default() } + } +} + +/// 连接配置 +#[derive(Debug, Clone)] +pub struct ConnectionConfig { + /// 连接超时时间(秒,默认:10) + pub connect_timeout: u64, + /// 请求超时时间(秒,默认:60) + pub request_timeout: u64, + /// 最大解码消息大小(字节,默认:10MB) + pub max_decoding_message_size: usize, +} + +impl Default for ConnectionConfig { + fn default() -> Self { + Self { + connect_timeout: DEFAULT_CONNECT_TIMEOUT, + request_timeout: DEFAULT_REQUEST_TIMEOUT, + max_decoding_message_size: DEFAULT_MAX_DECODING_MESSAGE_SIZE, + } + } +} + +/// 通用客户端配置 +#[derive(Debug, Clone)] +pub struct StreamClientConfig { + /// 连接配置 + pub connection: ConnectionConfig, + /// 批处理配置 + pub batch: BatchConfig, + /// 背压配置 + pub backpressure: BackpressureConfig, + /// 是否启用性能监控(默认:false) + pub enable_metrics: bool, +} + +impl Default for StreamClientConfig { + fn default() -> Self { + Self { + connection: ConnectionConfig::default(), + batch: BatchConfig::default(), + backpressure: BackpressureConfig::default(), + enable_metrics: false, + } + } +} + +impl StreamClientConfig { + /// 创建高性能配置(适合高并发场景) + pub fn high_performance() -> Self { + Self { + connection: ConnectionConfig::default(), + batch: BatchConfig { batch_size: 200, batch_timeout_ms: 5, enabled: true }, + backpressure: BackpressureConfig { + channel_size: 20000, + strategy: BackpressureStrategy::Drop, + }, + enable_metrics: true, + } + } + + /// 创建低延迟配置(适合实时场景) + pub fn low_latency() -> Self { + Self { + connection: ConnectionConfig::default(), + batch: BatchConfig { + batch_size: 10, + batch_timeout_ms: 1, + enabled: false, // 禁用批处理,即时处理 + }, + backpressure: BackpressureConfig { + channel_size: 1000, + strategy: BackpressureStrategy::Block, + }, + enable_metrics: false, + } + } +} diff --git a/src/streaming/common/constants.rs b/src/streaming/common/constants.rs new file mode 100644 index 0000000..d4b3d18 --- /dev/null +++ b/src/streaming/common/constants.rs @@ -0,0 +1,14 @@ +// 流处理相关的常量定义 + +// 默认配置常量 +pub const DEFAULT_CONNECT_TIMEOUT: u64 = 10; +pub const DEFAULT_REQUEST_TIMEOUT: u64 = 60; +pub const DEFAULT_CHANNEL_SIZE: usize = 1000; +pub const DEFAULT_MAX_DECODING_MESSAGE_SIZE: usize = 1024 * 1024 * 10; +pub const DEFAULT_BATCH_SIZE: usize = 100; +pub const DEFAULT_BATCH_TIMEOUT_MS: u64 = 5; + +// 性能监控相关常量 +pub const DEFAULT_METRICS_WINDOW_SECONDS: u64 = 5; +pub const DEFAULT_METRICS_PRINT_INTERVAL_SECONDS: u64 = 10; +pub const SLOW_PROCESSING_THRESHOLD_MS: f64 = 10.0; diff --git a/src/streaming/common/metrics.rs b/src/streaming/common/metrics.rs new file mode 100644 index 0000000..ecaaa9c --- /dev/null +++ b/src/streaming/common/metrics.rs @@ -0,0 +1,177 @@ +use std::sync::Arc; +use tokio::sync::Mutex; + +use super::constants::*; +use super::config::StreamClientConfig; + +/// 通用性能监控指标 +#[derive(Debug, Clone)] +pub struct PerformanceMetrics { + 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 cache_hit_rate: f64, + pub last_update_time: std::time::Instant, + pub events_in_window: u64, + pub window_start_time: std::time::Instant, +} + +impl Default for PerformanceMetrics { + fn default() -> Self { + Self::new() + } +} + +impl PerformanceMetrics { + 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, + cache_hit_rate: 0.0, + last_update_time: now, + events_in_window: 0, + window_start_time: now, + } + } +} + +/// 通用性能监控管理器 +pub struct MetricsManager { + metrics: Arc>, + config: Arc, + stream_name: String, +} + +impl MetricsManager { + /// 创建新的性能监控管理器 + pub fn new( + metrics: Arc>, + config: Arc, + stream_name: String, + ) -> Self { + Self { metrics, config, stream_name } + } + + /// 获取性能指标 + pub async fn get_metrics(&self) -> PerformanceMetrics { + let metrics = self.metrics.lock().await; + metrics.clone() + } + + /// 打印性能指标 + pub async fn print_metrics(&self) { + let metrics = self.get_metrics().await; + println!("📊 {} Performance Metrics:", self.stream_name); + 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); + if metrics.cache_hit_rate > 0.0 { + println!(" Cache Hit Rate: {:.2}%", metrics.cache_hit_rate * 100.0); + } + println!("---"); + } + + /// 启动自动性能监控任务 + pub async fn start_auto_monitoring(&self) { + // 检查是否启用性能监控 + if !self.config.enable_metrics { + return; // 如果未启用性能监控,不启动监控任务 + } + + let metrics_manager = self.clone(); + tokio::spawn(async move { + let mut interval = tokio::time::interval( + tokio::time::Duration::from_secs(DEFAULT_METRICS_PRINT_INTERVAL_SECONDS) + ); + loop { + interval.tick().await; + metrics_manager.print_metrics().await; + } + }); + } + + /// 更新性能指标 + pub 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; + } + + // 基于时间窗口计算每秒处理事件数 + 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; + } else { + // 如果窗口内没有事件,保持之前的速率或设为0 + metrics.events_per_second = 0.0; + } + + // 重置窗口 + metrics.events_in_window = 0; + metrics.window_start_time = now; + } + + } + + /// 更新缓存命中率 + pub async fn update_cache_hit_rate(&self, hit_rate: f64) { + if !self.config.enable_metrics { + return; + } + + let mut metrics = self.metrics.lock().await; + metrics.cache_hit_rate = hit_rate; + } + + /// 记录慢处理操作 + pub fn log_slow_processing(&self, processing_time_ms: f64, event_count: usize) { + if processing_time_ms > SLOW_PROCESSING_THRESHOLD_MS { + log::warn!( + "{} slow processing: {processing_time_ms}ms for {event_count} events", + self.stream_name + ); + } + } +} + +impl Clone for MetricsManager { + fn clone(&self) -> Self { + Self { + metrics: self.metrics.clone(), + config: self.config.clone(), + stream_name: self.stream_name.clone(), + } + } +} diff --git a/src/streaming/common/mod.rs b/src/streaming/common/mod.rs new file mode 100644 index 0000000..1132b0e --- /dev/null +++ b/src/streaming/common/mod.rs @@ -0,0 +1,11 @@ +// 公用模块 - 包含流处理相关的通用功能 +pub mod config; +pub mod metrics; +pub mod batch; +pub mod constants; + +// 重新导出主要类型 +pub use config::*; +pub use metrics::*; +pub use batch::*; +pub use constants::*; diff --git a/src/streaming/event_parser/common/mod.rs b/src/streaming/event_parser/common/mod.rs index 2984990..aa905b6 100755 --- a/src/streaming/event_parser/common/mod.rs +++ b/src/streaming/event_parser/common/mod.rs @@ -55,8 +55,8 @@ macro_rules! impl_unified_event { } } - fn set_transfer_datas(&mut self, transfer_datas: Vec<$crate::streaming::event_parser::common::types::TransferData>) { - self.metadata.transfer_datas = transfer_datas; + fn set_transfer_datas(&mut self, transfer_datas: Vec<$crate::streaming::event_parser::common::types::TransferData>, swap_data: Option<$crate::streaming::event_parser::common::types::SwapData>) { + self.metadata.set_transfer_datas(transfer_datas, swap_data); } fn index(&self) -> String { diff --git a/src/streaming/event_parser/common/types.rs b/src/streaming/event_parser/common/types.rs index 942b547..4b72619 100755 --- a/src/streaming/event_parser/common/types.rs +++ b/src/streaming/event_parser/common/types.rs @@ -2,9 +2,27 @@ use borsh::{BorshDeserialize, BorshSerialize}; use serde::{Deserialize, Serialize}; use solana_sdk::pubkey::Pubkey; use solana_transaction_status::UiInstruction; -use std::{hash::{DefaultHasher, Hash, Hasher}, sync::Arc}; +use std::{ + hash::{DefaultHasher, Hash, Hasher}, + str::FromStr, + sync::Arc, +}; use tokio::sync::Mutex; +use crate::{ + match_event, + streaming::event_parser::{ + protocols::{ + bonk::BonkTradeEvent, + pumpfun::PumpFunTradeEvent, + pumpswap::{PumpSwapBuyEvent, PumpSwapSellEvent}, + raydium_clmm::{RaydiumClmmSwapEvent, RaydiumClmmSwapV2Event}, + raydium_cpmm::RaydiumCpmmSwapEvent, + }, + UnifiedEvent, + }, +}; + // Object pool size configuration const EVENT_METADATA_POOL_SIZE: usize = 1000; const TRANSFER_DATA_POOL_SIZE: usize = 2000; @@ -22,9 +40,7 @@ impl Default for EventMetadataPool { impl EventMetadataPool { pub fn new() -> Self { - Self { - pool: Arc::new(Mutex::new(Vec::with_capacity(EVENT_METADATA_POOL_SIZE))), - } + Self { pool: Arc::new(Mutex::new(Vec::with_capacity(EVENT_METADATA_POOL_SIZE))) } } pub async fn acquire(&self) -> Option { @@ -53,9 +69,7 @@ impl Default for TransferDataPool { impl TransferDataPool { pub fn new() -> Self { - Self { - pool: Arc::new(Mutex::new(Vec::with_capacity(TRANSFER_DATA_POOL_SIZE))), - } + Self { pool: Arc::new(Mutex::new(Vec::with_capacity(TRANSFER_DATA_POOL_SIZE))) } } pub async fn acquire(&self) -> Option { @@ -87,7 +101,7 @@ pub enum ProtocolType { Bonk, RaydiumCpmm, RaydiumClmm, - SDKSystem, + Common, } /// Event type enumeration @@ -126,7 +140,7 @@ pub enum EventType { RaydiumClmmSwapV2, // Common events - SDKSystem, + BlockMeta, Unknown, } @@ -153,7 +167,7 @@ impl EventType { EventType::RaydiumCpmmSwapBaseOutput => "RaydiumCpmmSwapBaseOutput".to_string(), EventType::RaydiumClmmSwap => "RaydiumClmmSwap".to_string(), EventType::RaydiumClmmSwapV2 => "RaydiumClmmSwapV2".to_string(), - EventType::SDKSystem => "SDKSystem".to_string(), + EventType::BlockMeta => "BlockMeta".to_string(), EventType::Unknown => "Unknown".to_string(), } } @@ -169,19 +183,11 @@ pub struct ParseResult { impl ParseResult { pub fn success(data: T) -> Self { - Self { - success: true, - data: Some(data), - error: None, - } + Self { success: true, data: Some(data), error: None } } pub fn failure(error: String) -> Self { - Self { - success: false, - data: None, - error: Some(error), - } + Self { success: false, data: None, error: Some(error) } } pub fn is_success(&self) -> bool { @@ -224,6 +230,17 @@ pub struct TransferData { pub mint: Option, } +#[derive( + Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, BorshSerialize, BorshDeserialize, +)] +pub struct SwapData { + pub from_mint: Pubkey, + pub to_mint: Pubkey, + pub from_amount: u64, + pub to_amount: u64, + pub description: Option, +} + /// Event metadata #[derive( Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, BorshSerialize, BorshDeserialize, @@ -240,6 +257,7 @@ pub struct EventMetadata { pub event_type: EventType, pub program_id: Pubkey, pub transfer_datas: Vec, + pub swap_data: Option, pub index: String, } @@ -269,56 +287,11 @@ impl EventMetadata { event_type, program_id, transfer_datas: Vec::with_capacity(4), // Pre-allocate capacity + swap_data: None, index, } } - /// Create EventMetadata using object pool - #[allow(clippy::too_many_arguments)] - pub async fn new_with_pool( - id: String, - signature: String, - slot: u64, - block_time: i64, - block_time_ms: i64, - protocol: ProtocolType, - event_type: EventType, - program_id: Pubkey, - index: String, - program_received_time_ms: i64, - ) -> Self { - // Try to get from object pool - if let Some(mut metadata) = EVENT_METADATA_POOL.acquire().await { - metadata.id = id; - metadata.signature = signature; - metadata.slot = slot; - metadata.block_time = block_time; - metadata.block_time_ms = block_time_ms; - metadata.program_received_time_ms = program_received_time_ms; - metadata.program_handle_time_consuming_ms = 0; - metadata.protocol = protocol; - metadata.event_type = event_type; - metadata.program_id = program_id; - metadata.index = index; - metadata.transfer_datas.clear(); - return metadata; - } - - // If object pool is empty, create new one - Self::new( - id, - signature, - slot, - block_time, - block_time_ms, - protocol, - event_type, - program_id, - index, - program_received_time_ms, - ) - } - pub fn set_id(&mut self, id: String) { let _id = format!("{}-{}-{}", self.signature, self.event_type.to_string(), id); let mut hasher = DefaultHasher::new(); @@ -327,8 +300,13 @@ impl EventMetadata { self.id = format!("{:x}", hash_value); } - pub fn set_transfer_datas(&mut self, transfer_datas: Vec) { + pub fn set_transfer_datas( + &mut self, + transfer_datas: Vec, + swap_data: Option, + ) { self.transfer_datas = transfer_datas; + self.swap_data = swap_data; } /// Recycle EventMetadata to object pool @@ -339,11 +317,12 @@ impl EventMetadata { /// Parse token transfer data from next instructions pub fn parse_transfer_datas_from_next_instructions( + event: Box, inner_instruction: &solana_transaction_status::UiInnerInstructions, current_index: i8, accounts: &[Pubkey], event_type: EventType, -) -> Vec { +) -> (Vec, Option) { let take = match event_type { EventType::PumpFunBuy => 4, EventType::PumpFunSell => 1, @@ -360,7 +339,7 @@ pub fn parse_transfer_datas_from_next_instructions( _ => 0, }; if take == 0 { - return vec![]; + return (vec![], None); } let mut transfer_datas = vec![]; // Get the next two instructions after the current instruction @@ -377,11 +356,8 @@ pub fn parse_transfer_datas_from_next_instructions( // Token Program: transferChecked // Token 2022 Program: transferChecked if data[0] == 12 { - let account_pubkeys: Vec = compiled - .accounts - .iter() - .map(|a| accounts[*a as usize]) - .collect(); + let account_pubkeys: Vec = + compiled.accounts.iter().map(|a| accounts[*a as usize]).collect(); if account_pubkeys.len() < 4 { continue; } @@ -406,11 +382,8 @@ pub fn parse_transfer_datas_from_next_instructions( } // Token Program: transfer else if data[0] == 3 { - let account_pubkeys: Vec = compiled - .accounts - .iter() - .map(|a| accounts[*a as usize]) - .collect(); + let account_pubkeys: Vec = + compiled.accounts.iter().map(|a| accounts[*a as usize]).collect(); if account_pubkeys.len() < 3 { continue; } @@ -430,11 +403,8 @@ pub fn parse_transfer_datas_from_next_instructions( } //System Program: transfer else if data[0] == 2 { - let account_pubkeys: Vec = compiled - .accounts - .iter() - .map(|a| accounts[*a as usize]) - .collect(); + let account_pubkeys: Vec = + compiled.accounts.iter().map(|a| accounts[*a as usize]).collect(); if account_pubkeys.len() < 2 { continue; } @@ -454,5 +424,111 @@ pub fn parse_transfer_datas_from_next_instructions( } } } - transfer_datas + let mut swap_data: SwapData = SwapData { + from_mint: Pubkey::default(), + to_mint: Pubkey::default(), + from_amount: 0, + to_amount: 0, + description: None, + }; + let sol_mint = Pubkey::from_str("So11111111111111111111111111111111111111111").unwrap(); + if transfer_datas.len() > 0 { + let mut user: Option = None; + let mut from_mint: Option = None; + let mut to_mint: Option = None; + let mut user_from_token: Option = None; + let mut user_to_token: Option = None; + let mut from_vault: Option = None; + let mut to_vault: Option = None; + match_event!(event, { + BonkTradeEvent => |e: BonkTradeEvent| { + user = Some(e.payer); + from_mint = Some(e.base_token_mint); + to_mint = Some(e.quote_token_mint); + user_from_token = Some(e.user_base_token); + user_to_token = Some(e.user_quote_token); + from_vault = Some(e.base_vault); + to_vault = Some(e.quote_vault); + }, + PumpFunTradeEvent => |e: PumpFunTradeEvent| { + swap_data.from_mint = if e.is_buy { + sol_mint + } else { + e.mint + }; + swap_data.to_mint = if e.is_buy { + e.mint + } else { + sol_mint + }; + }, + PumpSwapBuyEvent => |e: PumpSwapBuyEvent| { + swap_data.from_mint = e.quote_mint; + swap_data.to_mint = e.base_mint; + }, + PumpSwapSellEvent => |e: PumpSwapSellEvent| { + swap_data.from_mint = e.base_mint; + swap_data.to_mint = e.quote_mint; + }, + RaydiumCpmmSwapEvent => |e: RaydiumCpmmSwapEvent| { + user = Some(e.payer); + from_mint = Some(e.input_token_mint); + to_mint = Some(e.output_token_mint); + user_from_token = Some(e.input_token_account); + user_to_token = Some(e.output_token_account); + from_vault = Some(e.input_vault); + to_vault = Some(e.output_vault); + }, + RaydiumClmmSwapEvent => |e: RaydiumClmmSwapEvent| { + user = Some(e.payer); + swap_data.description = Some("Unable to get from_mint and to_mint from RaydiumClmmSwapEvent".to_string()); + user_from_token = Some(e.input_token_account); + user_to_token = Some(e.output_token_account); + from_vault = Some(e.input_vault); + to_vault = Some(e.output_vault); + }, + RaydiumClmmSwapV2Event => |e: RaydiumClmmSwapV2Event| { + user = Some(e.payer); + from_mint = Some(e.input_vault_mint); + to_mint = Some(e.output_vault_mint); + user_from_token = Some(e.input_token_account); + user_to_token = Some(e.output_token_account); + from_vault = Some(e.input_vault); + to_vault = Some(e.output_vault); + } + }); + + for transfer_data in transfer_datas.clone() { + if transfer_data.source == user_to_token.unwrap_or_default() + && transfer_data.destination == to_vault.unwrap_or_default() + { + swap_data.from_mint = to_mint.unwrap_or_default(); + swap_data.from_amount = transfer_data.amount; + } else if transfer_data.source == from_vault.unwrap_or_default() + && transfer_data.destination == user_from_token.unwrap_or_default() + { + swap_data.to_mint = from_mint.unwrap_or_default(); + swap_data.to_amount = transfer_data.amount; + } else if transfer_data.source == user_from_token.unwrap_or_default() + && transfer_data.destination == from_vault.unwrap_or_default() + { + swap_data.from_mint = from_mint.unwrap_or_default(); + swap_data.from_amount = transfer_data.amount; + } else if transfer_data.source == to_vault.unwrap_or_default() + && transfer_data.destination == user_to_token.unwrap_or_default() + { + swap_data.to_mint = to_mint.unwrap_or_default(); + swap_data.to_amount = transfer_data.amount; + } + } + } + if swap_data.from_mint != Pubkey::default() + || swap_data.to_mint != Pubkey::default() + || swap_data.from_amount != 0 + || swap_data.to_amount != 0 + { + (transfer_datas, Some(swap_data)) + } else { + (transfer_datas, None) + } } diff --git a/src/streaming/event_parser/core/common_event_parser.rs b/src/streaming/event_parser/core/common_event_parser.rs new file mode 100644 index 0000000..8442446 --- /dev/null +++ b/src/streaming/event_parser/core/common_event_parser.rs @@ -0,0 +1,15 @@ +use crate::streaming::event_parser::core::traits::UnifiedEvent; +use crate::streaming::event_parser::protocols::block::block_meta_event::BlockMetaEvent; + +pub struct CommonEventParser {} + +impl CommonEventParser { + pub fn generate_block_meta_event( + slot: u64, + block_hash: &str, + block_time_ms: i64, + ) -> Box { + let block_meta_event = BlockMetaEvent::new(slot, block_hash.to_string(), block_time_ms); + Box::new(block_meta_event) + } +} diff --git a/src/streaming/event_parser/core/mod.rs b/src/streaming/event_parser/core/mod.rs index 663ef85..c81525e 100755 --- a/src/streaming/event_parser/core/mod.rs +++ b/src/streaming/event_parser/core/mod.rs @@ -1,2 +1,3 @@ +pub mod common_event_parser; pub mod traits; pub use traits::{EventParser, UnifiedEvent}; diff --git a/src/streaming/event_parser/core/traits.rs b/src/streaming/event_parser/core/traits.rs index b3cc6e1..7439735 100755 --- a/src/streaming/event_parser/core/traits.rs +++ b/src/streaming/event_parser/core/traits.rs @@ -8,12 +8,11 @@ use solana_transaction_status::{ }; use std::fmt::Debug; use std::{collections::HashMap, str::FromStr}; -use std::sync::Arc; -use tokio::sync::RwLock; use crate::streaming::event_parser::common::{ - parse_transfer_datas_from_next_instructions, TransferData, + parse_transfer_datas_from_next_instructions, SwapData, TransferData, }; +use crate::streaming::event_parser::protocols::pumpswap::{PumpSwapBuyEvent, PumpSwapSellEvent}; use crate::streaming::event_parser::{ common::{utils::*, EventMetadata, EventType, ProtocolType}, protocols::{ @@ -22,78 +21,6 @@ use crate::streaming::event_parser::{ }, }; -// 解析缓存配置 -const PARSE_CACHE_SIZE: usize = 10000; -const CACHE_TTL_SECONDS: u64 = 300; // 5分钟 - -/// 解析结果缓存 -#[derive(Clone)] -pub struct ParseCacheEntry { - pub events: Vec>, - pub timestamp: u64, -} - -/// 事件解析缓存 -pub struct EventParseCache { - cache: Arc>>, - max_size: usize, - ttl_seconds: u64, -} - -impl EventParseCache { - pub fn new(max_size: usize, ttl_seconds: u64) -> Self { - Self { - cache: Arc::new(RwLock::new(HashMap::with_capacity(max_size))), - max_size, - ttl_seconds, - } - } - - pub async fn get(&self, key: &str) -> Option>> { - let cache = self.cache.read().await; - if let Some(entry) = cache.get(key) { - let now = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap() - .as_secs(); - - if now - entry.timestamp < self.ttl_seconds { - return Some(entry.events.clone()); - } - } - None - } - - pub async fn set(&self, key: String, events: Vec>) { - let mut cache = self.cache.write().await; - - // 清理过期条目 - if cache.len() >= self.max_size { - let now = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap() - .as_secs(); - - cache.retain(|_, entry| now - entry.timestamp < self.ttl_seconds); - } - - let entry = ParseCacheEntry { - events, - timestamp: std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap() - .as_secs(), - }; - - cache.insert(key, entry); - } -} - -// 全局解析缓存实例 -lazy_static::lazy_static! { - pub static ref PARSE_CACHE: EventParseCache = EventParseCache::new(PARSE_CACHE_SIZE, CACHE_TTL_SECONDS); -} - /// Unified Event Interface - All protocol events must implement this trait pub trait UnifiedEvent: Debug + Send + Sync { /// Get event ID @@ -132,7 +59,11 @@ pub trait UnifiedEvent: Debug + Send + Sync { } /// Set transfer datas - fn set_transfer_datas(&mut self, transfer_datas: Vec); + fn set_transfer_datas( + &mut self, + transfer_datas: Vec, + swap_data: Option, + ); /// Get index fn index(&self) -> String; @@ -141,6 +72,10 @@ pub trait UnifiedEvent: Debug + Send + Sync { /// 事件解析器trait - 定义了事件解析的核心方法 #[async_trait::async_trait] pub trait EventParser: Send + Sync { + /// 获取内联指令解析配置 + fn inner_instruction_configs(&self) -> HashMap<&'static str, Vec>; + /// 获取指令解析配置 + fn instruction_configs(&self) -> HashMap, Vec>; /// 从内联指令中解析事件数据 #[allow(clippy::too_many_arguments)] fn parse_events_from_inner_instruction( @@ -217,14 +152,15 @@ pub trait EventParser: Send + Sync { }) { events.iter_mut().for_each(|event| { - let transfer_datas = + let (transfer_datas, swap_data) = parse_transfer_datas_from_next_instructions( + event.clone_boxed(), inn, -1_i8, &accounts, event.event_type(), ); - event.set_transfer_datas(transfer_datas); + event.set_transfer_datas(transfer_datas, swap_data); }); } instruction_events.extend(events); @@ -271,31 +207,34 @@ pub trait EventParser: Send + Sync { program_received_time_ms: i64, bot_wallet: Option, ) -> Result>> { - // TODO: bug - 待优化 // // 生成缓存键 // let cache_key = format!("{}_{}_{}", signature, slot.unwrap_or(0), program_received_time_ms); - + // // 尝试从缓存获取 // if let Some(cached_events) = PARSE_CACHE.get(&cache_key).await { // return Ok(cached_events); // } - + let transaction = tx.transaction; // 检查交易元数据 - let meta = tx - .meta - .as_ref() - .ok_or_else(|| anyhow::anyhow!("Missing transaction metadata"))?; + let meta = + tx.meta.as_ref().ok_or_else(|| anyhow::anyhow!("Missing transaction metadata"))?; let mut address_table_lookups: Vec = vec![]; let mut inner_instructions: Vec = vec![]; if meta.err.is_none() { // 正确处理OptionSerializer类型 - if let solana_transaction_status::option_serializer::OptionSerializer::Some(meta_inner_instructions) = &meta.inner_instructions { + if let solana_transaction_status::option_serializer::OptionSerializer::Some( + meta_inner_instructions, + ) = &meta.inner_instructions + { inner_instructions = meta_inner_instructions.clone(); } - if let solana_transaction_status::option_serializer::OptionSerializer::Some(loaded_addresses) = &meta.loaded_addresses { + if let solana_transaction_status::option_serializer::OptionSerializer::Some( + loaded_addresses, + ) = &meta.loaded_addresses + { for lookup in &loaded_addresses.writable { if let Ok(pubkey) = Pubkey::from_str(lookup) { address_table_lookups.push(pubkey); @@ -364,14 +303,15 @@ pub trait EventParser: Send + Sync { { if !events.is_empty() { events.iter_mut().for_each(|event| { - let transfer_datas = + let (transfer_datas, swap_data) = parse_transfer_datas_from_next_instructions( + event.clone_boxed(), &inner_instruction, index as i8, &accounts, event.event_type(), ); - event.set_transfer_datas(transfer_datas); + event.set_transfer_datas(transfer_datas, swap_data); }); instruction_events.extend(events); } @@ -389,14 +329,15 @@ pub trait EventParser: Send + Sync { { if !events.is_empty() { events.iter_mut().for_each(|event| { - let transfer_datas = + let (transfer_datas, swap_data) = parse_transfer_datas_from_next_instructions( + event.clone_boxed(), &inner_instruction, index as i8, &accounts, event.event_type(), ); - event.set_transfer_datas(transfer_datas); + event.set_transfer_datas(transfer_datas, swap_data); }); inner_instruction_events.extend(events); } @@ -422,13 +363,17 @@ pub trait EventParser: Send + Sync { // 嵌套指令 let i_index_parts: Vec<&str> = i_index.split(".").collect(); let in_index_parts: Vec<&str> = in_index.split(".").collect(); - - if !i_index_parts.is_empty() && !in_index_parts.is_empty() - && i_index_parts[0] == in_index_parts[0] { - let i_index_child_index = i_index_parts.get(1) + + if !i_index_parts.is_empty() + && !in_index_parts.is_empty() + && i_index_parts[0] == in_index_parts[0] + { + let i_index_child_index = i_index_parts + .get(1) .and_then(|s| s.parse::().ok()) .unwrap_or(0); - let in_index_child_index = in_index_parts.get(1) + let in_index_child_index = in_index_parts + .get(1) .and_then(|s| s.parse::().ok()) .unwrap_or(0); if in_index_child_index > i_index_child_index { @@ -441,12 +386,12 @@ pub trait EventParser: Send + Sync { } } } - + let result = self.process_events(instruction_events, bot_wallet); - + // 缓存结果 // PARSE_CACHE.set(cache_key, result.clone()).await; - + Ok(result) } @@ -476,8 +421,36 @@ pub trait EventParser: Send + Sync { } else { trade_info.is_dev_create_token_trade = false; } - } - if let Some(pool_info) = event.as_any().downcast_ref::() { + if trade_info.metadata.swap_data.is_some() { + trade_info.metadata.swap_data.as_mut().unwrap().from_amount = + if trade_info.is_buy { + trade_info.sol_amount + } else { + trade_info.token_amount + }; + trade_info.metadata.swap_data.as_mut().unwrap().to_amount = if trade_info.is_buy + { + trade_info.token_amount + } else { + trade_info.sol_amount + }; + } + } else if let Some(trade_info) = event.as_any_mut().downcast_mut::() { + if trade_info.metadata.swap_data.is_some() { + trade_info.metadata.swap_data.as_mut().unwrap().from_amount = + trade_info.user_quote_amount_in; + trade_info.metadata.swap_data.as_mut().unwrap().to_amount = + trade_info.base_amount_out; + } + } else if let Some(trade_info) = event.as_any_mut().downcast_mut::() + { + if trade_info.metadata.swap_data.is_some() { + trade_info.metadata.swap_data.as_mut().unwrap().from_amount = + trade_info.base_amount_in; + trade_info.metadata.swap_data.as_mut().unwrap().to_amount = + trade_info.user_quote_amount_out; + } + } else if let Some(pool_info) = event.as_any().downcast_ref::() { bonk_dev_address = Some(pool_info.creator); } else if let Some(trade_info) = event.as_any_mut().downcast_mut::() { if Some(trade_info.payer) == bonk_dev_address { @@ -491,14 +464,17 @@ pub trait EventParser: Send + Sync { let now = chrono::Utc::now().timestamp_millis(); event.set_program_handle_time_consuming_ms(now - event.program_received_time_ms()); } - + // 记录处理时间 let processing_time = start_time.elapsed(); if processing_time.as_millis() > 10 { - log::warn!("Event processing took {}ms for {} events", - processing_time.as_millis(), events.len()); + log::warn!( + "Event processing took {}ms for {} events", + processing_time.as_millis(), + events.len() + ); } - + events } @@ -564,6 +540,8 @@ impl Clone for Box { /// 通用事件解析器配置 #[derive(Debug, Clone)] pub struct GenericEventParseConfig { + pub program_id: Pubkey, + pub protocol_type: ProtocolType, pub inner_instruction_discriminator: &'static str, pub instruction_discriminator: &'static [u8], pub event_type: EventType, @@ -581,19 +559,14 @@ pub type InstructionEventParser = /// 通用事件解析器基类 pub struct GenericEventParser { - program_id: Pubkey, - protocol_type: ProtocolType, - inner_instruction_configs: HashMap<&'static str, Vec>, - instruction_configs: HashMap, Vec>, + pub program_ids: Vec, + pub inner_instruction_configs: HashMap<&'static str, Vec>, + pub instruction_configs: HashMap, Vec>, } impl GenericEventParser { /// 创建新的通用事件解析器 - pub fn new( - program_id: Pubkey, - protocol_type: ProtocolType, - configs: Vec, - ) -> Self { + pub fn new(program_ids: Vec, configs: Vec) -> Self { // 预分配容量,避免动态扩容 let mut inner_instruction_configs = HashMap::with_capacity(configs.len()); let mut instruction_configs = HashMap::with_capacity(configs.len()); @@ -609,12 +582,7 @@ impl GenericEventParser { .push(config); } - Self { - program_id, - protocol_type, - inner_instruction_configs, - instruction_configs, - } + Self { program_ids, inner_instruction_configs, instruction_configs } } /// 通用的内联指令解析方法 @@ -629,10 +597,7 @@ impl GenericEventParser { program_received_time_ms: i64, index: String, ) -> Option> { - let timestamp = block_time.unwrap_or(Timestamp { - seconds: 0, - nanos: 0, - }); + let timestamp = block_time.unwrap_or(Timestamp { seconds: 0, nanos: 0 }); let block_time_ms = timestamp.seconds * 1000 + (timestamp.nanos as i64) / 1_000_000; let metadata = EventMetadata::new( signature.to_string(), @@ -640,9 +605,9 @@ impl GenericEventParser { slot, timestamp.seconds, block_time_ms, - self.protocol_type.clone(), + config.protocol_type.clone(), config.event_type.clone(), - self.program_id, + config.program_id, index, program_received_time_ms, ); @@ -662,10 +627,7 @@ impl GenericEventParser { program_received_time_ms: i64, index: String, ) -> Option> { - let timestamp = block_time.unwrap_or(Timestamp { - seconds: 0, - nanos: 0, - }); + let timestamp = block_time.unwrap_or(Timestamp { seconds: 0, nanos: 0 }); let block_time_ms = timestamp.seconds * 1000 + (timestamp.nanos as i64) / 1_000_000; let metadata = EventMetadata::new( signature.to_string(), @@ -673,9 +635,9 @@ impl GenericEventParser { slot, timestamp.seconds, block_time_ms, - self.protocol_type.clone(), + config.protocol_type.clone(), config.event_type.clone(), - self.program_id, + config.program_id, index, program_received_time_ms, ); @@ -685,6 +647,12 @@ impl GenericEventParser { #[async_trait::async_trait] impl EventParser for GenericEventParser { + fn inner_instruction_configs(&self) -> HashMap<&'static str, Vec> { + self.inner_instruction_configs.clone() + } + fn instruction_configs(&self) -> HashMap, Vec> { + self.instruction_configs.clone() + } /// 从内联指令中解析事件数据 #[allow(clippy::too_many_arguments)] fn parse_events_from_inner_instruction( @@ -697,9 +665,8 @@ impl EventParser for GenericEventParser { index: String, ) -> Vec> { let inner_instruction_data = inner_instruction.data.clone(); - let inner_instruction_data_decoded = bs58::decode(inner_instruction_data) - .into_vec() - .unwrap_or_else(|_| vec![]); + let inner_instruction_data_decoded = + bs58::decode(inner_instruction_data).into_vec().unwrap_or_else(|_| vec![]); if inner_instruction_data_decoded.len() < 16 { return Vec::new(); } @@ -756,12 +723,12 @@ impl EventParser for GenericEventParser { continue; } - let account_pubkeys: Vec = instruction - .accounts - .iter() - .map(|&idx| accounts[idx as usize]) - .collect(); + let account_pubkeys: Vec = + instruction.accounts.iter().map(|&idx| accounts[idx as usize]).collect(); for config in configs { + if config.program_id != program_id { + continue; + } if let Some(event) = self.parse_instruction_event( config, data, @@ -782,13 +749,10 @@ impl EventParser for GenericEventParser { } fn should_handle(&self, program_id: &Pubkey) -> bool { - *program_id == self.program_id + self.program_ids.contains(program_id) } fn supported_program_ids(&self) -> Vec { - vec![self.program_id] + self.program_ids.clone() } } - -pub struct SDKSystemEventParser {} -impl SDKSystemEventParser {} diff --git a/src/streaming/event_parser/protocols/block/block_meta_event.rs b/src/streaming/event_parser/protocols/block/block_meta_event.rs new file mode 100644 index 0000000..f4bee6d --- /dev/null +++ b/src/streaming/event_parser/protocols/block/block_meta_event.rs @@ -0,0 +1,34 @@ +use crate::impl_unified_event; +use crate::streaming::event_parser::common::{types::EventType, EventMetadata}; +use borsh::BorshDeserialize; +use serde::{Deserialize, Serialize}; + +/// Block元数据事件 +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] +pub struct BlockMetaEvent { + #[borsh(skip)] + pub metadata: EventMetadata, + pub slot: u64, + pub block_hash: String, +} + +impl BlockMetaEvent { + pub fn new(slot: u64, block_hash: String, block_time_ms: i64) -> Self { + let metadata = EventMetadata::new( + format!("block_{}_{}", slot, block_hash), + "".to_string(), + slot, + block_time_ms / 1000, + block_time_ms, + crate::streaming::event_parser::common::types::ProtocolType::Common, + EventType::BlockMeta, + solana_sdk::pubkey::Pubkey::default(), + "".to_string(), + chrono::Utc::now().timestamp_millis(), + ); + Self { metadata, slot, block_hash } + } +} + +// 使用macro生成UnifiedEvent实现 +impl_unified_event!(BlockMetaEvent,); diff --git a/src/streaming/event_parser/protocols/block/mod.rs b/src/streaming/event_parser/protocols/block/mod.rs new file mode 100644 index 0000000..a8d6d32 --- /dev/null +++ b/src/streaming/event_parser/protocols/block/mod.rs @@ -0,0 +1 @@ +pub mod block_meta_event; \ No newline at end of file diff --git a/src/streaming/event_parser/protocols/bonk/parser.rs b/src/streaming/event_parser/protocols/bonk/parser.rs index 10e44be..5d34a3d 100755 --- a/src/streaming/event_parser/protocols/bonk/parser.rs +++ b/src/streaming/event_parser/protocols/bonk/parser.rs @@ -1,3 +1,5 @@ +use std::collections::HashMap; + use prost_types::Timestamp; use solana_sdk::{instruction::CompiledInstruction, pubkey::Pubkey}; use solana_transaction_status::UiCompiledInstruction; @@ -32,6 +34,8 @@ impl BonkEventParser { // Configure all event types let configs = vec![ GenericEventParseConfig { + program_id: BONK_PROGRAM_ID, + protocol_type: ProtocolType::Bonk, inner_instruction_discriminator: discriminators::TRADE_EVENT, instruction_discriminator: discriminators::BUY_EXACT_IN, event_type: EventType::BonkBuyExactIn, @@ -39,6 +43,8 @@ impl BonkEventParser { instruction_parser: Self::parse_buy_exact_in_instruction, }, GenericEventParseConfig { + program_id: BONK_PROGRAM_ID, + protocol_type: ProtocolType::Bonk, inner_instruction_discriminator: discriminators::TRADE_EVENT, instruction_discriminator: discriminators::BUY_EXACT_OUT, event_type: EventType::BonkBuyExactOut, @@ -46,6 +52,8 @@ impl BonkEventParser { instruction_parser: Self::parse_buy_exact_out_instruction, }, GenericEventParseConfig { + program_id: BONK_PROGRAM_ID, + protocol_type: ProtocolType::Bonk, inner_instruction_discriminator: discriminators::TRADE_EVENT, instruction_discriminator: discriminators::SELL_EXACT_IN, event_type: EventType::BonkSellExactIn, @@ -53,6 +61,8 @@ impl BonkEventParser { instruction_parser: Self::parse_sell_exact_in_instruction, }, GenericEventParseConfig { + program_id: BONK_PROGRAM_ID, + protocol_type: ProtocolType::Bonk, inner_instruction_discriminator: discriminators::TRADE_EVENT, instruction_discriminator: discriminators::SELL_EXACT_OUT, event_type: EventType::BonkSellExactOut, @@ -60,6 +70,8 @@ impl BonkEventParser { instruction_parser: Self::parse_sell_exact_out_instruction, }, GenericEventParseConfig { + program_id: BONK_PROGRAM_ID, + protocol_type: ProtocolType::Bonk, inner_instruction_discriminator: discriminators::POOL_CREATE_EVENT, instruction_discriminator: discriminators::INITIALIZE, event_type: EventType::BonkInitialize, @@ -67,6 +79,8 @@ impl BonkEventParser { instruction_parser: Self::parse_initialize_instruction, }, GenericEventParseConfig { + program_id: BONK_PROGRAM_ID, + protocol_type: ProtocolType::Bonk, inner_instruction_discriminator: "", instruction_discriminator: discriminators::MIGRATE_TO_AMM, event_type: EventType::BonkMigrateToAmm, @@ -74,6 +88,8 @@ impl BonkEventParser { instruction_parser: Self::parse_migrate_to_amm_instruction, }, GenericEventParseConfig { + program_id: BONK_PROGRAM_ID, + protocol_type: ProtocolType::Bonk, inner_instruction_discriminator: "", instruction_discriminator: discriminators::MIGRATE_TO_CP_SWAP, event_type: EventType::BonkMigrateToCpswap, @@ -82,7 +98,7 @@ impl BonkEventParser { }, ]; - let inner = GenericEventParser::new(BONK_PROGRAM_ID, ProtocolType::Bonk, configs); + let inner = GenericEventParser::new(vec![BONK_PROGRAM_ID], configs); Self { inner } } @@ -526,6 +542,12 @@ impl BonkEventParser { #[async_trait::async_trait] impl EventParser for BonkEventParser { + fn inner_instruction_configs(&self) -> HashMap<&'static str, Vec> { + self.inner.inner_instruction_configs() + } + fn instruction_configs(&self) -> HashMap, Vec> { + self.inner.instruction_configs() + } fn parse_events_from_inner_instruction( &self, inner_instruction: &UiCompiledInstruction, diff --git a/src/streaming/event_parser/protocols/mod.rs b/src/streaming/event_parser/protocols/mod.rs index e0d8aa5..ece390b 100755 --- a/src/streaming/event_parser/protocols/mod.rs +++ b/src/streaming/event_parser/protocols/mod.rs @@ -3,9 +3,13 @@ pub mod pumpswap; pub mod bonk; pub mod raydium_cpmm; pub mod raydium_clmm; +pub mod block; +pub mod mutil; pub use pumpfun::PumpFunEventParser; pub use pumpswap::PumpSwapEventParser; pub use bonk::BonkEventParser; pub use raydium_cpmm::RaydiumCpmmEventParser; -pub use raydium_clmm::RaydiumClmmEventParser; \ No newline at end of file +pub use raydium_clmm::RaydiumClmmEventParser; +pub use block::block_meta_event::BlockMetaEvent; +pub use mutil::MutilEventParser; \ No newline at end of file diff --git a/src/streaming/event_parser/protocols/mutil/mod.rs b/src/streaming/event_parser/protocols/mutil/mod.rs new file mode 100644 index 0000000..852450f --- /dev/null +++ b/src/streaming/event_parser/protocols/mutil/mod.rs @@ -0,0 +1,3 @@ +pub mod parser; + +pub use parser::MutilEventParser; \ No newline at end of file diff --git a/src/streaming/event_parser/protocols/mutil/parser.rs b/src/streaming/event_parser/protocols/mutil/parser.rs new file mode 100755 index 0000000..9b5d75b --- /dev/null +++ b/src/streaming/event_parser/protocols/mutil/parser.rs @@ -0,0 +1,95 @@ +use std::collections::HashMap; + +use prost_types::Timestamp; +use solana_sdk::{instruction::CompiledInstruction, pubkey::Pubkey}; +use solana_transaction_status::UiCompiledInstruction; + +use crate::streaming::event_parser::{ + core::traits::{EventParser, GenericEventParseConfig, GenericEventParser, UnifiedEvent}, + EventParserFactory, Protocol, +}; + +pub struct MutilEventParser { + inner: GenericEventParser, +} + +impl MutilEventParser { + pub fn new(protocols: Vec) -> Self { + let mut inner = GenericEventParser::new(vec![], vec![]); + // Configure all event types + for protocol in protocols { + let parse = EventParserFactory::create_parser(protocol); + + // 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); + } + + // 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); + } + + // Append program_ids (this is already appending) + inner.program_ids.extend(parse.supported_program_ids().clone()); + } + Self { inner } + } +} + +#[async_trait::async_trait] +impl EventParser for MutilEventParser { + fn inner_instruction_configs(&self) -> HashMap<&'static str, Vec> { + self.inner.inner_instruction_configs() + } + fn instruction_configs(&self) -> HashMap, Vec> { + self.inner.instruction_configs() + } + fn parse_events_from_inner_instruction( + &self, + inner_instruction: &UiCompiledInstruction, + signature: &str, + slot: u64, + block_time: Option, + program_received_time_ms: i64, + index: String, + ) -> Vec> { + self.inner.parse_events_from_inner_instruction( + inner_instruction, + signature, + slot, + block_time, + program_received_time_ms, + index, + ) + } + + fn parse_events_from_instruction( + &self, + instruction: &CompiledInstruction, + accounts: &[Pubkey], + signature: &str, + slot: u64, + block_time: Option, + program_received_time_ms: i64, + index: String, + ) -> Vec> { + self.inner.parse_events_from_instruction( + instruction, + accounts, + signature, + slot, + block_time, + program_received_time_ms, + index, + ) + } + + fn should_handle(&self, program_id: &Pubkey) -> bool { + self.inner.should_handle(program_id) + } + + fn supported_program_ids(&self) -> Vec { + self.inner.supported_program_ids() + } +} diff --git a/src/streaming/event_parser/protocols/pumpfun/parser.rs b/src/streaming/event_parser/protocols/pumpfun/parser.rs index 5fdc88a..5ebfefb 100755 --- a/src/streaming/event_parser/protocols/pumpfun/parser.rs +++ b/src/streaming/event_parser/protocols/pumpfun/parser.rs @@ -1,3 +1,5 @@ +use std::collections::HashMap; + use prost_types::Timestamp; use solana_sdk::{instruction::CompiledInstruction, pubkey::Pubkey}; use solana_transaction_status::UiCompiledInstruction; @@ -28,6 +30,8 @@ impl PumpFunEventParser { // 配置所有事件类型 let configs = vec![ GenericEventParseConfig { + program_id: PUMPFUN_PROGRAM_ID, + protocol_type: ProtocolType::PumpFun, inner_instruction_discriminator: discriminators::CREATE_TOKEN_EVENT, instruction_discriminator: discriminators::CREATE_TOKEN_IX, event_type: EventType::PumpFunCreateToken, @@ -35,6 +39,8 @@ impl PumpFunEventParser { instruction_parser: Self::parse_create_token_instruction, }, GenericEventParseConfig { + program_id: PUMPFUN_PROGRAM_ID, + protocol_type: ProtocolType::PumpFun, inner_instruction_discriminator: discriminators::TRADE_EVENT, instruction_discriminator: discriminators::BUY_IX, event_type: EventType::PumpFunBuy, @@ -42,6 +48,8 @@ impl PumpFunEventParser { instruction_parser: Self::parse_buy_instruction, }, GenericEventParseConfig { + program_id: PUMPFUN_PROGRAM_ID, + protocol_type: ProtocolType::PumpFun, inner_instruction_discriminator: discriminators::TRADE_EVENT, instruction_discriminator: discriminators::SELL_IX, event_type: EventType::PumpFunSell, @@ -50,7 +58,7 @@ impl PumpFunEventParser { }, ]; - let inner = GenericEventParser::new(PUMPFUN_PROGRAM_ID, ProtocolType::PumpFun, configs); + let inner = GenericEventParser::new(vec![PUMPFUN_PROGRAM_ID], configs); Self { inner } } @@ -228,6 +236,12 @@ impl PumpFunEventParser { #[async_trait::async_trait] impl EventParser for PumpFunEventParser { + fn inner_instruction_configs(&self) -> HashMap<&'static str, Vec> { + self.inner.inner_instruction_configs() + } + fn instruction_configs(&self) -> HashMap, Vec> { + self.inner.instruction_configs() + } fn parse_events_from_inner_instruction( &self, inner_instruction: &UiCompiledInstruction, diff --git a/src/streaming/event_parser/protocols/pumpswap/parser.rs b/src/streaming/event_parser/protocols/pumpswap/parser.rs index 4dc4fd9..ce59bb3 100755 --- a/src/streaming/event_parser/protocols/pumpswap/parser.rs +++ b/src/streaming/event_parser/protocols/pumpswap/parser.rs @@ -1,3 +1,5 @@ +use std::collections::HashMap; + use prost_types::Timestamp; use solana_sdk::{instruction::CompiledInstruction, pubkey::Pubkey}; use solana_transaction_status::UiCompiledInstruction; @@ -31,6 +33,8 @@ impl PumpSwapEventParser { // 配置所有事件类型 let configs = vec![ GenericEventParseConfig { + program_id: PUMPSWAP_PROGRAM_ID, + protocol_type: ProtocolType::PumpSwap, inner_instruction_discriminator: discriminators::BUY_EVENT, instruction_discriminator: discriminators::BUY_IX, event_type: EventType::PumpSwapBuy, @@ -38,6 +42,8 @@ impl PumpSwapEventParser { instruction_parser: Self::parse_buy_instruction, }, GenericEventParseConfig { + program_id: PUMPSWAP_PROGRAM_ID, + protocol_type: ProtocolType::PumpSwap, inner_instruction_discriminator: discriminators::SELL_EVENT, instruction_discriminator: discriminators::SELL_IX, event_type: EventType::PumpSwapSell, @@ -45,6 +51,8 @@ impl PumpSwapEventParser { instruction_parser: Self::parse_sell_instruction, }, GenericEventParseConfig { + program_id: PUMPSWAP_PROGRAM_ID, + protocol_type: ProtocolType::PumpSwap, inner_instruction_discriminator: discriminators::CREATE_POOL_EVENT, instruction_discriminator: discriminators::CREATE_POOL_IX, event_type: EventType::PumpSwapCreatePool, @@ -52,6 +60,8 @@ impl PumpSwapEventParser { instruction_parser: Self::parse_create_pool_instruction, }, GenericEventParseConfig { + program_id: PUMPSWAP_PROGRAM_ID, + protocol_type: ProtocolType::PumpSwap, inner_instruction_discriminator: discriminators::DEPOSIT_EVENT, instruction_discriminator: discriminators::DEPOSIT_IX, event_type: EventType::PumpSwapDeposit, @@ -59,6 +69,8 @@ impl PumpSwapEventParser { instruction_parser: Self::parse_deposit_instruction, }, GenericEventParseConfig { + program_id: PUMPSWAP_PROGRAM_ID, + protocol_type: ProtocolType::PumpSwap, inner_instruction_discriminator: discriminators::WITHDRAW_EVENT, instruction_discriminator: discriminators::WITHDRAW_IX, event_type: EventType::PumpSwapWithdraw, @@ -67,7 +79,7 @@ impl PumpSwapEventParser { }, ]; - let inner = GenericEventParser::new(PUMPSWAP_PROGRAM_ID, ProtocolType::PumpSwap, configs); + let inner = GenericEventParser::new(vec![PUMPSWAP_PROGRAM_ID], configs); Self { inner } } @@ -374,6 +386,12 @@ impl PumpSwapEventParser { #[async_trait::async_trait] impl EventParser for PumpSwapEventParser { + fn inner_instruction_configs(&self) -> HashMap<&'static str, Vec> { + self.inner.inner_instruction_configs() + } + fn instruction_configs(&self) -> HashMap, Vec> { + self.inner.instruction_configs() + } fn parse_events_from_inner_instruction( &self, inner_instruction: &UiCompiledInstruction, diff --git a/src/streaming/event_parser/protocols/raydium_clmm/parser.rs b/src/streaming/event_parser/protocols/raydium_clmm/parser.rs index 9a39826..c73d5d7 100755 --- a/src/streaming/event_parser/protocols/raydium_clmm/parser.rs +++ b/src/streaming/event_parser/protocols/raydium_clmm/parser.rs @@ -1,3 +1,5 @@ +use std::collections::HashMap; + use prost_types::Timestamp; use solana_sdk::{instruction::CompiledInstruction, pubkey::Pubkey}; use solana_transaction_status::UiCompiledInstruction; @@ -28,6 +30,8 @@ impl RaydiumClmmEventParser { // 配置所有事件类型 let configs = vec![ GenericEventParseConfig { + program_id: RAYDIUM_CLMM_PROGRAM_ID, + protocol_type: ProtocolType::RaydiumClmm, inner_instruction_discriminator: "", instruction_discriminator: discriminators::SWAP, event_type: EventType::RaydiumClmmSwap, @@ -35,6 +39,8 @@ impl RaydiumClmmEventParser { instruction_parser: Self::parse_swap_instruction, }, GenericEventParseConfig { + program_id: RAYDIUM_CLMM_PROGRAM_ID, + protocol_type: ProtocolType::RaydiumClmm, inner_instruction_discriminator: "", instruction_discriminator: discriminators::SWAP_V2, event_type: EventType::RaydiumClmmSwapV2, @@ -43,8 +49,7 @@ impl RaydiumClmmEventParser { }, ]; - let inner = - GenericEventParser::new(RAYDIUM_CLMM_PROGRAM_ID, ProtocolType::RaydiumClmm, configs); + let inner = GenericEventParser::new(vec![RAYDIUM_CLMM_PROGRAM_ID], configs); Self { inner } } @@ -144,6 +149,12 @@ impl RaydiumClmmEventParser { #[async_trait::async_trait] impl EventParser for RaydiumClmmEventParser { + fn inner_instruction_configs(&self) -> HashMap<&'static str, Vec> { + self.inner.inner_instruction_configs() + } + fn instruction_configs(&self) -> HashMap, Vec> { + self.inner.instruction_configs() + } fn parse_events_from_inner_instruction( &self, inner_instruction: &UiCompiledInstruction, diff --git a/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs b/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs index a8650f1..420f020 100755 --- a/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs +++ b/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs @@ -1,3 +1,5 @@ +use std::collections::HashMap; + use prost_types::Timestamp; use solana_sdk::{instruction::CompiledInstruction, pubkey::Pubkey}; use solana_transaction_status::UiCompiledInstruction; @@ -28,6 +30,8 @@ impl RaydiumCpmmEventParser { // 配置所有事件类型 let configs = vec![ GenericEventParseConfig { + program_id: RAYDIUM_CPMM_PROGRAM_ID, + protocol_type: ProtocolType::RaydiumCpmm, inner_instruction_discriminator: "", instruction_discriminator: discriminators::SWAP_BASE_IN, event_type: EventType::RaydiumCpmmSwapBaseInput, @@ -35,6 +39,8 @@ impl RaydiumCpmmEventParser { instruction_parser: Self::parse_swap_base_input_instruction, }, GenericEventParseConfig { + program_id: RAYDIUM_CPMM_PROGRAM_ID, + protocol_type: ProtocolType::RaydiumCpmm, inner_instruction_discriminator: "", instruction_discriminator: discriminators::SWAP_BASE_OUT, event_type: EventType::RaydiumCpmmSwapBaseOutput, @@ -43,8 +49,7 @@ impl RaydiumCpmmEventParser { }, ]; - let inner = - GenericEventParser::new(RAYDIUM_CPMM_PROGRAM_ID, ProtocolType::RaydiumCpmm, configs); + let inner = GenericEventParser::new(vec![RAYDIUM_CPMM_PROGRAM_ID], configs); Self { inner } } @@ -135,6 +140,12 @@ impl RaydiumCpmmEventParser { #[async_trait::async_trait] impl EventParser for RaydiumCpmmEventParser { + fn inner_instruction_configs(&self) -> HashMap<&'static str, Vec> { + self.inner.inner_instruction_configs() + } + fn instruction_configs(&self) -> HashMap, Vec> { + self.inner.instruction_configs() + } fn parse_events_from_inner_instruction( &self, inner_instruction: &UiCompiledInstruction, diff --git a/src/streaming/grpc/connection.rs b/src/streaming/grpc/connection.rs new file mode 100644 index 0000000..bfbc8f5 --- /dev/null +++ b/src/streaming/grpc/connection.rs @@ -0,0 +1,33 @@ +use std::time::Duration; +use tonic::transport::channel::ClientTlsConfig; +use yellowstone_grpc_client::{GeyserGrpcClient, Interceptor}; +use crate::common::AnyResult; +use crate::streaming::common::constants::{ + DEFAULT_CONNECT_TIMEOUT, DEFAULT_REQUEST_TIMEOUT, DEFAULT_MAX_DECODING_MESSAGE_SIZE +}; + +/// gRPC连接池 - 简化版本 +pub struct GrpcConnectionPool { + endpoint: String, + x_token: Option, +} + +impl GrpcConnectionPool { + pub fn new(endpoint: String, x_token: Option) -> Self { + Self { + endpoint, + x_token, + } + } + + pub async fn create_connection(&self) -> AnyResult> { + let builder = GeyserGrpcClient::build_from_shared(self.endpoint.clone())? + .x_token(self.x_token.clone())? + .tls_config(ClientTlsConfig::new().with_native_roots())? + .max_decoding_message_size(DEFAULT_MAX_DECODING_MESSAGE_SIZE) + .connect_timeout(Duration::from_secs(DEFAULT_CONNECT_TIMEOUT)) + .timeout(Duration::from_secs(DEFAULT_REQUEST_TIMEOUT)); + + Ok(builder.connect().await?) + } +} diff --git a/src/streaming/grpc/event_processor.rs b/src/streaming/grpc/event_processor.rs new file mode 100644 index 0000000..0780bf3 --- /dev/null +++ b/src/streaming/grpc/event_processor.rs @@ -0,0 +1,201 @@ +use std::sync::Arc; + +use solana_sdk::pubkey::Pubkey; + +use super::types::EventPretty; +use crate::common::AnyResult; +use crate::streaming::common::{ + EventBatchProcessor as EventBatchCollector, MetricsManager, StreamClientConfig as ClientConfig, +}; +use crate::streaming::event_parser::core::common_event_parser::CommonEventParser; +use crate::streaming::event_parser::EventParser; +use crate::streaming::event_parser::{ + core::traits::UnifiedEvent, protocols::mutil::parser::MutilEventParser, Protocol, +}; + +/// 事件处理器 +pub struct EventProcessor { + pub(crate) metrics_manager: MetricsManager, + pub(crate) config: ClientConfig, +} + +impl EventProcessor { + /// 创建新的事件处理器 + pub fn new(metrics_manager: MetricsManager, config: ClientConfig) -> Self { + Self { metrics_manager, config } + } + + /// 使用性能监控处理事件交易 + pub async fn process_event_transaction_with_metrics( + &self, + event_pretty: EventPretty, + callback: &F, + bot_wallet: Option, + protocols: Vec, + ) -> AnyResult<()> + where + F: Fn(Box) + Send + Sync, + { + match event_pretty { + EventPretty::Transaction(transaction_pretty) => { + 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 all_events = parser + .parse_transaction( + transaction_pretty.tx.clone(), + &signature, + Some(slot), + transaction_pretty.block_time.map(|ts| prost_types::Timestamp { + seconds: ts.seconds, + nanos: ts.nanos, + }), + program_received_time_ms, + bot_wallet, + ) + .await + .unwrap_or_else(|_e| vec![]); + + // 保存事件数量用于日志记录 + let event_count = all_events.len(); + + // 批量处理事件 + if !all_events.is_empty() { + for event in all_events { + callback(event); + } + } + + // 更新性能指标 + let processing_time = start_time.elapsed(); + let processing_time_ms = processing_time.as_millis() as f64; + + // 更新性能指标(如果启用) + if self.config.enable_metrics { + self.metrics_manager + .update_metrics(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 block_time_ms = block_meta_pretty + .block_time + .map(|ts| ts.seconds * 1000 + ts.nanos as i64 / 1_000_000) + .unwrap_or_else(|| chrono::Utc::now().timestamp_millis()); + let block_meta_event = CommonEventParser::generate_block_meta_event( + block_meta_pretty.slot, + &block_meta_pretty.block_hash, + block_time_ms, + ); + callback(block_meta_event); + } + } + + Ok(()) + } + + /// 使用批处理处理事件交易 + pub async fn process_event_transaction_with_batch( + &self, + event_pretty: EventPretty, + batch_processor: &mut EventBatchCollector, + bot_wallet: Option, + protocols: Vec, + ) -> AnyResult<()> + where + F: Fn(Vec>) + Send + Sync + 'static, + { + match event_pretty { + EventPretty::Transaction(transaction_pretty) => { + 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 result = parser + .parse_transaction( + transaction_pretty.tx.clone(), + &signature, + Some(slot), + transaction_pretty.block_time.map(|ts| prost_types::Timestamp { + seconds: ts.seconds, + nanos: ts.nanos, + }), + program_received_time_ms, + bot_wallet, + ) + .await; + + // 处理解析结果并使用批处理器 + let total_events = match result { + Ok(events) => { + let event_count = events.len(); + if !events.is_empty() { + log::info!("Parsed {} events", event_count); + log::info!("Adding {} events to batch processor", event_count); + for event in events { + if self.config.batch.enabled { + batch_processor.add_event(event); + } else { + // 如果批处理被禁用,直接调用回调 + // 这里需要将单个事件包装成Vec来调用批处理回调 + let single_event_batch = vec![event]; + (batch_processor.callback)(single_event_batch); + } + } + } + event_count + } + Err(e) => { + log::warn!("Failed to parse transaction: {:?}", e); + 0 + } + }; + + // 添加调试信息 + if total_events > 0 { + log::info!( + "Total events parsed: {} for transaction {}", + total_events, + signature + ); + } + + // 更新性能指标 + let processing_time = start_time.elapsed(); + 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.log_slow_processing(processing_time_ms, total_events); + } + EventPretty::BlockMeta(block_meta_pretty) => { + let block_time_ms = block_meta_pretty + .block_time + .map(|ts| ts.seconds * 1000 + ts.nanos as i64 / 1_000_000) + .unwrap_or_else(|| chrono::Utc::now().timestamp_millis()); + let block_meta_event = CommonEventParser::generate_block_meta_event( + block_meta_pretty.slot, + &block_meta_pretty.block_hash, + block_time_ms, + ); + (batch_processor.callback)(vec![block_meta_event]); + } + } + + Ok(()) + } +} diff --git a/src/streaming/grpc/mod.rs b/src/streaming/grpc/mod.rs new file mode 100644 index 0000000..44713a9 --- /dev/null +++ b/src/streaming/grpc/mod.rs @@ -0,0 +1,25 @@ +// gRPC 相关模块 +pub mod connection; +pub mod types; +pub mod subscription; +pub mod stream_handler; +pub mod event_processor; + +// 重新导出主要类型 +pub use connection::*; +pub use types::*; +pub use subscription::*; +pub use stream_handler::*; +pub use event_processor::*; + +// 从公用模块重新导出 +pub use crate::streaming::common::{ + StreamClientConfig as ClientConfig, + PerformanceMetrics, + MetricsManager, + EventBatchProcessor as EventBatchCollector, + BackpressureStrategy, + BatchConfig, + BackpressureConfig, + ConnectionConfig, +}; \ No newline at end of file diff --git a/src/streaming/grpc/stream_handler.rs b/src/streaming/grpc/stream_handler.rs new file mode 100644 index 0000000..222995d --- /dev/null +++ b/src/streaming/grpc/stream_handler.rs @@ -0,0 +1,124 @@ +use chrono::Local; +use futures::{channel::mpsc, sink::Sink, SinkExt}; +use log::info; +use yellowstone_grpc_proto::geyser::{ + subscribe_update::UpdateOneof, SubscribeRequest, SubscribeRequestPing, SubscribeUpdate, +}; + +use super::types::{BlockMetaPretty, EventPretty, TransactionPretty}; +use crate::common::AnyResult; +use crate::streaming::common::BackpressureStrategy; + +/// 流消息处理器 +pub struct StreamHandler; + +impl StreamHandler { + /// 处理单个流消息 + pub async fn handle_stream_message( + msg: SubscribeUpdate, + tx: &mut mpsc::Sender, + subscribe_tx: &mut (impl Sink + Unpin), + backpressure_strategy: BackpressureStrategy, + ) -> AnyResult<()> { + let created_at = msg.created_at; + match msg.update_oneof { + Some(UpdateOneof::BlockMeta(sut)) => { + let block_meta_pretty = BlockMetaPretty::from((sut, created_at)); + log::info!("Received block meta: {:?}", block_meta_pretty); + Self::handle_backpressure( + tx, + EventPretty::BlockMeta(block_meta_pretty), + backpressure_strategy, + ) + .await?; + } + Some(UpdateOneof::Transaction(sut)) => { + let transaction_pretty = TransactionPretty::from((sut, created_at)); + log::info!( + "Received transaction: {} at slot {}", + transaction_pretty.signature, + transaction_pretty.slot + ); + + // 根据背压策略处理发送 + Self::handle_backpressure( + tx, + EventPretty::Transaction(transaction_pretty), + backpressure_strategy, + ) + .await?; + } + Some(UpdateOneof::Ping(_)) => { + subscribe_tx + .send(SubscribeRequest { + ping: Some(SubscribeRequestPing { id: 1 }), + ..Default::default() + }) + .await?; + info!("service is ping: {}", Local::now()); + } + Some(UpdateOneof::Pong(_)) => { + info!("service is pong: {}", Local::now()); + } + _ => { + log::debug!("Received other message type"); + } + } + Ok(()) + } + + /// 处理背压策略 + async fn handle_backpressure( + tx: &mut mpsc::Sender, + event_pretty: EventPretty, + backpressure_strategy: BackpressureStrategy, + ) -> AnyResult<()> { + match backpressure_strategy { + BackpressureStrategy::Block => { + // 阻塞等待,直到有空间 + if let Err(e) = tx.send(event_pretty).await { + log::error!("Failed to send transaction to channel: {:?}", e); + return Err(anyhow::anyhow!("Channel send failed: {:?}", e)); + } + } + BackpressureStrategy::Drop => { + // 尝试发送,如果失败则丢弃 + if let Err(e) = tx.try_send(event_pretty) { + if e.is_full() { + log::warn!("Channel is full, dropping transaction"); + } else { + log::error!("Channel is closed: {:?}", e); + return Err(anyhow::anyhow!("Channel is closed: {:?}", e)); + } + } + } + BackpressureStrategy::Retry { max_attempts, wait_ms } => { + // 重试有限次数 + let mut retry_count = 0; + loop { + match tx.try_send(event_pretty.clone()) { + Ok(_) => break, + Err(e) => { + if e.is_full() { + retry_count += 1; + if retry_count >= max_attempts { + log::warn!( + "Channel is full after {} attempts, dropping transaction", + retry_count + ); + break; + } + tokio::time::sleep(tokio::time::Duration::from_millis(wait_ms)) + .await; + } else { + log::error!("Channel is closed: {:?}", e); + return Err(anyhow::anyhow!("Channel is closed: {:?}", e)); + } + } + } + } + } + } + Ok(()) + } +} diff --git a/src/streaming/grpc/subscription.rs b/src/streaming/grpc/subscription.rs new file mode 100644 index 0000000..2969a1c --- /dev/null +++ b/src/streaming/grpc/subscription.rs @@ -0,0 +1,106 @@ +use futures::{channel::mpsc, sink::Sink, Stream}; +use maplit::hashmap; +use std::{collections::HashMap, time::Duration}; +use tonic::{transport::channel::ClientTlsConfig, Status}; +use yellowstone_grpc_client::{GeyserGrpcClient, Interceptor}; +use yellowstone_grpc_proto::geyser::{ + CommitmentLevel, SubscribeRequest, SubscribeRequestFilterBlocksMeta, + SubscribeRequestFilterTransactions, SubscribeUpdate, +}; + +use super::types::TransactionsFilterMap; +use crate::common::AnyResult; +use crate::streaming::common::StreamClientConfig as ClientConfig; + +/// 订阅管理器 +#[derive(Clone)] +pub struct SubscriptionManager { + endpoint: String, + x_token: Option, + config: ClientConfig, +} + +impl SubscriptionManager { + /// 创建新的订阅管理器 + pub fn new(endpoint: String, x_token: Option, config: ClientConfig) -> Self { + Self { endpoint, x_token, config } + } + + /// 创建 gRPC 连接 + pub async fn connect(&self) -> AnyResult> { + let builder = GeyserGrpcClient::build_from_shared(self.endpoint.clone())? + .x_token(self.x_token.clone())? + .tls_config(ClientTlsConfig::new().with_native_roots())? + .max_decoding_message_size(self.config.connection.max_decoding_message_size) + .connect_timeout(Duration::from_secs(self.config.connection.connect_timeout)) + .timeout(Duration::from_secs(self.config.connection.request_timeout)); + Ok(builder.connect().await?) + } + + /// 创建订阅请求并返回流 + pub async fn subscribe_with_request( + &self, + transactions: TransactionsFilterMap, + commitment: Option, + ) -> AnyResult<( + impl Sink, + impl Stream>, + )> { + let subscribe_request = SubscribeRequest { + transactions, + blocks_meta: hashmap! { "".to_owned() => SubscribeRequestFilterBlocksMeta {} }, + commitment: if let Some(commitment) = commitment { + Some(commitment as i32) + } else { + Some(CommitmentLevel::Processed.into()) + }, + ..Default::default() + }; + + let mut client = self.connect().await?; + let (sink, stream) = client.subscribe_with_request(Some(subscribe_request)).await?; + Ok((sink, stream)) + } + + /// 生成订阅请求过滤器 + pub fn get_subscribe_request_filter( + &self, + account_include: Vec, + account_exclude: Vec, + account_required: Vec, + ) -> TransactionsFilterMap { + let mut transactions = HashMap::new(); + transactions.insert( + "client".to_string(), + SubscribeRequestFilterTransactions { + vote: Some(false), + failed: Some(false), + signature: None, + account_include, + account_exclude, + account_required, + }, + ); + transactions + } + + /// 验证订阅参数 + pub fn validate_subscription_params( + &self, + account_include: &[String], + account_exclude: &[String], + account_required: &[String], + ) -> AnyResult<()> { + if account_include.is_empty() && account_exclude.is_empty() && account_required.is_empty() { + return Err(anyhow::anyhow!( + "account_include or account_exclude or account_required cannot be empty" + )); + } + Ok(()) + } + + /// 获取配置 + pub fn get_config(&self) -> &ClientConfig { + &self.config + } +} diff --git a/src/streaming/grpc/types.rs b/src/streaming/grpc/types.rs new file mode 100644 index 0000000..cd555d7 --- /dev/null +++ b/src/streaming/grpc/types.rs @@ -0,0 +1,96 @@ +use solana_sdk::signature::Signature; +use solana_transaction_status::{EncodedTransactionWithStatusMeta, UiTransactionEncoding}; +use std::{collections::HashMap, fmt}; +use yellowstone_grpc_proto::{ + geyser::{ + SubscribeRequestFilterTransactions, SubscribeUpdateBlockMeta, SubscribeUpdateTransaction, + }, + prost_types::Timestamp, +}; + +pub type TransactionsFilterMap = HashMap; + +#[derive(Clone)] +pub enum EventPretty { + BlockMeta(BlockMetaPretty), + Transaction(TransactionPretty), +} + +#[derive(Clone)] +pub struct BlockMetaPretty { + pub slot: u64, + pub block_hash: String, + pub block_time: Option, +} + +impl fmt::Debug for BlockMetaPretty { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("BlockMetaPretty") + .field("slot", &self.slot) + .field("block_hash", &self.block_hash) + .field("block_time", &self.block_time) + .finish() + } +} + +#[derive(Clone)] +pub struct TransactionPretty { + pub slot: u64, + pub block_hash: String, + pub block_time: Option, + pub signature: Signature, + pub is_vote: bool, + pub tx: EncodedTransactionWithStatusMeta, +} + +impl fmt::Debug for TransactionPretty { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + struct TxWrap<'a>(&'a EncodedTransactionWithStatusMeta); + impl<'a> fmt::Debug for TxWrap<'a> { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let serialized = serde_json::to_string(self.0).expect("failed to serialize"); + fmt::Display::fmt(&serialized, f) + } + } + + f.debug_struct("TransactionPretty") + .field("slot", &self.slot) + .field("signature", &self.signature) + .field("is_vote", &self.is_vote) + .field("tx", &TxWrap(&self.tx)) + .finish() + } +} + +impl From<(SubscribeUpdateBlockMeta, Option)> for BlockMetaPretty { + fn from( + (SubscribeUpdateBlockMeta { slot, blockhash, .. }, block_time): ( + SubscribeUpdateBlockMeta, + Option, + ), + ) -> Self { + Self { block_hash: blockhash.to_string(), block_time, slot } + } +} + +impl From<(SubscribeUpdateTransaction, Option)> for TransactionPretty { + fn from( + (SubscribeUpdateTransaction { transaction, slot }, block_time): ( + SubscribeUpdateTransaction, + Option, + ), + ) -> Self { + let tx = transaction.expect("should be defined"); + Self { + slot, + block_time, + block_hash: "".to_string(), + signature: Signature::try_from(tx.signature.as_slice()).expect("valid signature"), + is_vote: tx.is_vote, + tx: yellowstone_grpc_proto::convert_from::create_tx_with_meta(tx) + .expect("valid tx with meta") + .encode(UiTransactionEncoding::Base64, Some(u8::MAX), true) + .expect("failed to encode"), + } + } +} diff --git a/src/streaming/mod.rs b/src/streaming/mod.rs index 019ab5f..9fd07b9 100755 --- a/src/streaming/mod.rs +++ b/src/streaming/mod.rs @@ -1,8 +1,10 @@ -pub mod yellowstone_grpc; -pub mod yellowstone_sub_system; -pub mod shred_stream; +pub mod common; pub mod event_parser; +pub mod grpc; +pub mod shred_stream; +pub mod yellowstone_grpc; +pub mod yellowstone_sub_system; +pub use shred_stream::ShredStreamGrpc; pub use yellowstone_grpc::YellowstoneGrpc; pub use yellowstone_sub_system::{SystemEvent, TransferInfo}; -pub use shred_stream::ShredStreamGrpc; \ No newline at end of file diff --git a/src/streaming/shred_stream.rs b/src/streaming/shred_stream.rs index 3fe05d4..3cf52b4 100755 --- a/src/streaming/shred_stream.rs +++ b/src/streaming/shred_stream.rs @@ -15,6 +15,10 @@ use crate::protos::shredstream::shredstream_proxy_client::ShredstreamProxyClient use crate::protos::shredstream::SubscribeEntriesRequest; use solana_sdk::pubkey::Pubkey; +// -------------- TODO 待重构 -------------- +// +// -------------- -------------- + // 默认配置常量 const DEFAULT_CHANNEL_SIZE: usize = 1000; const DEFAULT_BATCH_SIZE: usize = 100; @@ -50,9 +54,7 @@ pub struct ShredBackpressureConfig { impl Default for ShredBackpressureConfig { fn default() -> Self { - Self { - channel_size: DEFAULT_CHANNEL_SIZE, - } + Self { channel_size: DEFAULT_CHANNEL_SIZE } } } @@ -81,14 +83,8 @@ impl ShredClientConfig { /// 创建高性能配置(适合高并发场景) pub fn high_performance() -> Self { Self { - batch: ShredBatchConfig { - batch_size: 200, - batch_timeout_ms: 5, - enabled: true, - }, - backpressure: ShredBackpressureConfig { - channel_size: 20000, - }, + batch: ShredBatchConfig { batch_size: 200, batch_timeout_ms: 5, enabled: true }, + backpressure: ShredBackpressureConfig { channel_size: 20000 }, enable_metrics: true, } } @@ -101,9 +97,7 @@ impl ShredClientConfig { batch_timeout_ms: 1, enabled: false, // 禁用批处理,即时处理 }, - backpressure: ShredBackpressureConfig { - channel_size: 1000, - }, + backpressure: ShredBackpressureConfig { channel_size: 1000 }, enable_metrics: false, } } @@ -117,7 +111,6 @@ pub struct ShredPerformanceMetrics { pub average_processing_time_ms: f64, pub min_processing_time_ms: f64, pub max_processing_time_ms: f64, - pub memory_usage_mb: f64, pub last_update_time: std::time::Instant, pub events_in_window: u64, pub window_start_time: std::time::Instant, @@ -136,9 +129,8 @@ impl ShredPerformanceMetrics { events_processed: 0, events_per_second: 0.0, average_processing_time_ms: 0.0, - min_processing_time_ms: f64::MAX, + min_processing_time_ms: 0.0, max_processing_time_ms: 0.0, - memory_usage_mb: 0.0, last_update_time: now, events_in_window: 0, window_start_time: now, @@ -186,7 +178,7 @@ where 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(); @@ -262,7 +254,6 @@ impl ShredStreamGrpc { 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!(" Memory Usage: {:.2}MB", metrics.memory_usage_mb); println!("---"); } @@ -292,26 +283,27 @@ impl ShredStreamGrpc { 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 { + 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.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 { @@ -322,7 +314,7 @@ impl ShredStreamGrpc { // 如果窗口内没有事件,保持之前的速率或设为0 metrics.events_per_second = 0.0; } - + // 重置窗口 metrics.events_in_window = 0; metrics.window_start_time = now; @@ -330,9 +322,7 @@ impl ShredStreamGrpc { // 如果窗口还没满,不更新 events_per_second,保持之前的计算值 // 这样可以避免因为单次批处理时间波动导致的指标跳跃 } - - // 估算内存使用(基于处理的事件数量) - metrics.memory_usage_mb = metrics.events_processed as f64 * 0.001; // 每个事件约1KB + } /// 订阅ShredStream事件(支持批处理和即时处理) @@ -349,12 +339,12 @@ impl ShredStreamGrpc { if self.config.enable_metrics { self.start_auto_metrics_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); - + // 根据配置选择处理模式 if self.config.batch.enabled { // 批处理模式 @@ -384,13 +374,13 @@ impl ShredStreamGrpc { callback(event); } }; - + let mut batch_processor = ShredBatchProcessor::new( - batch_callback, - self.config.batch.batch_size, - self.config.batch.batch_timeout_ms + 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 { @@ -416,18 +406,19 @@ impl ShredStreamGrpc { let self_clone = self.clone(); while let Some(transaction_with_slot) = rx.next().await { - if let Err(e) = self_clone.process_transaction_with_batch( - transaction_with_slot, - protocols.clone(), - bot_wallet, - &mut batch_processor, - ) - .await + if let Err(e) = self_clone + .process_transaction_with_batch( + transaction_with_slot, + protocols.clone(), + bot_wallet, + &mut batch_processor, + ) + .await { error!("Error processing transaction: {e:?}"); } } - + // 处理剩余的事件 batch_processor.flush(); @@ -472,13 +463,14 @@ impl ShredStreamGrpc { let self_clone = self.clone(); while let Some(transaction_with_slot) = rx.next().await { - if let Err(e) = self_clone.process_transaction_immediate( - transaction_with_slot, - protocols.clone(), - bot_wallet, - &callback, - ) - .await + if let Err(e) = self_clone + .process_transaction_immediate( + transaction_with_slot, + protocols.clone(), + bot_wallet, + &callback, + ) + .await { error!("Error processing transaction: {e:?}"); } @@ -506,7 +498,7 @@ impl ShredStreamGrpc { // 预分配向量容量 let mut all_events = Vec::with_capacity(protocols.len() * 2); - + for protocol in protocols { let parser = EventParserFactory::create_parser(protocol.clone()); let events = parser @@ -522,26 +514,29 @@ impl ShredStreamGrpc { .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); + log::warn!( + "ShredStream transaction processing took {}ms for {} events", + processing_time_ms, + event_count + ); } Ok(()) @@ -565,7 +560,7 @@ impl ShredStreamGrpc { // 预分配向量容量 let mut all_events = Vec::with_capacity(protocols.len() * 2); - + for protocol in protocols { let parser = EventParserFactory::create_parser(protocol.clone()); let events = parser @@ -581,28 +576,31 @@ impl ShredStreamGrpc { .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); + log::warn!( + "ShredStream transaction processing took {}ms for {} events", + processing_time_ms, + event_count + ); } Ok(()) } -} \ No newline at end of file +} diff --git a/src/streaming/yellowstone_grpc.rs b/src/streaming/yellowstone_grpc.rs old mode 100755 new mode 100644 index c6c027f..1efcfb5 --- a/src/streaming/yellowstone_grpc.rs +++ b/src/streaming/yellowstone_grpc.rs @@ -1,442 +1,98 @@ -use std::{collections::HashMap, fmt, time::Duration}; - -use chrono::Local; -use futures::{channel::mpsc, sink::Sink, SinkExt, Stream, StreamExt}; -use log::{error, info}; -use yellowstone_grpc_proto::prost_types::Timestamp; -use rustls::crypto::{ring::default_provider, CryptoProvider}; -use solana_sdk::{pubkey::Pubkey, signature::Signature}; -use solana_transaction_status::{EncodedTransactionWithStatusMeta, UiTransactionEncoding}; -use tonic::{transport::channel::ClientTlsConfig, Status}; -use yellowstone_grpc_client::{GeyserGrpcClient, Interceptor}; -use yellowstone_grpc_proto::geyser::{ - subscribe_update::UpdateOneof, CommitmentLevel, SubscribeRequest, - SubscribeRequestFilterTransactions, SubscribeRequestPing, SubscribeUpdate, - SubscribeUpdateTransaction, -}; +use futures::{channel::mpsc, StreamExt}; +use log::error; +use solana_sdk::pubkey::Pubkey; use std::sync::Arc; use tokio::sync::Mutex; +use yellowstone_grpc_proto::geyser::CommitmentLevel; use crate::common::AnyResult; -use crate::streaming::event_parser::{EventParserFactory, Protocol, UnifiedEvent}; - -type TransactionsFilterMap = HashMap; - -// 默认配置常量 -const DEFAULT_CONNECT_TIMEOUT: u64 = 10; -const DEFAULT_REQUEST_TIMEOUT: u64 = 60; -const DEFAULT_CHANNEL_SIZE: usize = 1000; -const DEFAULT_MAX_DECODING_MESSAGE_SIZE: usize = 1024 * 1024 * 10; -const DEFAULT_BATCH_SIZE: usize = 100; -const DEFAULT_BATCH_TIMEOUT_MS: u64 = 5; - -// 背压处理策略 -#[derive(Debug, Clone, Copy)] -pub enum BackpressureStrategy { - /// 阻塞等待(默认) - Block, - /// 丢弃消息 - Drop, - /// 重试有限次数后丢弃 - Retry { max_attempts: usize, wait_ms: u64 }, - /// 有序处理(确保按 slot 顺序处理) - Ordered { max_pending_slots: usize }, -} - -impl Default for BackpressureStrategy { - fn default() -> Self { - Self::Block - } -} - -/// 批处理配置 -#[derive(Debug, Clone)] -pub struct BatchConfig { - /// 批处理大小(默认:100) - pub batch_size: usize, - /// 批处理超时时间(毫秒,默认:10ms) - pub batch_timeout_ms: u64, - /// 是否启用批处理(默认:true) - pub enabled: bool, -} - -impl Default for BatchConfig { - fn default() -> Self { - Self { - batch_size: DEFAULT_BATCH_SIZE, - batch_timeout_ms: DEFAULT_BATCH_TIMEOUT_MS, - enabled: true, - } - } -} - -/// 背压配置 -#[derive(Debug, Clone)] -pub struct BackpressureConfig { - /// 通道大小(默认:10000) - pub channel_size: usize, - /// 背压处理策略(默认:Block) - pub strategy: BackpressureStrategy, -} - -impl Default for BackpressureConfig { - fn default() -> Self { - Self { - channel_size: DEFAULT_CHANNEL_SIZE, - strategy: BackpressureStrategy::default(), - } - } -} - -/// 连接配置 -#[derive(Debug, Clone)] -pub struct ConnectionConfig { - /// 连接超时时间(秒,默认:10) - pub connect_timeout: u64, - /// 请求超时时间(秒,默认:60) - pub request_timeout: u64, - /// 最大解码消息大小(字节,默认:10MB) - pub max_decoding_message_size: usize, -} - -impl Default for ConnectionConfig { - fn default() -> Self { - Self { - connect_timeout: DEFAULT_CONNECT_TIMEOUT, - request_timeout: DEFAULT_REQUEST_TIMEOUT, - max_decoding_message_size: DEFAULT_MAX_DECODING_MESSAGE_SIZE, - } - } -} - -/// 完整的客户端配置 -#[derive(Debug, Clone)] -pub struct ClientConfig { - /// 连接配置 - pub connection: ConnectionConfig, - /// 批处理配置 - pub batch: BatchConfig, - /// 背压配置 - pub backpressure: BackpressureConfig, - /// 是否启用性能监控(默认:false) - pub enable_metrics: bool, -} - -impl Default for ClientConfig { - fn default() -> Self { - Self { - connection: ConnectionConfig::default(), - batch: BatchConfig::default(), - backpressure: BackpressureConfig::default(), - enable_metrics: false, - } - } -} - -impl ClientConfig { - /// 创建高性能配置(适合高并发场景) - pub fn high_performance() -> Self { - Self { - connection: ConnectionConfig::default(), - batch: BatchConfig { - batch_size: 200, - batch_timeout_ms: 5, - enabled: true, - }, - backpressure: BackpressureConfig { - channel_size: 20000, - strategy: BackpressureStrategy::Drop, - }, - enable_metrics: true, - } - } - - /// 创建低延迟配置(适合实时场景) - pub fn low_latency() -> Self { - Self { - connection: ConnectionConfig::default(), - batch: BatchConfig { - batch_size: 10, - batch_timeout_ms: 1, - enabled: false, - }, - backpressure: BackpressureConfig { - channel_size: 1000, - strategy: BackpressureStrategy::Block, - }, - enable_metrics: false, - } - } - - /// 创建有序处理配置(确保事件按顺序处理) - pub fn ordered_processing(max_pending_slots: usize) -> Self { - Self { - connection: ConnectionConfig::default(), - batch: BatchConfig { - batch_size: 50, - batch_timeout_ms: 5, - enabled: true, - }, - backpressure: BackpressureConfig { - channel_size: 15000, - strategy: BackpressureStrategy::Ordered { max_pending_slots }, - }, - enable_metrics: true, - } - } -} - -/// 性能监控指标 -#[derive(Debug, Clone)] -pub struct PerformanceMetrics { - 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 cache_hit_rate: f64, - pub memory_usage_mb: f64, - pub last_update_time: std::time::Instant, - pub events_in_window: u64, - pub window_start_time: std::time::Instant, -} - -impl Default for PerformanceMetrics { - fn default() -> Self { - Self::new() - } -} - -impl PerformanceMetrics { - 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: f64::MAX, - max_processing_time_ms: 0.0, - cache_hit_rate: 0.0, - memory_usage_mb: 0.0, - last_update_time: now, - events_in_window: 0, - window_start_time: now, - } - } -} - -/// gRPC连接池 - 简化版本 -pub struct GrpcConnectionPool { - endpoint: String, - x_token: Option, -} - -impl GrpcConnectionPool { - pub fn new(endpoint: String, x_token: Option) -> Self { - Self { - endpoint, - x_token, - } - } - - pub async fn create_connection(&self) -> AnyResult> { - let builder = GeyserGrpcClient::build_from_shared(self.endpoint.clone())? - .x_token(self.x_token.clone())? - .tls_config(ClientTlsConfig::new().with_native_roots())? - .max_decoding_message_size(DEFAULT_MAX_DECODING_MESSAGE_SIZE) - .connect_timeout(Duration::from_secs(DEFAULT_CONNECT_TIMEOUT)) - .timeout(Duration::from_secs(DEFAULT_REQUEST_TIMEOUT)); - - Ok(builder.connect().await?) - } -} - -/// 批处理事件收集器 -pub struct EventBatchCollector -where - F: Fn(Vec>) + Send + Sync + 'static, -{ - pub(crate) callback: F, - batch: Vec>, - batch_size: usize, - timeout_ms: u64, - last_flush_time: std::time::Instant, -} - -impl EventBatchCollector -where - F: Fn(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) { - log::debug!("Adding event to batch: {} (type: {:?})", event.id(), event.event_type()); - self.batch.push(event); - - // 检查是否需要刷新批次 - if self.batch.len() >= self.batch_size || self.should_flush_by_timeout() { - log::info!("Flushing batch: size={}, timeout={}", self.batch.len(), 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)); - log::info!("Flushing {} events from batch processor", events.len()); - - // 添加更详细的调试信息 - for (i, event) in events.iter().enumerate() { - log::info!("Event {}: Type={:?}, ID={}", i, event.event_type(), event.id()); - } - - // 执行回调并捕获可能的错误 - log::info!("About to execute batch callback with {} events", events.len()); - match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { - (self.callback)(events); - })) { - Ok(_) => { - log::info!("Batch callback executed successfully"); - } - Err(e) => { - log::error!("Batch callback panicked: {:?}", e); - } - } - - self.last_flush_time = std::time::Instant::now(); - } else { - log::debug!("No events to flush"); - } - } - - fn should_flush_by_timeout(&self) -> bool { - self.last_flush_time.elapsed().as_millis() >= self.timeout_ms as u128 - } -} - -#[derive(Clone)] -pub struct TransactionPretty { - pub slot: u64, - pub block_time: Option, - pub signature: Signature, - pub is_vote: bool, - pub tx: EncodedTransactionWithStatusMeta, -} - -impl fmt::Debug for TransactionPretty { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - struct TxWrap<'a>(&'a EncodedTransactionWithStatusMeta); - impl<'a> fmt::Debug for TxWrap<'a> { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - let serialized = serde_json::to_string(self.0).expect("failed to serialize"); - fmt::Display::fmt(&serialized, f) - } - } - - f.debug_struct("TransactionPretty") - .field("slot", &self.slot) - .field("signature", &self.signature) - .field("is_vote", &self.is_vote) - .field("tx", &TxWrap(&self.tx)) - .finish() - } -} - -impl From<(SubscribeUpdateTransaction, Option)> for TransactionPretty { - fn from( - (SubscribeUpdateTransaction { transaction, slot }, block_time): ( - SubscribeUpdateTransaction, - Option, - ), - ) -> Self { - let tx = transaction.expect("should be defined"); - Self { - slot, - block_time, - signature: Signature::try_from(tx.signature.as_slice()).expect("valid signature"), - is_vote: tx.is_vote, - tx: yellowstone_grpc_proto::convert_from::create_tx_with_meta(tx) - .expect("valid tx with meta") - .encode(UiTransactionEncoding::Base64, Some(u8::MAX), true) - .expect("failed to encode"), - } - } -} +use crate::streaming::common::{ + EventBatchProcessor, MetricsManager, PerformanceMetrics, StreamClientConfig, +}; +use crate::streaming::event_parser::{Protocol, UnifiedEvent}; +use crate::streaming::grpc::{EventPretty, EventProcessor, StreamHandler, SubscriptionManager}; #[derive(Clone)] pub struct YellowstoneGrpc { - endpoint: String, - x_token: Option, - config: ClientConfig, - metrics: Arc>, + pub endpoint: String, + pub x_token: Option, + pub config: StreamClientConfig, + pub metrics: Arc>, + pub subscription_manager: SubscriptionManager, + pub metrics_manager: MetricsManager, + pub event_processor: EventProcessor, } impl YellowstoneGrpc { /// 创建客户端,使用默认配置 pub fn new(endpoint: String, x_token: Option) -> AnyResult { - Self::new_with_config(endpoint, x_token, ClientConfig::default()) + Self::new_with_config(endpoint, x_token, StreamClientConfig::default()) } /// 创建客户端,使用自定义配置 - pub fn new_with_config(endpoint: String, x_token: Option, config: ClientConfig) -> AnyResult { - if CryptoProvider::get_default().is_none() { - default_provider() - .install_default() - .map_err(|e| anyhow::anyhow!("Failed to install crypto provider: {:?}", e))?; - } + pub fn new_with_config( + endpoint: String, + x_token: Option, + config: StreamClientConfig, + ) -> AnyResult { + let _ = rustls::crypto::ring::default_provider() + .install_default() + .map_err(|_| anyhow::anyhow!("Failed to install rustls crypto provider"))?; + let metrics = Arc::new(Mutex::new(PerformanceMetrics::new())); + let config_arc = Arc::new(config.clone()); - Ok(Self { - endpoint, + let subscription_manager = + SubscriptionManager::new(endpoint.clone(), x_token.clone(), config.clone()); + let metrics_manager = + MetricsManager::new(metrics.clone(), config_arc.clone(), "YellowstoneGrpc".to_string()); + let event_processor = EventProcessor::new(metrics_manager.clone(), config.clone()); + + Ok(Self { + endpoint, x_token, config, - metrics: Arc::new(Mutex::new(PerformanceMetrics::new())), + metrics, + subscription_manager, + metrics_manager, + event_processor, }) } - /// 创建高性能客户端(适合高并发场景) + /// 创建高性能客户端 pub fn new_high_performance(endpoint: String, x_token: Option) -> AnyResult { - Self::new_with_config(endpoint, x_token, ClientConfig::high_performance()) + Self::new_with_config(endpoint, x_token, StreamClientConfig::high_performance()) } - /// 创建低延迟客户端(适合实时场景) + /// 创建低延迟客户端 pub fn new_low_latency(endpoint: String, x_token: Option) -> AnyResult { - Self::new_with_config(endpoint, x_token, ClientConfig::low_latency()) + Self::new_with_config(endpoint, x_token, StreamClientConfig::low_latency()) } - /// 创建有序处理客户端(确保事件按顺序处理) - pub fn new_ordered_processing(endpoint: String, x_token: Option, max_pending_slots: usize) -> AnyResult { - Self::new_with_config(endpoint, x_token, ClientConfig::ordered_processing(max_pending_slots)) - } - - /// 创建简化的即时处理客户端(推荐用于简单场景) + /// 创建即时处理客户端 pub fn new_immediate(endpoint: String, x_token: Option) -> AnyResult { - let mut config = ClientConfig::low_latency(); - config.enable_metrics = false; // 即时模式默认关闭性能监控 + let mut config = StreamClientConfig::low_latency(); + config.enable_metrics = false; Self::new_with_config(endpoint, x_token, config) } - /// 获取当前配置 - pub fn get_config(&self) -> &ClientConfig { + /// 获取配置 + pub fn get_config(&self) -> &StreamClientConfig { &self.config } /// 更新配置 - pub fn update_config(&mut self, config: ClientConfig) { + pub fn update_config(&mut self, config: StreamClientConfig) { self.config = config; } /// 获取性能指标 pub async fn get_metrics(&self) -> PerformanceMetrics { - let metrics = self.metrics.lock().await; - metrics.clone() + self.metrics_manager.get_metrics().await + } + + /// 打印性能指标 + pub async fn print_metrics(&self) { + self.metrics_manager.print_metrics().await; } /// 启用或禁用性能监控 @@ -444,276 +100,6 @@ impl YellowstoneGrpc { self.config.enable_metrics = enabled; } - - - /// 打印性能指标 - pub async fn print_metrics(&self) { - let metrics = self.get_metrics().await; - println!("📊 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!(" Cache Hit Rate: {:.2}%", metrics.cache_hit_rate * 100.0); - println!(" Memory Usage: {:.2}MB", metrics.memory_usage_mb); - 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 = 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,保持之前的计算值 - // 这样可以避免因为单次批处理时间波动导致的指标跳跃 - } - - // 估算内存使用(基于处理的事件数量) - metrics.memory_usage_mb = metrics.events_processed as f64 * 0.001; // 每个事件约1KB - } - - pub async fn connect(&self) -> AnyResult> { - let builder = GeyserGrpcClient::build_from_shared(self.endpoint.clone())? - .x_token(self.x_token.clone())? - .tls_config(ClientTlsConfig::new().with_native_roots())? - .max_decoding_message_size(self.config.connection.max_decoding_message_size) - .connect_timeout(Duration::from_secs(self.config.connection.connect_timeout)) - .timeout(Duration::from_secs(self.config.connection.request_timeout)); - - Ok(builder.connect().await?) - } - - pub async fn subscribe_with_request( - &self, - transactions: TransactionsFilterMap, - commitment: Option, - ) -> AnyResult<( - impl Sink, - impl Stream>, - )> { - let subscribe_request = SubscribeRequest { - transactions, - commitment: if let Some(commitment) = commitment { - Some(commitment as i32) - } else { - Some(CommitmentLevel::Processed.into()) - }, - ..Default::default() - }; - - let mut client = self.connect().await?; - let (sink, stream) = client - .subscribe_with_request(Some(subscribe_request)) - .await?; - Ok((sink, stream)) - } - - pub fn get_subscribe_request_filter( - &self, - account_include: Vec, - account_exclude: Vec, - account_required: Vec, - ) -> TransactionsFilterMap { - let mut transactions = HashMap::new(); - transactions.insert( - "client".to_string(), - SubscribeRequestFilterTransactions { - vote: Some(false), - failed: Some(false), - signature: None, - account_include, - account_exclude, - account_required, - }, - ); - transactions - } - - pub async fn handle_stream_message( - msg: SubscribeUpdate, - tx: &mut mpsc::Sender, - subscribe_tx: &mut (impl Sink + Unpin), - backpressure_strategy: BackpressureStrategy, - ) -> AnyResult<()> { - let created_at = msg.created_at; - match msg.update_oneof { - Some(UpdateOneof::Transaction(sut)) => { - let transaction_pretty = TransactionPretty::from((sut, created_at)); - log::info!("Received transaction: {} at slot {}", transaction_pretty.signature, transaction_pretty.slot); - - // 根据背压策略处理发送 - match backpressure_strategy { - BackpressureStrategy::Block => { - // 阻塞等待,直到有空间 - if let Err(e) = tx.send(transaction_pretty).await { - log::error!("Failed to send transaction to channel: {:?}", e); - return Err(anyhow::anyhow!("Channel send failed: {:?}", e)); - } - } - BackpressureStrategy::Drop => { - // 尝试发送,如果失败则丢弃 - if let Err(e) = tx.try_send(transaction_pretty) { - if e.is_full() { - log::warn!("Channel is full, dropping transaction"); - } else { - log::error!("Channel is closed: {:?}", e); - return Err(anyhow::anyhow!("Channel is closed: {:?}", e)); - } - } - } - BackpressureStrategy::Retry { max_attempts, wait_ms } => { - // 重试有限次数 - let mut retry_count = 0; - loop { - match tx.try_send(transaction_pretty.clone()) { - Ok(_) => break, - Err(e) => { - if e.is_full() { - retry_count += 1; - if retry_count >= max_attempts { - log::warn!("Channel is full after {} attempts, dropping transaction", retry_count); - break; - } - tokio::time::sleep(tokio::time::Duration::from_millis(wait_ms)).await; - } else { - log::error!("Channel is closed: {:?}", e); - return Err(anyhow::anyhow!("Channel is closed: {:?}", e)); - } - } - } - } - } - BackpressureStrategy::Ordered { max_pending_slots: _ } => { - // 有序处理策略 - 这里暂时使用阻塞策略,实际的有序处理在接收端实现 - if let Err(e) = tx.send(transaction_pretty).await { - log::error!("Failed to send transaction to channel: {:?}", e); - return Err(anyhow::anyhow!("Channel send failed: {:?}", e)); - } - } - } - } - Some(UpdateOneof::Ping(_)) => { - subscribe_tx - .send(SubscribeRequest { - ping: Some(SubscribeRequestPing { id: 1 }), - ..Default::default() - }) - .await?; - info!("service is ping: {}", Local::now()); - } - Some(UpdateOneof::Pong(_)) => { - info!("service is pong: {}", Local::now()); - } - _ => { - log::debug!("Received other message type"); - } - } - Ok(()) - } - - /// Subscribe to Yellowstone GRPC service events with advanced filtering options - /// - /// This method allows subscribing to specific protocol events with more granular account filtering. - /// It processes transactions in real-time and calls the provided callback function when matching events are found. - /// - /// # Parameters - /// - /// * `protocols` - List of protocols to parse (e.g., PumpFun, PumpSwap, Bonk, RaydiumCpmm) - /// * `bot_wallet` - Optional bot wallet address. If passed: in PumpFunTradeEvent if user is in the address, is_bot=true will be set. In BonkTradeEvent if payer is in the address, is_bot=true will be set. Default is false. - /// * `account_include` - List of account addresses to include in the subscription - /// * `account_exclude` - List of account addresses to exclude from the subscription - /// * `account_required` - List of account addresses that must be present in transactions - /// * `commitment` - Optional commitment level for the subscription - /// * `callback` - Function to call when matching events are found - #[allow(clippy::too_many_arguments)] - pub async fn subscribe_events_v2( - &self, - protocols: Vec, - bot_wallet: Option, - account_include: Vec, - account_exclude: Vec, - account_required: Vec, - commitment: Option, - callback: F, - ) -> AnyResult<()> - where - F: Fn(Box) + Send + Sync + 'static, - { - // 启动自动性能监控(如果启用) - if self.config.enable_metrics { - self.start_auto_metrics_monitoring().await; - } - - // 默认使用即时处理模式 - self.subscribe_events_immediate( - protocols, - bot_wallet, - account_include, - account_exclude, - account_required, - commitment, - callback, - ) - .await - } - /// 简化的即时事件订阅(推荐用于简单场景) pub async fn subscribe_events_immediate( &self, @@ -730,25 +116,28 @@ impl YellowstoneGrpc { { // 启动自动性能监控(如果启用) if self.config.enable_metrics { - self.start_auto_metrics_monitoring().await; - } - - if account_include.is_empty() && account_exclude.is_empty() && account_required.is_empty() { - return Err(anyhow::anyhow!( - "account_include or account_exclude or account_required cannot be empty" - )); + self.metrics_manager.start_auto_monitoring().await; } - let transactions = - self.get_subscribe_request_filter(account_include, account_exclude, account_required); - + // 验证订阅参数 + self.subscription_manager.validate_subscription_params( + &account_include, + &account_exclude, + &account_required, + )?; + + let transactions = self.subscription_manager.get_subscribe_request_filter( + account_include, + account_exclude, + account_required, + ); + // 订阅事件 - let (mut subscribe_tx, mut stream) = self - .subscribe_with_request(transactions, commitment) - .await?; + let (mut subscribe_tx, mut stream) = + self.subscription_manager.subscribe_with_request(transactions, commitment).await?; // 创建通道,使用配置中的通道大小 - let (mut tx, mut rx) = mpsc::channel::(self.config.backpressure.channel_size); + let (mut tx, mut rx) = mpsc::channel::(self.config.backpressure.channel_size); // 启动流处理任务 let backpressure_strategy = self.config.backpressure.strategy; @@ -756,8 +145,13 @@ impl YellowstoneGrpc { while let Some(message) = stream.next().await { match message { Ok(msg) => { - if let Err(e) = - Self::handle_stream_message(msg, &mut tx, &mut subscribe_tx, backpressure_strategy).await + if let Err(e) = StreamHandler::handle_stream_message( + msg, + &mut tx, + &mut subscribe_tx, + backpressure_strategy, + ) + .await { error!("Error handling message: {e:?}"); break; @@ -772,16 +166,17 @@ impl YellowstoneGrpc { }); // 即时处理交易,无批处理 - let self_clone = self.clone(); + let event_processor = self.event_processor.clone(); tokio::spawn(async move { - while let Some(transaction_pretty) = rx.next().await { - if let Err(e) = self_clone.process_event_transaction_with_metrics( - transaction_pretty, - &callback, - bot_wallet, - protocols.clone(), - ) - .await + while let Some(event_pretty) = rx.next().await { + if let Err(e) = event_processor + .process_event_transaction_with_metrics( + event_pretty, + &callback, + bot_wallet, + protocols.clone(), + ) + .await { error!("Error processing transaction: {e:?}"); } @@ -808,24 +203,28 @@ impl YellowstoneGrpc { { // 启动自动性能监控(如果启用) if self.config.enable_metrics { - self.start_auto_metrics_monitoring().await; - } - - if account_include.is_empty() && account_exclude.is_empty() && account_required.is_empty() { - return Err(anyhow::anyhow!( - "account_include or account_exclude or account_required cannot be empty" - )); + self.metrics_manager.start_auto_monitoring().await; } - let transactions = - self.get_subscribe_request_filter(account_include, account_exclude, account_required); + // 验证订阅参数 + self.subscription_manager.validate_subscription_params( + &account_include, + &account_exclude, + &account_required, + )?; + + let transactions = self.subscription_manager.get_subscribe_request_filter( + account_include, + account_exclude, + account_required, + ); + // Subscribe to events - let (mut subscribe_tx, mut stream) = self - .subscribe_with_request(transactions, commitment) - .await?; + let (mut subscribe_tx, mut stream) = + self.subscription_manager.subscribe_with_request(transactions, commitment).await?; // Create channel - let (mut tx, mut rx) = mpsc::channel::(self.config.backpressure.channel_size); + let (mut tx, mut rx) = mpsc::channel::(self.config.backpressure.channel_size); // 创建批处理器,将单个事件回调转换为批量回调 let batch_callback = move |events: Vec>| { @@ -833,11 +232,11 @@ impl YellowstoneGrpc { callback(event); } }; - - let mut batch_processor = EventBatchCollector::new( - batch_callback, - self.config.batch.batch_size, - self.config.batch.batch_timeout_ms + + let mut batch_processor = EventBatchProcessor::new( + batch_callback, + self.config.batch.batch_size, + self.config.batch.batch_timeout_ms, ); // Start task to process the stream @@ -846,8 +245,13 @@ impl YellowstoneGrpc { while let Some(message) = stream.next().await { match message { Ok(msg) => { - if let Err(e) = - Self::handle_stream_message(msg, &mut tx, &mut subscribe_tx, backpressure_strategy).await + if let Err(e) = StreamHandler::handle_stream_message( + msg, + &mut tx, + &mut subscribe_tx, + backpressure_strategy, + ) + .await { error!("Error handling message: {e:?}"); break; @@ -862,335 +266,61 @@ impl YellowstoneGrpc { }); // Process transactions with batch processing - let self_clone = self.clone(); - - // 根据背压策略选择处理方式 - match self.config.backpressure.strategy { - BackpressureStrategy::Ordered { max_pending_slots: _ } => { - // 使用有序处理 - 暂时使用普通的批处理方式 - tokio::spawn(async move { - while let Some(transaction_pretty) = rx.next().await { - if let Err(e) = self_clone.process_event_transaction_with_batch( - transaction_pretty, - &mut batch_processor, - bot_wallet, - protocols.clone(), - ) - .await - { - error!("Error processing transaction: {e:?}"); - } - } - - // 处理剩余的事件 - batch_processor.flush(); - }); + let event_processor = self.event_processor.clone(); + tokio::spawn(async move { + while let Some(event_pretty) = rx.next().await { + if let Err(e) = event_processor + .process_event_transaction_with_batch( + event_pretty, + &mut batch_processor, + bot_wallet, + protocols.clone(), + ) + .await + { + error!("Error processing transaction: {e:?}"); + } } - _ => { - // 使用原有的批处理方式 - tokio::spawn(async move { - while let Some(transaction_pretty) = rx.next().await { - if let Err(e) = self_clone.process_event_transaction_with_batch( - transaction_pretty, - &mut batch_processor, - bot_wallet, - protocols.clone(), - ) - .await - { - error!("Error processing transaction: {e:?}"); - } - } - - // 处理剩余的事件 - batch_processor.flush(); - }); - } - } + + // 处理剩余的事件 + batch_processor.flush(); + }); tokio::signal::ctrl_c().await?; Ok(()) } - /// 订阅事件 - #[deprecated( - since = "0.1.5", - note = "This method will be removed, please use the new API: subscribe_events_v2" - )] - #[allow(clippy::too_many_arguments)] - pub async fn subscribe_events( + /// 默认订阅方法 - 委托给即时处理模式 + #[deprecated(since = "0.1.12", note = "Use subscribe_events_immediate instead")] + pub async fn subscribe_events_v2( &self, protocols: Vec, bot_wallet: Option, - account_include: Option>, - account_exclude: Option>, - account_required: Option>, + account_include: Vec, + account_exclude: Vec, + account_required: Vec, commitment: Option, callback: F, ) -> AnyResult<()> where F: Fn(Box) + Send + Sync + 'static, { - // 启动自动性能监控(如果启用) - if self.config.enable_metrics { - self.start_auto_metrics_monitoring().await; - } - - // 创建过滤器 - let protocol_accounts = protocols - .iter() - .flat_map(|p| p.get_program_id()) - .map(|p| p.to_string()) - .collect::>(); - let mut account_include = account_include.unwrap_or_default(); - let account_exclude = account_exclude.unwrap_or_default(); - let account_required = account_required.unwrap_or_default(); - - account_include.extend(protocol_accounts.clone()); - - let transactions = - self.get_subscribe_request_filter(account_include, account_exclude, account_required); - - // 订阅事件 - let (mut subscribe_tx, mut stream) = self - .subscribe_with_request(transactions, commitment) - .await?; - - // 创建通道 - let (mut tx, mut rx) = mpsc::channel::(self.config.backpressure.channel_size); - - // 创建回调函数,使用 Arc 包装以便在多个任务中共享 - let callback = std::sync::Arc::new(Box::new(callback)); - - // 启动处理流的任务 - let backpressure_strategy = self.config.backpressure.strategy; - tokio::spawn(async move { - while let Some(message) = stream.next().await { - match message { - Ok(msg) => { - if let Err(e) = - Self::handle_stream_message(msg, &mut tx, &mut subscribe_tx, backpressure_strategy).await - { - error!("Error handling message: {e:?}"); - break; - } - } - Err(error) => { - error!("Stream error: {error:?}"); - break; - } - } - } - }); - - // 处理交易 - let self_clone = self.clone(); - tokio::spawn(async move { - while let Some(transaction_pretty) = rx.next().await { - if let Err(e) = self_clone.process_event_transaction_with_metrics( - transaction_pretty, - &**callback, - bot_wallet, - protocols.clone(), - ) - .await - { - error!("Error processing transaction: {e:?}"); - } - } - }); - - tokio::signal::ctrl_c().await?; - Ok(()) - } - - async fn process_event_transaction_with_metrics( - &self, - transaction_pretty: TransactionPretty, - callback: &F, - bot_wallet: Option, - protocols: Vec, - ) -> 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_pretty.slot; - let signature = transaction_pretty.signature.to_string(); - - // 预分配向量容量,避免动态扩容 - let mut futures = Vec::with_capacity(protocols.len()); - - for protocol in protocols { - let parser = EventParserFactory::create_parser(protocol); - // 在异步任务中需要克隆值 - let tx_clone = transaction_pretty.tx.clone(); - let signature_clone = signature.clone(); - let bot_wallet_clone = bot_wallet; - - futures.push(tokio::spawn(async move { - parser - .parse_transaction( - tx_clone, - &signature_clone, - Some(slot), - transaction_pretty.block_time.map(|ts| prost_types::Timestamp { - seconds: ts.seconds, - nanos: ts.nanos, - }), - program_received_time_ms, - bot_wallet_clone, - ) - .await - .unwrap_or_else(|_e| vec![]) - })); - } - - let results = futures::future::join_all(futures).await; - - // 收集所有事件 - let mut all_events = Vec::new(); - for events in results.into_iter().flatten() { - all_events.extend(events); - } - - // 保存事件数量用于日志记录 - let event_count = all_events.len(); - - // 批量处理事件 - if !all_events.is_empty() { - for event in all_events { - callback(event); - } - } - - // 更新性能指标 - let processing_time = start_time.elapsed(); - let processing_time_ms = processing_time.as_millis() as f64; - - // 更新性能指标(如果启用) - if self.config.enable_metrics { - self.update_metrics(event_count as u64, processing_time_ms).await; - } - - // 记录慢处理操作 - if processing_time_ms > 10.0 { - log::warn!("Slow event processing: {processing_time_ms}ms for {event_count} events"); - } - - Ok(()) - } - - async fn process_event_transaction_with_batch( - &self, - transaction_pretty: TransactionPretty, - batch_processor: &mut EventBatchCollector, - bot_wallet: Option, - protocols: Vec, - ) -> AnyResult<()> - where - F: Fn(Vec>) + Send + Sync + 'static, - { - 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 mut futures: Vec>, anyhow::Error>>> = Vec::with_capacity(protocols.len()); - - for protocol in protocols { - let parser = EventParserFactory::create_parser(protocol.clone()); - // 在异步任务中需要克隆值 - let tx_clone = transaction_pretty.tx.clone(); - let signature_clone = signature.clone(); - let bot_wallet_clone = bot_wallet; - let protocol_clone = protocol.clone(); - - futures.push(tokio::spawn(async move { - let result = parser - .parse_transaction( - tx_clone, - &signature_clone, - Some(slot), - transaction_pretty.block_time.map(|ts| prost_types::Timestamp { - seconds: ts.seconds, - nanos: ts.nanos, - }), - program_received_time_ms, - bot_wallet_clone, - ) - .await; - - match result { - Ok(events) => { - if !events.is_empty() { - log::info!("Parsed {} events for protocol {:?}", events.len(), protocol_clone); - } - Ok(events) - } - Err(e) => { - log::warn!("Failed to parse transaction for protocol {:?}: {:?}", protocol_clone, e); - Ok(vec![]) - } - } - })); - } - - let results = futures::future::join_all(futures).await; - - // 收集所有事件并使用批处理器 - let mut total_events = 0; - for result in results { - match result { - Ok(parse_result) => { - match parse_result { - Ok(events) => { - total_events += events.len(); - log::info!("Adding {} events to batch processor", events.len()); - for event in events { - if self.config.batch.enabled { - batch_processor.add_event(event); - } else { - // 如果批处理被禁用,直接调用回调 - // 这里需要将单个事件包装成Vec来调用批处理回调 - let single_event_batch = vec![event]; - (batch_processor.callback)(single_event_batch); - } - } - } - Err(e) => { - log::warn!("Failed to parse transaction: {:?}", e); - } - } - } - Err(e) => { - log::warn!("Failed to get events from async task: {:?}", e); - } - } - } - - // 添加调试信息 - if total_events > 0 { - log::info!("Total events parsed: {} for transaction {}", total_events, signature); - } - - // 更新性能指标 - let processing_time = start_time.elapsed(); - let processing_time_ms = processing_time.as_millis() as f64; - - // 实际调用性能指标更新 - self.update_metrics(total_events as u64, processing_time_ms).await; - - // 记录慢处理操作 - if processing_time_ms > 10.0 { - log::warn!("Slow event processing: {processing_time_ms}ms for {total_events} events"); - } - - Ok(()) + self.subscribe_events_immediate( + protocols, + bot_wallet, + account_include, + account_exclude, + account_required, + commitment, + callback, + ) + .await } } - - +// 实现 Clone trait 以支持模块间共享 +impl Clone for EventProcessor { + fn clone(&self) -> Self { + Self { metrics_manager: self.metrics_manager.clone(), config: self.config.clone() } + } +} diff --git a/src/streaming/yellowstone_sub_system.rs b/src/streaming/yellowstone_sub_system.rs index 2062ce5..3347239 100755 --- a/src/streaming/yellowstone_sub_system.rs +++ b/src/streaming/yellowstone_sub_system.rs @@ -1,6 +1,9 @@ use crate::{ common::AnyResult, - streaming::yellowstone_grpc::{TransactionPretty, YellowstoneGrpc, BackpressureStrategy}, + streaming::{ + grpc::{BackpressureStrategy, EventPretty, StreamHandler}, + yellowstone_grpc::YellowstoneGrpc, + }, }; use futures::{channel::mpsc, StreamExt}; use log::error; @@ -10,7 +13,7 @@ use solana_transaction_status::EncodedTransactionWithStatusMeta; const SYSTEM_PROGRAM_ID: Pubkey = pubkey!("11111111111111111111111111111111"); // 根据实际并发量调整通道大小,避免背压 -const CHANNEL_SIZE: usize = 50000; // 增加到 50000 +const CHANNEL_SIZE: usize = 50000; // 增加到 50000 #[derive(Debug)] pub enum SystemEvent { @@ -38,11 +41,14 @@ impl YellowstoneGrpc { let addrs = vec![SYSTEM_PROGRAM_ID.to_string()]; let account_include = account_include.unwrap_or_default(); let account_exclude = account_exclude.unwrap_or_default(); - let transactions = - self.get_subscribe_request_filter(account_include, account_exclude, addrs); + let transactions = self.subscription_manager.get_subscribe_request_filter( + account_include, + account_exclude, + addrs, + ); let (mut subscribe_tx, mut stream) = - self.subscribe_with_request(transactions, None).await?; - let (mut tx, mut rx) = mpsc::channel::(CHANNEL_SIZE); + self.subscription_manager.subscribe_with_request(transactions, None).await?; + let (mut tx, mut rx) = mpsc::channel::(CHANNEL_SIZE); let callback = Box::new(callback); @@ -50,8 +56,13 @@ impl YellowstoneGrpc { while let Some(message) = stream.next().await { match message { Ok(msg) => { - if let Err(e) = - Self::handle_stream_message(msg, &mut tx, &mut subscribe_tx, BackpressureStrategy::Block).await + if let Err(e) = StreamHandler::handle_stream_message( + msg, + &mut tx, + &mut subscribe_tx, + BackpressureStrategy::Block, + ) + .await { error!("Error handling message: {e:?}"); break; @@ -65,37 +76,38 @@ impl YellowstoneGrpc { } }); - while let Some(transaction_pretty) = rx.next().await { - if let Err(e) = Self::process_system_transaction(transaction_pretty, &*callback).await { + while let Some(event_pretty) = rx.next().await { + if let Err(e) = Self::process_system_transaction(event_pretty, &*callback).await { error!("Error processing transaction: {e:?}"); } } Ok(()) } - async fn process_system_transaction( - transaction_pretty: TransactionPretty, - callback: &F, - ) -> AnyResult<()> + async fn process_system_transaction(event_pretty: EventPretty, callback: &F) -> AnyResult<()> where F: Fn(SystemEvent) + Send + Sync, { - let trade_raw: EncodedTransactionWithStatusMeta = transaction_pretty.tx; - let meta = trade_raw - .meta - .as_ref() - .ok_or_else(|| anyhow::anyhow!("Missing transaction metadata"))?; + match event_pretty { + EventPretty::Transaction(transaction_pretty) => { + let trade_raw: EncodedTransactionWithStatusMeta = transaction_pretty.tx; + let meta = trade_raw + .meta + .as_ref() + .ok_or_else(|| anyhow::anyhow!("Missing transaction metadata"))?; - if meta.err.is_some() { - return Ok(()); + if meta.err.is_some() { + return Ok(()); + } + + callback(SystemEvent::NewTransfer(TransferInfo { + slot: transaction_pretty.slot, + signature: transaction_pretty.signature.to_string(), + tx: trade_raw.transaction.decode(), + })); + } + _ => {} } - - callback(SystemEvent::NewTransfer(TransferInfo { - slot: transaction_pretty.slot, - signature: transaction_pretty.signature.to_string(), - tx: trade_raw.transaction.decode(), - })); - Ok(()) } }