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

feat: Add account event parser and protocol type definitions

Browse Source 
- Add account event parser (account_event_parser.rs) for account-level event handling
- Introduce type definitions for pumpfun, pumpswap, raydium_amm_v4, raydium_clmm, raydium_cpmm protocols
- Enhance event processor to support account event processing alongside transactions
- Improve subscription system with separate transaction and account filters
- Optimize parser configuration using Option types for cleaner code structure
- Update examples and documentation to reflect new capabilities

Changes include:
- 6 new types.rs files for protocol-specific data structures
- Updated event processor for account, transaction, and block meta events
- Refactored subscription manager with account filtering support
- Enhanced metrics and batch processing logic
This commit is contained in:
ysq
2025-08-13 23:30:59 +08:00
parent c753c8418d
commit aee8921ad5
40 changed files with 1845 additions and 349 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/streaming/grpc/event_processor.rs
+64 -9
View File
@@ -7,6 +7,7 @@ use crate::common::AnyResult;
use crate::streaming::common::{
EventBatchProcessor as EventBatchCollector, MetricsManager, StreamClientConfig as ClientConfig,
};
use crate::streaming::event_parser::core::account_event_parser::AccountEventParser;
use crate::streaming::event_parser::core::common_event_parser::CommonEventParser;
use crate::streaming::event_parser::EventParser;
use crate::streaming::event_parser::{
@@ -37,7 +38,28 @@ impl EventProcessor {
F: Fn(Box<dyn UnifiedEvent>) + Send + Sync,
{
match event_pretty {
EventPretty::Account(account_pretty) => {
self.metrics_manager.add_process_count().await;
let start_time = std::time::Instant::now();
let program_received_time_ms = chrono::Utc::now().timestamp_millis();
let account_event = AccountEventParser::parse_account_event(
protocols.clone(),
account_pretty,
program_received_time_ms,
);
if let Some(event) = account_event {
callback(event);
// 更新性能指标
let processing_time = start_time.elapsed();
let processing_time_ms = processing_time.as_millis() as f64;
// 更新性能指标（如果启用）
self.metrics_manager.update_metrics(1, processing_time_ms).await;
// 记录慢处理操作
self.metrics_manager.log_slow_processing(processing_time_ms, 1);
}
}
EventPretty::Transaction(transaction_pretty) => {
self.metrics_manager.add_process_count().await;
let start_time = std::time::Instant::now();
let program_received_time_ms = chrono::Utc::now().timestamp_millis();
let slot = transaction_pretty.slot;
@@ -76,16 +98,13 @@ impl EventProcessor {
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.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 start_time = std::time::Instant::now();
self.metrics_manager.add_process_count().await;
let block_time_ms = block_meta_pretty
.block_time
.map(|ts| ts.seconds * 1000 + ts.nanos as i64 / 1_000_000)
@@ -96,6 +115,13 @@ impl EventProcessor {
block_time_ms,
);
callback(block_meta_event);
// 更新性能指标
let processing_time = start_time.elapsed();
let processing_time_ms = processing_time.as_millis() as f64;
// 更新性能指标（如果启用）
self.metrics_manager.update_metrics(1, processing_time_ms).await;
// 记录慢处理操作
self.metrics_manager.log_slow_processing(processing_time_ms, 1);
}
}
@@ -114,7 +140,28 @@ impl EventProcessor {
F: Fn(Vec<Box<dyn UnifiedEvent>>) + Send + Sync + 'static,
{
match event_pretty {
EventPretty::Account(account_pretty) => {
self.metrics_manager.add_process_count().await;
let start_time = std::time::Instant::now();
let program_received_time_ms = chrono::Utc::now().timestamp_millis();
let account_event = AccountEventParser::parse_account_event(
protocols.clone(),
account_pretty,
program_received_time_ms,
);
if let Some(event) = account_event {
(batch_processor.callback)(vec![event]);
// 更新性能指标
let processing_time = start_time.elapsed();
let processing_time_ms = processing_time.as_millis() as f64;
// 实际调用性能指标更新
self.metrics_manager.update_metrics(1, processing_time_ms).await;
// 记录慢处理操作
self.metrics_manager.log_slow_processing(processing_time_ms, 1);
}
}
EventPretty::Transaction(transaction_pretty) => {
self.metrics_manager.add_process_count().await;
let start_time = std::time::Instant::now();
let program_received_time_ms = chrono::Utc::now().timestamp_millis();
let slot = transaction_pretty.slot;
@@ -142,8 +189,8 @@ impl EventProcessor {
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);
log::debug!("Parsed {} events", event_count);
log::debug!("Adding {} events to batch processor", event_count);
for event in events {
if self.config.batch.enabled {
batch_processor.add_event(event);
@@ -165,7 +212,7 @@ impl EventProcessor {
// 添加调试信息
if total_events > 0 {
log::info!(
log::debug!(
"Total events parsed: {} for transaction {}",
total_events,
signature
@@ -183,6 +230,7 @@ impl EventProcessor {
self.metrics_manager.log_slow_processing(processing_time_ms, total_events);
}
EventPretty::BlockMeta(block_meta_pretty) => {
let start_time = std::time::Instant::now();
let block_time_ms = block_meta_pretty
.block_time
.map(|ts| ts.seconds * 1000 + ts.nanos as i64 / 1_000_000)
@@ -193,6 +241,13 @@ impl EventProcessor {
block_time_ms,
);
(batch_processor.callback)(vec![block_meta_event]);
// 更新性能指标
let processing_time = start_time.elapsed();
let processing_time_ms = processing_time.as_millis() as f64;
// 更新性能指标（如果启用）
self.metrics_manager.update_metrics(1, processing_time_ms).await;
// 记录慢处理操作
self.metrics_manager.log_slow_processing(processing_time_ms, 1);
}
}
src/streaming/grpc/stream_handler.rs
+15 -5
View File
@@ -1,6 +1,5 @@
use chrono::Local;
use futures::{channel::mpsc, sink::Sink, SinkExt};
use log::info;
use yellowstone_grpc_proto::geyser::{
subscribe_update::UpdateOneof, SubscribeRequest, SubscribeRequestPing, SubscribeUpdate,
};
@@ -8,6 +7,7 @@ use yellowstone_grpc_proto::geyser::{
use super::types::{BlockMetaPretty, EventPretty, TransactionPretty};
use crate::common::AnyResult;
use crate::streaming::common::BackpressureStrategy;
use crate::streaming::grpc::AccountPretty;
/// 流消息处理器
pub struct StreamHandler;
@@ -22,9 +22,19 @@ impl StreamHandler {
) -> AnyResult<()> {
let created_at = msg.created_at;
match msg.update_oneof {
Some(UpdateOneof::Account(account)) => {
let account_pretty = AccountPretty::from(account);
log::debug!("Received account: {:?}", account_pretty);
Self::handle_backpressure(
tx,
EventPretty::Account(account_pretty),
backpressure_strategy,
)
.await?;
}
Some(UpdateOneof::BlockMeta(sut)) => {
let block_meta_pretty = BlockMetaPretty::from((sut, created_at));
log::info!("Received block meta: {:?}", block_meta_pretty);
log::debug!("Received block meta: {:?}", block_meta_pretty);
Self::handle_backpressure(
tx,
EventPretty::BlockMeta(block_meta_pretty),
@@ -34,7 +44,7 @@ impl StreamHandler {
}
Some(UpdateOneof::Transaction(sut)) => {
let transaction_pretty = TransactionPretty::from((sut, created_at));
log::info!(
log::debug!(
"Received transaction: {} at slot {}",
transaction_pretty.signature,
transaction_pretty.slot
@@ -55,10 +65,10 @@ impl StreamHandler {
..Default::default()
})
.await?;
info!("service is ping: {}", Local::now());
log::debug!("service is ping: {}", Local::now());
}
Some(UpdateOneof::Pong(_)) => {
info!("service is pong: {}", Local::now());
log::debug!("service is pong: {}", Local::now());
}
_ => {
log::debug!("Received other message type");
src/streaming/grpc/subscription.rs
+27 -18
View File
@@ -4,10 +4,11 @@ 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,
CommitmentLevel, SubscribeRequest, SubscribeRequestFilterAccounts,
SubscribeRequestFilterBlocksMeta, SubscribeRequestFilterTransactions, SubscribeUpdate,
};
use super::types::AccountsFilterMap;
use super::types::TransactionsFilterMap;
use crate::common::AnyResult;
use crate::streaming::common::StreamClientConfig as ClientConfig;
@@ -41,12 +42,14 @@ impl SubscriptionManager {
pub async fn subscribe_with_request(
&self,
transactions: TransactionsFilterMap,
accounts: Option<AccountsFilterMap>,
commitment: Option<CommitmentLevel>,
) -> AnyResult<(
impl Sink<SubscribeRequest, Error = mpsc::SendError>,
impl Stream<Item = Result<SubscribeUpdate, Status>>,
)> {
let subscribe_request = SubscribeRequest {
accounts: accounts.unwrap_or_default(),
transactions,
blocks_meta: hashmap! { "".to_owned() => SubscribeRequestFilterBlocksMeta {} },
commitment: if let Some(commitment) = commitment {
@@ -56,12 +59,33 @@ impl SubscriptionManager {
},
..Default::default()
};
let mut client = self.connect().await?;
let (sink, stream) = client.subscribe_with_request(Some(subscribe_request)).await?;
Ok((sink, stream))
}
/// 创建账户订阅请求并返回流
pub fn subscribe_with_account_request(
&self,
account: Vec<String>,
owner: Vec<String>,
) -> Option<AccountsFilterMap> {
if account.len() == 0 && owner.len() == 0 {
return None;
}
let mut accounts = HashMap::new();
accounts.insert(
"".to_owned(),
SubscribeRequestFilterAccounts {
account: account,
owner: owner,
filters: vec![],
nonempty_txn_signature: None,
},
);
Some(accounts)
}
/// 生成订阅请求过滤器
pub fn get_subscribe_request_filter(
&self,
@@ -84,21 +108,6 @@ impl SubscriptionManager {
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
+47 -1
View File
@@ -3,17 +3,47 @@ use solana_transaction_status::{EncodedTransactionWithStatusMeta, UiTransactionE
use std::{collections::HashMap, fmt};
use yellowstone_grpc_proto::{
geyser::{
SubscribeRequestFilterTransactions, SubscribeUpdateBlockMeta, SubscribeUpdateTransaction,
SubscribeRequestFilterAccounts, SubscribeRequestFilterTransactions, SubscribeUpdateAccount,
SubscribeUpdateBlockMeta, SubscribeUpdateTransaction,
},
prost_types::Timestamp,
};
pub type TransactionsFilterMap = HashMap<String, SubscribeRequestFilterTransactions>;
pub type AccountsFilterMap = HashMap<String, SubscribeRequestFilterAccounts>;
#[derive(Clone)]
pub enum EventPretty {
BlockMeta(BlockMetaPretty),
Transaction(TransactionPretty),
Account(AccountPretty),
}
#[derive(Clone)]
pub struct AccountPretty {
pub slot: u64,
pub signature: String,
pub pubkey: String,
pub executable: bool,
pub lamports: u64,
pub owner: String,
pub rent_epoch: u64,
pub data: Vec<u8>,
}
impl fmt::Debug for AccountPretty {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("AccountPretty")
.field("slot", &self.slot)
.field("signature", &self.signature)
.field("pubkey", &self.pubkey)
.field("executable", &self.executable)
.field("lamports", &self.lamports)
.field("owner", &self.owner)
.field("rent_epoch", &self.rent_epoch)
.field("data", &self.data)
.finish()
}
}
#[derive(Clone)]
@@ -62,6 +92,22 @@ impl fmt::Debug for TransactionPretty {
}
}
impl From<SubscribeUpdateAccount> for AccountPretty {
fn from(account: SubscribeUpdateAccount) -> Self {
let account_info = account.account.unwrap();
Self {
slot: account.slot,
signature: bs58::encode(&account_info.txn_signature.unwrap_or_default()).into_string(),
pubkey: bs58::encode(&account_info.pubkey).into_string(),
executable: account_info.executable,
lamports: account_info.lamports,
owner: bs58::encode(&account_info.owner).into_string(),
rent_epoch: account_info.rent_epoch,
data: account_info.data,
}
}
}
impl From<(SubscribeUpdateBlockMeta, Option<Timestamp>)> for BlockMetaPretty {
fn from(
(SubscribeUpdateBlockMeta { slot, blockhash, .. }, block_time): (