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

feat: Major refactor with batch processing and performance optimization

Browse Source 
- Refactor streaming architecture with modular grpc components
- Add batch processing system with backpressure strategies
- Implement performance monitoring and metrics collection
- Add multi-protocol parser with block metadata support
- Enhance configuration system with preset profiles
- Add parse_tx_events example and update documentation
- Optimize yellowstone_grpc client complexity
This commit is contained in:
ysq
2025-08-11 01:04:16 +08:00
parent 87465cbfdb
commit 97faba9406
35 changed files with 2067 additions and 1434 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.toml
+2 -2
View File
@@ -1,6 +1,6 @@
[package]
name = "solana-streamer-sdk"
version = "0.1.11"
version = "0.2.0"
edition = "2021"
authors = ["William <byteblock6@gmail.com>", "sgxiang <sgxiang@gmail.com>", "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"
README.md
+48 -30
View File
@@ -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<dyn std::error::Error>> {
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<dyn std::error::Error>> {
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<dyn std::error::Error>> {
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<dyn std::error::Error>> {
];
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<dyn UnifiedEvent>) {
|event: Box<dyn UnifiedEvent>| {
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<dyn UnifiedEvent>) {
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
README_CN.md
+58 -27
View File
@@ -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<dyn std::error::Error>> {
println!("Starting Solana Streamer...");
test_grpc().await?;
test_shreds().await?;
Ok(())
}
async fn test_grpc() -> Result<(), Box<dyn std::error::Error>> {
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<dyn std::error::Error>> {
config,
)?;
println!("GRPC client created successfully");
let callback = create_event_callback();
// 将会从交易中尝试解析对应的协议事件
@@ -107,6 +129,8 @@ async fn test_grpc() -> Result<(), Box<dyn std::error::Error>> {
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<dyn std::error::Error>> {
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<dyn std::error::Error>> {
}
async fn test_shreds() -> Result<(), Box<dyn std::error::Error>> {
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<dyn std::error::Error>> {
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<dyn UnifiedEvent>) {
|event: Box<dyn UnifiedEvent>| {
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<dyn UnifiedEvent>) {
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. **合规性**: 确保遵守相关法律法规
## 语言版本
examples/parse_tx_events.rs
+124
View File
@@ -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<dyn EventParser> = 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(())
}
src/main.rs
+6 -6
View File
@@ -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<dyn std::error::Error>> {
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<dyn UnifiedEvent>) {
|event: Box<dyn UnifiedEvent>| {
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);
src/streaming/common/batch.rs
+131
View File
@@ -0,0 +1,131 @@
use crate::streaming::event_parser::UnifiedEvent;
/// 通用批处理事件收集器
pub struct EventBatchProcessor<F>
where
F: FnMut(Vec<Box<dyn UnifiedEvent>>) + Send + Sync + 'static,
{
pub(crate) callback: F,
batch: Vec<Box<dyn UnifiedEvent>>,
batch_size: usize,
timeout_ms: u64,
last_flush_time: std::time::Instant,
}
impl<F> EventBatchProcessor<F>
where
F: FnMut(Vec<Box<dyn UnifiedEvent>>) + 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<dyn UnifiedEvent>) {
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<F>
where
F: Fn(Box<dyn UnifiedEvent>) + Send + Sync + 'static,
{
callback: F,
}
impl<F> SimpleEventBatchProcessor<F>
where
F: Fn(Box<dyn UnifiedEvent>) + Send + Sync + 'static,
{
pub fn new(callback: F) -> Self {
Self { callback }
}
/// 将批量事件拆分为单个事件处理
pub fn process_batch(&self, events: Vec<Box<dyn UnifiedEvent>>) {
for event in events {
(self.callback)(event);
}
}
}
/// 批处理器包装器，用于将单个事件回调适配为批量处理
pub fn create_batch_callback_adapter<F>(
single_event_callback: F,
) -> impl FnMut(Vec<Box<dyn UnifiedEvent>>) + Send + Sync + 'static
where
F: Fn(Box<dyn UnifiedEvent>) + Send + Sync + 'static,
{
move |events: Vec<Box<dyn UnifiedEvent>>| {
for event in events {
single_event_callback(event);
}
}
}
src/streaming/common/config.rs
+131
View File
@@ -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,
}
}
}
src/streaming/common/constants.rs
+14
View File
@@ -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;
src/streaming/common/metrics.rs
+177
View File
@@ -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<Mutex<PerformanceMetrics>>,
config: Arc<StreamClientConfig>,
stream_name: String,
}
impl MetricsManager {
/// 创建新的性能监控管理器
pub fn new(
metrics: Arc<Mutex<PerformanceMetrics>>,
config: Arc<StreamClientConfig>,
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(),
}
}
}
src/streaming/common/mod.rs
+11
View File
@@ -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::*;
src/streaming/event_parser/common/mod.rs
+2 -2
View File
@@ -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 {
src/streaming/event_parser/common/types.rs
+161 -85
View File
@@ -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<EventMetadata> {
@@ -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<TransferData> {
@@ -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<T> {
impl<T> ParseResult<T> {
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<Pubkey>,
}
#[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<String>,
}
/// 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<TransferData>,
pub swap_data: Option<SwapData>,
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<TransferData>) {
pub fn set_transfer_datas(
&mut self,
transfer_datas: Vec<TransferData>,
swap_data: Option<SwapData>,
) {
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<dyn UnifiedEvent>,
inner_instruction: &solana_transaction_status::UiInnerInstructions,
current_index: i8,
accounts: &[Pubkey],
event_type: EventType,
) -> Vec<TransferData> {
) -> (Vec<TransferData>, Option<SwapData>) {
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<Pubkey> = compiled
.accounts
.iter()
.map(|a| accounts[*a as usize])
.collect();
let account_pubkeys: Vec<Pubkey> =
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<Pubkey> = compiled
.accounts
.iter()
.map(|a| accounts[*a as usize])
.collect();
let account_pubkeys: Vec<Pubkey> =
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<Pubkey> = compiled
.accounts
.iter()
.map(|a| accounts[*a as usize])
.collect();
let account_pubkeys: Vec<Pubkey> =
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<Pubkey> = None;
let mut from_mint: Option<Pubkey> = None;
let mut to_mint: Option<Pubkey> = None;
let mut user_from_token: Option<Pubkey> = None;
let mut user_to_token: Option<Pubkey> = None;
let mut from_vault: Option<Pubkey> = None;
let mut to_vault: Option<Pubkey> = 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)
}
}
src/streaming/event_parser/core/common_event_parser.rs
+15
View File
@@ -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<dyn UnifiedEvent> {
let block_meta_event = BlockMetaEvent::new(slot, block_hash.to_string(), block_time_ms);
Box::new(block_meta_event)
}
}
src/streaming/event_parser/core/mod.rs
+1
View File
@@ -1,2 +1,3 @@
pub mod common_event_parser;
pub mod traits;
pub use traits::{EventParser, UnifiedEvent};
src/streaming/event_parser/core/traits.rs
+109 -145
View File
@@ -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<Box<dyn UnifiedEvent>>,
pub timestamp: u64,
}
/// 事件解析缓存
pub struct EventParseCache {
cache: Arc<RwLock<HashMap<String, ParseCacheEntry>>>,
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<Vec<Box<dyn UnifiedEvent>>> {
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<Box<dyn UnifiedEvent>>) {
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<TransferData>);
fn set_transfer_datas(
&mut self,
transfer_datas: Vec<TransferData>,
swap_data: Option<SwapData>,
);
/// 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<GenericEventParseConfig>>;
/// 获取指令解析配置
fn instruction_configs(&self) -> HashMap<Vec<u8>, Vec<GenericEventParseConfig>>;
/// 从内联指令中解析事件数据
#[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<Pubkey>,
) -> Result<Vec<Box<dyn UnifiedEvent>>> {
// 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<Pubkey> = vec![];
let mut inner_instructions: Vec<UiInnerInstructions> = 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::<u32>().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::<u32>().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::<BonkPoolCreateEvent>() {
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::<PumpSwapBuyEvent>() {
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::<PumpSwapSellEvent>()
{
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::<BonkPoolCreateEvent>() {
bonk_dev_address = Some(pool_info.creator);
} else if let Some(trade_info) = event.as_any_mut().downcast_mut::<BonkTradeEvent>() {
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<dyn UnifiedEvent> {
/// 通用事件解析器配置
#[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<GenericEventParseConfig>>,
instruction_configs: HashMap<Vec<u8>, Vec<GenericEventParseConfig>>,
pub program_ids: Vec<Pubkey>,
pub inner_instruction_configs: HashMap<&'static str, Vec<GenericEventParseConfig>>,
pub instruction_configs: HashMap<Vec<u8>, Vec<GenericEventParseConfig>>,
}
impl GenericEventParser {
/// 创建新的通用事件解析器
pub fn new(
program_id: Pubkey,
protocol_type: ProtocolType,
configs: Vec<GenericEventParseConfig>,
) -> Self {
pub fn new(program_ids: Vec<Pubkey>, configs: Vec<GenericEventParseConfig>) -> 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<Box<dyn UnifiedEvent>> {
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<Box<dyn UnifiedEvent>> {
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<GenericEventParseConfig>> {
self.inner_instruction_configs.clone()
}
fn instruction_configs(&self) -> HashMap<Vec<u8>, Vec<GenericEventParseConfig>> {
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<Box<dyn UnifiedEvent>> {
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<Pubkey> = instruction
.accounts
.iter()
.map(|&idx| accounts[idx as usize])
.collect();
let account_pubkeys: Vec<Pubkey> =
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<Pubkey> {
vec![self.program_id]
self.program_ids.clone()
}
}
pub struct SDKSystemEventParser {}
impl SDKSystemEventParser {}
src/streaming/event_parser/protocols/block/block_meta_event.rs
+34
View File
@@ -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,);
src/streaming/event_parser/protocols/block/mod.rs
+1
View File
@@ -0,0 +1 @@
pub mod block_meta_event;
src/streaming/event_parser/protocols/bonk/parser.rs
+23 -1
View File
@@ -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<GenericEventParseConfig>> {
self.inner.inner_instruction_configs()
}
fn instruction_configs(&self) -> HashMap<Vec<u8>, Vec<GenericEventParseConfig>> {
self.inner.instruction_configs()
}
fn parse_events_from_inner_instruction(
&self,
inner_instruction: &UiCompiledInstruction,
src/streaming/event_parser/protocols/mod.rs
+5 -1
View File
@@ -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;
pub use raydium_clmm::RaydiumClmmEventParser;
pub use block::block_meta_event::BlockMetaEvent;
pub use mutil::MutilEventParser;
src/streaming/event_parser/protocols/mutil/mod.rs
+3
View File
@@ -0,0 +1,3 @@
pub mod parser;
pub use parser::MutilEventParser;
src/streaming/event_parser/protocols/mutil/parser.rs
Executable
+95
View File
@@ -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<Protocol>) -> 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<GenericEventParseConfig>> {
self.inner.inner_instruction_configs()
}
fn instruction_configs(&self) -> HashMap<Vec<u8>, Vec<GenericEventParseConfig>> {
self.inner.instruction_configs()
}
fn parse_events_from_inner_instruction(
&self,
inner_instruction: &UiCompiledInstruction,
signature: &str,
slot: u64,
block_time: Option<Timestamp>,
program_received_time_ms: i64,
index: String,
) -> Vec<Box<dyn UnifiedEvent>> {
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<Timestamp>,
program_received_time_ms: i64,
index: String,
) -> Vec<Box<dyn UnifiedEvent>> {
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<Pubkey> {
self.inner.supported_program_ids()
}
}
src/streaming/event_parser/protocols/pumpfun/parser.rs
+15 -1
View File
@@ -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<GenericEventParseConfig>> {
self.inner.inner_instruction_configs()
}
fn instruction_configs(&self) -> HashMap<Vec<u8>, Vec<GenericEventParseConfig>> {
self.inner.instruction_configs()
}
fn parse_events_from_inner_instruction(
&self,
inner_instruction: &UiCompiledInstruction,
src/streaming/event_parser/protocols/pumpswap/parser.rs
+19 -1
View File
@@ -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<GenericEventParseConfig>> {
self.inner.inner_instruction_configs()
}
fn instruction_configs(&self) -> HashMap<Vec<u8>, Vec<GenericEventParseConfig>> {
self.inner.instruction_configs()
}
fn parse_events_from_inner_instruction(
&self,
inner_instruction: &UiCompiledInstruction,
src/streaming/event_parser/protocols/raydium_clmm/parser.rs
+13 -2
View File
@@ -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<GenericEventParseConfig>> {
self.inner.inner_instruction_configs()
}
fn instruction_configs(&self) -> HashMap<Vec<u8>, Vec<GenericEventParseConfig>> {
self.inner.instruction_configs()
}
fn parse_events_from_inner_instruction(
&self,
inner_instruction: &UiCompiledInstruction,
src/streaming/event_parser/protocols/raydium_cpmm/parser.rs
+13 -2
View File
@@ -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<GenericEventParseConfig>> {
self.inner.inner_instruction_configs()
}
fn instruction_configs(&self) -> HashMap<Vec<u8>, Vec<GenericEventParseConfig>> {
self.inner.instruction_configs()
}
fn parse_events_from_inner_instruction(
&self,
inner_instruction: &UiCompiledInstruction,
src/streaming/grpc/connection.rs
+33
View File
@@ -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<String>,
}
impl GrpcConnectionPool {
pub fn new(endpoint: String, x_token: Option<String>) -> Self {
Self {
endpoint,
x_token,
}
}
pub async fn create_connection(&self) -> AnyResult<GeyserGrpcClient<impl Interceptor>> {
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?)
}
}
src/streaming/grpc/event_processor.rs
+201
View File
@@ -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<F>(
&self,
event_pretty: EventPretty,
callback: &F,
bot_wallet: Option<Pubkey>,
protocols: Vec<Protocol>,
) -> AnyResult<()>
where
F: Fn(Box<dyn UnifiedEvent>) + 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<dyn EventParser> =
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<F>(
&self,
event_pretty: EventPretty,
batch_processor: &mut EventBatchCollector<F>,
bot_wallet: Option<Pubkey>,
protocols: Vec<Protocol>,
) -> AnyResult<()>
where
F: Fn(Vec<Box<dyn UnifiedEvent>>) + 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<dyn EventParser> =
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(())
}
}
src/streaming/grpc/mod.rs
+25
View File
@@ -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,
};
src/streaming/grpc/stream_handler.rs
+124
View File
@@ -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<EventPretty>,
subscribe_tx: &mut (impl Sink<SubscribeRequest, Error = mpsc::SendError> + 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<EventPretty>,
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(())
}
}
src/streaming/grpc/subscription.rs
+106
View File
@@ -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<String>,
config: ClientConfig,
}
impl SubscriptionManager {
/// 创建新的订阅管理器
pub fn new(endpoint: String, x_token: Option<String>, config: ClientConfig) -> Self {
Self { endpoint, x_token, config }
}
/// 创建 gRPC 连接
pub async fn connect(&self) -> AnyResult<GeyserGrpcClient<impl Interceptor>> {
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<CommitmentLevel>,
) -> AnyResult<(
impl Sink<SubscribeRequest, Error = mpsc::SendError>,
impl Stream<Item = Result<SubscribeUpdate, Status>>,
)> {
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<String>,
account_exclude: Vec<String>,
account_required: Vec<String>,
) -> 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
}
}
src/streaming/grpc/types.rs
+96
View File
@@ -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<String, SubscribeRequestFilterTransactions>;
#[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<Timestamp>,
}
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<Timestamp>,
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<Timestamp>)> for BlockMetaPretty {
fn from(
(SubscribeUpdateBlockMeta { slot, blockhash, .. }, block_time): (
SubscribeUpdateBlockMeta,
Option<Timestamp>,
),
) -> Self {
Self { block_hash: blockhash.to_string(), block_time, slot }
}
}
impl From<(SubscribeUpdateTransaction, Option<Timestamp>)> for TransactionPretty {
fn from(
(SubscribeUpdateTransaction { transaction, slot }, block_time): (
SubscribeUpdateTransaction,
Option<Timestamp>,
),
) -> 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"),
}
}
}
src/streaming/mod.rs
+6 -4
View File
@@ -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;
src/streaming/shred_stream.rs
+67 -69
View File
@@ -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<dyn UnifiedEvent>) {
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::<TransactionWithSlot>(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(())
}
}
}
src/streaming/yellowstone_grpc.rs
Executable → Regular
+158 -1028
View File
File diff suppressed because it is too large Load Diff
src/streaming/yellowstone_sub_system.rs
+40 -28
View File
@@ -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::<TransactionPretty>(CHANNEL_SIZE);
self.subscription_manager.subscribe_with_request(transactions, None).await?;
let (mut tx, mut rx) = mpsc::channel::<EventPretty>(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<F>(
transaction_pretty: TransactionPretty,
callback: &F,
) -> AnyResult<()>
async fn process_system_transaction<F>(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(())
}
}