diff --git a/README.md b/README.md index d4fb8c6..2707c48 100755 --- a/README.md +++ b/README.md @@ -296,7 +296,7 @@ Note: Multiple subscription attempts on the same client return an error. ### Unified Event Interface -- **UnifiedEvent Trait**: All protocol events implement a common interface +- **DexEvent Trait**: All protocol events implement a common interface - **Protocol Enum**: Easy identification of event sources - **Event Factory**: Automatic event parsing and categorization diff --git a/README_CN.md b/README_CN.md index 495d104..f009537 100644 --- a/README_CN.md +++ b/README_CN.md @@ -295,7 +295,7 @@ grpc.update_subscription( ### 统一事件接口 -- **UnifiedEvent Trait**: 所有协议事件实现通用接口 +- **DexEvent Trait**: 所有协议事件实现通用接口 - **Protocol Enum**: 轻松识别事件来源 - **Event Factory**: 自动事件解析和分类 diff --git a/examples/dynamic_subscription.rs b/examples/dynamic_subscription.rs index bdae3a0..38edeb2 100644 --- a/examples/dynamic_subscription.rs +++ b/examples/dynamic_subscription.rs @@ -31,7 +31,7 @@ async fn main() -> Result<()> { let counter = event_counter.clone(); let callback = - move |event: solana_streamer_sdk::streaming::event_parser::UnifiedEvent| { + move |event: solana_streamer_sdk::streaming::event_parser::DexEvent| { let count = counter.fetch_add(1, Ordering::Relaxed); let protocol = match event.metadata().event_type { @@ -254,7 +254,7 @@ async fn main() -> Result<()> { let shutdown_event_counter = Arc::new(AtomicU64::new(0)); let shutdown_counter = shutdown_event_counter.clone(); let shutdown_callback = - move |_event: solana_streamer_sdk::streaming::event_parser::UnifiedEvent| { + move |_event: solana_streamer_sdk::streaming::event_parser::DexEvent| { shutdown_counter.fetch_add(1, Ordering::Relaxed); }; @@ -327,7 +327,7 @@ async fn main() -> Result<()> { println!("\n=== Subscription enforcement ==="); let test_callback = - |_event: solana_streamer_sdk::streaming::event_parser::UnifiedEvent| {}; + |_event: solana_streamer_sdk::streaming::event_parser::DexEvent| {}; match client .subscribe_events_immediate( @@ -358,7 +358,7 @@ async fn main() -> Result<()> { let client2_counter = Arc::new(AtomicU64::new(0)); let counter2 = client2_counter.clone(); let client2_callback = - move |_event: solana_streamer_sdk::streaming::event_parser::UnifiedEvent| { + move |_event: solana_streamer_sdk::streaming::event_parser::DexEvent| { counter2.fetch_add(1, Ordering::Relaxed); }; @@ -390,7 +390,7 @@ async fn main() -> Result<()> { println!("\n=== Advanced subscription enforcement ==="); let test_callback_advanced = - |_event: solana_streamer_sdk::streaming::event_parser::UnifiedEvent| {}; + |_event: solana_streamer_sdk::streaming::event_parser::DexEvent| {}; let client3 = Arc::new(YellowstoneGrpc::new(GRPC_ENDPOINT.to_string(), API_KEY.map(|s| s.to_string()))?); @@ -447,7 +447,7 @@ async fn main() -> Result<()> { let client4_counter = Arc::new(AtomicU64::new(0)); let counter4 = client4_counter.clone(); let client4_callback = - move |_event: solana_streamer_sdk::streaming::event_parser::UnifiedEvent| { + move |_event: solana_streamer_sdk::streaming::event_parser::DexEvent| { counter4.fetch_add(1, Ordering::Relaxed); }; diff --git a/examples/grpc_example.rs b/examples/grpc_example.rs index 2af1f60..73c61d8 100644 --- a/examples/grpc_example.rs +++ b/examples/grpc_example.rs @@ -7,7 +7,7 @@ use solana_streamer_sdk::streaming::{ raydium_clmm::parser::RAYDIUM_CLMM_PROGRAM_ID, raydium_cpmm::parser::RAYDIUM_CPMM_PROGRAM_ID, }, - Protocol, UnifiedEvent, + Protocol, DexEvent, }, grpc::ClientConfig, yellowstone_grpc::{AccountFilter, TransactionFilter}, @@ -108,15 +108,15 @@ async fn test_grpc() -> Result<(), Box> { Ok(()) } -fn create_event_callback() -> impl Fn(UnifiedEvent) { - |event: UnifiedEvent| { +fn create_event_callback() -> impl Fn(DexEvent) { + |event: DexEvent| { println!( "🎉 Event received! Type: {:?}, transaction_index: {:?}", event.metadata().event_type, event.metadata().transaction_index ); match event { - UnifiedEvent::BlockMetaEvent(e) => { + DexEvent::BlockMetaEvent(e) => { println!("{:?}", e); } // .... other events diff --git a/examples/mint_all_ata_account_listen_example.rs b/examples/mint_all_ata_account_listen_example.rs index 9952a5b..86368ea 100644 --- a/examples/mint_all_ata_account_listen_example.rs +++ b/examples/mint_all_ata_account_listen_example.rs @@ -4,7 +4,7 @@ use solana_sdk::pubkey::Pubkey; use solana_streamer_sdk::streaming::{ event_parser::{ common::{filter::EventTypeFilter, EventType}, - UnifiedEvent, + DexEvent, }, grpc::ClientConfig, yellowstone_grpc::{AccountFilter, TransactionFilter}, @@ -102,9 +102,9 @@ async fn test_grpc() -> Result<(), Box> { Ok(()) } -fn create_event_callback() -> impl Fn(UnifiedEvent) { - |event: UnifiedEvent| match event { - UnifiedEvent::TokenAccountEvent(e) => { +fn create_event_callback() -> impl Fn(DexEvent) { + |event: DexEvent| match event { + DexEvent::TokenAccountEvent(e) => { println!("TokenAccount: {:?} amount: {:?}", e.pubkey, e.amount); } _ => {} diff --git a/examples/nonce_listen_example.rs b/examples/nonce_listen_example.rs index e1fb2b8..32e2a4e 100644 --- a/examples/nonce_listen_example.rs +++ b/examples/nonce_listen_example.rs @@ -1,7 +1,7 @@ use solana_streamer_sdk::streaming::{ event_parser::{ common::{filter::EventTypeFilter, EventType}, - UnifiedEvent, + DexEvent, }, grpc::ClientConfig, yellowstone_grpc::{AccountFilter, TransactionFilter}, @@ -75,8 +75,8 @@ async fn test_grpc() -> Result<(), Box> { Ok(()) } -fn create_event_callback() -> impl Fn(UnifiedEvent) { - |event: UnifiedEvent| { +fn create_event_callback() -> impl Fn(DexEvent) { + |event: DexEvent| { println!("🎉 Event received! {:?}", event); } } diff --git a/examples/parse_tx_events.rs b/examples/parse_tx_events.rs index e191b80..677e281 100644 --- a/examples/parse_tx_events.rs +++ b/examples/parse_tx_events.rs @@ -2,7 +2,7 @@ use anyhow::Result; use solana_commitment_config::CommitmentConfig; use solana_streamer_sdk::streaming::event_parser::core::event_parser::EventParser; use solana_streamer_sdk::streaming::event_parser::Protocol; -use solana_streamer_sdk::streaming::event_parser::UnifiedEvent; +use solana_streamer_sdk::streaming::event_parser::DexEvent; use std::str::FromStr; use std::sync::Arc; /// Get transaction data based on transaction signature @@ -101,7 +101,7 @@ async fn get_single_transaction_details(signature_str: &str) -> Result<()> { None, signature, transaction, - Arc::new(move |event: &UnifiedEvent| { + Arc::new(move |event: &DexEvent| { println!("{:?}\n", event); }), ) diff --git a/examples/pumpswap_pool_account_listen_example.rs b/examples/pumpswap_pool_account_listen_example.rs index a25f45c..72f83e2 100644 --- a/examples/pumpswap_pool_account_listen_example.rs +++ b/examples/pumpswap_pool_account_listen_example.rs @@ -4,7 +4,7 @@ use solana_sdk::pubkey::Pubkey; use solana_streamer_sdk::streaming::{ event_parser::{ common::{filter::EventTypeFilter, EventType}, - UnifiedEvent, + DexEvent, }, grpc::ClientConfig, yellowstone_grpc::{AccountFilter, TransactionFilter}, @@ -104,9 +104,9 @@ async fn test_grpc() -> Result<(), Box> { Ok(()) } -fn create_event_callback() -> impl Fn(UnifiedEvent) { - |event: UnifiedEvent| match event { - UnifiedEvent::TokenAccountEvent(e) => { +fn create_event_callback() -> impl Fn(DexEvent) { + |event: DexEvent| match event { + DexEvent::TokenAccountEvent(e) => { println!("TokenAccount: {:?} amount: {:?}", e.pubkey, e.amount); } _ => {} diff --git a/examples/shred_example.rs b/examples/shred_example.rs index 31459c2..c01b74d 100644 --- a/examples/shred_example.rs +++ b/examples/shred_example.rs @@ -1,5 +1,5 @@ use solana_streamer_sdk::streaming::{ - event_parser::{Protocol, UnifiedEvent}, + event_parser::{Protocol, DexEvent}, shred::StreamClientConfig, ShredStreamGrpc, }; @@ -54,15 +54,15 @@ async fn test_shreds() -> Result<(), Box> { Ok(()) } -fn create_event_callback() -> impl Fn(UnifiedEvent) { - |event: UnifiedEvent| { +fn create_event_callback() -> impl Fn(DexEvent) { + |event: DexEvent| { println!( "🎉 Event received! Type: {:?}, transaction_index: {:?}", event.metadata().event_type, event.metadata().transaction_index ); match event { - UnifiedEvent::BlockMetaEvent(e) => { + DexEvent::BlockMetaEvent(e) => { println!("BlockMetaEvent: {:?}", e.metadata.handle_us); } _ => {} diff --git a/examples/token_balance_listen_example.rs b/examples/token_balance_listen_example.rs index 033ef4f..b36e750 100644 --- a/examples/token_balance_listen_example.rs +++ b/examples/token_balance_listen_example.rs @@ -1,7 +1,7 @@ use solana_streamer_sdk::streaming::{ event_parser::{ common::{filter::EventTypeFilter, EventType}, - UnifiedEvent, + DexEvent, }, grpc::ClientConfig, yellowstone_grpc::{AccountFilter, TransactionFilter}, @@ -76,8 +76,8 @@ async fn test_grpc() -> Result<(), Box> { Ok(()) } -fn create_event_callback() -> impl Fn(UnifiedEvent) { - |event: UnifiedEvent| { +fn create_event_callback() -> impl Fn(DexEvent) { + |event: DexEvent| { println!("🎉 Event received! {:?}", event); } } diff --git a/examples/token_decimals_listen_example.rs b/examples/token_decimals_listen_example.rs index bc0d5c5..159db96 100644 --- a/examples/token_decimals_listen_example.rs +++ b/examples/token_decimals_listen_example.rs @@ -1,7 +1,7 @@ use solana_streamer_sdk::streaming::{ event_parser::{ common::{filter::EventTypeFilter, EventType}, - UnifiedEvent, + DexEvent, }, grpc::ClientConfig, yellowstone_grpc::{AccountFilter, TransactionFilter}, @@ -76,9 +76,9 @@ async fn test_grpc() -> Result<(), Box> { Ok(()) } -fn create_event_callback() -> impl Fn(UnifiedEvent) { - |event: UnifiedEvent| match event { - UnifiedEvent::TokenInfoEvent(e) => { +fn create_event_callback() -> impl Fn(DexEvent) { + |event: DexEvent| match event { + DexEvent::TokenInfoEvent(e) => { println!("TokenInfoEvent: {:?}", e.decimals); } _ => {} diff --git a/src/streaming/common/event_processor.rs b/src/streaming/common/event_processor.rs index ba5e856..7e7c81e 100644 --- a/src/streaming/common/event_processor.rs +++ b/src/streaming/common/event_processor.rs @@ -4,7 +4,7 @@ use crate::streaming::event_parser::common::filter::EventTypeFilter; 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::core::event_parser::EventParser; -use crate::streaming::event_parser::{core::traits::UnifiedEvent, Protocol}; +use crate::streaming::event_parser::{core::traits::DexEvent, Protocol}; use crate::streaming::grpc::{EventPretty, MetricsManager}; use crate::streaming::shred::TransactionWithSlot; use solana_sdk::pubkey::Pubkey; @@ -15,9 +15,9 @@ use std::sync::Arc; /// 用于 Transaction 事件处理,在调用原始 callback 的同时更新 metrics #[inline] fn create_metrics_callback( - callback: Arc, -) -> Arc { - Arc::new(move |event: UnifiedEvent| { + callback: Arc, +) -> Arc { + Arc::new(move |event: DexEvent| { let processing_time_us = event.metadata().handle_us as f64; callback(event); MetricsManager::global().update_metrics( @@ -33,7 +33,7 @@ pub async fn process_grpc_transaction( event_pretty: EventPretty, protocols: &[Protocol], event_type_filter: Option<&EventTypeFilter>, - callback: Arc, + callback: Arc, bot_wallet: Option, ) -> AnyResult<()> { match event_pretty { @@ -107,7 +107,7 @@ pub async fn process_shred_transaction( transaction_with_slot: TransactionWithSlot, protocols: &[Protocol], event_type_filter: Option<&EventTypeFilter>, - callback: Arc, + callback: Arc, bot_wallet: Option, ) -> AnyResult<()> { MetricsManager::global().add_tx_process_count(); diff --git a/src/streaming/event_parser/common/types.rs b/src/streaming/event_parser/common/types.rs index f9bd2a9..904da19 100755 --- a/src/streaming/event_parser/common/types.rs +++ b/src/streaming/event_parser/common/types.rs @@ -4,7 +4,7 @@ use serde::{Deserialize, Serialize}; use solana_sdk::{pubkey::Pubkey, signature::Signature}; use std::{borrow::Cow, fmt, str::FromStr, sync::Arc}; -use crate::streaming::{common::SimdUtils, event_parser::UnifiedEvent}; +use crate::streaming::{common::SimdUtils, event_parser::DexEvent}; // Object pool size configuration const EVENT_METADATA_POOL_SIZE: usize = 1000; @@ -348,7 +348,7 @@ lazy_static::lazy_static! { /// Parse token transfer data from next instructions pub fn parse_swap_data_from_next_instructions( - event: &UnifiedEvent, + event: &DexEvent, inner_instruction: &solana_transaction_status::InnerInstructions, current_index: i8, accounts: &[Pubkey], @@ -371,7 +371,7 @@ pub fn parse_swap_data_from_next_instructions( let mut to_vault: Option = None; match event { - UnifiedEvent::BonkTradeEvent(e) => { + DexEvent::BonkTradeEvent(e) => { // user = Some(e.payer); from_mint = Some(e.base_token_mint); to_mint = Some(e.quote_token_mint); @@ -380,19 +380,19 @@ pub fn parse_swap_data_from_next_instructions( from_vault = Some(e.base_vault); to_vault = Some(e.quote_vault); } - UnifiedEvent::PumpFunTradeEvent(e) => { + DexEvent::PumpFunTradeEvent(e) => { 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 }; } - UnifiedEvent::PumpSwapBuyEvent(e) => { + DexEvent::PumpSwapBuyEvent(e) => { swap_data.from_mint = e.quote_mint; swap_data.to_mint = e.base_mint; } - UnifiedEvent::PumpSwapSellEvent(e) => { + DexEvent::PumpSwapSellEvent(e) => { swap_data.from_mint = e.base_mint; swap_data.to_mint = e.quote_mint; } - UnifiedEvent::RaydiumCpmmSwapEvent(e) => { + DexEvent::RaydiumCpmmSwapEvent(e) => { // user = Some(e.payer); from_mint = Some(e.input_token_mint); to_mint = Some(e.output_token_mint); @@ -401,7 +401,7 @@ pub fn parse_swap_data_from_next_instructions( from_vault = Some(e.input_vault); to_vault = Some(e.output_vault); } - UnifiedEvent::RaydiumClmmSwapEvent(e) => { + DexEvent::RaydiumClmmSwapEvent(e) => { // user = Some(e.payer); swap_data.description = Some("Unable to get from_mint and to_mint from RaydiumClmmSwapEvent".into()); @@ -410,7 +410,7 @@ pub fn parse_swap_data_from_next_instructions( from_vault = Some(e.input_vault); to_vault = Some(e.output_vault); } - UnifiedEvent::RaydiumClmmSwapV2Event(e) => { + DexEvent::RaydiumClmmSwapV2Event(e) => { // user = Some(e.payer); from_mint = Some(e.input_vault_mint); to_mint = Some(e.output_vault_mint); @@ -419,7 +419,7 @@ pub fn parse_swap_data_from_next_instructions( from_vault = Some(e.input_vault); to_vault = Some(e.output_vault); } - UnifiedEvent::RaydiumAmmV4SwapEvent(e) => { + DexEvent::RaydiumAmmV4SwapEvent(e) => { // user = Some(e.user_source_owner); swap_data.description = Some("Unable to get from_mint and to_mint from RaydiumAmmV4SwapEvent".into()); @@ -518,7 +518,7 @@ pub fn parse_swap_data_from_next_instructions( /// Parse token transfer data from next instructions /// TODO: - wait refactor pub fn parse_swap_data_from_next_grpc_instructions( - event: &UnifiedEvent, + event: &DexEvent, inner_instruction: &yellowstone_grpc_proto::prelude::InnerInstructions, current_index: i8, accounts: &[Pubkey], @@ -541,7 +541,7 @@ pub fn parse_swap_data_from_next_grpc_instructions( let mut to_vault: Option = None; match event { - UnifiedEvent::BonkTradeEvent(e) => { + DexEvent::BonkTradeEvent(e) => { // user = Some(e.payer); from_mint = Some(e.base_token_mint); to_mint = Some(e.quote_token_mint); @@ -550,19 +550,19 @@ pub fn parse_swap_data_from_next_grpc_instructions( from_vault = Some(e.base_vault); to_vault = Some(e.quote_vault); } - UnifiedEvent::PumpFunTradeEvent(e) => { + DexEvent::PumpFunTradeEvent(e) => { 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 }; } - UnifiedEvent::PumpSwapBuyEvent(e) => { + DexEvent::PumpSwapBuyEvent(e) => { swap_data.from_mint = e.quote_mint; swap_data.to_mint = e.base_mint; } - UnifiedEvent::PumpSwapSellEvent(e) => { + DexEvent::PumpSwapSellEvent(e) => { swap_data.from_mint = e.base_mint; swap_data.to_mint = e.quote_mint; } - UnifiedEvent::RaydiumCpmmSwapEvent(e) => { + DexEvent::RaydiumCpmmSwapEvent(e) => { // user = Some(e.payer); from_mint = Some(e.input_token_mint); to_mint = Some(e.output_token_mint); @@ -571,7 +571,7 @@ pub fn parse_swap_data_from_next_grpc_instructions( from_vault = Some(e.input_vault); to_vault = Some(e.output_vault); } - UnifiedEvent::RaydiumClmmSwapEvent(e) => { + DexEvent::RaydiumClmmSwapEvent(e) => { // user = Some(e.payer); swap_data.description = Some("Unable to get from_mint and to_mint from RaydiumClmmSwapEvent".into()); @@ -580,7 +580,7 @@ pub fn parse_swap_data_from_next_grpc_instructions( from_vault = Some(e.input_vault); to_vault = Some(e.output_vault); } - UnifiedEvent::RaydiumClmmSwapV2Event(e) => { + DexEvent::RaydiumClmmSwapV2Event(e) => { // user = Some(e.payer); from_mint = Some(e.input_vault_mint); to_mint = Some(e.output_vault_mint); @@ -589,7 +589,7 @@ pub fn parse_swap_data_from_next_grpc_instructions( from_vault = Some(e.input_vault); to_vault = Some(e.output_vault); } - UnifiedEvent::RaydiumAmmV4SwapEvent(e) => { + DexEvent::RaydiumAmmV4SwapEvent(e) => { // user = Some(e.user_source_owner); swap_data.description = Some("Unable to get from_mint and to_mint from RaydiumAmmV4SwapEvent".into()); diff --git a/src/streaming/event_parser/core/account_event_parser.rs b/src/streaming/event_parser/core/account_event_parser.rs index 4f29f19..22e3d20 100644 --- a/src/streaming/event_parser/core/account_event_parser.rs +++ b/src/streaming/event_parser/core/account_event_parser.rs @@ -3,7 +3,7 @@ use crate::streaming::event_parser::common::filter::EventTypeFilter; use crate::streaming::event_parser::common::high_performance_clock::elapsed_micros_since; use crate::streaming::event_parser::common::EventMetadata; use crate::streaming::event_parser::core::parser_cache::get_account_configs; -use crate::streaming::event_parser::core::traits::UnifiedEvent; +use crate::streaming::event_parser::core::traits::DexEvent; use crate::streaming::event_parser::Protocol; use crate::streaming::grpc::AccountPretty; use serde::{Deserialize, Serialize}; @@ -62,7 +62,7 @@ impl AccountEventParser { protocols: &[Protocol], account: AccountPretty, event_type_filter: Option<&EventTypeFilter>, - ) -> Option { + ) -> Option { // 直接从 parser_cache 获取配置 let configs = get_account_configs( protocols, @@ -102,7 +102,7 @@ impl AccountEventParser { pub fn parse_token_account_event( account: &AccountPretty, metadata: EventMetadata, - ) -> Option { + ) -> Option { let pubkey = account.pubkey; let executable = account.executable; let lamports = account.lamports; @@ -123,7 +123,7 @@ impl AccountEventParser { }; let recv_delta = elapsed_micros_since(account.recv_us); event.metadata.handle_us = recv_delta; - return Some(UnifiedEvent::TokenInfoEvent(event)); + return Some(DexEvent::TokenInfoEvent(event)); } } // Spl Token2022 Mint @@ -141,7 +141,7 @@ impl AccountEventParser { }; let recv_delta = elapsed_micros_since(account.recv_us); event.metadata.handle_us = recv_delta; - return Some(UnifiedEvent::TokenInfoEvent(event)); + return Some(DexEvent::TokenInfoEvent(event)); } } let amount = if account.owner.to_bytes() == spl_token_2022::ID.to_bytes() { @@ -164,13 +164,13 @@ impl AccountEventParser { }; let recv_delta = elapsed_micros_since(account.recv_us); event.metadata.handle_us = recv_delta; - Some(UnifiedEvent::TokenAccountEvent(event)) + Some(DexEvent::TokenAccountEvent(event)) } pub fn parse_nonce_account_event( account: &AccountPretty, metadata: EventMetadata, - ) -> Option { + ) -> Option { if let Ok(info) = parse_nonce(&account.data) { match info { solana_account_decoder::parse_nonce::UiNonceState::Initialized(details) => { @@ -185,7 +185,7 @@ impl AccountEventParser { authority: details.authority, }; event.metadata.handle_us = elapsed_micros_since(account.recv_us); - return Some(UnifiedEvent::NonceAccountEvent(event)); + return Some(DexEvent::NonceAccountEvent(event)); } solana_account_decoder::parse_nonce::UiNonceState::Uninitialized => {} } diff --git a/src/streaming/event_parser/core/common_event_parser.rs b/src/streaming/event_parser/core/common_event_parser.rs index 1775270..6f34fd3 100644 --- a/src/streaming/event_parser/core/common_event_parser.rs +++ b/src/streaming/event_parser/core/common_event_parser.rs @@ -1,5 +1,5 @@ use crate::streaming::event_parser::common::high_performance_clock::elapsed_micros_since; -use crate::streaming::event_parser::core::traits::UnifiedEvent; +use crate::streaming::event_parser::core::traits::DexEvent; use crate::streaming::event_parser::protocols::block::block_meta_event::BlockMetaEvent; pub struct CommonEventParser {} @@ -10,9 +10,9 @@ impl CommonEventParser { block_hash: String, block_time_ms: i64, recv_us: i64, - ) -> UnifiedEvent { + ) -> DexEvent { let mut block_meta_event = BlockMetaEvent::new(slot, block_hash, block_time_ms, recv_us); block_meta_event.metadata.handle_us = elapsed_micros_since(recv_us); - UnifiedEvent::BlockMetaEvent(block_meta_event) + DexEvent::BlockMetaEvent(block_meta_event) } } diff --git a/src/streaming/event_parser/core/event_parser.rs b/src/streaming/event_parser/core/event_parser.rs index 90436cf..e155885 100644 --- a/src/streaming/event_parser/core/event_parser.rs +++ b/src/streaming/event_parser/core/event_parser.rs @@ -12,12 +12,13 @@ use crate::streaming::{ add_bonk_dev_address, add_dev_address, is_bonk_dev_address_in_signature, is_dev_address_in_signature, }, + merger_event::merge, parser_cache::{ build_account_pubkeys_with_cache, get_global_instruction_configs, get_global_program_ids, GenericEventParseConfig, }, }, - Protocol, UnifiedEvent, + Protocol, DexEvent, }, }; use prost_types::Timestamp; @@ -47,7 +48,7 @@ impl EventParser { inner_instructions: &[yellowstone_grpc_proto::prelude::InnerInstructions], bot_wallet: Option, transaction_index: Option, - callback: Arc Fn(&'a UnifiedEvent) + Send + Sync>, + callback: Arc Fn(&'a DexEvent) + Send + Sync>, ) -> anyhow::Result<()> { // 获取交易的指令和账户 let mut accounts = accounts.to_vec(); @@ -137,7 +138,7 @@ impl EventParser { inner_instructions: &[InnerInstructions], bot_wallet: Option, transaction_index: Option, - callback: Arc Fn(&'a UnifiedEvent) + Send + Sync>, + callback: Arc Fn(&'a DexEvent) + Send + Sync>, ) -> anyhow::Result<()> { // 获取交易的指令和账户 let compiled_instructions = transaction.message.instructions(); @@ -217,10 +218,10 @@ impl EventParser { bot_wallet: Option, transaction_index: Option, inner_instructions: &[InnerInstructions], - callback: Arc, + callback: Arc, ) -> anyhow::Result<()> { // 创建适配器回调,将所有权回调转换为引用回调 - let adapter_callback = Arc::new(move |event: &UnifiedEvent| { + let adapter_callback = Arc::new(move |event: &DexEvent| { callback(event.clone()); }); Self::parse_versioned_transaction( @@ -251,7 +252,7 @@ impl EventParser { bot_wallet: Option, transaction_index: Option, inner_instructions: &[InnerInstructions], - callback: Arc Fn(&'a UnifiedEvent) + Send + Sync>, + callback: Arc Fn(&'a DexEvent) + Send + Sync>, ) -> anyhow::Result<()> { let accounts: Vec = versioned_tx.message.static_account_keys().to_vec(); Self::parse_instruction_events_from_versioned_transaction( @@ -282,10 +283,10 @@ impl EventParser { recv_us: i64, bot_wallet: Option, transaction_index: Option, - callback: Arc, + callback: Arc, ) -> anyhow::Result<()> { // 创建适配器回调,将所有权回调转换为引用回调 - let adapter_callback = Arc::new(move |event: &UnifiedEvent| { + let adapter_callback = Arc::new(move |event: &DexEvent| { callback(event.clone()); }); // 调用原始方法 @@ -314,7 +315,7 @@ impl EventParser { recv_us: i64, bot_wallet: Option, transaction_index: Option, - callback: Arc Fn(&'a UnifiedEvent) + Send + Sync>, + callback: Arc Fn(&'a DexEvent) + Send + Sync>, ) -> anyhow::Result<()> { if let Some(transition) = grpc_tx.transaction { if let Some(message) = &transition.message { @@ -381,7 +382,7 @@ impl EventParser { event_type_filter: Option<&EventTypeFilter>, signature: Signature, transaction: EncodedConfirmedTransactionWithStatusMeta, - callback: Arc Fn(&'a UnifiedEvent) + Send + Sync>, + callback: Arc Fn(&'a DexEvent) + Send + Sync>, ) -> anyhow::Result<()> { let versioned_tx = match transaction.transaction.transaction.decode() { Some(tx) => tx, @@ -502,7 +503,7 @@ impl EventParser { outer_index: i64, inner_index: Option, transaction_index: Option, - ) -> Option { + ) -> Option { if let Some(parser) = config.inner_instruction_parser { 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; @@ -538,7 +539,7 @@ impl EventParser { outer_index: i64, inner_index: Option, transaction_index: Option, - ) -> Option { + ) -> Option { if let Some(parser) = config.instruction_parser { 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; @@ -573,7 +574,7 @@ impl EventParser { inner_index: Option, transaction_index: Option, config: &GenericEventParseConfig, - ) -> Vec { + ) -> Vec { // Use SIMD-optimized data validation with correct discriminator length let discriminator_len = config.inner_instruction_discriminator.len(); if !SimdUtils::validate_instruction_data_simd( @@ -622,7 +623,7 @@ impl EventParser { inner_index: Option, transaction_index: Option, config: &GenericEventParseConfig, - ) -> Vec { + ) -> Vec { // Use SIMD-optimized data validation with correct discriminator length let discriminator_len = config.inner_instruction_discriminator.len(); if !SimdUtils::validate_instruction_data_simd( @@ -675,7 +676,7 @@ impl EventParser { bot_wallet: Option, transaction_index: Option, inner_instructions: Option<&InnerInstructions>, - callback: Arc Fn(&'a UnifiedEvent) + Send + Sync>, + callback: Arc Fn(&'a DexEvent) + Send + Sync>, ) -> anyhow::Result<()> { // 添加边界检查以防止越界访问 let program_id_index = instruction.program_id_index as usize; @@ -734,7 +735,7 @@ impl EventParser { for (_disc, config, mut event) in all_results { // 阻塞处理:原有的同步逻辑 - let mut inner_instruction_event: Option = None; + let mut inner_instruction_event: Option = None; if inner_instructions.is_some() { let inner_instructions_ref = inner_instructions.unwrap(); @@ -790,7 +791,7 @@ impl EventParser { // 合并事件 if let Some(inner_instruction_event) = inner_instruction_event { - // event.merge(&*inner_instruction_event); + merge(&mut event, inner_instruction_event); } // 设置处理时间(使用高性能时钟) event.metadata_mut().handle_us = elapsed_micros_since(recv_us); @@ -817,7 +818,7 @@ impl EventParser { bot_wallet: Option, transaction_index: Option, inner_instructions: Option<&yellowstone_grpc_proto::prelude::InnerInstructions>, - callback: Arc Fn(&'a UnifiedEvent) + Send + Sync>, + callback: Arc Fn(&'a DexEvent) + Send + Sync>, ) -> anyhow::Result<()> { // 添加边界检查以防止越界访问 let program_id_index = instruction.program_id_index as usize; @@ -876,7 +877,7 @@ impl EventParser { for (_disc, config, mut event) in all_results { // 阻塞处理:原有的同步逻辑 - let mut inner_instruction_event: Option = None; + let mut inner_instruction_event: Option = None; if inner_instructions.is_some() { let inner_instructions_ref = inner_instructions.unwrap(); @@ -932,7 +933,7 @@ impl EventParser { // 合并事件 if let Some(inner_instruction_event) = inner_instruction_event { - // event.merge(&*inner_instruction_event); + merge(&mut event, inner_instruction_event); } // 设置处理时间(使用高性能时钟) event.metadata_mut().handle_us = elapsed_micros_since(recv_us); @@ -950,18 +951,18 @@ impl EventParser { get_global_program_ids(protocols, event_type_filter).contains(program_id) } - fn process_event(mut event: UnifiedEvent, bot_wallet: Option) -> UnifiedEvent { + fn process_event(event: DexEvent, bot_wallet: Option) -> DexEvent { let signature = event.metadata().signature; // Copy the signature to avoid borrowing issues match event { - UnifiedEvent::PumpFunCreateTokenEvent(token_info) => { + DexEvent::PumpFunCreateTokenEvent(token_info) => { add_dev_address(&signature, token_info.user); if token_info.creator != Pubkey::default() && token_info.creator != token_info.user { add_dev_address(&signature, token_info.creator); } - UnifiedEvent::PumpFunCreateTokenEvent(token_info) + DexEvent::PumpFunCreateTokenEvent(token_info) } - UnifiedEvent::PumpFunTradeEvent(mut trade_info) => { + DexEvent::PumpFunTradeEvent(mut trade_info) => { if is_dev_address_in_signature(&signature, &trade_info.user) || is_dev_address_in_signature(&signature, &trade_info.creator) { @@ -985,31 +986,31 @@ impl EventParser { trade_info.sol_amount }; } - UnifiedEvent::PumpFunTradeEvent(trade_info) + DexEvent::PumpFunTradeEvent(trade_info) } - UnifiedEvent::PumpSwapBuyEvent(mut trade_info) => { + DexEvent::PumpSwapBuyEvent(mut trade_info) => { 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; } - UnifiedEvent::PumpSwapBuyEvent(trade_info) + DexEvent::PumpSwapBuyEvent(trade_info) } - UnifiedEvent::PumpSwapSellEvent(mut trade_info) => { + DexEvent::PumpSwapSellEvent(mut trade_info) => { 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; } - UnifiedEvent::PumpSwapSellEvent(trade_info) + DexEvent::PumpSwapSellEvent(trade_info) } - UnifiedEvent::BonkPoolCreateEvent(pool_info) => { + DexEvent::BonkPoolCreateEvent(pool_info) => { add_bonk_dev_address(&signature, pool_info.creator); - UnifiedEvent::BonkPoolCreateEvent(pool_info) + DexEvent::BonkPoolCreateEvent(pool_info) } - UnifiedEvent::BonkTradeEvent(mut trade_info) => { + DexEvent::BonkTradeEvent(mut trade_info) => { if is_bonk_dev_address_in_signature(&signature, &trade_info.payer) { trade_info.is_dev_create_token_trade = true; } else if Some(trade_info.payer) == bot_wallet { @@ -1017,7 +1018,7 @@ impl EventParser { } else { trade_info.is_dev_create_token_trade = false; } - UnifiedEvent::BonkTradeEvent(trade_info) + DexEvent::BonkTradeEvent(trade_info) } _ => event, } diff --git a/src/streaming/event_parser/core/merger_event.rs b/src/streaming/event_parser/core/merger_event.rs new file mode 100644 index 0000000..2497d87 --- /dev/null +++ b/src/streaming/event_parser/core/merger_event.rs @@ -0,0 +1,226 @@ +use crate::streaming::event_parser::DexEvent; + +pub fn merge(instruction_event: &mut DexEvent, cpi_log_event: DexEvent) { + match instruction_event { + // PumpFun events + DexEvent::PumpFunTradeEvent(e) => match cpi_log_event { + DexEvent::PumpFunTradeEvent(cpie) => { + e.mint = cpie.mint; + e.sol_amount = cpie.sol_amount; + e.token_amount = cpie.token_amount; + e.is_buy = cpie.is_buy; + e.user = cpie.user; + e.timestamp = cpie.timestamp; + e.virtual_sol_reserves = cpie.virtual_sol_reserves; + e.virtual_token_reserves = cpie.virtual_token_reserves; + e.real_sol_reserves = cpie.real_sol_reserves; + e.real_token_reserves = cpie.real_token_reserves; + e.fee_recipient = cpie.fee_recipient; + e.fee_basis_points = cpie.fee_basis_points; + e.fee = cpie.fee; + e.creator = cpie.creator; + e.creator_fee_basis_points = cpie.creator_fee_basis_points; + e.creator_fee = cpie.creator_fee; + } + _ => {} + }, + DexEvent::PumpFunCreateTokenEvent(e) => match cpi_log_event { + DexEvent::PumpFunCreateTokenEvent(cpie) => { + e.mint = cpie.mint; + e.bonding_curve = cpie.bonding_curve; + e.user = cpie.user; + e.creator = cpie.creator; + e.timestamp = cpie.timestamp; + e.virtual_token_reserves = cpie.virtual_token_reserves; + e.virtual_sol_reserves = cpie.virtual_sol_reserves; + e.real_token_reserves = cpie.real_token_reserves; + e.token_total_supply = cpie.token_total_supply; + } + _ => {} + }, + DexEvent::PumpFunMigrateEvent(e) => match cpi_log_event { + DexEvent::PumpFunMigrateEvent(cpie) => { + e.user = cpie.user; + e.mint = cpie.mint; + e.mint_amount = cpie.mint_amount; + e.sol_amount = cpie.sol_amount; + e.pool_migration_fee = cpie.pool_migration_fee; + e.bonding_curve = cpie.bonding_curve; + e.timestamp = cpie.timestamp; + e.pool = cpie.pool; + } + _ => {} + }, + + // Bonk events + DexEvent::BonkTradeEvent(e) => match cpi_log_event { + DexEvent::BonkTradeEvent(cpie) => { + e.pool_state = cpie.pool_state; + e.total_base_sell = cpie.total_base_sell; + e.virtual_base = cpie.virtual_base; + e.virtual_quote = cpie.virtual_quote; + e.real_base_before = cpie.real_base_before; + e.real_quote_before = cpie.real_quote_before; + e.real_base_after = cpie.real_base_after; + e.real_quote_after = cpie.real_quote_after; + e.amount_in = cpie.amount_in; + e.amount_out = cpie.amount_out; + e.protocol_fee = cpie.protocol_fee; + e.platform_fee = cpie.platform_fee; + e.creator_fee = cpie.creator_fee; + e.share_fee = cpie.share_fee; + e.trade_direction = cpie.trade_direction; + e.pool_status = cpie.pool_status; + e.exact_in = cpie.exact_in; + } + _ => {} + }, + DexEvent::BonkPoolCreateEvent(e) => match cpi_log_event { + DexEvent::BonkPoolCreateEvent(cpie) => { + e.pool_state = cpie.pool_state; + e.creator = cpie.creator; + e.config = cpie.config; + e.base_mint_param = cpie.base_mint_param; + e.curve_param = cpie.curve_param; + e.vesting_param = cpie.vesting_param; + e.amm_fee_on = cpie.amm_fee_on; + } + _ => {} + }, + DexEvent::BonkMigrateToAmmEvent(e) => match cpi_log_event { + DexEvent::BonkMigrateToAmmEvent(cpie) => { + e.base_lot_size = cpie.base_lot_size; + e.quote_lot_size = cpie.quote_lot_size; + e.market_vault_signer_nonce = cpie.market_vault_signer_nonce; + } + _ => {} + }, + + // PumpSwap events + DexEvent::PumpSwapBuyEvent(e) => match cpi_log_event { + DexEvent::PumpSwapBuyEvent(cpie) => { + e.timestamp = cpie.timestamp; + e.base_amount_out = cpie.base_amount_out; + e.max_quote_amount_in = cpie.max_quote_amount_in; + e.user_base_token_reserves = cpie.user_base_token_reserves; + e.user_quote_token_reserves = cpie.user_quote_token_reserves; + e.pool_base_token_reserves = cpie.pool_base_token_reserves; + e.pool_quote_token_reserves = cpie.pool_quote_token_reserves; + e.quote_amount_in = cpie.quote_amount_in; + e.lp_fee_basis_points = cpie.lp_fee_basis_points; + e.lp_fee = cpie.lp_fee; + e.protocol_fee_basis_points = cpie.protocol_fee_basis_points; + e.protocol_fee = cpie.protocol_fee; + e.quote_amount_in_with_lp_fee = cpie.quote_amount_in_with_lp_fee; + e.user_quote_amount_in = cpie.user_quote_amount_in; + e.pool = cpie.pool; + e.user = cpie.user; + e.user_base_token_account = cpie.user_base_token_account; + e.user_quote_token_account = cpie.user_quote_token_account; + e.protocol_fee_recipient = cpie.protocol_fee_recipient; + e.protocol_fee_recipient_token_account = cpie.protocol_fee_recipient_token_account; + e.coin_creator = cpie.coin_creator; + e.coin_creator_fee_basis_points = cpie.coin_creator_fee_basis_points; + e.coin_creator_fee = cpie.coin_creator_fee; + } + _ => {} + }, + DexEvent::PumpSwapSellEvent(e) => match cpi_log_event { + DexEvent::PumpSwapSellEvent(cpie) => { + e.timestamp = cpie.timestamp; + e.base_amount_in = cpie.base_amount_in; + e.min_quote_amount_out = cpie.min_quote_amount_out; + e.user_base_token_reserves = cpie.user_base_token_reserves; + e.user_quote_token_reserves = cpie.user_quote_token_reserves; + e.pool_base_token_reserves = cpie.pool_base_token_reserves; + e.pool_quote_token_reserves = cpie.pool_quote_token_reserves; + e.quote_amount_out = cpie.quote_amount_out; + e.lp_fee_basis_points = cpie.lp_fee_basis_points; + e.lp_fee = cpie.lp_fee; + e.protocol_fee_basis_points = cpie.protocol_fee_basis_points; + e.protocol_fee = cpie.protocol_fee; + e.quote_amount_out_without_lp_fee = cpie.quote_amount_out_without_lp_fee; + e.user_quote_amount_out = cpie.user_quote_amount_out; + e.pool = cpie.pool; + e.user = cpie.user; + e.user_base_token_account = cpie.user_base_token_account; + e.user_quote_token_account = cpie.user_quote_token_account; + e.protocol_fee_recipient = cpie.protocol_fee_recipient; + e.protocol_fee_recipient_token_account = cpie.protocol_fee_recipient_token_account; + e.coin_creator = cpie.coin_creator; + e.coin_creator_fee_basis_points = cpie.coin_creator_fee_basis_points; + e.coin_creator_fee = cpie.coin_creator_fee; + } + _ => {} + }, + DexEvent::PumpSwapCreatePoolEvent(e) => match cpi_log_event { + DexEvent::PumpSwapCreatePoolEvent(cpie) => { + e.timestamp = cpie.timestamp; + e.index = cpie.index; + e.creator = cpie.creator; + e.base_mint = cpie.base_mint; + e.quote_mint = cpie.quote_mint; + e.base_mint_decimals = cpie.base_mint_decimals; + e.quote_mint_decimals = cpie.quote_mint_decimals; + e.base_amount_in = cpie.base_amount_in; + e.quote_amount_in = cpie.quote_amount_in; + e.pool_base_amount = cpie.pool_base_amount; + e.pool_quote_amount = cpie.pool_quote_amount; + e.minimum_liquidity = cpie.minimum_liquidity; + e.initial_liquidity = cpie.initial_liquidity; + e.lp_token_amount_out = cpie.lp_token_amount_out; + e.pool_bump = cpie.pool_bump; + e.pool = cpie.pool; + e.lp_mint = cpie.lp_mint; + e.user_base_token_account = cpie.user_base_token_account; + e.user_quote_token_account = cpie.user_quote_token_account; + e.coin_creator = cpie.coin_creator; + } + _ => {} + }, + DexEvent::PumpSwapDepositEvent(e) => match cpi_log_event { + DexEvent::PumpSwapDepositEvent(cpie) => { + e.timestamp = cpie.timestamp; + e.lp_token_amount_out = cpie.lp_token_amount_out; + e.max_base_amount_in = cpie.max_base_amount_in; + e.max_quote_amount_in = cpie.max_quote_amount_in; + e.user_base_token_reserves = cpie.user_base_token_reserves; + e.user_quote_token_reserves = cpie.user_quote_token_reserves; + e.pool_base_token_reserves = cpie.pool_base_token_reserves; + e.pool_quote_token_reserves = cpie.pool_quote_token_reserves; + e.base_amount_in = cpie.base_amount_in; + e.quote_amount_in = cpie.quote_amount_in; + e.lp_mint_supply = cpie.lp_mint_supply; + e.pool = cpie.pool; + e.user = cpie.user; + e.user_base_token_account = cpie.user_base_token_account; + e.user_quote_token_account = cpie.user_quote_token_account; + e.user_pool_token_account = cpie.user_pool_token_account; + } + _ => {} + }, + DexEvent::PumpSwapWithdrawEvent(e) => match cpi_log_event { + DexEvent::PumpSwapWithdrawEvent(cpie) => { + e.timestamp = cpie.timestamp; + e.lp_token_amount_in = cpie.lp_token_amount_in; + e.min_base_amount_out = cpie.min_base_amount_out; + e.min_quote_amount_out = cpie.min_quote_amount_out; + e.user_base_token_reserves = cpie.user_base_token_reserves; + e.user_quote_token_reserves = cpie.user_quote_token_reserves; + e.pool_base_token_reserves = cpie.pool_base_token_reserves; + e.pool_quote_token_reserves = cpie.pool_quote_token_reserves; + e.base_amount_out = cpie.base_amount_out; + e.quote_amount_out = cpie.quote_amount_out; + e.lp_mint_supply = cpie.lp_mint_supply; + e.pool = cpie.pool; + e.user = cpie.user; + e.user_base_token_account = cpie.user_base_token_account; + e.user_quote_token_account = cpie.user_quote_token_account; + e.user_pool_token_account = cpie.user_pool_token_account; + } + _ => {} + }, + + _ => {} + } +} diff --git a/src/streaming/event_parser/core/mod.rs b/src/streaming/event_parser/core/mod.rs index dcce995..2fcba6a 100755 --- a/src/streaming/event_parser/core/mod.rs +++ b/src/streaming/event_parser/core/mod.rs @@ -3,10 +3,11 @@ pub mod common_event_parser; pub mod global_state; pub mod parser_cache; pub mod traits; -pub use traits::UnifiedEvent; +pub use traits::DexEvent; pub use parser_cache::{ AccountEventParseConfig, AccountEventParserFn, GenericEventParseConfig, InnerInstructionEventParser, InstructionEventParser, get_account_configs, }; pub mod event_parser; +pub mod merger_event; \ No newline at end of file diff --git a/src/streaming/event_parser/core/parser_cache.rs b/src/streaming/event_parser/core/parser_cache.rs index affcedd..d13cb8c 100644 --- a/src/streaming/event_parser/core/parser_cache.rs +++ b/src/streaming/event_parser/core/parser_cache.rs @@ -21,7 +21,7 @@ use crate::streaming::{ raydium_clmm::parser::RAYDIUM_CLMM_PROGRAM_ID, raydium_cpmm::parser::RAYDIUM_CPMM_PROGRAM_ID, }, - Protocol, UnifiedEvent, + Protocol, DexEvent, }, grpc::AccountPretty, }; @@ -39,13 +39,13 @@ use std::{ /// /// 用于解析内联指令(Inner Instruction)生成的事件 pub type InnerInstructionEventParser = - fn(data: &[u8], metadata: EventMetadata) -> Option; + fn(data: &[u8], metadata: EventMetadata) -> Option; /// 指令事件解析器函数类型 /// /// 用于解析普通指令(Instruction)生成的事件,需要账户信息 pub type InstructionEventParser = - fn(data: &[u8], accounts: &[Pubkey], metadata: EventMetadata) -> Option; + fn(data: &[u8], accounts: &[Pubkey], metadata: EventMetadata) -> Option; /// 指令事件解析器配置 /// @@ -327,9 +327,6 @@ impl Default for AccountPubkeyCache { } } -/// 线程局部账户公钥缓存 -/// -/// 每个线程独立维护一个缓存实例,避免锁竞争 thread_local! { static THREAD_LOCAL_ACCOUNT_CACHE: std::cell::RefCell = std::cell::RefCell::new(AccountPubkeyCache::new()); @@ -365,7 +362,7 @@ pub fn build_account_pubkeys_with_cache( /// /// 用于解析账户状态变更生成的事件 pub type AccountEventParserFn = - fn(account: &AccountPretty, metadata: EventMetadata) -> Option; + fn(account: &AccountPretty, metadata: EventMetadata) -> Option; /// 账户事件解析器配置 /// diff --git a/src/streaming/event_parser/core/traits.rs b/src/streaming/event_parser/core/traits.rs index f5f6475..a2f6f00 100755 --- a/src/streaming/event_parser/core/traits.rs +++ b/src/streaming/event_parser/core/traits.rs @@ -14,7 +14,7 @@ use std::fmt::Debug; /// Unified Event Enum - Replaces the trait-based approach with a type-safe enum #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] -pub enum UnifiedEvent { +pub enum DexEvent { // Bonk events BonkTradeEvent(BonkTradeEvent), BonkPoolCreateEvent(BonkPoolCreateEvent), @@ -76,106 +76,106 @@ pub enum UnifiedEvent { BlockMetaEvent(BlockMetaEvent), } -impl UnifiedEvent { +impl DexEvent { pub fn metadata(&self) -> &EventMetadata { match self { - UnifiedEvent::BonkTradeEvent(e) => &e.metadata, - UnifiedEvent::BonkPoolCreateEvent(e) => &e.metadata, - UnifiedEvent::BonkMigrateToAmmEvent(e) => &e.metadata, - UnifiedEvent::BonkMigrateToCpswapEvent(e) => &e.metadata, - UnifiedEvent::BonkPoolStateAccountEvent(e) => &e.metadata, - UnifiedEvent::BonkGlobalConfigAccountEvent(e) => &e.metadata, - UnifiedEvent::BonkPlatformConfigAccountEvent(e) => &e.metadata, - UnifiedEvent::PumpFunCreateTokenEvent(e) => &e.metadata, - UnifiedEvent::PumpFunTradeEvent(e) => &e.metadata, - UnifiedEvent::PumpFunMigrateEvent(e) => &e.metadata, - UnifiedEvent::PumpFunBondingCurveAccountEvent(e) => &e.metadata, - UnifiedEvent::PumpFunGlobalAccountEvent(e) => &e.metadata, - UnifiedEvent::PumpSwapBuyEvent(e) => &e.metadata, - UnifiedEvent::PumpSwapSellEvent(e) => &e.metadata, - UnifiedEvent::PumpSwapCreatePoolEvent(e) => &e.metadata, - UnifiedEvent::PumpSwapDepositEvent(e) => &e.metadata, - UnifiedEvent::PumpSwapWithdrawEvent(e) => &e.metadata, - UnifiedEvent::PumpSwapGlobalConfigAccountEvent(e) => &e.metadata, - UnifiedEvent::PumpSwapPoolAccountEvent(e) => &e.metadata, - UnifiedEvent::RaydiumAmmV4SwapEvent(e) => &e.metadata, - UnifiedEvent::RaydiumAmmV4DepositEvent(e) => &e.metadata, - UnifiedEvent::RaydiumAmmV4WithdrawEvent(e) => &e.metadata, - UnifiedEvent::RaydiumAmmV4WithdrawPnlEvent(e) => &e.metadata, - UnifiedEvent::RaydiumAmmV4Initialize2Event(e) => &e.metadata, - UnifiedEvent::RaydiumAmmV4AmmInfoAccountEvent(e) => &e.metadata, - UnifiedEvent::RaydiumClmmSwapEvent(e) => &e.metadata, - UnifiedEvent::RaydiumClmmSwapV2Event(e) => &e.metadata, - UnifiedEvent::RaydiumClmmClosePositionEvent(e) => &e.metadata, - UnifiedEvent::RaydiumClmmIncreaseLiquidityV2Event(e) => &e.metadata, - UnifiedEvent::RaydiumClmmDecreaseLiquidityV2Event(e) => &e.metadata, - UnifiedEvent::RaydiumClmmCreatePoolEvent(e) => &e.metadata, - UnifiedEvent::RaydiumClmmOpenPositionWithToken22NftEvent(e) => &e.metadata, - UnifiedEvent::RaydiumClmmOpenPositionV2Event(e) => &e.metadata, - UnifiedEvent::RaydiumClmmAmmConfigAccountEvent(e) => &e.metadata, - UnifiedEvent::RaydiumClmmPoolStateAccountEvent(e) => &e.metadata, - UnifiedEvent::RaydiumClmmTickArrayStateAccountEvent(e) => &e.metadata, - UnifiedEvent::RaydiumCpmmSwapEvent(e) => &e.metadata, - UnifiedEvent::RaydiumCpmmDepositEvent(e) => &e.metadata, - UnifiedEvent::RaydiumCpmmWithdrawEvent(e) => &e.metadata, - UnifiedEvent::RaydiumCpmmInitializeEvent(e) => &e.metadata, - UnifiedEvent::RaydiumCpmmAmmConfigAccountEvent(e) => &e.metadata, - UnifiedEvent::RaydiumCpmmPoolStateAccountEvent(e) => &e.metadata, - UnifiedEvent::TokenAccountEvent(e) => &e.metadata, - UnifiedEvent::NonceAccountEvent(e) => &e.metadata, - UnifiedEvent::TokenInfoEvent(e) => &e.metadata, - UnifiedEvent::BlockMetaEvent(e) => &e.metadata, + DexEvent::BonkTradeEvent(e) => &e.metadata, + DexEvent::BonkPoolCreateEvent(e) => &e.metadata, + DexEvent::BonkMigrateToAmmEvent(e) => &e.metadata, + DexEvent::BonkMigrateToCpswapEvent(e) => &e.metadata, + DexEvent::BonkPoolStateAccountEvent(e) => &e.metadata, + DexEvent::BonkGlobalConfigAccountEvent(e) => &e.metadata, + DexEvent::BonkPlatformConfigAccountEvent(e) => &e.metadata, + DexEvent::PumpFunCreateTokenEvent(e) => &e.metadata, + DexEvent::PumpFunTradeEvent(e) => &e.metadata, + DexEvent::PumpFunMigrateEvent(e) => &e.metadata, + DexEvent::PumpFunBondingCurveAccountEvent(e) => &e.metadata, + DexEvent::PumpFunGlobalAccountEvent(e) => &e.metadata, + DexEvent::PumpSwapBuyEvent(e) => &e.metadata, + DexEvent::PumpSwapSellEvent(e) => &e.metadata, + DexEvent::PumpSwapCreatePoolEvent(e) => &e.metadata, + DexEvent::PumpSwapDepositEvent(e) => &e.metadata, + DexEvent::PumpSwapWithdrawEvent(e) => &e.metadata, + DexEvent::PumpSwapGlobalConfigAccountEvent(e) => &e.metadata, + DexEvent::PumpSwapPoolAccountEvent(e) => &e.metadata, + DexEvent::RaydiumAmmV4SwapEvent(e) => &e.metadata, + DexEvent::RaydiumAmmV4DepositEvent(e) => &e.metadata, + DexEvent::RaydiumAmmV4WithdrawEvent(e) => &e.metadata, + DexEvent::RaydiumAmmV4WithdrawPnlEvent(e) => &e.metadata, + DexEvent::RaydiumAmmV4Initialize2Event(e) => &e.metadata, + DexEvent::RaydiumAmmV4AmmInfoAccountEvent(e) => &e.metadata, + DexEvent::RaydiumClmmSwapEvent(e) => &e.metadata, + DexEvent::RaydiumClmmSwapV2Event(e) => &e.metadata, + DexEvent::RaydiumClmmClosePositionEvent(e) => &e.metadata, + DexEvent::RaydiumClmmIncreaseLiquidityV2Event(e) => &e.metadata, + DexEvent::RaydiumClmmDecreaseLiquidityV2Event(e) => &e.metadata, + DexEvent::RaydiumClmmCreatePoolEvent(e) => &e.metadata, + DexEvent::RaydiumClmmOpenPositionWithToken22NftEvent(e) => &e.metadata, + DexEvent::RaydiumClmmOpenPositionV2Event(e) => &e.metadata, + DexEvent::RaydiumClmmAmmConfigAccountEvent(e) => &e.metadata, + DexEvent::RaydiumClmmPoolStateAccountEvent(e) => &e.metadata, + DexEvent::RaydiumClmmTickArrayStateAccountEvent(e) => &e.metadata, + DexEvent::RaydiumCpmmSwapEvent(e) => &e.metadata, + DexEvent::RaydiumCpmmDepositEvent(e) => &e.metadata, + DexEvent::RaydiumCpmmWithdrawEvent(e) => &e.metadata, + DexEvent::RaydiumCpmmInitializeEvent(e) => &e.metadata, + DexEvent::RaydiumCpmmAmmConfigAccountEvent(e) => &e.metadata, + DexEvent::RaydiumCpmmPoolStateAccountEvent(e) => &e.metadata, + DexEvent::TokenAccountEvent(e) => &e.metadata, + DexEvent::NonceAccountEvent(e) => &e.metadata, + DexEvent::TokenInfoEvent(e) => &e.metadata, + DexEvent::BlockMetaEvent(e) => &e.metadata, } } pub fn metadata_mut(&mut self) -> &mut EventMetadata { match self { - UnifiedEvent::BonkTradeEvent(e) => &mut e.metadata, - UnifiedEvent::BonkPoolCreateEvent(e) => &mut e.metadata, - UnifiedEvent::BonkMigrateToAmmEvent(e) => &mut e.metadata, - UnifiedEvent::BonkMigrateToCpswapEvent(e) => &mut e.metadata, - UnifiedEvent::BonkPoolStateAccountEvent(e) => &mut e.metadata, - UnifiedEvent::BonkGlobalConfigAccountEvent(e) => &mut e.metadata, - UnifiedEvent::BonkPlatformConfigAccountEvent(e) => &mut e.metadata, - UnifiedEvent::PumpFunCreateTokenEvent(e) => &mut e.metadata, - UnifiedEvent::PumpFunTradeEvent(e) => &mut e.metadata, - UnifiedEvent::PumpFunMigrateEvent(e) => &mut e.metadata, - UnifiedEvent::PumpFunBondingCurveAccountEvent(e) => &mut e.metadata, - UnifiedEvent::PumpFunGlobalAccountEvent(e) => &mut e.metadata, - UnifiedEvent::PumpSwapBuyEvent(e) => &mut e.metadata, - UnifiedEvent::PumpSwapSellEvent(e) => &mut e.metadata, - UnifiedEvent::PumpSwapCreatePoolEvent(e) => &mut e.metadata, - UnifiedEvent::PumpSwapDepositEvent(e) => &mut e.metadata, - UnifiedEvent::PumpSwapWithdrawEvent(e) => &mut e.metadata, - UnifiedEvent::PumpSwapGlobalConfigAccountEvent(e) => &mut e.metadata, - UnifiedEvent::PumpSwapPoolAccountEvent(e) => &mut e.metadata, - UnifiedEvent::RaydiumAmmV4SwapEvent(e) => &mut e.metadata, - UnifiedEvent::RaydiumAmmV4DepositEvent(e) => &mut e.metadata, - UnifiedEvent::RaydiumAmmV4WithdrawEvent(e) => &mut e.metadata, - UnifiedEvent::RaydiumAmmV4WithdrawPnlEvent(e) => &mut e.metadata, - UnifiedEvent::RaydiumAmmV4Initialize2Event(e) => &mut e.metadata, - UnifiedEvent::RaydiumAmmV4AmmInfoAccountEvent(e) => &mut e.metadata, - UnifiedEvent::RaydiumClmmSwapEvent(e) => &mut e.metadata, - UnifiedEvent::RaydiumClmmSwapV2Event(e) => &mut e.metadata, - UnifiedEvent::RaydiumClmmClosePositionEvent(e) => &mut e.metadata, - UnifiedEvent::RaydiumClmmIncreaseLiquidityV2Event(e) => &mut e.metadata, - UnifiedEvent::RaydiumClmmDecreaseLiquidityV2Event(e) => &mut e.metadata, - UnifiedEvent::RaydiumClmmCreatePoolEvent(e) => &mut e.metadata, - UnifiedEvent::RaydiumClmmOpenPositionWithToken22NftEvent(e) => &mut e.metadata, - UnifiedEvent::RaydiumClmmOpenPositionV2Event(e) => &mut e.metadata, - UnifiedEvent::RaydiumClmmAmmConfigAccountEvent(e) => &mut e.metadata, - UnifiedEvent::RaydiumClmmPoolStateAccountEvent(e) => &mut e.metadata, - UnifiedEvent::RaydiumClmmTickArrayStateAccountEvent(e) => &mut e.metadata, - UnifiedEvent::RaydiumCpmmSwapEvent(e) => &mut e.metadata, - UnifiedEvent::RaydiumCpmmDepositEvent(e) => &mut e.metadata, - UnifiedEvent::RaydiumCpmmWithdrawEvent(e) => &mut e.metadata, - UnifiedEvent::RaydiumCpmmInitializeEvent(e) => &mut e.metadata, - UnifiedEvent::RaydiumCpmmAmmConfigAccountEvent(e) => &mut e.metadata, - UnifiedEvent::RaydiumCpmmPoolStateAccountEvent(e) => &mut e.metadata, - UnifiedEvent::TokenAccountEvent(e) => &mut e.metadata, - UnifiedEvent::NonceAccountEvent(e) => &mut e.metadata, - UnifiedEvent::TokenInfoEvent(e) => &mut e.metadata, - UnifiedEvent::BlockMetaEvent(e) => &mut e.metadata, + DexEvent::BonkTradeEvent(e) => &mut e.metadata, + DexEvent::BonkPoolCreateEvent(e) => &mut e.metadata, + DexEvent::BonkMigrateToAmmEvent(e) => &mut e.metadata, + DexEvent::BonkMigrateToCpswapEvent(e) => &mut e.metadata, + DexEvent::BonkPoolStateAccountEvent(e) => &mut e.metadata, + DexEvent::BonkGlobalConfigAccountEvent(e) => &mut e.metadata, + DexEvent::BonkPlatformConfigAccountEvent(e) => &mut e.metadata, + DexEvent::PumpFunCreateTokenEvent(e) => &mut e.metadata, + DexEvent::PumpFunTradeEvent(e) => &mut e.metadata, + DexEvent::PumpFunMigrateEvent(e) => &mut e.metadata, + DexEvent::PumpFunBondingCurveAccountEvent(e) => &mut e.metadata, + DexEvent::PumpFunGlobalAccountEvent(e) => &mut e.metadata, + DexEvent::PumpSwapBuyEvent(e) => &mut e.metadata, + DexEvent::PumpSwapSellEvent(e) => &mut e.metadata, + DexEvent::PumpSwapCreatePoolEvent(e) => &mut e.metadata, + DexEvent::PumpSwapDepositEvent(e) => &mut e.metadata, + DexEvent::PumpSwapWithdrawEvent(e) => &mut e.metadata, + DexEvent::PumpSwapGlobalConfigAccountEvent(e) => &mut e.metadata, + DexEvent::PumpSwapPoolAccountEvent(e) => &mut e.metadata, + DexEvent::RaydiumAmmV4SwapEvent(e) => &mut e.metadata, + DexEvent::RaydiumAmmV4DepositEvent(e) => &mut e.metadata, + DexEvent::RaydiumAmmV4WithdrawEvent(e) => &mut e.metadata, + DexEvent::RaydiumAmmV4WithdrawPnlEvent(e) => &mut e.metadata, + DexEvent::RaydiumAmmV4Initialize2Event(e) => &mut e.metadata, + DexEvent::RaydiumAmmV4AmmInfoAccountEvent(e) => &mut e.metadata, + DexEvent::RaydiumClmmSwapEvent(e) => &mut e.metadata, + DexEvent::RaydiumClmmSwapV2Event(e) => &mut e.metadata, + DexEvent::RaydiumClmmClosePositionEvent(e) => &mut e.metadata, + DexEvent::RaydiumClmmIncreaseLiquidityV2Event(e) => &mut e.metadata, + DexEvent::RaydiumClmmDecreaseLiquidityV2Event(e) => &mut e.metadata, + DexEvent::RaydiumClmmCreatePoolEvent(e) => &mut e.metadata, + DexEvent::RaydiumClmmOpenPositionWithToken22NftEvent(e) => &mut e.metadata, + DexEvent::RaydiumClmmOpenPositionV2Event(e) => &mut e.metadata, + DexEvent::RaydiumClmmAmmConfigAccountEvent(e) => &mut e.metadata, + DexEvent::RaydiumClmmPoolStateAccountEvent(e) => &mut e.metadata, + DexEvent::RaydiumClmmTickArrayStateAccountEvent(e) => &mut e.metadata, + DexEvent::RaydiumCpmmSwapEvent(e) => &mut e.metadata, + DexEvent::RaydiumCpmmDepositEvent(e) => &mut e.metadata, + DexEvent::RaydiumCpmmWithdrawEvent(e) => &mut e.metadata, + DexEvent::RaydiumCpmmInitializeEvent(e) => &mut e.metadata, + DexEvent::RaydiumCpmmAmmConfigAccountEvent(e) => &mut e.metadata, + DexEvent::RaydiumCpmmPoolStateAccountEvent(e) => &mut e.metadata, + DexEvent::TokenAccountEvent(e) => &mut e.metadata, + DexEvent::NonceAccountEvent(e) => &mut e.metadata, + DexEvent::TokenInfoEvent(e) => &mut e.metadata, + DexEvent::BlockMetaEvent(e) => &mut e.metadata, } } } diff --git a/src/streaming/event_parser/mod.rs b/src/streaming/event_parser/mod.rs index a53ba60..b981df0 100755 --- a/src/streaming/event_parser/mod.rs +++ b/src/streaming/event_parser/mod.rs @@ -2,5 +2,5 @@ pub mod common; pub mod core; pub mod protocols; -pub use core::traits::UnifiedEvent; +pub use core::traits::DexEvent; pub use protocols::types::Protocol; \ No newline at end of file diff --git a/src/streaming/event_parser/protocols/bonk/events.rs b/src/streaming/event_parser/protocols/bonk/events.rs index 42457a7..8172a59 100755 --- a/src/streaming/event_parser/protocols/bonk/events.rs +++ b/src/streaming/event_parser/protocols/bonk/events.rs @@ -80,28 +80,6 @@ pub fn bonk_trade_event_log_decode(data: &[u8]) -> Option { borsh::from_slice::(&data[..BONK_TRADE_EVENT_LOG_SIZE]).ok() } -// Macro to generate UnifiedEvent implementation, specifying the fields to be merged -// impl_unified_event!( -// BonkTradeEvent, -// pool_state, -// total_base_sell, -// virtual_base, -// virtual_quote, -// real_base_before, -// real_quote_before, -// real_base_after, -// real_quote_after, -// amount_in, -// amount_out, -// protocol_fee, -// platform_fee, -// creator_fee, -// share_fee, -// trade_direction, -// pool_status, -// exact_in -// ); - /// Create pool event #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct BonkPoolCreateEvent { @@ -139,18 +117,6 @@ pub fn bonk_pool_create_event_log_decode(data: &[u8]) -> Option(&data[..BONK_POOL_CREATE_EVENT_LOG_SIZE]).ok() } -// Macro to generate UnifiedEvent implementation, specifying the fields to be merged -// impl_unified_event!( -// BonkPoolCreateEvent, -// pool_state, -// creator, -// config, -// base_mint_param, -// curve_param, -// vesting_param, -// amm_fee_on -// ); - /// Create pool event #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct BonkMigrateToAmmEvent { @@ -225,14 +191,6 @@ pub struct BonkMigrateToAmmEvent { pub rent_program: Pubkey, } -// Macro to generate UnifiedEvent implementation, specifying the fields to be merged -// impl_unified_event!( -// BonkMigrateToAmmEvent, -// base_lot_size, -// quote_lot_size, -// market_vault_signer_nonce -// ); - // Migrate to CP Swap event #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct BonkMigrateToCpswapEvent { diff --git a/src/streaming/event_parser/protocols/bonk/parser.rs b/src/streaming/event_parser/protocols/bonk/parser.rs index fc22718..07829d6 100755 --- a/src/streaming/event_parser/protocols/bonk/parser.rs +++ b/src/streaming/event_parser/protocols/bonk/parser.rs @@ -9,7 +9,7 @@ use crate::streaming::event_parser::{ ConstantCurve, CurveParams, FixedCurve, LinearCurve, MintParams, TradeDirection, VestingParams, }, - UnifiedEvent, + DexEvent, }; /// Bonk Program ID @@ -113,16 +113,16 @@ pub const CONFIGS: &[GenericEventParseConfig] = &[ fn parse_pool_create_inner_instruction( data: &[u8], metadata: EventMetadata, -) -> Option { +) -> Option { if let Some(event) = bonk_pool_create_event_log_decode(data) { - Some(UnifiedEvent::BonkPoolCreateEvent(BonkPoolCreateEvent { metadata, ..event })) + Some(DexEvent::BonkPoolCreateEvent(BonkPoolCreateEvent { metadata, ..event })) } else { None } } /// Parse trade event -fn parse_trade_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option { +fn parse_trade_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option { if let Some(event) = bonk_trade_event_log_decode(data) { if metadata.event_type == EventType::BonkBuyExactIn || metadata.event_type == EventType::BonkBuyExactOut @@ -136,7 +136,7 @@ fn parse_trade_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option { return None; } - Some(UnifiedEvent::BonkTradeEvent(BonkTradeEvent { metadata, ..event })) + Some(DexEvent::BonkTradeEvent(BonkTradeEvent { metadata, ..event })) } else { None } @@ -147,7 +147,7 @@ fn parse_buy_exact_in_instruction( data: &[u8], accounts: &[Pubkey], metadata: EventMetadata, -) -> Option { +) -> Option { if data.len() < 16 || accounts.len() < 18 { return None; } @@ -156,7 +156,7 @@ fn parse_buy_exact_in_instruction( let minimum_amount_out = read_u64_le(data, 8)?; let share_fee_rate = read_u64_le(data, 16)?; - Some(UnifiedEvent::BonkTradeEvent(BonkTradeEvent { + Some(DexEvent::BonkTradeEvent(BonkTradeEvent { metadata, amount_in, minimum_amount_out, @@ -185,7 +185,7 @@ fn parse_buy_exact_out_instruction( data: &[u8], accounts: &[Pubkey], metadata: EventMetadata, -) -> Option { +) -> Option { if data.len() < 16 || accounts.len() < 18 { return None; } @@ -194,7 +194,7 @@ fn parse_buy_exact_out_instruction( let maximum_amount_in = read_u64_le(data, 8)?; let share_fee_rate = read_u64_le(data, 16)?; - Some(UnifiedEvent::BonkTradeEvent(BonkTradeEvent { + Some(DexEvent::BonkTradeEvent(BonkTradeEvent { metadata, amount_out, maximum_amount_in, @@ -223,7 +223,7 @@ fn parse_sell_exact_in_instruction( data: &[u8], accounts: &[Pubkey], metadata: EventMetadata, -) -> Option { +) -> Option { if data.len() < 16 || accounts.len() < 18 { return None; } @@ -232,7 +232,7 @@ fn parse_sell_exact_in_instruction( let minimum_amount_out = read_u64_le(data, 8)?; let share_fee_rate = read_u64_le(data, 16)?; - Some(UnifiedEvent::BonkTradeEvent(BonkTradeEvent { + Some(DexEvent::BonkTradeEvent(BonkTradeEvent { metadata, amount_in, minimum_amount_out, @@ -261,7 +261,7 @@ fn parse_sell_exact_out_instruction( data: &[u8], accounts: &[Pubkey], metadata: EventMetadata, -) -> Option { +) -> Option { if data.len() < 16 || accounts.len() < 18 { return None; } @@ -270,7 +270,7 @@ fn parse_sell_exact_out_instruction( let maximum_amount_in = read_u64_le(data, 8)?; let share_fee_rate = read_u64_le(data, 16)?; - Some(UnifiedEvent::BonkTradeEvent(BonkTradeEvent { + Some(DexEvent::BonkTradeEvent(BonkTradeEvent { metadata, amount_out, maximum_amount_in, @@ -300,7 +300,7 @@ fn parse_initialize_instruction( data: &[u8], accounts: &[Pubkey], metadata: EventMetadata, -) -> Option { +) -> Option { if data.len() < 24 { return None; } @@ -310,7 +310,7 @@ fn parse_initialize_instruction( let curve_param = parse_curve_params(data, &mut offset)?; let vesting_param = parse_vesting_params(data, &mut offset)?; - Some(UnifiedEvent::BonkPoolCreateEvent(BonkPoolCreateEvent { + Some(DexEvent::BonkPoolCreateEvent(BonkPoolCreateEvent { metadata, payer: accounts[0], creator: accounts[1], @@ -333,7 +333,7 @@ fn parse_initialize_v2_instruction( data: &[u8], accounts: &[Pubkey], metadata: EventMetadata, -) -> Option { +) -> Option { if data.len() < 24 { return None; } @@ -344,7 +344,7 @@ fn parse_initialize_v2_instruction( let vesting_param = parse_vesting_params(data, &mut offset)?; let amm_fee_on = data[offset]; - Some(UnifiedEvent::BonkPoolCreateEvent(BonkPoolCreateEvent { + Some(DexEvent::BonkPoolCreateEvent(BonkPoolCreateEvent { metadata, payer: accounts[0], creator: accounts[1], @@ -372,7 +372,7 @@ fn parse_initialize_with_token_2022_instruction( data: &[u8], accounts: &[Pubkey], metadata: EventMetadata, -) -> Option { +) -> Option { if data.len() < 24 { return None; } @@ -383,7 +383,7 @@ fn parse_initialize_with_token_2022_instruction( let vesting_param = parse_vesting_params(data, &mut offset)?; let amm_fee_on = data[offset]; - Some(UnifiedEvent::BonkPoolCreateEvent(BonkPoolCreateEvent { + Some(DexEvent::BonkPoolCreateEvent(BonkPoolCreateEvent { metadata, payer: accounts[0], creator: accounts[1], @@ -516,7 +516,7 @@ fn parse_migrate_to_amm_instruction( data: &[u8], accounts: &[Pubkey], metadata: EventMetadata, -) -> Option { +) -> Option { if data.len() < 16 { return None; } @@ -525,7 +525,7 @@ fn parse_migrate_to_amm_instruction( let quote_lot_size = u64::from_le_bytes(data[8..16].try_into().unwrap()); let market_vault_signer_nonce = data[16]; - Some(UnifiedEvent::BonkMigrateToAmmEvent(BonkMigrateToAmmEvent { + Some(DexEvent::BonkMigrateToAmmEvent(BonkMigrateToAmmEvent { metadata, base_lot_size, quote_lot_size, @@ -571,8 +571,8 @@ fn parse_migrate_to_cpswap_instruction( _data: &[u8], accounts: &[Pubkey], metadata: EventMetadata, -) -> Option { - Some(UnifiedEvent::BonkMigrateToCpswapEvent(BonkMigrateToCpswapEvent { +) -> Option { + Some(DexEvent::BonkMigrateToCpswapEvent(BonkMigrateToCpswapEvent { metadata, payer: accounts[0], base_mint: accounts[1], diff --git a/src/streaming/event_parser/protocols/bonk/types.rs b/src/streaming/event_parser/protocols/bonk/types.rs index 7374063..5386852 100755 --- a/src/streaming/event_parser/protocols/bonk/types.rs +++ b/src/streaming/event_parser/protocols/bonk/types.rs @@ -8,7 +8,7 @@ use crate::streaming::{ protocols::bonk::{ BonkGlobalConfigAccountEvent, BonkPlatformConfigAccountEvent, BonkPoolStateAccountEvent, }, - UnifiedEvent, + DexEvent, }, grpc::AccountPretty, }; @@ -132,12 +132,12 @@ pub fn pool_state_decode(data: &[u8]) -> Option { borsh::from_slice::(&data[..POOL_STATE_SIZE]).ok() } -pub fn pool_state_parser(account: &AccountPretty, metadata: EventMetadata) -> Option { +pub fn pool_state_parser(account: &AccountPretty, metadata: EventMetadata) -> Option { if account.data.len() < POOL_STATE_SIZE + 8 { return None; } if let Some(pool_state) = pool_state_decode(&account.data[8..POOL_STATE_SIZE + 8]) { - Some(UnifiedEvent::BonkPoolStateAccountEvent(BonkPoolStateAccountEvent { + Some(DexEvent::BonkPoolStateAccountEvent(BonkPoolStateAccountEvent { metadata, pubkey: account.pubkey, executable: account.executable, @@ -183,12 +183,12 @@ pub fn global_config_decode(data: &[u8]) -> Option { pub fn global_config_parser( account: &AccountPretty, metadata: EventMetadata, -) -> Option { +) -> Option { if account.data.len() < GLOBAL_CONFIG_SIZE + 8 { return None; } if let Some(global_config) = global_config_decode(&account.data[8..GLOBAL_CONFIG_SIZE + 8]) { - Some(UnifiedEvent::BonkGlobalConfigAccountEvent(BonkGlobalConfigAccountEvent { + Some(DexEvent::BonkGlobalConfigAccountEvent(BonkGlobalConfigAccountEvent { metadata, pubkey: account.pubkey, executable: account.executable, @@ -229,14 +229,14 @@ pub fn platform_config_decode(data: &[u8]) -> Option { pub fn platform_config_parser( account: &AccountPretty, metadata: EventMetadata, -) -> Option { +) -> Option { if account.data.len() < PLATFORM_CONFIG_SIZE + 8 { return None; } if let Some(platform_config) = platform_config_decode(&account.data[8..PLATFORM_CONFIG_SIZE + 8]) { - Some(UnifiedEvent::BonkPlatformConfigAccountEvent(BonkPlatformConfigAccountEvent { + Some(DexEvent::BonkPlatformConfigAccountEvent(BonkPlatformConfigAccountEvent { metadata, pubkey: account.pubkey, executable: account.executable, diff --git a/src/streaming/event_parser/protocols/pumpfun/events.rs b/src/streaming/event_parser/protocols/pumpfun/events.rs index 9125d7c..1e176ce 100755 --- a/src/streaming/event_parser/protocols/pumpfun/events.rs +++ b/src/streaming/event_parser/protocols/pumpfun/events.rs @@ -36,19 +36,6 @@ pub fn pumpfun_create_token_event_log_decode(data: &[u8]) -> Option(&data[..PUMPFUN_CREATE_TOKEN_EVENT_LOG_SIZE]).ok() } -// impl_unified_event!( -// PumpFunCreateTokenEvent, -// mint, -// bonding_curve, -// user, -// creator, -// timestamp, -// virtual_token_reserves, -// virtual_sol_reserves, -// real_token_reserves, -// token_total_supply -// ); - #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct PumpFunTradeEvent { #[borsh(skip)] @@ -125,26 +112,6 @@ pub fn pumpfun_trade_event_log_decode(data: &[u8]) -> Option borsh::from_slice::(&data[..PUMPFUN_TRADE_EVENT_LOG_SIZE]).ok() } -// impl_unified_event!( -// PumpFunTradeEvent, -// mint, -// sol_amount, -// token_amount, -// is_buy, -// user, -// timestamp, -// virtual_sol_reserves, -// virtual_token_reserves, -// real_sol_reserves, -// real_token_reserves, -// fee_recipient, -// fee_basis_points, -// fee, -// creator, -// creator_fee_basis_points, -// creator_fee -// ); - #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct PumpFunMigrateEvent { #[borsh(skip)] @@ -210,18 +177,6 @@ pub fn pumpfun_migrate_event_log_decode(data: &[u8]) -> Option(&data[..PUMPFUN_MIGRATE_EVENT_LOG_SIZE]).ok() } -// impl_unified_event!( -// PumpFunMigrateEvent, -// user, -// mint, -// mint_amount, -// sol_amount, -// pool_migration_fee, -// bonding_curve, -// timestamp, -// pool -// ); - /// 铸币曲线 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct PumpFunBondingCurveAccountEvent { diff --git a/src/streaming/event_parser/protocols/pumpfun/parser.rs b/src/streaming/event_parser/protocols/pumpfun/parser.rs index b47992f..d92decd 100755 --- a/src/streaming/event_parser/protocols/pumpfun/parser.rs +++ b/src/streaming/event_parser/protocols/pumpfun/parser.rs @@ -6,7 +6,7 @@ use crate::streaming::event_parser::{ pumpfun_trade_event_log_decode, PumpFunCreateTokenEvent, PumpFunMigrateEvent, PumpFunTradeEvent, }, - UnifiedEvent, + DexEvent, }; use solana_sdk::pubkey::Pubkey; @@ -60,9 +60,9 @@ pub const CONFIGS: &[GenericEventParseConfig] = &[ ]; /// 解析迁移事件 -fn parse_migrate_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option { +fn parse_migrate_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option { if let Some(event) = pumpfun_migrate_event_log_decode(data) { - Some(UnifiedEvent::PumpFunMigrateEvent(PumpFunMigrateEvent { metadata, ..event })) + Some(DexEvent::PumpFunMigrateEvent(PumpFunMigrateEvent { metadata, ..event })) } else { None } @@ -72,18 +72,18 @@ fn parse_migrate_inner_instruction(data: &[u8], metadata: EventMetadata) -> Opti fn parse_create_token_inner_instruction( data: &[u8], metadata: EventMetadata, -) -> Option { +) -> Option { if let Some(event) = pumpfun_create_token_event_log_decode(data) { - Some(UnifiedEvent::PumpFunCreateTokenEvent(PumpFunCreateTokenEvent { metadata, ..event })) + Some(DexEvent::PumpFunCreateTokenEvent(PumpFunCreateTokenEvent { metadata, ..event })) } else { None } } /// 解析交易事件 -fn parse_trade_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option { +fn parse_trade_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option { if let Some(event) = pumpfun_trade_event_log_decode(data) { - Some(UnifiedEvent::PumpFunTradeEvent(PumpFunTradeEvent { metadata, ..event })) + Some(DexEvent::PumpFunTradeEvent(PumpFunTradeEvent { metadata, ..event })) } else { None } @@ -94,7 +94,7 @@ fn parse_create_token_instruction( data: &[u8], accounts: &[Pubkey], metadata: EventMetadata, -) -> Option { +) -> Option { if data.len() < 16 || accounts.len() < 11 { return None; } @@ -135,7 +135,7 @@ fn parse_create_token_instruction( Pubkey::default() }; - Some(UnifiedEvent::PumpFunCreateTokenEvent(PumpFunCreateTokenEvent { + Some(DexEvent::PumpFunCreateTokenEvent(PumpFunCreateTokenEvent { metadata, name: name.to_string(), symbol: symbol.to_string(), @@ -155,13 +155,13 @@ fn parse_buy_instruction( data: &[u8], accounts: &[Pubkey], metadata: EventMetadata, -) -> Option { +) -> Option { if data.len() < 16 || accounts.len() < 13 { return None; } let amount = u64::from_le_bytes(data[0..8].try_into().unwrap()); let max_sol_cost = u64::from_le_bytes(data[8..16].try_into().unwrap()); - Some(UnifiedEvent::PumpFunTradeEvent(PumpFunTradeEvent { + Some(DexEvent::PumpFunTradeEvent(PumpFunTradeEvent { metadata, global: accounts[0], fee_recipient: accounts[1], @@ -189,13 +189,13 @@ fn parse_sell_instruction( data: &[u8], accounts: &[Pubkey], metadata: EventMetadata, -) -> Option { +) -> Option { if data.len() < 16 || accounts.len() < 11 { return None; } let amount = u64::from_le_bytes(data[0..8].try_into().unwrap()); let min_sol_output = u64::from_le_bytes(data[8..16].try_into().unwrap()); - Some(UnifiedEvent::PumpFunTradeEvent(PumpFunTradeEvent { + Some(DexEvent::PumpFunTradeEvent(PumpFunTradeEvent { metadata, global: accounts[0], fee_recipient: accounts[1], @@ -223,11 +223,11 @@ fn parse_migrate_instruction( _data: &[u8], accounts: &[Pubkey], metadata: EventMetadata, -) -> Option { +) -> Option { if accounts.len() < 24 { return None; } - Some(UnifiedEvent::PumpFunMigrateEvent(PumpFunMigrateEvent { + Some(DexEvent::PumpFunMigrateEvent(PumpFunMigrateEvent { metadata, global: accounts[0], withdraw_authority: accounts[1], diff --git a/src/streaming/event_parser/protocols/pumpfun/types.rs b/src/streaming/event_parser/protocols/pumpfun/types.rs index fbcefaf..15a0e23 100644 --- a/src/streaming/event_parser/protocols/pumpfun/types.rs +++ b/src/streaming/event_parser/protocols/pumpfun/types.rs @@ -6,7 +6,7 @@ use crate::streaming::{ event_parser::{ common::EventMetadata, protocols::pumpfun::{PumpFunBondingCurveAccountEvent, PumpFunGlobalAccountEvent}, - UnifiedEvent, + DexEvent, }, grpc::AccountPretty, }; @@ -34,12 +34,12 @@ pub fn bonding_curve_decode(data: &[u8]) -> Option { pub fn bonding_curve_parser( account: &AccountPretty, metadata: EventMetadata, -) -> Option { +) -> Option { if account.data.len() < BONDING_CURVE_SIZE + 8 { return None; } if let Some(bonding_curve) = bonding_curve_decode(&account.data[8..BONDING_CURVE_SIZE + 8]) { - Some(UnifiedEvent::PumpFunBondingCurveAccountEvent(PumpFunBondingCurveAccountEvent { + Some(DexEvent::PumpFunBondingCurveAccountEvent(PumpFunBondingCurveAccountEvent { metadata, pubkey: account.pubkey, executable: account.executable, @@ -81,12 +81,12 @@ pub fn global_decode(data: &[u8]) -> Option { borsh::from_slice::(&data[..GLOBAL_SIZE]).ok() } -pub fn global_parser(account: &AccountPretty, metadata: EventMetadata) -> Option { +pub fn global_parser(account: &AccountPretty, metadata: EventMetadata) -> Option { if account.data.len() < GLOBAL_SIZE + 8 { return None; } if let Some(global) = global_decode(&account.data[8..GLOBAL_SIZE + 8]) { - Some(UnifiedEvent::PumpFunGlobalAccountEvent(PumpFunGlobalAccountEvent { + Some(DexEvent::PumpFunGlobalAccountEvent(PumpFunGlobalAccountEvent { metadata, pubkey: account.pubkey, executable: account.executable, diff --git a/src/streaming/event_parser/protocols/pumpswap/events.rs b/src/streaming/event_parser/protocols/pumpswap/events.rs index 27c4738..67b1188 100755 --- a/src/streaming/event_parser/protocols/pumpswap/events.rs +++ b/src/streaming/event_parser/protocols/pumpswap/events.rs @@ -65,34 +65,6 @@ pub fn pump_swap_buy_event_log_decode(data: &[u8]) -> Option { borsh::from_slice::(&data[..PUMP_SWAP_BUY_EVENT_LOG_SIZE]).ok() } -// 使用宏生成UnifiedEvent实现,指定需要合并的字段 -// impl_unified_event!( -// PumpSwapBuyEvent, -// timestamp, -// base_amount_out, -// max_quote_amount_in, -// user_base_token_reserves, -// user_quote_token_reserves, -// pool_base_token_reserves, -// pool_quote_token_reserves, -// quote_amount_in, -// lp_fee_basis_points, -// lp_fee, -// protocol_fee_basis_points, -// protocol_fee, -// quote_amount_in_with_lp_fee, -// user_quote_amount_in, -// pool, -// user, -// user_base_token_account, -// user_quote_token_account, -// protocol_fee_recipient, -// protocol_fee_recipient_token_account, -// coin_creator, -// coin_creator_fee_basis_points, -// coin_creator_fee -// ); - /// 卖出事件 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct PumpSwapSellEvent { @@ -148,34 +120,6 @@ pub fn pump_swap_sell_event_log_decode(data: &[u8]) -> Option borsh::from_slice::(&data[..PUMP_SWAP_SELL_EVENT_LOG_SIZE]).ok() } -// 使用宏生成UnifiedEvent实现,指定需要合并的字段 -// impl_unified_event!( -// PumpSwapSellEvent, -// timestamp, -// base_amount_in, -// min_quote_amount_out, -// user_base_token_reserves, -// user_quote_token_reserves, -// pool_base_token_reserves, -// pool_quote_token_reserves, -// quote_amount_out, -// lp_fee_basis_points, -// lp_fee, -// protocol_fee_basis_points, -// protocol_fee, -// quote_amount_out_without_lp_fee, -// user_quote_amount_out, -// pool, -// user, -// user_base_token_account, -// user_quote_token_account, -// protocol_fee_recipient, -// protocol_fee_recipient_token_account, -// coin_creator, -// coin_creator_fee_basis_points, -// coin_creator_fee -// ); - /// 创建池子事件 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct PumpSwapCreatePoolEvent { @@ -218,30 +162,6 @@ pub fn pump_swap_create_pool_event_log_decode(data: &[u8]) -> Option(&data[..PUMP_SWAP_CREATE_POOL_EVENT_LOG_SIZE]).ok() } -// impl_unified_event!( -// PumpSwapCreatePoolEvent, -// timestamp, -// index, -// creator, -// base_mint, -// quote_mint, -// base_mint_decimals, -// quote_mint_decimals, -// base_amount_in, -// quote_amount_in, -// pool_base_amount, -// pool_quote_amount, -// minimum_liquidity, -// initial_liquidity, -// lp_token_amount_out, -// pool_bump, -// pool, -// lp_mint, -// user_base_token_account, -// user_quote_token_account, -// coin_creator -// ); - /// 存款事件 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct PumpSwapDepositEvent { @@ -282,26 +202,6 @@ pub fn pump_swap_deposit_event_log_decode(data: &[u8]) -> Option(&data[..PUMP_SWAP_DEPOSIT_EVENT_LOG_SIZE]).ok() } -// impl_unified_event!( -// PumpSwapDepositEvent, -// timestamp, -// lp_token_amount_out, -// max_base_amount_in, -// max_quote_amount_in, -// user_base_token_reserves, -// user_quote_token_reserves, -// pool_base_token_reserves, -// pool_quote_token_reserves, -// base_amount_in, -// quote_amount_in, -// lp_mint_supply, -// pool, -// user, -// user_base_token_account, -// user_quote_token_account, -// user_pool_token_account -// ); - /// 提款事件 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct PumpSwapWithdrawEvent { @@ -342,26 +242,6 @@ pub fn pump_swap_withdraw_event_log_decode(data: &[u8]) -> Option(&data[..PUMP_SWAP_WITHDRAW_EVENT_LOG_SIZE]).ok() } -// impl_unified_event!( -// PumpSwapWithdrawEvent, -// timestamp, -// lp_token_amount_in, -// min_base_amount_out, -// min_quote_amount_out, -// user_base_token_reserves, -// user_quote_token_reserves, -// pool_base_token_reserves, -// pool_quote_token_reserves, -// base_amount_out, -// quote_amount_out, -// lp_mint_supply, -// pool, -// user, -// user_base_token_account, -// user_quote_token_account, -// user_pool_token_account -// ); - /// 全局配置 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct PumpSwapGlobalConfigAccountEvent { diff --git a/src/streaming/event_parser/protocols/pumpswap/parser.rs b/src/streaming/event_parser/protocols/pumpswap/parser.rs index 5df431d..7b9d2e8 100755 --- a/src/streaming/event_parser/protocols/pumpswap/parser.rs +++ b/src/streaming/event_parser/protocols/pumpswap/parser.rs @@ -7,7 +7,7 @@ use crate::streaming::event_parser::{ pump_swap_withdraw_event_log_decode, PumpSwapBuyEvent, PumpSwapCreatePoolEvent, PumpSwapDepositEvent, PumpSwapSellEvent, PumpSwapWithdrawEvent, }, - UnifiedEvent, + DexEvent, }; use solana_sdk::pubkey::Pubkey; @@ -70,18 +70,18 @@ pub const CONFIGS: &[GenericEventParseConfig] = &[ ]; /// 解析买入日志事件 -fn parse_buy_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option { +fn parse_buy_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option { if let Some(event) = pump_swap_buy_event_log_decode(data) { - Some(UnifiedEvent::PumpSwapBuyEvent(PumpSwapBuyEvent { metadata, ..event })) + Some(DexEvent::PumpSwapBuyEvent(PumpSwapBuyEvent { metadata, ..event })) } else { None } } /// 解析卖出日志事件 -fn parse_sell_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option { +fn parse_sell_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option { if let Some(event) = pump_swap_sell_event_log_decode(data) { - Some(UnifiedEvent::PumpSwapSellEvent(PumpSwapSellEvent { metadata, ..event })) + Some(DexEvent::PumpSwapSellEvent(PumpSwapSellEvent { metadata, ..event })) } else { None } @@ -91,27 +91,27 @@ fn parse_sell_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option< fn parse_create_pool_inner_instruction( data: &[u8], metadata: EventMetadata, -) -> Option { +) -> Option { if let Some(event) = pump_swap_create_pool_event_log_decode(data) { - Some(UnifiedEvent::PumpSwapCreatePoolEvent(PumpSwapCreatePoolEvent { metadata, ..event })) + Some(DexEvent::PumpSwapCreatePoolEvent(PumpSwapCreatePoolEvent { metadata, ..event })) } else { None } } /// 解析存款日志事件 -fn parse_deposit_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option { +fn parse_deposit_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option { if let Some(event) = pump_swap_deposit_event_log_decode(data) { - Some(UnifiedEvent::PumpSwapDepositEvent(PumpSwapDepositEvent { metadata, ..event })) + Some(DexEvent::PumpSwapDepositEvent(PumpSwapDepositEvent { metadata, ..event })) } else { None } } /// 解析提款日志事件 -fn parse_withdraw_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option { +fn parse_withdraw_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option { if let Some(event) = pump_swap_withdraw_event_log_decode(data) { - Some(UnifiedEvent::PumpSwapWithdrawEvent(PumpSwapWithdrawEvent { metadata, ..event })) + Some(DexEvent::PumpSwapWithdrawEvent(PumpSwapWithdrawEvent { metadata, ..event })) } else { None } @@ -122,7 +122,7 @@ fn parse_buy_instruction( data: &[u8], accounts: &[Pubkey], metadata: EventMetadata, -) -> Option { +) -> Option { if data.len() < 16 || accounts.len() < 11 { return None; } @@ -130,7 +130,7 @@ fn parse_buy_instruction( let base_amount_out = read_u64_le(data, 0)?; let max_quote_amount_in = read_u64_le(data, 8)?; - Some(UnifiedEvent::PumpSwapBuyEvent(PumpSwapBuyEvent { + Some(DexEvent::PumpSwapBuyEvent(PumpSwapBuyEvent { metadata, base_amount_out, max_quote_amount_in, @@ -157,7 +157,7 @@ fn parse_sell_instruction( data: &[u8], accounts: &[Pubkey], metadata: EventMetadata, -) -> Option { +) -> Option { if data.len() < 16 || accounts.len() < 11 { return None; } @@ -165,7 +165,7 @@ fn parse_sell_instruction( let base_amount_in = read_u64_le(data, 0)?; let min_quote_amount_out = read_u64_le(data, 8)?; - Some(UnifiedEvent::PumpSwapSellEvent(PumpSwapSellEvent { + Some(DexEvent::PumpSwapSellEvent(PumpSwapSellEvent { metadata, base_amount_in, min_quote_amount_out, @@ -192,7 +192,7 @@ fn parse_create_pool_instruction( data: &[u8], accounts: &[Pubkey], metadata: EventMetadata, -) -> Option { +) -> Option { if data.len() < 18 || accounts.len() < 11 { return None; } @@ -206,7 +206,7 @@ fn parse_create_pool_instruction( Pubkey::default() }; - Some(UnifiedEvent::PumpSwapCreatePoolEvent(PumpSwapCreatePoolEvent { + Some(DexEvent::PumpSwapCreatePoolEvent(PumpSwapCreatePoolEvent { metadata, index, base_amount_in, @@ -231,7 +231,7 @@ fn parse_deposit_instruction( data: &[u8], accounts: &[Pubkey], metadata: EventMetadata, -) -> Option { +) -> Option { if data.len() < 24 || accounts.len() < 11 { return None; } @@ -240,7 +240,7 @@ fn parse_deposit_instruction( let max_base_amount_in = u64::from_le_bytes(data[8..16].try_into().ok()?); let max_quote_amount_in = u64::from_le_bytes(data[16..24].try_into().ok()?); - Some(UnifiedEvent::PumpSwapDepositEvent(PumpSwapDepositEvent { + Some(DexEvent::PumpSwapDepositEvent(PumpSwapDepositEvent { metadata, lp_token_amount_out, max_base_amount_in, @@ -263,7 +263,7 @@ fn parse_withdraw_instruction( data: &[u8], accounts: &[Pubkey], metadata: EventMetadata, -) -> Option { +) -> Option { if data.len() < 24 || accounts.len() < 11 { return None; } @@ -272,7 +272,7 @@ fn parse_withdraw_instruction( let min_base_amount_out = u64::from_le_bytes(data[8..16].try_into().ok()?); let min_quote_amount_out = u64::from_le_bytes(data[16..24].try_into().ok()?); - Some(UnifiedEvent::PumpSwapWithdrawEvent(PumpSwapWithdrawEvent { + Some(DexEvent::PumpSwapWithdrawEvent(PumpSwapWithdrawEvent { metadata, lp_token_amount_in, min_base_amount_out, diff --git a/src/streaming/event_parser/protocols/pumpswap/types.rs b/src/streaming/event_parser/protocols/pumpswap/types.rs index c428ec7..e66d007 100644 --- a/src/streaming/event_parser/protocols/pumpswap/types.rs +++ b/src/streaming/event_parser/protocols/pumpswap/types.rs @@ -6,7 +6,7 @@ use crate::streaming::{ event_parser::{ common::EventMetadata, protocols::pumpswap::{PumpSwapGlobalConfigAccountEvent, PumpSwapPoolAccountEvent}, - UnifiedEvent, + DexEvent, }, grpc::AccountPretty, }; @@ -34,12 +34,12 @@ pub fn global_config_decode(data: &[u8]) -> Option { pub fn global_config_parser( account: &AccountPretty, metadata: EventMetadata, -) -> Option { +) -> Option { if account.data.len() < GLOBAL_CONFIG_SIZE + 8 { return None; } if let Some(config) = global_config_decode(&account.data[8..GLOBAL_CONFIG_SIZE + 8]) { - Some(UnifiedEvent::PumpSwapGlobalConfigAccountEvent(PumpSwapGlobalConfigAccountEvent { + Some(DexEvent::PumpSwapGlobalConfigAccountEvent(PumpSwapGlobalConfigAccountEvent { metadata, pubkey: account.pubkey, executable: account.executable, @@ -76,12 +76,12 @@ pub fn pool_decode(data: &[u8]) -> Option { borsh::from_slice::(&data[..POOL_SIZE]).ok() } -pub fn pool_parser(account: &AccountPretty, metadata: EventMetadata) -> Option { +pub fn pool_parser(account: &AccountPretty, metadata: EventMetadata) -> Option { if account.data.len() < POOL_SIZE + 8 { return None; } if let Some(pool) = pool_decode(&account.data[8..POOL_SIZE + 8]) { - Some(UnifiedEvent::PumpSwapPoolAccountEvent(PumpSwapPoolAccountEvent { + Some(DexEvent::PumpSwapPoolAccountEvent(PumpSwapPoolAccountEvent { metadata, pubkey: account.pubkey, executable: account.executable, diff --git a/src/streaming/event_parser/protocols/raydium_amm_v4/parser.rs b/src/streaming/event_parser/protocols/raydium_amm_v4/parser.rs index 94f1957..d19fb2c 100755 --- a/src/streaming/event_parser/protocols/raydium_amm_v4/parser.rs +++ b/src/streaming/event_parser/protocols/raydium_amm_v4/parser.rs @@ -5,7 +5,7 @@ use crate::streaming::event_parser::{ discriminators, RaydiumAmmV4DepositEvent, RaydiumAmmV4Initialize2Event, RaydiumAmmV4SwapEvent, RaydiumAmmV4WithdrawEvent, RaydiumAmmV4WithdrawPnlEvent, }, - UnifiedEvent, + DexEvent, }; use solana_sdk::pubkey::Pubkey; @@ -82,12 +82,12 @@ fn parse_withdraw_pnl_instruction( _data: &[u8], accounts: &[Pubkey], metadata: EventMetadata, -) -> Option { +) -> Option { if accounts.len() < 17 { return None; } - Some(UnifiedEvent::RaydiumAmmV4WithdrawPnlEvent(RaydiumAmmV4WithdrawPnlEvent { + Some(DexEvent::RaydiumAmmV4WithdrawPnlEvent(RaydiumAmmV4WithdrawPnlEvent { metadata, token_program: accounts[0], amm: accounts[1], @@ -114,13 +114,13 @@ fn parse_withdraw_instruction( data: &[u8], accounts: &[Pubkey], metadata: EventMetadata, -) -> Option { +) -> Option { if data.len() < 8 || accounts.len() < 22 { return None; } let amount = read_u64_le(data, 0)?; - Some(UnifiedEvent::RaydiumAmmV4WithdrawEvent(RaydiumAmmV4WithdrawEvent { + Some(DexEvent::RaydiumAmmV4WithdrawEvent(RaydiumAmmV4WithdrawEvent { metadata, amount, @@ -154,7 +154,7 @@ fn parse_initialize2_instruction( data: &[u8], accounts: &[Pubkey], metadata: EventMetadata, -) -> Option { +) -> Option { if data.len() < 25 || accounts.len() < 21 { return None; } @@ -163,7 +163,7 @@ fn parse_initialize2_instruction( let init_pc_amount = read_u64_le(data, 9)?; let init_coin_amount = read_u64_le(data, 17)?; - Some(UnifiedEvent::RaydiumAmmV4Initialize2Event(RaydiumAmmV4Initialize2Event { + Some(DexEvent::RaydiumAmmV4Initialize2Event(RaydiumAmmV4Initialize2Event { metadata, nonce, open_time, @@ -199,7 +199,7 @@ fn parse_deposit_instruction( data: &[u8], accounts: &[Pubkey], metadata: EventMetadata, -) -> Option { +) -> Option { if data.len() < 24 || accounts.len() < 14 { return None; } @@ -207,7 +207,7 @@ fn parse_deposit_instruction( let max_pc_amount = read_u64_le(data, 8)?; let base_side = read_u64_le(data, 16)?; - Some(UnifiedEvent::RaydiumAmmV4DepositEvent(RaydiumAmmV4DepositEvent { + Some(DexEvent::RaydiumAmmV4DepositEvent(RaydiumAmmV4DepositEvent { metadata, max_coin_amount, max_pc_amount, @@ -235,7 +235,7 @@ fn parse_swap_base_output_instruction( data: &[u8], accounts: &[Pubkey], metadata: EventMetadata, -) -> Option { +) -> Option { if data.len() < 16 || accounts.len() < 17 { return None; } @@ -249,7 +249,7 @@ fn parse_swap_base_output_instruction( accounts.insert(4, Pubkey::default()); } - Some(UnifiedEvent::RaydiumAmmV4SwapEvent(RaydiumAmmV4SwapEvent { + Some(DexEvent::RaydiumAmmV4SwapEvent(RaydiumAmmV4SwapEvent { metadata, max_amount_in, amount_out, @@ -282,7 +282,7 @@ fn parse_swap_base_input_instruction( data: &[u8], accounts: &[Pubkey], metadata: EventMetadata, -) -> Option { +) -> Option { if data.len() < 16 || accounts.len() < 17 { return None; } @@ -296,7 +296,7 @@ fn parse_swap_base_input_instruction( accounts.insert(4, Pubkey::default()); } - Some(UnifiedEvent::RaydiumAmmV4SwapEvent(RaydiumAmmV4SwapEvent { + Some(DexEvent::RaydiumAmmV4SwapEvent(RaydiumAmmV4SwapEvent { metadata, amount_in, minimum_amount_out, diff --git a/src/streaming/event_parser/protocols/raydium_amm_v4/types.rs b/src/streaming/event_parser/protocols/raydium_amm_v4/types.rs index 7f34804..ba8d7ce 100644 --- a/src/streaming/event_parser/protocols/raydium_amm_v4/types.rs +++ b/src/streaming/event_parser/protocols/raydium_amm_v4/types.rs @@ -5,7 +5,7 @@ use solana_sdk::pubkey::Pubkey; use crate::streaming::{ event_parser::{ common::EventMetadata, protocols::raydium_amm_v4::RaydiumAmmV4AmmInfoAccountEvent, - UnifiedEvent, + DexEvent, }, grpc::AccountPretty, }; @@ -86,12 +86,12 @@ pub fn amm_info_decode(data: &[u8]) -> Option { borsh::from_slice::(&data[..AMM_INFO_SIZE]).ok() } -pub fn amm_info_parser(account: &AccountPretty, metadata: EventMetadata) -> Option { +pub fn amm_info_parser(account: &AccountPretty, metadata: EventMetadata) -> Option { if account.data.len() < AMM_INFO_SIZE { return None; } if let Some(amm_info) = amm_info_decode(&account.data[..AMM_INFO_SIZE]) { - Some(UnifiedEvent::RaydiumAmmV4AmmInfoAccountEvent(RaydiumAmmV4AmmInfoAccountEvent { + Some(DexEvent::RaydiumAmmV4AmmInfoAccountEvent(RaydiumAmmV4AmmInfoAccountEvent { metadata, pubkey: account.pubkey, executable: account.executable, diff --git a/src/streaming/event_parser/protocols/raydium_clmm/parser.rs b/src/streaming/event_parser/protocols/raydium_clmm/parser.rs index 77e37cd..8d7e9cf 100755 --- a/src/streaming/event_parser/protocols/raydium_clmm/parser.rs +++ b/src/streaming/event_parser/protocols/raydium_clmm/parser.rs @@ -10,7 +10,7 @@ use crate::streaming::event_parser::{ RaydiumClmmOpenPositionV2Event, RaydiumClmmOpenPositionWithToken22NftEvent, RaydiumClmmSwapEvent, RaydiumClmmSwapV2Event, }, - UnifiedEvent, + DexEvent, }; use solana_sdk::pubkey::Pubkey; @@ -107,11 +107,11 @@ fn parse_open_position_v2_instruction( data: &[u8], accounts: &[Pubkey], metadata: EventMetadata, -) -> Option { +) -> Option { if data.len() < 51 || accounts.len() < 22 { return None; } - Some(UnifiedEvent::RaydiumClmmOpenPositionV2Event(RaydiumClmmOpenPositionV2Event { + Some(DexEvent::RaydiumClmmOpenPositionV2Event(RaydiumClmmOpenPositionV2Event { metadata, tick_lower_index: read_i32_le(data, 0)?, tick_upper_index: read_i32_le(data, 4)?, @@ -153,11 +153,11 @@ fn parse_open_position_with_token_22_nft_instruction( data: &[u8], accounts: &[Pubkey], metadata: EventMetadata, -) -> Option { +) -> Option { if data.len() < 51 || accounts.len() < 20 { return None; } - Some(UnifiedEvent::RaydiumClmmOpenPositionWithToken22NftEvent( + Some(DexEvent::RaydiumClmmOpenPositionWithToken22NftEvent( RaydiumClmmOpenPositionWithToken22NftEvent { metadata, tick_lower_index: read_i32_le(data, 0)?, @@ -198,11 +198,11 @@ fn parse_increase_liquidity_v2_instruction( data: &[u8], accounts: &[Pubkey], metadata: EventMetadata, -) -> Option { +) -> Option { if data.len() < 34 || accounts.len() < 15 { return None; } - Some(UnifiedEvent::RaydiumClmmIncreaseLiquidityV2Event(RaydiumClmmIncreaseLiquidityV2Event { + Some(DexEvent::RaydiumClmmIncreaseLiquidityV2Event(RaydiumClmmIncreaseLiquidityV2Event { metadata, liquidity: read_u128_le(data, 0)?, amount0_max: read_u64_le(data, 16)?, @@ -231,11 +231,11 @@ fn parse_create_pool_instruction( data: &[u8], accounts: &[Pubkey], metadata: EventMetadata, -) -> Option { +) -> Option { if data.len() < 24 || accounts.len() < 13 { return None; } - Some(UnifiedEvent::RaydiumClmmCreatePoolEvent(RaydiumClmmCreatePoolEvent { + Some(DexEvent::RaydiumClmmCreatePoolEvent(RaydiumClmmCreatePoolEvent { metadata, sqrt_price_x64: read_u128_le(data, 0)?, open_time: read_u64_le(data, 16)?, @@ -260,11 +260,11 @@ fn parse_decrease_liquidity_v2_instruction( data: &[u8], accounts: &[Pubkey], metadata: EventMetadata, -) -> Option { +) -> Option { if data.len() < 32 || accounts.len() < 16 { return None; } - Some(UnifiedEvent::RaydiumClmmDecreaseLiquidityV2Event(RaydiumClmmDecreaseLiquidityV2Event { + Some(DexEvent::RaydiumClmmDecreaseLiquidityV2Event(RaydiumClmmDecreaseLiquidityV2Event { metadata, liquidity: read_u128_le(data, 0)?, amount0_min: read_u64_le(data, 16)?, @@ -294,11 +294,11 @@ fn parse_close_position_instruction( _data: &[u8], accounts: &[Pubkey], metadata: EventMetadata, -) -> Option { +) -> Option { if accounts.len() < 6 { return None; } - Some(UnifiedEvent::RaydiumClmmClosePositionEvent(RaydiumClmmClosePositionEvent { + Some(DexEvent::RaydiumClmmClosePositionEvent(RaydiumClmmClosePositionEvent { metadata, nft_owner: accounts[0], position_nft_mint: accounts[1], @@ -314,7 +314,7 @@ fn parse_swap_instruction( data: &[u8], accounts: &[Pubkey], metadata: EventMetadata, -) -> Option { +) -> Option { if data.len() < 33 || accounts.len() < 10 { return None; } @@ -324,7 +324,7 @@ fn parse_swap_instruction( let sqrt_price_limit_x64 = read_u128_le(data, 16)?; let is_base_input = read_u8_le(data, 32)?; - Some(UnifiedEvent::RaydiumClmmSwapEvent(RaydiumClmmSwapEvent { + Some(DexEvent::RaydiumClmmSwapEvent(RaydiumClmmSwapEvent { metadata, amount, other_amount_threshold, @@ -348,7 +348,7 @@ fn parse_swap_v2_instruction( data: &[u8], accounts: &[Pubkey], metadata: EventMetadata, -) -> Option { +) -> Option { if data.len() < 33 || accounts.len() < 13 { return None; } @@ -358,7 +358,7 @@ fn parse_swap_v2_instruction( let sqrt_price_limit_x64 = read_u128_le(data, 16)?; let is_base_input = read_u8_le(data, 32)?; - Some(UnifiedEvent::RaydiumClmmSwapV2Event(RaydiumClmmSwapV2Event { + Some(DexEvent::RaydiumClmmSwapV2Event(RaydiumClmmSwapV2Event { metadata, amount, other_amount_threshold, diff --git a/src/streaming/event_parser/protocols/raydium_clmm/types.rs b/src/streaming/event_parser/protocols/raydium_clmm/types.rs index 523f1d0..a3731b5 100644 --- a/src/streaming/event_parser/protocols/raydium_clmm/types.rs +++ b/src/streaming/event_parser/protocols/raydium_clmm/types.rs @@ -9,7 +9,7 @@ use crate::streaming::{ RaydiumClmmAmmConfigAccountEvent, RaydiumClmmPoolStateAccountEvent, RaydiumClmmTickArrayStateAccountEvent, }, - UnifiedEvent, + DexEvent, }, grpc::AccountPretty, }; @@ -37,12 +37,12 @@ pub fn amm_config_decode(data: &[u8]) -> Option { borsh::from_slice::(&data[..AMM_CONFIG_SIZE]).ok() } -pub fn amm_config_parser(account: &AccountPretty, metadata: EventMetadata) -> Option { +pub fn amm_config_parser(account: &AccountPretty, metadata: EventMetadata) -> Option { if account.data.len() < AMM_CONFIG_SIZE + 8 { return None; } if let Some(amm_config) = amm_config_decode(&account.data[8..AMM_CONFIG_SIZE + 8]) { - Some(UnifiedEvent::RaydiumClmmAmmConfigAccountEvent(RaydiumClmmAmmConfigAccountEvent { + Some(DexEvent::RaydiumClmmAmmConfigAccountEvent(RaydiumClmmAmmConfigAccountEvent { metadata, pubkey: account.pubkey, executable: account.executable, @@ -122,12 +122,12 @@ pub fn pool_state_decode(data: &[u8]) -> Option { borsh::from_slice::(&data[..POOL_STATE_SIZE]).ok() } -pub fn pool_state_parser(account: &AccountPretty, metadata: EventMetadata) -> Option { +pub fn pool_state_parser(account: &AccountPretty, metadata: EventMetadata) -> Option { if account.data.len() < POOL_STATE_SIZE + 8 { return None; } if let Some(pool_state) = pool_state_decode(&account.data[8..POOL_STATE_SIZE + 8]) { - Some(UnifiedEvent::RaydiumClmmPoolStateAccountEvent(RaydiumClmmPoolStateAccountEvent { + Some(DexEvent::RaydiumClmmPoolStateAccountEvent(RaydiumClmmPoolStateAccountEvent { metadata, pubkey: account.pubkey, executable: account.executable, @@ -203,14 +203,14 @@ pub fn tick_array_state_decode(data: &[u8]) -> Option { pub fn tick_array_state_parser( account: &AccountPretty, metadata: EventMetadata, -) -> Option { +) -> Option { if account.data.len() < TICK_ARRAY_STATE_SIZE + 8 { return None; } if let Some(tick_array_state) = tick_array_state_decode(&account.data[8..TICK_ARRAY_STATE_SIZE + 8]) { - Some(UnifiedEvent::RaydiumClmmTickArrayStateAccountEvent( + Some(DexEvent::RaydiumClmmTickArrayStateAccountEvent( RaydiumClmmTickArrayStateAccountEvent { metadata, pubkey: account.pubkey, diff --git a/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs b/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs index 5571a33..2e59eb1 100755 --- a/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs +++ b/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs @@ -7,7 +7,7 @@ use crate::streaming::event_parser::{ discriminators, RaydiumCpmmDepositEvent, RaydiumCpmmInitializeEvent, RaydiumCpmmSwapEvent, RaydiumCpmmWithdrawEvent, }, - UnifiedEvent, + DexEvent, }; /// Raydium CPMM程序ID @@ -73,11 +73,11 @@ fn parse_withdraw_instruction( data: &[u8], accounts: &[Pubkey], metadata: EventMetadata, -) -> Option { +) -> Option { if data.len() < 24 || accounts.len() < 14 { return None; } - Some(UnifiedEvent::RaydiumCpmmWithdrawEvent(RaydiumCpmmWithdrawEvent { + Some(DexEvent::RaydiumCpmmWithdrawEvent(RaydiumCpmmWithdrawEvent { metadata, lp_token_amount: read_u64_le(data, 0)?, minimum_token0_amount: read_u64_le(data, 8)?, @@ -104,11 +104,11 @@ fn parse_initialize_instruction( data: &[u8], accounts: &[Pubkey], metadata: EventMetadata, -) -> Option { +) -> Option { if data.len() < 24 || accounts.len() < 20 { return None; } - Some(UnifiedEvent::RaydiumCpmmInitializeEvent(RaydiumCpmmInitializeEvent { + Some(DexEvent::RaydiumCpmmInitializeEvent(RaydiumCpmmInitializeEvent { metadata, init_amount0: read_u64_le(data, 0)?, init_amount1: read_u64_le(data, 8)?, @@ -141,11 +141,11 @@ fn parse_deposit_instruction( data: &[u8], accounts: &[Pubkey], metadata: EventMetadata, -) -> Option { +) -> Option { if data.len() < 24 || accounts.len() < 13 { return None; } - Some(UnifiedEvent::RaydiumCpmmDepositEvent(RaydiumCpmmDepositEvent { + Some(DexEvent::RaydiumCpmmDepositEvent(RaydiumCpmmDepositEvent { metadata, lp_token_amount: read_u64_le(data, 0)?, maximum_token0_amount: read_u64_le(data, 8)?, @@ -171,7 +171,7 @@ fn parse_swap_base_input_instruction( data: &[u8], accounts: &[Pubkey], metadata: EventMetadata, -) -> Option { +) -> Option { if data.len() < 16 || accounts.len() < 13 { return None; } @@ -179,7 +179,7 @@ fn parse_swap_base_input_instruction( let amount_in = read_u64_le(data, 0)?; let minimum_amount_out = read_u64_le(data, 8)?; - Some(UnifiedEvent::RaydiumCpmmSwapEvent(RaydiumCpmmSwapEvent { + Some(DexEvent::RaydiumCpmmSwapEvent(RaydiumCpmmSwapEvent { metadata, amount_in, minimum_amount_out, @@ -204,7 +204,7 @@ fn parse_swap_base_output_instruction( data: &[u8], accounts: &[Pubkey], metadata: EventMetadata, -) -> Option { +) -> Option { if data.len() < 16 || accounts.len() < 13 { return None; } @@ -212,7 +212,7 @@ fn parse_swap_base_output_instruction( let max_amount_in = read_u64_le(data, 0)?; let amount_out = read_u64_le(data, 8)?; - Some(UnifiedEvent::RaydiumCpmmSwapEvent(RaydiumCpmmSwapEvent { + Some(DexEvent::RaydiumCpmmSwapEvent(RaydiumCpmmSwapEvent { metadata, max_amount_in, amount_out, diff --git a/src/streaming/event_parser/protocols/raydium_cpmm/types.rs b/src/streaming/event_parser/protocols/raydium_cpmm/types.rs index c428f16..21601a2 100644 --- a/src/streaming/event_parser/protocols/raydium_cpmm/types.rs +++ b/src/streaming/event_parser/protocols/raydium_cpmm/types.rs @@ -8,7 +8,7 @@ use crate::streaming::{ protocols::raydium_cpmm::{ RaydiumCpmmAmmConfigAccountEvent, RaydiumCpmmPoolStateAccountEvent, }, - UnifiedEvent, + DexEvent, }, grpc::AccountPretty, }; @@ -36,12 +36,12 @@ pub fn amm_config_decode(data: &[u8]) -> Option { borsh::from_slice::(&data[..AMM_CONFIG_SIZE]).ok() } -pub fn amm_config_parser(account: &AccountPretty, metadata: EventMetadata) -> Option { +pub fn amm_config_parser(account: &AccountPretty, metadata: EventMetadata) -> Option { if account.data.len() < AMM_CONFIG_SIZE + 8 { return None; } if let Some(amm_config) = amm_config_decode(&account.data[8..AMM_CONFIG_SIZE + 8]) { - Some(UnifiedEvent::RaydiumCpmmAmmConfigAccountEvent(RaydiumCpmmAmmConfigAccountEvent { + Some(DexEvent::RaydiumCpmmAmmConfigAccountEvent(RaydiumCpmmAmmConfigAccountEvent { metadata, pubkey: account.pubkey, executable: account.executable, @@ -91,12 +91,12 @@ pub fn pool_state_decode(data: &[u8]) -> Option { borsh::from_slice::(&data[..POOL_STATE_SIZE]).ok() } -pub fn pool_state_parser(account: &AccountPretty, metadata: EventMetadata) -> Option { +pub fn pool_state_parser(account: &AccountPretty, metadata: EventMetadata) -> Option { if account.data.len() < POOL_STATE_SIZE + 8 { return None; } if let Some(pool_state) = pool_state_decode(&account.data[8..POOL_STATE_SIZE + 8]) { - Some(UnifiedEvent::RaydiumCpmmPoolStateAccountEvent(RaydiumCpmmPoolStateAccountEvent { + Some(DexEvent::RaydiumCpmmPoolStateAccountEvent(RaydiumCpmmPoolStateAccountEvent { metadata, pubkey: account.pubkey, executable: account.executable, diff --git a/src/streaming/shred_stream.rs b/src/streaming/shred_stream.rs index 272c5f6..f98a412 100755 --- a/src/streaming/shred_stream.rs +++ b/src/streaming/shred_stream.rs @@ -8,7 +8,7 @@ use crate::protos::shredstream::SubscribeEntriesRequest; use crate::streaming::common::{process_shred_transaction, SubscriptionHandle}; use crate::streaming::event_parser::common::filter::EventTypeFilter; use crate::streaming::event_parser::common::high_performance_clock::get_high_perf_clock; -use crate::streaming::event_parser::{Protocol, UnifiedEvent}; +use crate::streaming::event_parser::{Protocol, DexEvent}; use crate::streaming::grpc::MetricsManager; use crate::streaming::shred::pool::factory; use log::error; @@ -26,7 +26,7 @@ impl ShredStreamGrpc { callback: F, ) -> AnyResult<()> where - F: Fn(UnifiedEvent) + Send + Sync + 'static, + F: Fn(DexEvent) + Send + Sync + 'static, { // 如果已有活跃订阅,先停止它 self.stop().await; diff --git a/src/streaming/yellowstone_grpc.rs b/src/streaming/yellowstone_grpc.rs index 8147028..2a6907f 100644 --- a/src/streaming/yellowstone_grpc.rs +++ b/src/streaming/yellowstone_grpc.rs @@ -4,7 +4,7 @@ use crate::streaming::common::{ SubscriptionHandle, }; use crate::streaming::event_parser::common::filter::EventTypeFilter; -use crate::streaming::event_parser::{Protocol, UnifiedEvent}; +use crate::streaming::event_parser::{Protocol, DexEvent}; use crate::streaming::grpc::pool::factory; use crate::streaming::grpc::{EventPretty, SubscriptionManager}; use anyhow::anyhow; @@ -141,7 +141,7 @@ impl YellowstoneGrpc { callback: F, ) -> AnyResult<()> where - F: Fn(UnifiedEvent) + Send + Sync + 'static, + F: Fn(DexEvent) + Send + Sync + 'static, { *self.event_type_filter.write().await = event_type_filter.clone(); if self