mirror of
https://github.com/0xfnzero/solana-streamer.git
synced 2026-08-20 04:18:06 +00:00
refactor: enhance account event parser with more precise event classification
- Rename CommonAccountEvent to TokenAccountEvent for better semantic clarity - Rename AccountNonce to NonceAccount for consistent naming convention - Add MintInfoEvent type to support SPL Token Mint account parsing - Enhance parse_token_account_event to automatically detect and parse Mint accounts - Update all example code to use new event types and structures - Improve event type semantics for better developer experience
This commit is contained in:
@@ -3,7 +3,7 @@ use solana_streamer_sdk::{
|
||||
streaming::{
|
||||
event_parser::{
|
||||
common::EventType,
|
||||
core::account_event_parser::CommonAccountEvent,
|
||||
core::account_event_parser::{TokenInfoEvent, NonceAccountEvent, TokenAccountEvent},
|
||||
protocols::{
|
||||
bonk::{
|
||||
parser::BONK_PROGRAM_ID, BonkGlobalConfigAccountEvent, BonkMigrateToAmmEvent,
|
||||
@@ -289,9 +289,15 @@ fn create_event_callback() -> impl Fn(Box<dyn UnifiedEvent>) {
|
||||
RaydiumCpmmPoolStateAccountEvent => |e: RaydiumCpmmPoolStateAccountEvent| {
|
||||
println!("RaydiumCpmmPoolStateAccountEvent: {e:?}");
|
||||
},
|
||||
CommonAccountEvent => |e: CommonAccountEvent| {
|
||||
println!("CommonAccountEvent: {e:?}");
|
||||
TokenAccountEvent => |e: TokenAccountEvent| {
|
||||
println!("TokenAccountEvent: {e:?}");
|
||||
},
|
||||
NonceAccountEvent => |e: NonceAccountEvent| {
|
||||
println!("NonceAccountEvent: {e:?}");
|
||||
},
|
||||
TokenInfoEvent => |e: TokenInfoEvent| {
|
||||
println!("TokenInfoEvent: {e:?}");
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -45,7 +45,7 @@ async fn test_grpc() -> Result<(), Box<dyn std::error::Error>> {
|
||||
let account_filter = AccountFilter { account: vec![nonce_account], owner: vec![] };
|
||||
|
||||
// Event filtering
|
||||
let event_type_filter = Some(EventTypeFilter { include: vec![EventType::AccountNonce] });
|
||||
let event_type_filter = Some(EventTypeFilter { include: vec![EventType::NonceAccount] });
|
||||
|
||||
println!("Starting to listen for events, press Ctrl+C to stop...");
|
||||
println!("Starting subscription...");
|
||||
|
||||
@@ -3,7 +3,7 @@ use solana_streamer_sdk::{
|
||||
streaming::{
|
||||
event_parser::{
|
||||
common::EventType,
|
||||
core::account_event_parser::CommonAccountEvent,
|
||||
core::account_event_parser::TokenAccountEvent,
|
||||
protocols::{
|
||||
bonk::{
|
||||
BonkGlobalConfigAccountEvent, BonkMigrateToAmmEvent,
|
||||
@@ -245,8 +245,8 @@ fn create_event_callback() -> impl Fn(Box<dyn UnifiedEvent>) {
|
||||
RaydiumCpmmPoolStateAccountEvent => |e: RaydiumCpmmPoolStateAccountEvent| {
|
||||
println!("RaydiumCpmmPoolStateAccountEvent: {e:?}");
|
||||
},
|
||||
CommonAccountEvent => |e: CommonAccountEvent| {
|
||||
println!("CommonAccountEvent: {e:?}");
|
||||
TokenAccountEvent => |e: TokenAccountEvent| {
|
||||
println!("TokenAccountEvent: {e:?}");
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
@@ -46,7 +46,7 @@ async fn test_grpc() -> Result<(), Box<dyn std::error::Error>> {
|
||||
let account_filter = AccountFilter { account: vec![account_to_listen], owner: vec![] };
|
||||
|
||||
// Event filtering
|
||||
let event_type_filter = Some(EventTypeFilter { include: vec![EventType::AccountCommon] });
|
||||
let event_type_filter = Some(EventTypeFilter { include: vec![EventType::TokenAccount] });
|
||||
|
||||
println!("Starting to listen for events, press Ctrl+C to stop...");
|
||||
println!("Starting subscription...");
|
||||
|
||||
@@ -0,0 +1,90 @@
|
||||
use solana_streamer_sdk::{
|
||||
match_event,
|
||||
streaming::{
|
||||
event_parser::{
|
||||
common::{filter::EventTypeFilter, EventType},
|
||||
core::account_event_parser::TokenInfoEvent,
|
||||
UnifiedEvent,
|
||||
},
|
||||
grpc::ClientConfig,
|
||||
yellowstone_grpc::{AccountFilter, TransactionFilter},
|
||||
YellowstoneGrpc,
|
||||
},
|
||||
};
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
println!("Starting Yellowstone gRPC Streamer...");
|
||||
test_grpc().await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn test_grpc() -> Result<(), Box<dyn std::error::Error>> {
|
||||
println!("Subscribing to Yellowstone gRPC events...");
|
||||
// Create low-latency configuration
|
||||
let mut config: ClientConfig = ClientConfig::low_latency();
|
||||
// Enable performance monitoring, has performance overhead, disabled by default
|
||||
config.enable_metrics = true;
|
||||
let grpc = YellowstoneGrpc::new_with_config(
|
||||
"https://solana-yellowstone-grpc.publicnode.com:443".to_string(),
|
||||
None,
|
||||
config,
|
||||
)?;
|
||||
println!("GRPC client created successfully");
|
||||
let callback = create_event_callback();
|
||||
// Will try to parse corresponding protocol events from transactions
|
||||
let protocols = vec![];
|
||||
println!("Protocols to monitor: {:?}", protocols);
|
||||
// Filter accounts
|
||||
let account_include = vec![];
|
||||
let account_exclude = vec![];
|
||||
let account_required = vec![];
|
||||
|
||||
// Listen to transaction data
|
||||
let transaction_filter =
|
||||
TransactionFilter { account_include, account_exclude, account_required };
|
||||
|
||||
let account_to_listen = "EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v".to_string();
|
||||
|
||||
// Listen to account data belonging to owner programs -> account event monitoring
|
||||
let account_filter = AccountFilter { account: vec![account_to_listen], owner: vec![] };
|
||||
|
||||
// Event filtering
|
||||
let event_type_filter = Some(EventTypeFilter { include: vec![EventType::TokenAccount] });
|
||||
|
||||
println!("Starting to listen for events, press Ctrl+C to stop...");
|
||||
println!("Starting subscription...");
|
||||
|
||||
grpc.subscribe_events_immediate(
|
||||
protocols.clone(),
|
||||
None,
|
||||
transaction_filter.clone(),
|
||||
account_filter.clone(),
|
||||
event_type_filter.clone(),
|
||||
None,
|
||||
callback,
|
||||
)
|
||||
.await?;
|
||||
|
||||
// 支持 stop 方法,测试代码 - 异步1000秒之后停止
|
||||
let grpc_clone = grpc.clone();
|
||||
tokio::spawn(async move {
|
||||
tokio::time::sleep(std::time::Duration::from_secs(1000)).await;
|
||||
grpc_clone.stop().await;
|
||||
});
|
||||
|
||||
println!("Waiting for Ctrl+C to stop...");
|
||||
tokio::signal::ctrl_c().await?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn create_event_callback() -> impl Fn(Box<dyn UnifiedEvent>) {
|
||||
|event: Box<dyn UnifiedEvent>| {
|
||||
match_event!(event, {
|
||||
TokenInfoEvent => |e: TokenInfoEvent| {
|
||||
println!("TokenInfoEvent: {:?}", e.decimals);
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user