diff --git a/Cargo.toml b/Cargo.toml index bb4ce0e..59da758 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "solana-streamer-sdk" -version = "0.1.4" +version = "0.1.5" edition = "2021" authors = ["William ", "sgxiang ", "wei <1415121722@qq.com>"] repository = "https://github.com/0xfnzero/solana-streamer" diff --git a/README.md b/README.md index be22147..88db3de 100755 --- a/README.md +++ b/README.md @@ -33,14 +33,14 @@ Add the dependency to your `Cargo.toml`: ```toml # Add to your Cargo.toml -solana-streamer-sdk = { path = "./solana-streamer", version = "0.1.4" } +solana-streamer-sdk = { path = "./solana-streamer", version = "0.1.5" } ``` ### Use crates.io ```toml # Add to your Cargo.toml -solana-streamer-sdk = "0.1.4" +solana-streamer-sdk = "0.1.5" ``` ## Usage Examples @@ -51,10 +51,16 @@ use solana_streamer_sdk::{ streaming::{ event_parser::{ protocols::{ - bonk::{BonkPoolCreateEvent, BonkTradeEvent}, pumpfun::{PumpFunCreateTokenEvent, PumpFunTradeEvent}, pumpswap::{ - PumpSwapBuyEvent, PumpSwapCreatePoolEvent, PumpSwapDepositEvent, - PumpSwapSellEvent, PumpSwapWithdrawEvent, - }, raydium_clmm::{RaydiumClmmSwapEvent, RaydiumClmmSwapV2Event}, raydium_cpmm::RaydiumCpmmSwapEvent + bonk::{parser::BONK_PROGRAM_ID, BonkPoolCreateEvent, BonkTradeEvent}, + pumpfun::{parser::PUMPFUN_PROGRAM_ID, PumpFunCreateTokenEvent, PumpFunTradeEvent}, + pumpswap::{ + parser::PUMPSWAP_PROGRAM_ID, PumpSwapBuyEvent, PumpSwapCreatePoolEvent, + PumpSwapDepositEvent, PumpSwapSellEvent, PumpSwapWithdrawEvent, + }, + raydium_clmm::{ + parser::RAYDIUM_CLMM_PROGRAM_ID, RaydiumClmmSwapEvent, RaydiumClmmSwapV2Event, + }, + raydium_cpmm::{parser::RAYDIUM_CPMM_PROGRAM_ID, RaydiumCpmmSwapEvent}, }, Protocol, UnifiedEvent, }, @@ -78,6 +84,8 @@ async fn test_grpc() -> Result<(), Box> { )?; let callback = create_event_callback(); + + // Will try to parse corresponding protocol events from transactions let protocols = vec![ Protocol::PumpFun, Protocol::PumpSwap, @@ -86,9 +94,29 @@ async fn test_grpc() -> Result<(), Box> { Protocol::RaydiumClmm, ]; - println!("Listening for events, press Ctrl+C to stop..."); - grpc.subscribe_events(protocols, None, None, None, None, None, callback) - .await?; + // Filter accounts + let account_include = vec![ + PUMPFUN_PROGRAM_ID.to_string(), // Listen to pumpfun program ID + PUMPSWAP_PROGRAM_ID.to_string(), // Listen to pumpswap program ID + BONK_PROGRAM_ID.to_string(), // Listen to bonk program ID + RAYDIUM_CPMM_PROGRAM_ID.to_string(), // Listen to raydium_cpmm program ID + RAYDIUM_CLMM_PROGRAM_ID.to_string(), // Listen to raydium_clmm program ID + "xxxxxxxx".to_string(), // Listen to xxxxx account + ]; + let account_exclude = vec![]; + let account_required = vec![]; + + println!("Starting to listen for events, press Ctrl+C to stop..."); + grpc.subscribe_events_v2( + protocols, + None, + account_include, + account_exclude, + account_required, + None, + callback, + ) + .await?; Ok(()) } diff --git a/README_CN.md b/README_CN.md index 0fceed8..c6f4f84 100644 --- a/README_CN.md +++ b/README_CN.md @@ -33,14 +33,14 @@ git clone https://github.com/0xfnzero/solana-streamer ```toml # 添加到您的 Cargo.toml -solana-streamer-sdk = { path = "./solana-streamer", version = "0.1.4" } +solana-streamer-sdk = { path = "./solana-streamer", version = "0.1.5" } ``` ### 使用 crates.io ```toml # 添加到您的 Cargo.toml -solana-streamer-sdk = "0.1.4" +solana-streamer-sdk = "0.1.5" ``` ## 使用示例 @@ -51,10 +51,16 @@ use solana_streamer_sdk::{ streaming::{ event_parser::{ protocols::{ - bonk::{BonkPoolCreateEvent, BonkTradeEvent}, pumpfun::{PumpFunCreateTokenEvent, PumpFunTradeEvent}, pumpswap::{ - PumpSwapBuyEvent, PumpSwapCreatePoolEvent, PumpSwapDepositEvent, - PumpSwapSellEvent, PumpSwapWithdrawEvent, - }, raydium_clmm::{RaydiumClmmSwapEvent, RaydiumClmmSwapV2Event}, raydium_cpmm::RaydiumCpmmSwapEvent + bonk::{parser::BONK_PROGRAM_ID, BonkPoolCreateEvent, BonkTradeEvent}, + pumpfun::{parser::PUMPFUN_PROGRAM_ID, PumpFunCreateTokenEvent, PumpFunTradeEvent}, + pumpswap::{ + parser::PUMPSWAP_PROGRAM_ID, PumpSwapBuyEvent, PumpSwapCreatePoolEvent, + PumpSwapDepositEvent, PumpSwapSellEvent, PumpSwapWithdrawEvent, + }, + raydium_clmm::{ + parser::RAYDIUM_CLMM_PROGRAM_ID, RaydiumClmmSwapEvent, RaydiumClmmSwapV2Event, + }, + raydium_cpmm::{parser::RAYDIUM_CPMM_PROGRAM_ID, RaydiumCpmmSwapEvent}, }, Protocol, UnifiedEvent, }, @@ -78,6 +84,8 @@ async fn test_grpc() -> Result<(), Box> { )?; let callback = create_event_callback(); + + // 将会从交易中尝试解析对应的协议事件 let protocols = vec![ Protocol::PumpFun, Protocol::PumpSwap, @@ -86,9 +94,29 @@ async fn test_grpc() -> Result<(), Box> { Protocol::RaydiumClmm, ]; + // 过滤账号 + let account_include = vec![ + PUMPFUN_PROGRAM_ID.to_string(), // 监听 pumpfun 程序ID + PUMPSWAP_PROGRAM_ID.to_string(), // 监听 pumpswap 程序ID + BONK_PROGRAM_ID.to_string(), // 监听 bonk 程序ID + RAYDIUM_CPMM_PROGRAM_ID.to_string(), // 监听 raydium_cpmm 程序ID + RAYDIUM_CLMM_PROGRAM_ID.to_string(), // 监听 raydium_clmm 程序ID + "xxxxxxxx".to_string(), // 监听 xxxxx 账号 + ]; + let account_exclude = vec![]; + let account_required = vec![]; + println!("开始监听事件，按 Ctrl+C 停止..."); - grpc.subscribe_events(protocols, None, None, None, None, None, callback) - .await?; + grpc.subscribe_events_v2( + protocols, + None, + account_include, + account_exclude, + account_required, + None, + callback, + ) + .await?; Ok(()) } diff --git a/src/main.rs b/src/main.rs index cb674c4..7842207 100755 --- a/src/main.rs +++ b/src/main.rs @@ -3,14 +3,16 @@ use solana_streamer_sdk::{ streaming::{ event_parser::{ protocols::{ - bonk::{BonkPoolCreateEvent, BonkTradeEvent}, - pumpfun::{PumpFunCreateTokenEvent, PumpFunTradeEvent}, + bonk::{parser::BONK_PROGRAM_ID, BonkPoolCreateEvent, BonkTradeEvent}, + pumpfun::{parser::PUMPFUN_PROGRAM_ID, PumpFunCreateTokenEvent, PumpFunTradeEvent}, pumpswap::{ - PumpSwapBuyEvent, PumpSwapCreatePoolEvent, PumpSwapDepositEvent, - PumpSwapSellEvent, PumpSwapWithdrawEvent, + parser::PUMPSWAP_PROGRAM_ID, PumpSwapBuyEvent, PumpSwapCreatePoolEvent, + PumpSwapDepositEvent, PumpSwapSellEvent, PumpSwapWithdrawEvent, }, - raydium_clmm::{RaydiumClmmSwapEvent, RaydiumClmmSwapV2Event}, - raydium_cpmm::RaydiumCpmmSwapEvent, + raydium_clmm::{ + parser::RAYDIUM_CLMM_PROGRAM_ID, RaydiumClmmSwapEvent, RaydiumClmmSwapV2Event, + }, + raydium_cpmm::{parser::RAYDIUM_CPMM_PROGRAM_ID, RaydiumCpmmSwapEvent}, }, Protocol, UnifiedEvent, }, @@ -26,7 +28,7 @@ async fn main() -> Result<(), Box> { } async fn test_grpc() -> Result<(), Box> { - println!("正在订阅 GRPC 事件..."); + println!("Subscribing to GRPC events..."); let grpc = YellowstoneGrpc::new( "https://solana-yellowstone-grpc.publicnode.com:443".to_string(), @@ -34,6 +36,8 @@ async fn test_grpc() -> Result<(), Box> { )?; let callback = create_event_callback(); + + // Will try to parse corresponding protocol events from transactions let protocols = vec![ Protocol::PumpFun, Protocol::PumpSwap, @@ -42,9 +46,29 @@ async fn test_grpc() -> Result<(), Box> { Protocol::RaydiumClmm, ]; - println!("开始监听事件，按 Ctrl+C 停止..."); - grpc.subscribe_events(protocols, None, None, None, None, None, callback) - .await?; + // Filter accounts + let account_include = vec![ + PUMPFUN_PROGRAM_ID.to_string(), // Listen to pumpfun program ID + PUMPSWAP_PROGRAM_ID.to_string(), // Listen to pumpswap program ID + BONK_PROGRAM_ID.to_string(), // Listen to bonk program ID + RAYDIUM_CPMM_PROGRAM_ID.to_string(), // Listen to raydium_cpmm program ID + RAYDIUM_CLMM_PROGRAM_ID.to_string(), // Listen to raydium_clmm program ID + "xxxxxxxx".to_string(), // Listen to xxxxx account + ]; + let account_exclude = vec![]; + let account_required = vec![]; + + println!("Starting to listen for events, press Ctrl+C to stop..."); + grpc.subscribe_events_v2( + protocols, + None, + account_include, + account_exclude, + account_required, + None, + callback, + ) + .await?; Ok(()) } diff --git a/src/streaming/event_parser/common/mod.rs b/src/streaming/event_parser/common/mod.rs index d25d73e..a3242ac 100755 --- a/src/streaming/event_parser/common/mod.rs +++ b/src/streaming/event_parser/common/mod.rs @@ -50,6 +50,10 @@ macro_rules! impl_unified_event { fn set_transfer_datas(&mut self, transfer_datas: Vec<$crate::streaming::event_parser::common::types::TransferData>) { self.metadata.transfer_datas = transfer_datas; } + + fn index(&self) -> String { + self.metadata.index.clone() + } } }; } diff --git a/src/streaming/event_parser/common/types.rs b/src/streaming/event_parser/common/types.rs index 28ccfb8..48eb811 100755 --- a/src/streaming/event_parser/common/types.rs +++ b/src/streaming/event_parser/common/types.rs @@ -162,6 +162,7 @@ pub struct EventMetadata { pub event_type: EventType, pub program_id: Pubkey, pub transfer_datas: Vec, + pub index: String, } impl EventMetadata { @@ -174,6 +175,7 @@ impl EventMetadata { protocol: ProtocolType, event_type: EventType, program_id: Pubkey, + index: String, ) -> Self { Self { id, @@ -186,6 +188,7 @@ impl EventMetadata { event_type, program_id, transfer_datas: vec![], + index, } } pub fn set_id(&mut self, id: String) { diff --git a/src/streaming/event_parser/core/traits.rs b/src/streaming/event_parser/core/traits.rs index 2bee766..ed95d10 100755 --- a/src/streaming/event_parser/core/traits.rs +++ b/src/streaming/event_parser/core/traits.rs @@ -52,6 +52,8 @@ pub trait UnifiedEvent: Debug + Send + Sync { } fn set_transfer_datas(&mut self, transfer_datas: Vec); + + fn index(&self) -> String; } /// 事件解析器trait - 定义了事件解析的核心方法 @@ -64,6 +66,7 @@ pub trait EventParser: Send + Sync { signature: &str, slot: u64, block_time: Option, + index: String, ) -> Vec>; /// 从指令中解析事件数据 @@ -74,6 +77,7 @@ pub trait EventParser: Send + Sync { signature: &str, slot: u64, block_time: Option, + index: String, ) -> Vec>; /// 从VersionedTransaction中解析指令事件的通用方法 @@ -106,7 +110,14 @@ pub trait EventParser: Send + Sync { } } if let Ok(mut events) = self - .parse_instruction(instruction, &accounts, signature, slot, block_time) + .parse_instruction( + instruction, + &accounts, + signature, + slot, + block_time, + format!("{}", index), + ) .await { if events.len() > 0 { @@ -231,6 +242,7 @@ pub trait EventParser: Send + Sync { signature, slot, block_time, + format!("{}.{}", inner_instruction.index, index), ) .await { @@ -249,7 +261,13 @@ pub trait EventParser: Send + Sync { } } if let Ok(mut events) = self - .parse_inner_instruction(compiled, signature, slot, block_time) + .parse_inner_instruction( + compiled, + signature, + slot, + block_time, + format!("{}.{}", inner_instruction.index, index), + ) .await { if events.len() > 0 { @@ -276,11 +294,28 @@ pub trait EventParser: Send + Sync { if instruction_events.len() > 0 && inner_instruction_events.len() > 0 { for instruction_event in &mut instruction_events { for inner_instruction_event in &inner_instruction_events { - if instruction_event.id() == inner_instruction_event.id() - && instruction_event.event_type() == inner_instruction_event.event_type() - { - instruction_event.merge(inner_instruction_event.clone_boxed()); - break; + if instruction_event.id() == inner_instruction_event.id() { + let i_index = instruction_event.index(); + let in_index = inner_instruction_event.index(); + if !i_index.contains(".") && in_index.contains(".") { + let in_index_parent_index = in_index.split(".").nth(0).unwrap(); + if in_index_parent_index == i_index { + instruction_event.merge(inner_instruction_event.clone_boxed()); + break; + } + } else if i_index.contains(".") && in_index.contains(".") { + // 嵌套指令 + let i_index_parent_index = i_index.split(".").nth(0).unwrap(); + let in_index_parent_index = in_index.split(".").nth(0).unwrap(); + if i_index_parent_index == in_index_parent_index { + let i_index_child_index = i_index.split(".").nth(1).unwrap(); + let in_index_child_index = in_index.split(".").nth(1).unwrap(); + if in_index_child_index > i_index_child_index { + instruction_event.merge(inner_instruction_event.clone_boxed()); + break; + } + } + } } } } @@ -329,10 +364,16 @@ pub trait EventParser: Send + Sync { signature: &str, slot: Option, block_time: Option, + index: String, ) -> Result>> { let slot = slot.unwrap_or(0); - let events = - self.parse_events_from_inner_instruction(instruction, signature, slot, block_time); + let events = self.parse_events_from_inner_instruction( + instruction, + signature, + slot, + block_time, + index, + ); Ok(events) } @@ -343,10 +384,17 @@ pub trait EventParser: Send + Sync { signature: &str, slot: Option, block_time: Option, + index: String, ) -> Result>> { let slot = slot.unwrap_or(0); - let events = - self.parse_events_from_instruction(instruction, accounts, signature, slot, block_time); + let events = self.parse_events_from_instruction( + instruction, + accounts, + signature, + slot, + block_time, + index, + ); Ok(events) } @@ -427,6 +475,7 @@ impl GenericEventParser { signature: &str, slot: u64, block_time: Option, + index: String, ) -> Option> { let timestamp = block_time.unwrap_or(Timestamp { seconds: 0, @@ -442,6 +491,7 @@ impl GenericEventParser { self.protocol_type.clone(), config.event_type.clone(), self.program_id, + index, ); (config.inner_instruction_parser)(data, metadata) } @@ -455,6 +505,7 @@ impl GenericEventParser { signature: &str, slot: u64, block_time: Option, + index: String, ) -> Option> { let timestamp = block_time.unwrap_or(Timestamp { seconds: 0, @@ -470,6 +521,7 @@ impl GenericEventParser { self.protocol_type.clone(), config.event_type.clone(), self.program_id, + index, ); (config.instruction_parser)(data, account_pubkeys, metadata) } @@ -484,6 +536,7 @@ impl EventParser for GenericEventParser { signature: &str, slot: u64, block_time: Option, + index: String, ) -> Vec> { let inner_instruction_data = inner_instruction.data.clone(); let inner_instruction_data_decoded = @@ -498,9 +551,14 @@ impl EventParser for GenericEventParser { for (disc, configs) in &self.inner_instruction_configs { if discriminator_matches(&inner_instruction_data_decoded_str, disc) { for config in configs { - if let Some(event) = self - .parse_inner_instruction_event(config, data, signature, slot, block_time) - { + if let Some(event) = self.parse_inner_instruction_event( + config, + data, + signature, + slot, + block_time, + index.clone(), + ) { events.push(event); } } @@ -517,6 +575,7 @@ impl EventParser for GenericEventParser { signature: &str, slot: u64, block_time: Option, + index: String, ) -> Vec> { let program_id = accounts[instruction.program_id_index as usize]; if !self.should_handle(&program_id) { @@ -548,6 +607,7 @@ impl EventParser for GenericEventParser { signature, slot, block_time, + index.clone(), ) { events.push(event); } diff --git a/src/streaming/event_parser/protocols/bonk/parser.rs b/src/streaming/event_parser/protocols/bonk/parser.rs index 855b09e..cf26f4f 100755 --- a/src/streaming/event_parser/protocols/bonk/parser.rs +++ b/src/streaming/event_parser/protocols/bonk/parser.rs @@ -421,9 +421,15 @@ impl EventParser for BonkEventParser { signature: &str, slot: u64, block_time: Option, + index: String, ) -> Vec> { - self.inner - .parse_events_from_inner_instruction(inner_instruction, signature, slot, block_time) + self.inner.parse_events_from_inner_instruction( + inner_instruction, + signature, + slot, + block_time, + index, + ) } fn parse_events_from_instruction( @@ -433,9 +439,16 @@ impl EventParser for BonkEventParser { signature: &str, slot: u64, block_time: Option, + index: String, ) -> Vec> { - self.inner - .parse_events_from_instruction(instruction, accounts, signature, slot, block_time) + self.inner.parse_events_from_instruction( + instruction, + accounts, + signature, + slot, + block_time, + index, + ) } fn should_handle(&self, program_id: &Pubkey) -> bool { diff --git a/src/streaming/event_parser/protocols/pumpfun/parser.rs b/src/streaming/event_parser/protocols/pumpfun/parser.rs index c277029..402937a 100755 --- a/src/streaming/event_parser/protocols/pumpfun/parser.rs +++ b/src/streaming/event_parser/protocols/pumpfun/parser.rs @@ -226,9 +226,15 @@ impl EventParser for PumpFunEventParser { signature: &str, slot: u64, block_time: Option, + index: String, ) -> Vec> { - self.inner - .parse_events_from_inner_instruction(inner_instruction, signature, slot, block_time) + self.inner.parse_events_from_inner_instruction( + inner_instruction, + signature, + slot, + block_time, + index, + ) } fn parse_events_from_instruction( @@ -238,9 +244,16 @@ impl EventParser for PumpFunEventParser { signature: &str, slot: u64, block_time: Option, + index: String, ) -> Vec> { - self.inner - .parse_events_from_instruction(instruction, accounts, signature, slot, block_time) + self.inner.parse_events_from_instruction( + instruction, + accounts, + signature, + slot, + block_time, + index, + ) } fn should_handle(&self, program_id: &Pubkey) -> bool { diff --git a/src/streaming/event_parser/protocols/pumpswap/parser.rs b/src/streaming/event_parser/protocols/pumpswap/parser.rs index 64a9c62..ce98fde 100755 --- a/src/streaming/event_parser/protocols/pumpswap/parser.rs +++ b/src/streaming/event_parser/protocols/pumpswap/parser.rs @@ -3,7 +3,7 @@ use solana_sdk::{instruction::CompiledInstruction, pubkey::Pubkey}; use solana_transaction_status::UiCompiledInstruction; use crate::streaming::event_parser::{ - common::{EventMetadata, EventType, ProtocolType, read_u64_le}, + common::{read_u64_le, EventMetadata, EventType, ProtocolType}, core::traits::{EventParser, GenericEventParseConfig, GenericEventParser, UnifiedEvent}, protocols::pumpswap::{ discriminators, PumpSwapBuyEvent, PumpSwapCreatePoolEvent, PumpSwapDepositEvent, @@ -67,7 +67,10 @@ impl PumpSwapEventParser { } /// 解析买入日志事件 - fn parse_buy_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option> { + fn parse_buy_inner_instruction( + data: &[u8], + metadata: EventMetadata, + ) -> Option> { if let Ok(event) = borsh::from_slice::(data) { let mut metadata = metadata; metadata.set_id(format!( @@ -84,7 +87,10 @@ impl PumpSwapEventParser { } /// 解析卖出日志事件 - fn parse_sell_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option> { + fn parse_sell_inner_instruction( + data: &[u8], + metadata: EventMetadata, + ) -> Option> { if let Ok(event) = borsh::from_slice::(data) { let mut metadata = metadata; metadata.set_id(format!( @@ -121,7 +127,10 @@ impl PumpSwapEventParser { } /// 解析存款日志事件 - fn parse_deposit_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option> { + fn parse_deposit_inner_instruction( + data: &[u8], + metadata: EventMetadata, + ) -> Option> { if let Ok(event) = borsh::from_slice::(data) { let mut metadata = metadata; metadata.set_id(format!( @@ -138,7 +147,10 @@ impl PumpSwapEventParser { } /// 解析提款日志事件 - fn parse_withdraw_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option> { + fn parse_withdraw_inner_instruction( + data: &[u8], + metadata: EventMetadata, + ) -> Option> { if let Ok(event) = borsh::from_slice::(data) { let mut metadata = metadata; metadata.set_id(format!( @@ -362,9 +374,15 @@ impl EventParser for PumpSwapEventParser { signature: &str, slot: u64, block_time: Option, + index: String, ) -> Vec> { - self.inner - .parse_events_from_inner_instruction(inner_instruction, signature, slot, block_time) + self.inner.parse_events_from_inner_instruction( + inner_instruction, + signature, + slot, + block_time, + index, + ) } fn parse_events_from_instruction( @@ -374,9 +392,16 @@ impl EventParser for PumpSwapEventParser { signature: &str, slot: u64, block_time: Option, + index: String, ) -> Vec> { - self.inner - .parse_events_from_instruction(instruction, accounts, signature, slot, block_time) + self.inner.parse_events_from_instruction( + instruction, + accounts, + signature, + slot, + block_time, + index, + ) } fn should_handle(&self, program_id: &Pubkey) -> bool { diff --git a/src/streaming/event_parser/protocols/raydium_clmm/parser.rs b/src/streaming/event_parser/protocols/raydium_clmm/parser.rs index 501d9b5..b504470 100755 --- a/src/streaming/event_parser/protocols/raydium_clmm/parser.rs +++ b/src/streaming/event_parser/protocols/raydium_clmm/parser.rs @@ -146,9 +146,15 @@ impl EventParser for RaydiumClmmEventParser { signature: &str, slot: u64, block_time: Option, + index: String, ) -> Vec> { - self.inner - .parse_events_from_inner_instruction(inner_instruction, signature, slot, block_time) + self.inner.parse_events_from_inner_instruction( + inner_instruction, + signature, + slot, + block_time, + index, + ) } fn parse_events_from_instruction( @@ -158,9 +164,16 @@ impl EventParser for RaydiumClmmEventParser { signature: &str, slot: u64, block_time: Option, + index: String, ) -> Vec> { - self.inner - .parse_events_from_instruction(instruction, accounts, signature, slot, block_time) + self.inner.parse_events_from_instruction( + instruction, + accounts, + signature, + slot, + block_time, + index, + ) } fn should_handle(&self, program_id: &Pubkey) -> bool { diff --git a/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs b/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs index f983d09..3de67cf 100755 --- a/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs +++ b/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs @@ -135,9 +135,15 @@ impl EventParser for RaydiumCpmmEventParser { signature: &str, slot: u64, block_time: Option, + index: String, ) -> Vec> { - self.inner - .parse_events_from_inner_instruction(inner_instruction, signature, slot, block_time) + self.inner.parse_events_from_inner_instruction( + inner_instruction, + signature, + slot, + block_time, + index, + ) } fn parse_events_from_instruction( @@ -147,9 +153,16 @@ impl EventParser for RaydiumCpmmEventParser { signature: &str, slot: u64, block_time: Option, + index: String, ) -> Vec> { - self.inner - .parse_events_from_instruction(instruction, accounts, signature, slot, block_time) + self.inner.parse_events_from_instruction( + instruction, + accounts, + signature, + slot, + block_time, + index, + ) } fn should_handle(&self, program_id: &Pubkey) -> bool { diff --git a/src/streaming/yellowstone_grpc.rs b/src/streaming/yellowstone_grpc.rs index ace83d2..3787017 100755 --- a/src/streaming/yellowstone_grpc.rs +++ b/src/streaming/yellowstone_grpc.rs @@ -175,7 +175,97 @@ impl YellowstoneGrpc { Ok(()) } + /// Subscribe to Yellowstone GRPC service events with advanced filtering options + /// + /// This method allows subscribing to specific protocol events with more granular account filtering. + /// It processes transactions in real-time and calls the provided callback function when matching events are found. + /// + /// # Parameters + /// + /// * `protocols` - List of protocols to parse (e.g., PumpFun, PumpSwap, Bonk, RaydiumCpmm) + /// * `bot_wallet` - Optional bot wallet address. If passed: in PumpFunTradeEvent if user is in the address, is_bot=true will be set. In BonkTradeEvent if payer is in the address, is_bot=true will be set. Default is false. + /// * `account_include` - List of account addresses to include in the subscription + /// * `account_exclude` - List of account addresses to exclude from the subscription + /// * `account_required` - List of account addresses that must be present in transactions + /// * `commitment` - Optional commitment level for the subscription + /// * `callback` - Function to call when matching events are found + pub async fn subscribe_events_v2( + &self, + protocols: Vec, + bot_wallet: Option, + account_include: Vec, + account_exclude: Vec, + account_required: Vec, + commitment: Option, + callback: F, + ) -> AnyResult<()> + where + F: Fn(Box) + Send + Sync + 'static, + { + if account_include.is_empty() && account_exclude.is_empty() && account_required.is_empty() { + return Err(anyhow::anyhow!( + "account_include or account_exclude or account_required cannot be empty" + )); + } + + let transactions = + self.get_subscribe_request_filter(account_include, account_exclude, account_required); + // Subscribe to events + let (mut subscribe_tx, mut stream) = self + .subscribe_with_request(transactions, commitment) + .await?; + + // Create channel + let (mut tx, mut rx) = mpsc::channel::(CHANNEL_SIZE); + + // Create callback function, wrap with Arc to share across multiple tasks + let callback = std::sync::Arc::new(Box::new(callback)); + + // Start task to process the stream + tokio::spawn(async move { + while let Some(message) = stream.next().await { + match message { + Ok(msg) => { + if let Err(e) = + Self::handle_stream_message(msg, &mut tx, &mut subscribe_tx).await + { + error!("Error handling message: {:?}", e); + break; + } + } + Err(error) => { + error!("Stream error: {error:?}"); + break; + } + } + } + }); + + // Process transactions + tokio::spawn(async move { + while let Some(transaction_pretty) = rx.next().await { + if let Err(e) = Self::process_event_transaction( + transaction_pretty, + &**callback, + bot_wallet, + protocols.clone(), + ) + .await + { + error!("Error processing transaction: {:?}", e); + } + } + }); + + tokio::signal::ctrl_c().await?; + Ok(()) + } + /// 订阅事件 + #[deprecated( + since = "0.1.5", + note = "This method will be removed, please use the new API: subscribe_events_v2" + )] pub async fn subscribe_events( &self, protocols: Vec, @@ -274,17 +364,18 @@ impl YellowstoneGrpc { let tx_clone = transaction_pretty.tx.clone(); let signature_clone = signature.clone(); let bot_wallet_clone = bot_wallet.clone(); - + futures.push(tokio::spawn(async move { - parser.parse_transaction( - tx_clone, - &signature_clone, - Some(slot), - transaction_pretty.block_time, - bot_wallet_clone, - ) - .await - .unwrap_or_else(|_e| vec![]) + parser + .parse_transaction( + tx_clone, + &signature_clone, + Some(slot), + transaction_pretty.block_time, + bot_wallet_clone, + ) + .await + .unwrap_or_else(|_e| vec![]) })); }