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

feat: enhance event parser with transfer data processing capabilities

Browse Source 
- Bump version to 0.1.2
- Add TransferData struct for transaction data parsing
- Implement parse_transfer_datas_from_next_instructions function to extract transfer data from subsequent instructions
- Enhance UnifiedEvent trait with set_transfer_datas method
- Optimize transfer data extraction when parsing inner instructions
- Support transfer data parsing for multiple protocols (PumpFun, PumpSwap, Bonk, Raydium, etc.)
- Improve support for Token Program, Token 2022 Program, and System Program transfer instructions
This commit is contained in:
ysq
2025-07-21 23:55:13 +08:00
parent f43d44b350
commit 7615102fd6
6 changed files with 210 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/streaming/event_parser/common/mod.rs
+4
View File
@@ -46,6 +46,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;
}
}
};
}
src/streaming/event_parser/common/types.rs
+140
View File
@@ -1,6 +1,10 @@
use borsh::{BorshDeserialize, BorshSerialize};
use serde::{Deserialize, Serialize};
use solana_sdk::instruction::CompiledInstruction;
use solana_sdk::pubkey::Pubkey;
use solana_transaction_status::{
UiCompiledInstruction, UiInstruction, UiTransactionStatusMeta, UiTransactionTokenBalance,
};
use std::collections::hash_map::DefaultHasher;
use std::hash::{Hash, Hasher};
@@ -129,6 +133,20 @@ impl ProtocolInfo {
}
}
/// 交易数据
#[derive(
Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, BorshSerialize, BorshDeserialize,
)]
pub struct TransferData {
pub token_program: Pubkey,
pub source: Pubkey,
pub destination: Pubkey,
pub authority: Option<Pubkey>,
pub amount: u64,
pub decimals: Option<u8>,
pub mint: Option<Pubkey>,
}
/// 事件元数据
#[derive(
Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, BorshSerialize, BorshDeserialize,
@@ -141,6 +159,7 @@ pub struct EventMetadata {
pub protocol: ProtocolType,
pub event_type: EventType,
pub program_id: Pubkey,
pub transfer_datas: Vec<TransferData>,
}
impl EventMetadata {
@@ -160,6 +179,7 @@ impl EventMetadata {
protocol,
event_type,
program_id,
transfer_datas: vec![],
}
}
pub fn set_id(&mut self, id: String) {
@@ -171,3 +191,123 @@ impl EventMetadata {
self.id = format!("{:x}", hash_value);
}
}
/// 解析接下来指令中的token转账数据
pub fn parse_transfer_datas_from_next_instructions(
inner_instruction: &solana_transaction_status::UiInnerInstructions,
current_index: i8,
accounts: &[Pubkey],
event_type: EventType,
) -> Vec<TransferData> {
let take = match event_type {
EventType::PumpFunBuy => 4,
EventType::PumpFunSell => 1,
EventType::PumpSwapBuy => 3,
EventType::PumpSwapSell => 3,
EventType::BonkBuyExactIn
| EventType::BonkBuyExactOut
| EventType::BonkSellExactIn
| EventType::BonkSellExactOut => 3,
EventType::RaydiumCpmmSwapBaseInput
| EventType::RaydiumCpmmSwapBaseOutput
| EventType::RaydiumClmmSwap
| EventType::RaydiumClmmSwapV2 => 2,
_ => 0,
};
if take == 0 {
return vec![];
}
let mut transfer_datas = vec![];
// 获取当前指令之后的两个指令
let next_instructions: Vec<&UiInstruction> = inner_instruction
.instructions
.iter()
.skip((current_index + 1) as usize)
.take(take)
.collect();
for instruction in next_instructions {
if let UiInstruction::Compiled(compiled) = instruction {
if let Ok(data) = bs58::decode(compiled.data.clone()).into_vec() {
// Token Program: transferChecked
// Token 2022 Program: transferChecked
if data[0] == 12 {
let account_pubkeys: Vec<Pubkey> = compiled
.accounts
.iter()
.map(|a| accounts[*a as usize])
.collect();
if account_pubkeys.len() < 4 {
continue;
}
let (source, mint, destination, authority) = (
account_pubkeys[0],
account_pubkeys[1],
account_pubkeys[2],
account_pubkeys[3],
);
let amount = u64::from_le_bytes(data[1..9].try_into().unwrap());
let decimals = data[9];
let token_program = accounts[compiled.program_id_index as usize];
transfer_datas.push(TransferData {
amount,
decimals: Some(decimals),
mint: Some(mint),
source,
destination,
authority: Some(authority),
token_program,
});
}
// Token Program: transfer
else if data[0] == 3 {
let account_pubkeys: Vec<Pubkey> = compiled
.accounts
.iter()
.map(|a| accounts[*a as usize])
.collect();
if account_pubkeys.len() < 3 {
continue;
}
let (source, destination, authority) =
(account_pubkeys[0], account_pubkeys[1], account_pubkeys[2]);
let amount = u64::from_le_bytes(data[1..9].try_into().unwrap());
let token_program = accounts[compiled.program_id_index as usize];
transfer_datas.push(TransferData {
amount,
decimals: None,
mint: None,
source,
destination,
authority: Some(authority),
token_program,
});
}
//System Program: transfer
else if data[0] == 2 {
let account_pubkeys: Vec<Pubkey> = compiled
.accounts
.iter()
.map(|a| accounts[*a as usize])
.collect();
if account_pubkeys.len() < 2 {
continue;
}
let (source, destination) = (account_pubkeys[0], account_pubkeys[1]);
let amount = u64::from_le_bytes(data[4..12].try_into().unwrap());
let token_program = accounts[compiled.program_id_index as usize];
transfer_datas.push(TransferData {
amount,
decimals: None,
mint: None,
source,
destination,
authority: None,
token_program,
});
}
}
}
}
transfer_datas
}
src/streaming/event_parser/core/traits.rs
+61 -10
View File
@@ -3,11 +3,14 @@ use solana_sdk::{
instruction::CompiledInstruction, pubkey::Pubkey, transaction::VersionedTransaction,
};
use solana_transaction_status::{
EncodedTransactionWithStatusMeta, UiCompiledInstruction, UiInstruction,
EncodedTransactionWithStatusMeta, UiCompiledInstruction, UiInnerInstructions, UiInstruction,
};
use std::fmt::Debug;
use std::{collections::HashMap, str::FromStr};
use crate::streaming::event_parser::common::{
parse_transfer_datas_from_next_instructions, TransferData,
};
use crate::streaming::event_parser::{
common::{utils::*, EventMetadata, EventType, ProtocolType},
protocols::{
@@ -46,6 +49,8 @@ pub trait UnifiedEvent: Debug + Send + Sync {
fn merge(&mut self, _other: Box<dyn UnifiedEvent>) {
// 默认实现：不进行任何合并操作
}
fn set_transfer_datas(&mut self, transfer_datas: Vec<TransferData>);
}
/// 事件解析器trait - 定义了事件解析的核心方法
@@ -75,6 +80,7 @@ pub trait EventParser: Send + Sync {
signature: &str,
slot: Option<u64>,
accounts: &[Pubkey],
inner_instructions: &[UiInnerInstructions],
) -> Result<Vec<Box<dyn UnifiedEvent>>> {
let mut instruction_events = Vec::new();
// 获取交易的指令和账户
@@ -85,7 +91,7 @@ pub trait EventParser: Send + Sync {
let has_program = accounts.iter().any(|account| self.should_handle(account));
if has_program {
// 解析每个指令
for instruction in compiled_instructions {
for (index, instruction) in compiled_instructions.iter().enumerate() {
if let Some(program_id) = accounts.get(instruction.program_id_index as usize) {
if self.should_handle(program_id) {
let max_idx = instruction.accounts.iter().max().unwrap_or(&0);
@@ -95,11 +101,29 @@ pub trait EventParser: Send + Sync {
accounts.push(Pubkey::default());
}
}
if let Ok(events) = self
if let Ok(mut events) = self
.parse_instruction(instruction, &accounts, signature, slot)
.await
{
instruction_events.extend(events);
if events.len() > 0 {
if let Some(inn) =
inner_instructions.iter().find(|inner_instruction| {
inner_instruction.index == index as u8
})
{
events.iter_mut().for_each(|event| {
let transfer_datas =
parse_transfer_datas_from_next_instructions(
&inn,
-1 as i8,
&accounts,
event.event_type(),
);
event.set_transfer_datas(transfer_datas.clone());
});
}
instruction_events.extend(events);
}
}
}
}
@@ -122,6 +146,7 @@ pub trait EventParser: Send + Sync {
signature,
slot,
&accounts,
&vec![],
)
.await
.unwrap_or_else(|_e| vec![]);
@@ -143,7 +168,9 @@ pub trait EventParser: Send + Sync {
.ok_or_else(|| anyhow::anyhow!("Missing transaction metadata"))?;
let mut address_table_lookups: Vec<Pubkey> = vec![];
let mut inner_instructions: Vec<UiInnerInstructions> = vec![];
if meta.err.is_none() {
inner_instructions = meta.inner_instructions.as_ref().unwrap().clone();
let loaded_addresses = meta.loaded_addresses.as_ref().unwrap();
for lookup in &loaded_addresses.writable {
address_table_lookups.push(Pubkey::from_str(lookup).unwrap());
@@ -167,6 +194,7 @@ pub trait EventParser: Send + Sync {
signature,
slot,
&accounts,
&inner_instructions,
)
.await
.unwrap_or_else(|_e| vec![]);
@@ -178,9 +206,8 @@ pub trait EventParser: Send + Sync {
let mut inner_instruction_events = Vec::new();
// 检查交易是否成功
if meta.err.is_none() {
let inner_instructions = meta.inner_instructions.as_ref().unwrap();
for inner_instruction in inner_instructions {
for instruction in &inner_instruction.instructions {
for (index, instruction) in inner_instruction.instructions.iter().enumerate() {
match instruction {
UiInstruction::Compiled(compiled) => {
// 解析嵌套指令
@@ -189,7 +216,7 @@ pub trait EventParser: Send + Sync {
accounts: compiled.accounts.clone(),
data: bs58::decode(compiled.data.clone()).into_vec().unwrap(),
};
if let Ok(events) = self
if let Ok(mut events) = self
.parse_instruction(
&compiled_instruction,
&accounts,
@@ -198,13 +225,37 @@ pub trait EventParser: Send + Sync {
)
.await
{
instruction_events.extend(events);
if events.len() > 0 {
events.iter_mut().for_each(|event| {
let transfer_datas =
parse_transfer_datas_from_next_instructions(
&inner_instruction,
index as i8,
&accounts,
event.event_type(),
);
event.set_transfer_datas(transfer_datas.clone());
});
instruction_events.extend(events);
}
}
if let Ok(events) = self
if let Ok(mut events) = self
.parse_inner_instruction(compiled, signature, slot)
.await
{
inner_instruction_events.extend(events);
if events.len() > 0 {
events.iter_mut().for_each(|event| {
let transfer_datas =
parse_transfer_datas_from_next_instructions(
&inner_instruction,
index as i8,
&accounts,
event.event_type(),
);
event.set_transfer_datas(transfer_datas.clone());
});
inner_instruction_events.extend(events);
}
}
}
_ => {}