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

Merge commit '89986e1b36a6bcb45db4b7c268330eb4c137c037'

Browse Source
This commit is contained in:
ysq
2025-11-06 23:19:02 +08:00
parent 72d5474dec 89986e1b36
commit ed290a5a69
19 changed files with 1501 additions and 1682 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/parse_tx_events.rs
+111 -8
View File
@@ -9,7 +9,7 @@ use std::sync::Arc;
#[tokio::main]
async fn main() -> Result<()> {
let signatures = vec![
"5sDWrTTkE69CNc6nrAX7SqPS7FiajJTg8TMog3Gve7KjVfrqYn8YZcX1kAoyKok976S4RTnK1EdCV8hRiDWg68Aj",
"4PsHYajH87x2zJPEGZczZtd2ksibuMCFPonC24jk5mTGZ46hzvjpzM5UZuLz9sRv79MkCBbtDqwJapGPTSkCFKoL",
];
// Validate signature format
let mut valid_signatures = Vec::new();
@@ -36,8 +36,9 @@ async fn main() -> Result<()> {
/// Get details of a single transaction
async fn get_single_transaction_details(signature_str: &str) -> Result<()> {
use solana_sdk::signature::Signature;
use solana_transaction_status::UiTransactionEncoding;
use solana_sdk::{signature::Signature, pubkey::Pubkey, message::compiled_instruction::CompiledInstruction};
use solana_transaction_status::{UiTransactionEncoding, InnerInstruction, InnerInstructions, UiInstruction};
use prost_types::Timestamp;
let signature = Signature::from_str(signature_str)?;
@@ -88,6 +89,96 @@ async fn get_single_transaction_details(signature_str: &str) -> Result<()> {
}
}
}
// Parse the transaction and extract necessary data
let versioned_tx = match transaction.transaction.transaction.decode() {
Some(tx) => tx,
None => {
println!("Failed to decode transaction");
return Ok(());
}
};
// Convert inner instructions from meta
let mut inner_instructions_vec: Vec<InnerInstructions> = Vec::new();
if let Some(meta) = &transaction.transaction.meta {
if let solana_transaction_status::option_serializer::OptionSerializer::Some(
ui_inner_insts,
) = &meta.inner_instructions
{
for ui_inner in ui_inner_insts {
let mut converted_instructions = Vec::new();
for ui_instruction in &ui_inner.instructions {
if let UiInstruction::Compiled(ui_compiled) = ui_instruction {
if let Ok(data) = solana_sdk::bs58::decode(&ui_compiled.data).into_vec() {
let compiled_instruction = CompiledInstruction {
program_id_index: ui_compiled.program_id_index,
accounts: ui_compiled.accounts.to_vec(),
data,
};
let inner_instruction = InnerInstruction {
instruction: compiled_instruction,
stack_height: ui_compiled.stack_height,
};
converted_instructions.push(inner_instruction);
}
}
}
let inner_instructions = InnerInstructions {
index: ui_inner.index,
instructions: converted_instructions,
};
inner_instructions_vec.push(inner_instructions);
}
}
}
// Extract address table lookups
let meta = transaction.transaction.meta;
let mut address_table_lookups: Vec<Pubkey> = vec![];
if let Some(meta) = meta {
if let solana_transaction_status::option_serializer::OptionSerializer::Some(
loaded_addresses,
) = &meta.loaded_addresses
{
address_table_lookups
.reserve(loaded_addresses.writable.len() + loaded_addresses.readonly.len());
address_table_lookups.extend(
loaded_addresses
.writable
.iter()
.filter_map(|s| s.parse::<Pubkey>().ok())
.chain(
loaded_addresses
.readonly
.iter()
.filter_map(|s| s.parse::<Pubkey>().ok()),
),
);
}
}
// Build complete accounts list
let mut accounts = Vec::with_capacity(
versioned_tx.message.static_account_keys().len() + address_table_lookups.len(),
);
accounts.extend_from_slice(versioned_tx.message.static_account_keys());
accounts.extend(address_table_lookups);
let slot = transaction.slot;
let block_time = transaction.block_time.map(|t| Timestamp { seconds: t as i64, nanos: 0 });
let recv_us = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_micros() as i64;
let bot_wallet = None;
let transaction_index = None;
let protocols = vec![
Protocol::Bonk,
Protocol::RaydiumClmm,
@@ -96,14 +187,26 @@ async fn get_single_transaction_details(signature_str: &str) -> Result<()> {
Protocol::RaydiumCpmm,
Protocol::RaydiumAmmV4,
];
EventParser::parse_encoded_confirmed_transaction_with_status_meta(
// Create callback
let callback = Arc::new(move |event: DexEvent| {
println!("{:?}\n", event);
});
// Call parse_instruction_events_from_versioned_transaction
EventParser::parse_instruction_events_from_versioned_transaction(
&protocols,
None,
&versioned_tx,
signature,
transaction,
Arc::new(move |event: &DexEvent| {
println!("{:?}\n", event);
}),
Some(slot),
block_time,
recv_us,
&accounts,
&inner_instructions_vec,
bot_wallet,
transaction_index,
callback,
)
.await?;
}
src/streaming/common/event_processor.rs
+6 -4
View File
@@ -64,7 +64,7 @@ pub async fn process_grpc_transaction(
let adapter_callback = create_metrics_callback(callback.clone());
EventParser::parse_grpc_transaction_owned(
EventParser::parse_grpc_transaction(
protocols,
event_type_filter,
grpc_tx,
@@ -123,18 +123,20 @@ pub async fn process_shred_transaction(
let recv_us = transaction_with_slot.recv_us;
let adapter_callback = create_metrics_callback(callback);
let accounts = tx.message.static_account_keys();
EventParser::parse_versioned_transaction_owned(
EventParser::parse_instruction_events_from_versioned_transaction(
protocols,
event_type_filter,
tx,
&tx,
signature,
Some(slot),
None,
recv_us,
accounts,
&[],
bot_wallet,
None,
&[],
adapter_callback,
)
.await?;
src/streaming/event_parser/core/account_event_parser.rs
+77 -30
View File
@@ -1,8 +1,6 @@
use crate::streaming::common::SimdUtils;
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::common::{EventMetadata, EventType, ProtocolType};
use crate::streaming::event_parser::core::traits::DexEvent;
use crate::streaming::event_parser::Protocol;
use crate::streaming::grpc::AccountPretty;
@@ -63,46 +61,93 @@ impl AccountEventParser {
account: AccountPretty,
event_type_filter: Option<&EventTypeFilter>,
) -> Option<DexEvent> {
// 直接从 parser_cache 获取配置
let configs = get_account_configs(
protocols,
event_type_filter,
Self::parse_nonce_account_event,
Self::parse_token_account_event,
);
for config in configs {
if config.program_id == Pubkey::default()
|| (account.owner == config.program_id
&& SimdUtils::fast_discriminator_match(
&account.data,
config.account_discriminator,
))
{
let event = (config.account_parser)(
&account,
EventMetadata {
use crate::streaming::event_parser::core::dispatcher::EventDispatcher;
// 1. 尝试从账户 discriminator 解析（协议特定账户）
if account.data.len() >= 8 {
let discriminator = &account.data[0..8];
// 尝试识别协议类型
if let Some(protocol) = EventDispatcher::match_protocol_by_program_id(&account.owner) {
// 检查是否在请求的协议列表中
if protocols.contains(&protocol) {
// 构建临时元数据（protocol会被dispatcher设置，event_type会在parser中设置）
let metadata = EventMetadata {
slot: account.slot,
signature: account.signature,
protocol: config.protocol_type,
event_type: config.event_type,
program_id: config.program_id,
protocol: ProtocolType::Common, // 会被 EventDispatcher::dispatch_account 设置
event_type: EventType::default(), // 会被具体 parser 设置
program_id: account.owner,
recv_us: account.recv_us,
handle_us: elapsed_micros_since(account.recv_us),
..Default::default()
},
);
if event.is_some() {
return event;
};
// 使用 dispatcher 解析
if let Some(event) = EventDispatcher::dispatch_account(
protocol,
discriminator,
&account,
metadata,
) {
// 应用事件类型过滤
if let Some(filter) = event_type_filter {
if filter.include.contains(&event.metadata().event_type) {
return Some(event);
}
// 不匹配过滤器，继续尝试其他解析方式
} else {
return Some(event);
}
}
}
}
}
// 2. 尝试解析特殊账户类型（Token、Nonce等）
// 这些是通用的，不属于特定协议
let metadata = EventMetadata {
slot: account.slot,
signature: account.signature,
protocol: ProtocolType::Common,
event_type: EventType::default(),
program_id: account.owner,
recv_us: account.recv_us,
handle_us: elapsed_micros_since(account.recv_us),
..Default::default()
};
// 尝试解析 Nonce 账户
if let Some(event) = Self::parse_nonce_account_event(&account, metadata.clone()) {
if let Some(filter) = event_type_filter {
if filter.include.contains(&event.metadata().event_type) {
return Some(event);
}
} else {
return Some(event);
}
}
// 尝试解析 Token 账户
if let Some(event) = Self::parse_token_account_event(&account, metadata) {
if let Some(filter) = event_type_filter {
if filter.include.contains(&event.metadata().event_type) {
return Some(event);
}
} else {
return Some(event);
}
}
None
}
pub fn parse_token_account_event(
account: &AccountPretty,
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::TokenAccount;
let pubkey = account.pubkey;
let executable = account.executable;
let lamports = account.lamports;
@@ -169,8 +214,10 @@ impl AccountEventParser {
pub fn parse_nonce_account_event(
account: &AccountPretty,
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::NonceAccount;
if let Ok(info) = parse_nonce(&account.data) {
match info {
solana_account_decoder::parse_nonce::UiNonceState::Initialized(details) => {
src/streaming/event_parser/core/dispatcher.rs
+245
View File
@@ -0,0 +1,245 @@
//! 中心事件解析调度器
//!
//! 根据协议类型路由到对应的解析函数，替代原有的静态 CONFIGS 数组架构
//!
//! ## 设计原则
//! - **单一职责**: 每个函数只负责一件事（路由、解析、合并分离）
//! - **灵活性**: 调用方可以选择是否合并，或自定义合并逻辑
//! - **可测试性**: 每个函数都可以独立测试
use crate::streaming::event_parser::{
common::EventMetadata,
protocols::{
bonk::parser as bonk, pumpfun::parser as pumpfun, pumpswap::parser as pumpswap,
raydium_amm_v4::parser as raydium_amm_v4, raydium_clmm::parser as raydium_clmm,
raydium_cpmm::parser as raydium_cpmm,
},
DexEvent, Protocol,
};
use solana_sdk::pubkey::Pubkey;
/// 中心事件解析调度器
///
/// 负责将解析请求路由到对应协议的解析函数
pub struct EventDispatcher;
impl EventDispatcher {
/// 解析 instruction 事件（只解析，不合并）
///
/// # 参数
/// - `protocol`: 协议类型
/// - `instruction_discriminator`: 指令判别器 (8 bytes)
/// - `instruction_data`: 指令数据
/// - `accounts`: 账户公钥列表
/// - `metadata`: 事件元数据
///
/// # 返回
/// 解析成功返回 `Some(DexEvent)`，否则返回 `None`
#[inline]
pub fn dispatch_instruction(
protocol: Protocol,
instruction_discriminator: &[u8],
instruction_data: &[u8],
accounts: &[Pubkey],
mut metadata: EventMetadata,
) -> Option<DexEvent> {
// 根据协议类型设置 metadata.protocol
use crate::streaming::event_parser::common::ProtocolType;
metadata.protocol = match protocol {
Protocol::PumpFun => ProtocolType::PumpFun,
Protocol::PumpSwap => ProtocolType::PumpSwap,
Protocol::Bonk => ProtocolType::Bonk,
Protocol::RaydiumCpmm => ProtocolType::RaydiumCpmm,
Protocol::RaydiumClmm => ProtocolType::RaydiumClmm,
Protocol::RaydiumAmmV4 => ProtocolType::RaydiumAmmV4,
};
match protocol {
Protocol::PumpFun => pumpfun::parse_pumpfun_instruction_data(
instruction_discriminator,
instruction_data,
accounts,
metadata,
),
Protocol::PumpSwap => pumpswap::parse_pumpswap_instruction_data(
instruction_discriminator,
instruction_data,
accounts,
metadata,
),
Protocol::Bonk => bonk::parse_bonk_instruction_data(
instruction_discriminator,
instruction_data,
accounts,
metadata,
),
Protocol::RaydiumCpmm => raydium_cpmm::parse_raydium_cpmm_instruction_data(
instruction_discriminator,
instruction_data,
accounts,
metadata,
),
Protocol::RaydiumClmm => raydium_clmm::parse_raydium_clmm_instruction_data(
instruction_discriminator,
instruction_data,
accounts,
metadata,
),
Protocol::RaydiumAmmV4 => raydium_amm_v4::parse_raydium_amm_v4_instruction_data(
instruction_discriminator,
instruction_data,
accounts,
metadata,
),
}
}
/// 解析 inner instruction 事件（只解析，不合并）
///
/// # 参数
/// - `protocol`: 协议类型
/// - `inner_instruction_discriminator`: 内联指令判别器 (16 bytes)
/// - `inner_instruction_data`: 内联指令数据
/// - `metadata`: 事件元数据
///
/// # 返回
/// 解析成功返回 `Some(DexEvent)`，否则返回 `None`
#[inline]
pub fn dispatch_inner_instruction(
protocol: Protocol,
inner_instruction_discriminator: &[u8],
inner_instruction_data: &[u8],
mut metadata: EventMetadata,
) -> Option<DexEvent> {
// 根据协议类型设置 metadata.protocol
use crate::streaming::event_parser::common::ProtocolType;
metadata.protocol = match protocol {
Protocol::PumpFun => ProtocolType::PumpFun,
Protocol::PumpSwap => ProtocolType::PumpSwap,
Protocol::Bonk => ProtocolType::Bonk,
Protocol::RaydiumCpmm => ProtocolType::RaydiumCpmm,
Protocol::RaydiumClmm => ProtocolType::RaydiumClmm,
Protocol::RaydiumAmmV4 => ProtocolType::RaydiumAmmV4,
};
match protocol {
Protocol::PumpFun => pumpfun::parse_pumpfun_inner_instruction_data(
inner_instruction_discriminator,
inner_instruction_data,
metadata,
),
Protocol::PumpSwap => pumpswap::parse_pumpswap_inner_instruction_data(
inner_instruction_discriminator,
inner_instruction_data,
metadata,
),
Protocol::Bonk => bonk::parse_bonk_inner_instruction_data(
inner_instruction_discriminator,
inner_instruction_data,
metadata,
),
Protocol::RaydiumCpmm => raydium_cpmm::parse_raydium_cpmm_inner_instruction_data(
inner_instruction_discriminator,
inner_instruction_data,
metadata,
),
Protocol::RaydiumClmm => raydium_clmm::parse_raydium_clmm_inner_instruction_data(
inner_instruction_discriminator,
inner_instruction_data,
metadata,
),
Protocol::RaydiumAmmV4 => raydium_amm_v4::parse_raydium_amm_v4_inner_instruction_data(
inner_instruction_discriminator,
inner_instruction_data,
metadata,
),
}
}
/// 通过 program_id 匹配协议类型
#[inline]
pub fn match_protocol_by_program_id(program_id: &Pubkey) -> Option<Protocol> {
if program_id == &pumpfun::PUMPFUN_PROGRAM_ID {
Some(Protocol::PumpFun)
} else if program_id == &pumpswap::PUMPSWAP_PROGRAM_ID {
Some(Protocol::PumpSwap)
} else if program_id == &bonk::BONK_PROGRAM_ID {
Some(Protocol::Bonk)
} else if program_id == &raydium_cpmm::RAYDIUM_CPMM_PROGRAM_ID {
Some(Protocol::RaydiumCpmm)
} else if program_id == &raydium_clmm::RAYDIUM_CLMM_PROGRAM_ID {
Some(Protocol::RaydiumClmm)
} else if program_id == &raydium_amm_v4::RAYDIUM_AMM_V4_PROGRAM_ID {
Some(Protocol::RaydiumAmmV4)
} else {
None
}
}
/// 获取指定协议的 program_id
#[inline]
pub fn get_program_id(protocol: Protocol) -> Pubkey {
match protocol {
Protocol::PumpFun => pumpfun::PUMPFUN_PROGRAM_ID,
Protocol::PumpSwap => pumpswap::PUMPSWAP_PROGRAM_ID,
Protocol::Bonk => bonk::BONK_PROGRAM_ID,
Protocol::RaydiumCpmm => raydium_cpmm::RAYDIUM_CPMM_PROGRAM_ID,
Protocol::RaydiumClmm => raydium_clmm::RAYDIUM_CLMM_PROGRAM_ID,
Protocol::RaydiumAmmV4 => raydium_amm_v4::RAYDIUM_AMM_V4_PROGRAM_ID,
}
}
/// 批量获取 program_ids
pub fn get_program_ids(protocols: &[Protocol]) -> Vec<Pubkey> {
protocols.iter().map(|p| Self::get_program_id(p.clone())).collect()
}
/// 解析账户数据
///
/// 根据账户的 discriminator 路由到对应协议的账户解析函数
///
/// # 参数
/// - `protocol`: 协议类型
/// - `discriminator`: 账户判别器
/// - `account`: 账户信息
/// - `metadata`: 事件元数据
///
/// # 返回
/// 解析成功返回 `Some(DexEvent)`，否则返回 `None`
pub fn dispatch_account(
protocol: Protocol,
discriminator: &[u8],
account: &crate::streaming::grpc::AccountPretty,
mut metadata: crate::streaming::event_parser::common::EventMetadata,
) -> Option<DexEvent> {
// 根据协议类型设置 metadata.protocol
use crate::streaming::event_parser::common::ProtocolType;
metadata.protocol = match protocol {
Protocol::PumpFun => ProtocolType::PumpFun,
Protocol::PumpSwap => ProtocolType::PumpSwap,
Protocol::Bonk => ProtocolType::Bonk,
Protocol::RaydiumCpmm => ProtocolType::RaydiumCpmm,
Protocol::RaydiumClmm => ProtocolType::RaydiumClmm,
Protocol::RaydiumAmmV4 => ProtocolType::RaydiumAmmV4,
};
match protocol {
Protocol::PumpFun => {
pumpfun::parse_pumpfun_account_data(discriminator, account, metadata)
}
Protocol::PumpSwap => {
pumpswap::parse_pumpswap_account_data(discriminator, account, metadata)
}
Protocol::Bonk => bonk::parse_bonk_account_data(discriminator, account, metadata),
Protocol::RaydiumCpmm => {
raydium_cpmm::parse_raydium_cpmm_account_data(discriminator, account, metadata)
}
Protocol::RaydiumClmm => {
raydium_clmm::parse_raydium_clmm_account_data(discriminator, account, metadata)
}
Protocol::RaydiumAmmV4 => {
raydium_amm_v4::parse_raydium_amm_v4_account_data(discriminator, account, metadata)
}
}
}
}
src/streaming/event_parser/core/event_parser.rs
+489 -769
View File
File diff suppressed because it is too large Load Diff
src/streaming/event_parser/core/mod.rs
+3 -4
View File
@@ -1,13 +1,12 @@
pub mod account_event_parser;
pub mod common_event_parser;
pub mod dispatcher;
pub mod global_state;
pub mod parser_cache;
pub mod traits;
pub use traits::DexEvent;
pub use parser_cache::{
AccountEventParseConfig, AccountEventParserFn, GenericEventParseConfig,
InnerInstructionEventParser, InstructionEventParser, get_account_configs,
};
pub use dispatcher::EventDispatcher;
pub mod event_parser;
pub mod merger_event;
src/streaming/event_parser/core/parser_cache.rs
+9 -406
View File
@@ -1,26 +1,18 @@
//! # 事件解析器配置缓存模块
//!
//! 本模块统一管理所有事件解析器的配置和缓存，包括：
//! - 指令事件解析器（Instruction Event Parser）
//! 本模块管理事件解析器的缓存，包括：
//! - 程序ID缓存
//! - 账户事件解析器（Account Event Parser）
//! - 高性能缓存工具
//!
//! ## 设计目标
//! - **统一配置管理**：所有协议的解析器配置集中管理
//! - **高性能缓存**：避免重复初始化和内存分配
//! - **易于扩展**：添加新协议只需修改配置映射
//! - **易于扩展**：通过 dispatcher 动态派发
use crate::streaming::{
event_parser::{
common::{filter::EventTypeFilter, EventMetadata, EventType, ProtocolType},
protocols::{
bonk::parser::BONK_PROGRAM_ID,
pumpfun::parser::PUMPFUN_PROGRAM_ID,
pumpswap::parser::PUMPSWAP_PROGRAM_ID,
raydium_amm_v4::parser::RAYDIUM_AMM_V4_PROGRAM_ID,
raydium_clmm::parser::RAYDIUM_CLMM_PROGRAM_ID,
raydium_cpmm::parser::RAYDIUM_CPMM_PROGRAM_ID,
},
core::dispatcher::EventDispatcher,
Protocol, DexEvent,
},
grpc::AccountPretty,
@@ -28,106 +20,14 @@ use crate::streaming::{
use solana_sdk::pubkey::Pubkey;
use std::{
collections::HashMap,
sync::{Arc, LazyLock, OnceLock},
sync::{Arc, LazyLock},
};
// ============================================================================
// 第一部分：指令事件解析器配置（Instruction Event Parser）
// 第一部分：程序ID缓存
// ============================================================================
/// 内联指令事件解析器函数类型
///
/// 用于解析内联指令（Inner Instruction）生成的事件
pub type InnerInstructionEventParser =
fn(data: &[u8], metadata: EventMetadata) -> Option<DexEvent>;
/// 指令事件解析器函数类型
///
/// 用于解析普通指令（Instruction）生成的事件，需要账户信息
pub type InstructionEventParser =
fn(data: &[u8], accounts: &[Pubkey], metadata: EventMetadata) -> Option<DexEvent>;
/// 指令事件解析器配置
///
/// 定义如何解析特定协议的指令事件
#[derive(Debug, Clone)]
pub struct GenericEventParseConfig {
/// 程序ID（Program ID）
pub program_id: Pubkey,
/// 协议类型
pub protocol_type: ProtocolType,
/// 内联指令判别器（8字节）
pub inner_instruction_discriminator: &'static [u8],
/// 普通指令判别器（8字节）
pub instruction_discriminator: &'static [u8],
/// 事件类型
pub event_type: EventType,
/// 内联指令解析器
pub inner_instruction_parser: Option<InnerInstructionEventParser>,
/// 普通指令解析器
pub instruction_parser: Option<InstructionEventParser>,
/// 是否需要内联指令数据
pub requires_inner_instruction: bool,
}
/// 全局指令事件解析器注册表
///
/// 存储所有协议的指令解析器配置，启动时初始化一次
pub static EVENT_PARSERS: LazyLock<HashMap<Protocol, (Pubkey, &[GenericEventParseConfig])>> =
LazyLock::new(|| {
let mut parsers = HashMap::with_capacity(6);
// 注册各协议的解析器配置
parsers.insert(
Protocol::PumpSwap,
(
PUMPSWAP_PROGRAM_ID,
crate::streaming::event_parser::protocols::pumpswap::parser::CONFIGS,
),
);
parsers.insert(
Protocol::PumpFun,
(
PUMPFUN_PROGRAM_ID,
crate::streaming::event_parser::protocols::pumpfun::parser::CONFIGS,
),
);
parsers.insert(
Protocol::Bonk,
(BONK_PROGRAM_ID, crate::streaming::event_parser::protocols::bonk::parser::CONFIGS),
);
parsers.insert(
Protocol::RaydiumCpmm,
(
RAYDIUM_CPMM_PROGRAM_ID,
crate::streaming::event_parser::protocols::raydium_cpmm::parser::CONFIGS,
),
);
parsers.insert(
Protocol::RaydiumClmm,
(
RAYDIUM_CLMM_PROGRAM_ID,
crate::streaming::event_parser::protocols::raydium_clmm::parser::CONFIGS,
),
);
parsers.insert(
Protocol::RaydiumAmmV4,
(
RAYDIUM_AMM_V4_PROGRAM_ID,
crate::streaming::event_parser::protocols::raydium_amm_v4::parser::CONFIGS,
),
);
parsers
});
// ----------------------------------------------------------------------------
// 指令解析器缓存工具
// ----------------------------------------------------------------------------
/// 缓存键：用于标识不同的协议和过滤器组合
///
/// 通过对协议和事件类型排序，确保相同组合生成相同的哈希键
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct CacheKey {
/// 协议列表（已排序）
@@ -138,13 +38,6 @@ pub struct CacheKey {
impl CacheKey {
/// 创建新的缓存键
///
/// # 参数
/// - `protocols`: 协议列表
/// - `filter`: 事件类型过滤器
///
/// # 优化
/// 使用 `sort_by_cached_key` 避免重复调用 `format!`
pub fn new(mut protocols: Vec<Protocol>, filter: Option<&EventTypeFilter>) -> Self {
// 排序协议列表，确保相同协议组合生成相同的key
protocols.sort_by_cached_key(|p| format!("{:?}", p));
@@ -164,23 +57,9 @@ static GLOBAL_PROGRAM_IDS_CACHE: LazyLock<
parking_lot::RwLock<HashMap<CacheKey, Arc<Vec<Pubkey>>>>,
> = LazyLock::new(|| parking_lot::RwLock::new(HashMap::new()));
/// 全局指令配置缓存（使用读写锁保护）
static GLOBAL_INSTRUCTION_CONFIGS_CACHE: LazyLock<
parking_lot::RwLock<HashMap<CacheKey, Arc<HashMap<Vec<u8>, Vec<GenericEventParseConfig>>>>>,
> = LazyLock::new(|| parking_lot::RwLock::new(HashMap::new()));
/// 获取指定协议的程序ID列表
///
/// # 参数
/// - `protocols`: 协议列表
/// - `filter`: 事件类型过滤器（可选）
///
/// # 返回
/// Arc包装的程序ID向量，支持零成本共享
///
/// # 缓存策略
/// - 快速路径：从缓存读取（读锁）
/// - 慢速路径：构建并缓存（写锁）
/// 使用 EventDispatcher 获取 program_ids，并缓存结果
pub fn get_global_program_ids(
protocols: &[Protocol],
filter: Option<&EventTypeFilter>,
@@ -195,14 +74,8 @@ pub fn get_global_program_ids(
}
}
// 慢速路径：构建并缓存
let mut program_ids = Vec::with_capacity(protocols.len());
for protocol in protocols {
if let Some(parse) = EVENT_PARSERS.get(protocol) {
program_ids.push(parse.0);
}
}
let program_ids = Arc::new(program_ids);
// 慢速路径：通过 EventDispatcher 获取并缓存
let program_ids = Arc::new(EventDispatcher::get_program_ids(protocols));
// 缓存结果（写锁）
GLOBAL_PROGRAM_IDS_CACHE.write().insert(cache_key, program_ids.clone());
@@ -210,58 +83,6 @@ pub fn get_global_program_ids(
program_ids
}
/// 获取指定协议的指令配置
///
/// # 参数
/// - `protocols`: 协议列表
/// - `filter`: 事件类型过滤器（可选）
///
/// # 返回
/// Arc包装的指令配置映射表，按判别器（Discriminator）索引
///
/// # 缓存策略
/// - 快速路径：从缓存读取（读锁）
/// - 慢速路径：构建并缓存（写锁）
pub fn get_global_instruction_configs(
protocols: &[Protocol],
filter: Option<&EventTypeFilter>,
) -> Arc<HashMap<Vec<u8>, Vec<GenericEventParseConfig>>> {
let cache_key = CacheKey::new(protocols.to_vec(), filter);
// 快速路径：尝试读取缓存
{
let cache = GLOBAL_INSTRUCTION_CONFIGS_CACHE.read();
if let Some(configs) = cache.get(&cache_key) {
return configs.clone();
}
}
// 慢速路径：构建并缓存
let mut instruction_configs = HashMap::with_capacity(protocols.len() * 4);
for protocol in protocols {
if let Some(parse) = EVENT_PARSERS.get(protocol) {
parse
.1
.iter()
.filter(|config| {
filter.as_ref().map(|f| f.include.contains(&config.event_type)).unwrap_or(true)
})
.for_each(|config| {
instruction_configs
.entry(config.instruction_discriminator.to_vec())
.or_insert_with(Vec::new)
.push(config.clone());
});
}
}
let instruction_configs = Arc::new(instruction_configs);
// 缓存结果（写锁）
GLOBAL_INSTRUCTION_CONFIGS_CACHE.write().insert(cache_key, instruction_configs.clone());
instruction_configs
}
// ============================================================================
// 第二部分：账户公钥缓存工具（Account Pubkey Cache）
// ============================================================================
@@ -381,221 +202,3 @@ pub struct AccountEventParseConfig {
pub account_parser: AccountEventParserFn,
}
/// 协议账户配置缓存
///
/// 存储各协议的账户解析器配置
static PROTOCOL_CONFIGS_CACHE: OnceLock<HashMap<Protocol, Vec<AccountEventParseConfig>>> =
OnceLock::new();
/// Nonce 账户配置缓存
static NONCE_CONFIG: OnceLock<AccountEventParseConfig> = OnceLock::new();
/// 通用 Token 账户配置缓存
static COMMON_CONFIG: OnceLock<AccountEventParseConfig> = OnceLock::new();
/// 获取账户解析器配置列表
///
/// # 参数
/// - `protocols`: 协议列表
/// - `event_type_filter`: 事件类型过滤器（可选）
/// - `nonce_parser`: Nonce账户解析器
/// - `token_parser`: Token账户解析器
///
/// # 返回
/// 账户解析器配置向量
///
/// # 配置组成
/// 1. 协议特定配置（根据protocols参数）
/// 2. Nonce账户配置（通用）
/// 3. Token账户配置（通用，作为兜底）
///
/// # 示例
/// ```ignore
/// let configs = get_account_configs(
/// &[Protocol::PumpFun, Protocol::Bonk],
/// None,
/// parse_nonce_account,
/// parse_token_account,
/// );
/// ```
pub fn get_account_configs(
protocols: &[Protocol],
event_type_filter: Option<&EventTypeFilter>,
nonce_parser: AccountEventParserFn,
token_parser: AccountEventParserFn,
) -> Vec<AccountEventParseConfig> {
// 初始化协议配置缓存（仅在首次调用时执行）
let protocols_map = PROTOCOL_CONFIGS_CACHE.get_or_init(|| {
let mut map = HashMap::new();
// PumpSwap 协议
map.insert(Protocol::PumpSwap, vec![
AccountEventParseConfig {
program_id: PUMPSWAP_PROGRAM_ID,
protocol_type: ProtocolType::PumpSwap,
event_type: EventType::AccountPumpSwapGlobalConfig,
account_discriminator: crate::streaming::event_parser::protocols::pumpswap::discriminators::GLOBAL_CONFIG_ACCOUNT,
account_parser: crate::streaming::event_parser::protocols::pumpswap::types::global_config_parser,
},
AccountEventParseConfig {
program_id: PUMPSWAP_PROGRAM_ID,
protocol_type: ProtocolType::PumpSwap,
event_type: EventType::AccountPumpSwapPool,
account_discriminator: crate::streaming::event_parser::protocols::pumpswap::discriminators::POOL_ACCOUNT,
account_parser: crate::streaming::event_parser::protocols::pumpswap::types::pool_parser,
},
]);
// PumpFun 协议
map.insert(Protocol::PumpFun, vec![
AccountEventParseConfig {
program_id: PUMPFUN_PROGRAM_ID,
protocol_type: ProtocolType::PumpFun,
event_type: EventType::AccountPumpFunBondingCurve,
account_discriminator: crate::streaming::event_parser::protocols::pumpfun::discriminators::BONDING_CURVE_ACCOUNT,
account_parser: crate::streaming::event_parser::protocols::pumpfun::types::bonding_curve_parser,
},
AccountEventParseConfig {
program_id: PUMPFUN_PROGRAM_ID,
protocol_type: ProtocolType::PumpFun,
event_type: EventType::AccountPumpFunGlobal,
account_discriminator: crate::streaming::event_parser::protocols::pumpfun::discriminators::GLOBAL_ACCOUNT,
account_parser: crate::streaming::event_parser::protocols::pumpfun::types::global_parser,
},
]);
// Bonk 协议
map.insert(Protocol::Bonk, vec![
AccountEventParseConfig {
program_id: BONK_PROGRAM_ID,
protocol_type: ProtocolType::Bonk,
event_type: EventType::AccountBonkPoolState,
account_discriminator: crate::streaming::event_parser::protocols::bonk::discriminators::POOL_STATE_ACCOUNT,
account_parser: crate::streaming::event_parser::protocols::bonk::types::pool_state_parser,
},
AccountEventParseConfig {
program_id: BONK_PROGRAM_ID,
protocol_type: ProtocolType::Bonk,
event_type: EventType::AccountBonkGlobalConfig,
account_discriminator: crate::streaming::event_parser::protocols::bonk::discriminators::GLOBAL_CONFIG_ACCOUNT,
account_parser: crate::streaming::event_parser::protocols::bonk::types::global_config_parser,
},
AccountEventParseConfig {
program_id: BONK_PROGRAM_ID,
protocol_type: ProtocolType::Bonk,
event_type: EventType::AccountBonkPlatformConfig,
account_discriminator: crate::streaming::event_parser::protocols::bonk::discriminators::PLATFORM_CONFIG_ACCOUNT,
account_parser: crate::streaming::event_parser::protocols::bonk::types::platform_config_parser,
},
]);
// Raydium CPMM 协议
map.insert(Protocol::RaydiumCpmm, vec![
AccountEventParseConfig {
program_id: RAYDIUM_CPMM_PROGRAM_ID,
protocol_type: ProtocolType::RaydiumCpmm,
event_type: EventType::AccountRaydiumCpmmAmmConfig,
account_discriminator: crate::streaming::event_parser::protocols::raydium_cpmm::discriminators::AMM_CONFIG,
account_parser: crate::streaming::event_parser::protocols::raydium_cpmm::types::amm_config_parser,
},
AccountEventParseConfig {
program_id: RAYDIUM_CPMM_PROGRAM_ID,
protocol_type: ProtocolType::RaydiumCpmm,
event_type: EventType::AccountRaydiumCpmmPoolState,
account_discriminator: crate::streaming::event_parser::protocols::raydium_cpmm::discriminators::POOL_STATE,
account_parser: crate::streaming::event_parser::protocols::raydium_cpmm::types::pool_state_parser,
},
]);
// Raydium CLMM 协议
map.insert(Protocol::RaydiumClmm, vec![
AccountEventParseConfig {
program_id: RAYDIUM_CLMM_PROGRAM_ID,
protocol_type: ProtocolType::RaydiumClmm,
event_type: EventType::AccountRaydiumClmmAmmConfig,
account_discriminator: crate::streaming::event_parser::protocols::raydium_clmm::discriminators::AMM_CONFIG,
account_parser: crate::streaming::event_parser::protocols::raydium_clmm::types::amm_config_parser,
},
AccountEventParseConfig {
program_id: RAYDIUM_CLMM_PROGRAM_ID,
protocol_type: ProtocolType::RaydiumClmm,
event_type: EventType::AccountRaydiumClmmPoolState,
account_discriminator: crate::streaming::event_parser::protocols::raydium_clmm::discriminators::POOL_STATE,
account_parser: crate::streaming::event_parser::protocols::raydium_clmm::types::pool_state_parser,
},
AccountEventParseConfig {
program_id: RAYDIUM_CLMM_PROGRAM_ID,
protocol_type: ProtocolType::RaydiumClmm,
event_type: EventType::AccountRaydiumClmmTickArrayState,
account_discriminator: crate::streaming::event_parser::protocols::raydium_clmm::discriminators::TICK_ARRAY_STATE,
account_parser: crate::streaming::event_parser::protocols::raydium_clmm::types::tick_array_state_parser,
},
]);
// Raydium AMM V4 协议
map.insert(Protocol::RaydiumAmmV4, vec![
AccountEventParseConfig {
program_id: RAYDIUM_AMM_V4_PROGRAM_ID,
protocol_type: ProtocolType::RaydiumAmmV4,
event_type: EventType::AccountRaydiumAmmV4AmmInfo,
account_discriminator: crate::streaming::event_parser::protocols::raydium_amm_v4::discriminators::AMM_INFO,
account_parser: crate::streaming::event_parser::protocols::raydium_amm_v4::types::amm_info_parser,
},
]);
map
});
// 构建配置列表
let mut configs = Vec::new();
let empty_vec = Vec::new();
// 预估容量（大多数协议有2-3个配置）
let estimated_capacity = protocols.len() * 3;
configs.reserve(estimated_capacity);
// 1. 添加协议特定配置
for protocol in protocols {
let protocol_configs = protocols_map.get(protocol).unwrap_or(&empty_vec);
if event_type_filter.is_none() {
// 无过滤器：添加所有配置
configs.extend(protocol_configs.iter().cloned());
} else {
// 有过滤器：仅添加匹配的配置
let filter = event_type_filter.unwrap();
configs.extend(
protocol_configs
.iter()
.filter(|config| filter.include.contains(&config.event_type))
.cloned(),
);
}
}
// 2. 添加 Nonce 账户配置（通用配置）
if event_type_filter.is_none()
|| event_type_filter.unwrap().include.contains(&EventType::NonceAccount)
{
let nonce_config = NONCE_CONFIG.get_or_init(|| AccountEventParseConfig {
program_id: Pubkey::default(),
protocol_type: ProtocolType::Common,
event_type: EventType::NonceAccount,
account_discriminator: &[1, 0, 0, 0, 1, 0, 0, 0],
account_parser: nonce_parser,
});
configs.push(nonce_config.clone());
}
// 3. 添加通用 Token 账户配置（兜底配置，始终添加）
let common_config = COMMON_CONFIG.get_or_init(|| AccountEventParseConfig {
program_id: Pubkey::default(),
protocol_type: ProtocolType::Common,
event_type: EventType::TokenAccount,
account_discriminator: &[],
account_parser: token_parser,
});
configs.push(common_config.clone());
configs
}
src/streaming/event_parser/protocols/bonk/parser.rs
+112 -104
View File
@@ -1,8 +1,7 @@
use solana_sdk::pubkey::Pubkey;
use crate::streaming::event_parser::{
common::{utils::*, EventMetadata, EventType, ProtocolType},
core::GenericEventParseConfig,
common::{utils::*, EventMetadata, EventType},
protocols::bonk::{
bonk_pool_create_event_log_decode, bonk_trade_event_log_decode, discriminators, AmmFeeOn,
BonkMigrateToAmmEvent, BonkMigrateToCpswapEvent, BonkPoolCreateEvent, BonkTradeEvent,
@@ -16,104 +15,95 @@ use crate::streaming::event_parser::{
pub const BONK_PROGRAM_ID: Pubkey =
solana_sdk::pubkey!("LanMV9sAd7wArD4vJFi2qDdfnVhFxYSUg6eADduJ3uj");
// Configure all event types
pub const CONFIGS: &[GenericEventParseConfig] = &[
GenericEventParseConfig {
program_id: BONK_PROGRAM_ID,
protocol_type: ProtocolType::Bonk,
inner_instruction_discriminator: discriminators::TRADE_EVENT,
instruction_discriminator: discriminators::BUY_EXACT_IN,
event_type: EventType::BonkBuyExactIn,
inner_instruction_parser: Some(parse_trade_inner_instruction),
instruction_parser: Some(parse_buy_exact_in_instruction),
requires_inner_instruction: false,
},
GenericEventParseConfig {
program_id: BONK_PROGRAM_ID,
protocol_type: ProtocolType::Bonk,
inner_instruction_discriminator: discriminators::TRADE_EVENT,
instruction_discriminator: discriminators::BUY_EXACT_OUT,
event_type: EventType::BonkBuyExactOut,
inner_instruction_parser: Some(parse_trade_inner_instruction),
instruction_parser: Some(parse_buy_exact_out_instruction),
requires_inner_instruction: false,
},
GenericEventParseConfig {
program_id: BONK_PROGRAM_ID,
protocol_type: ProtocolType::Bonk,
inner_instruction_discriminator: discriminators::TRADE_EVENT,
instruction_discriminator: discriminators::SELL_EXACT_IN,
event_type: EventType::BonkSellExactIn,
inner_instruction_parser: Some(parse_trade_inner_instruction),
instruction_parser: Some(parse_sell_exact_in_instruction),
requires_inner_instruction: false,
},
GenericEventParseConfig {
program_id: BONK_PROGRAM_ID,
protocol_type: ProtocolType::Bonk,
inner_instruction_discriminator: discriminators::TRADE_EVENT,
instruction_discriminator: discriminators::SELL_EXACT_OUT,
event_type: EventType::BonkSellExactOut,
inner_instruction_parser: Some(parse_trade_inner_instruction),
instruction_parser: Some(parse_sell_exact_out_instruction),
requires_inner_instruction: false,
},
GenericEventParseConfig {
program_id: BONK_PROGRAM_ID,
protocol_type: ProtocolType::Bonk,
inner_instruction_discriminator: discriminators::POOL_CREATE_EVENT,
instruction_discriminator: discriminators::INITIALIZE,
event_type: EventType::BonkInitialize,
inner_instruction_parser: Some(parse_pool_create_inner_instruction),
instruction_parser: Some(parse_initialize_instruction),
requires_inner_instruction: false,
},
GenericEventParseConfig {
program_id: BONK_PROGRAM_ID,
protocol_type: ProtocolType::Bonk,
inner_instruction_discriminator: discriminators::POOL_CREATE_EVENT,
instruction_discriminator: discriminators::INITIALIZE_V2,
event_type: EventType::BonkInitializeV2,
inner_instruction_parser: Some(parse_pool_create_inner_instruction),
instruction_parser: Some(parse_initialize_v2_instruction),
requires_inner_instruction: false,
},
GenericEventParseConfig {
program_id: BONK_PROGRAM_ID,
protocol_type: ProtocolType::Bonk,
inner_instruction_discriminator: discriminators::POOL_CREATE_EVENT,
instruction_discriminator: discriminators::INITIALIZE_WITH_TOKEN_2022,
event_type: EventType::BonkInitializeWithToken2022,
inner_instruction_parser: Some(parse_pool_create_inner_instruction),
instruction_parser: Some(parse_initialize_with_token_2022_instruction),
requires_inner_instruction: false,
},
GenericEventParseConfig {
program_id: BONK_PROGRAM_ID,
protocol_type: ProtocolType::Bonk,
inner_instruction_discriminator: &[],
instruction_discriminator: discriminators::MIGRATE_TO_AMM,
event_type: EventType::BonkMigrateToAmm,
inner_instruction_parser: None,
instruction_parser: Some(parse_migrate_to_amm_instruction),
requires_inner_instruction: false,
},
GenericEventParseConfig {
program_id: BONK_PROGRAM_ID,
protocol_type: ProtocolType::Bonk,
inner_instruction_discriminator: &[],
instruction_discriminator: discriminators::MIGRATE_TO_CP_SWAP,
event_type: EventType::BonkMigrateToCpswap,
inner_instruction_parser: None,
instruction_parser: Some(parse_migrate_to_cpswap_instruction),
requires_inner_instruction: false,
},
];
/// 解析 Bonk instruction data
///
/// 根据判别器路由到具体的 instruction 解析函数
pub fn parse_bonk_instruction_data(
discriminator: &[u8],
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
) -> Option<DexEvent> {
match discriminator {
discriminators::BUY_EXACT_IN => {
parse_buy_exact_in_instruction(data, accounts, metadata)
}
discriminators::BUY_EXACT_OUT => {
parse_buy_exact_out_instruction(data, accounts, metadata)
}
discriminators::SELL_EXACT_IN => {
parse_sell_exact_in_instruction(data, accounts, metadata)
}
discriminators::SELL_EXACT_OUT => {
parse_sell_exact_out_instruction(data, accounts, metadata)
}
discriminators::INITIALIZE => {
parse_initialize_instruction(data, accounts, metadata)
}
discriminators::INITIALIZE_V2 => {
parse_initialize_v2_instruction(data, accounts, metadata)
}
discriminators::INITIALIZE_WITH_TOKEN_2022 => {
parse_initialize_with_token_2022_instruction(data, accounts, metadata)
}
discriminators::MIGRATE_TO_AMM => {
parse_migrate_to_amm_instruction(data, accounts, metadata)
}
discriminators::MIGRATE_TO_CP_SWAP => {
parse_migrate_to_cpswap_instruction(data, accounts, metadata)
}
_ => None,
}
}
/// 解析 Bonk inner instruction data
///
/// 根据判别器路由到具体的 inner instruction 解析函数
pub fn parse_bonk_inner_instruction_data(
discriminator: &[u8],
data: &[u8],
metadata: EventMetadata,
) -> Option<DexEvent> {
match discriminator {
discriminators::TRADE_EVENT => {
parse_trade_inner_instruction(data, metadata)
}
discriminators::POOL_CREATE_EVENT => {
parse_pool_create_inner_instruction(data, metadata)
}
_ => None,
}
}
/// 解析 Bonk 账户数据
///
/// 根据判别器路由到具体的账户解析函数
pub fn parse_bonk_account_data(
discriminator: &[u8],
account: &crate::streaming::grpc::AccountPretty,
metadata: crate::streaming::event_parser::common::EventMetadata,
) -> Option<crate::streaming::event_parser::DexEvent> {
match discriminator {
discriminators::POOL_STATE_ACCOUNT => {
crate::streaming::event_parser::protocols::bonk::types::pool_state_parser(account, metadata)
}
discriminators::GLOBAL_CONFIG_ACCOUNT => {
crate::streaming::event_parser::protocols::bonk::types::global_config_parser(account, metadata)
}
discriminators::PLATFORM_CONFIG_ACCOUNT => {
crate::streaming::event_parser::protocols::bonk::types::platform_config_parser(account, metadata)
}
_ => None,
}
}
/// Parse pool creation event
fn parse_pool_create_inner_instruction(
data: &[u8],
metadata: EventMetadata,
) -> Option<DexEvent> {
// Note: event_type will be set by the instruction parser, not here
// Because different initialize instructions have different event types
if let Some(event) = bonk_pool_create_event_log_decode(data) {
Some(DexEvent::BonkPoolCreateEvent(BonkPoolCreateEvent { metadata, ..event }))
} else {
@@ -146,8 +136,10 @@ fn parse_trade_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option
fn parse_buy_exact_in_instruction(
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::BonkBuyExactIn;
if data.len() < 16 || accounts.len() < 18 {
return None;
}
@@ -184,8 +176,10 @@ fn parse_buy_exact_in_instruction(
fn parse_buy_exact_out_instruction(
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::BonkBuyExactOut;
if data.len() < 16 || accounts.len() < 18 {
return None;
}
@@ -222,8 +216,10 @@ fn parse_buy_exact_out_instruction(
fn parse_sell_exact_in_instruction(
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::BonkSellExactIn;
if data.len() < 16 || accounts.len() < 18 {
return None;
}
@@ -260,8 +256,10 @@ fn parse_sell_exact_in_instruction(
fn parse_sell_exact_out_instruction(
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::BonkSellExactOut;
if data.len() < 16 || accounts.len() < 18 {
return None;
}
@@ -299,8 +297,10 @@ fn parse_sell_exact_out_instruction(
fn parse_initialize_instruction(
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::BonkInitialize;
if data.len() < 24 {
return None;
}
@@ -332,8 +332,10 @@ fn parse_initialize_instruction(
fn parse_initialize_v2_instruction(
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::BonkInitializeV2;
if data.len() < 24 {
return None;
}
@@ -371,8 +373,10 @@ fn parse_initialize_v2_instruction(
fn parse_initialize_with_token_2022_instruction(
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::BonkInitializeWithToken2022;
if data.len() < 24 {
return None;
}
@@ -515,8 +519,10 @@ fn parse_vesting_params(data: &[u8], offset: &mut usize) -> Option<VestingParams
fn parse_migrate_to_amm_instruction(
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::BonkMigrateToAmm;
if data.len() < 16 {
return None;
}
@@ -570,8 +576,10 @@ fn parse_migrate_to_amm_instruction(
fn parse_migrate_to_cpswap_instruction(
_data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::BonkMigrateToCpswap;
Some(DexEvent::BonkMigrateToCpswapEvent(BonkMigrateToCpswapEvent {
metadata,
payer: accounts[0],
src/streaming/event_parser/protocols/bonk/types.rs
+10 -4
View File
@@ -4,7 +4,7 @@ use solana_sdk::pubkey::Pubkey;
use crate::streaming::{
event_parser::{
common::EventMetadata,
common::{EventMetadata, EventType},
protocols::bonk::{
BonkGlobalConfigAccountEvent, BonkPlatformConfigAccountEvent, BonkPoolStateAccountEvent,
},
@@ -132,7 +132,9 @@ pub fn pool_state_decode(data: &[u8]) -> Option<PoolState> {
borsh::from_slice::<PoolState>(&data[..POOL_STATE_SIZE]).ok()
}
pub fn pool_state_parser(account: &AccountPretty, metadata: EventMetadata) -> Option<DexEvent> {
pub fn pool_state_parser(account: &AccountPretty, mut metadata: EventMetadata) -> Option<DexEvent> {
metadata.event_type = EventType::AccountBonkPoolState;
if account.data.len() < POOL_STATE_SIZE + 8 {
return None;
}
@@ -182,8 +184,10 @@ pub fn global_config_decode(data: &[u8]) -> Option<GlobalConfig> {
pub fn global_config_parser(
account: &AccountPretty,
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::AccountBonkGlobalConfig;
if account.data.len() < GLOBAL_CONFIG_SIZE + 8 {
return None;
}
@@ -228,8 +232,10 @@ pub fn platform_config_decode(data: &[u8]) -> Option<PlatformConfig> {
pub fn platform_config_parser(
account: &AccountPretty,
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::AccountBonkPlatformConfig;
if account.data.len() < PLATFORM_CONFIG_SIZE + 8 {
return None;
}
src/streaming/event_parser/protocols/pumpfun/parser.rs
+88 -53
View File
@@ -1,6 +1,5 @@
use crate::streaming::event_parser::{
common::{EventMetadata, EventType, ProtocolType},
core::GenericEventParseConfig,
common::{EventMetadata, EventType},
protocols::pumpfun::{
discriminators, pumpfun_create_token_event_log_decode, pumpfun_migrate_event_log_decode,
pumpfun_trade_event_log_decode, PumpFunCreateTokenEvent, PumpFunMigrateEvent,
@@ -14,53 +13,78 @@ use solana_sdk::pubkey::Pubkey;
pub const PUMPFUN_PROGRAM_ID: Pubkey =
solana_sdk::pubkey!("6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P");
// 配置所有事件类型
pub const CONFIGS: &[GenericEventParseConfig] = &[
GenericEventParseConfig {
program_id: PUMPFUN_PROGRAM_ID,
protocol_type: ProtocolType::PumpFun,
inner_instruction_discriminator: discriminators::CREATE_TOKEN_EVENT,
instruction_discriminator: discriminators::CREATE_TOKEN_IX,
event_type: EventType::PumpFunCreateToken,
inner_instruction_parser: Some(parse_create_token_inner_instruction),
instruction_parser: Some(parse_create_token_instruction),
requires_inner_instruction: false,
},
GenericEventParseConfig {
program_id: PUMPFUN_PROGRAM_ID,
protocol_type: ProtocolType::PumpFun,
inner_instruction_discriminator: discriminators::TRADE_EVENT,
instruction_discriminator: discriminators::BUY_IX,
event_type: EventType::PumpFunBuy,
inner_instruction_parser: Some(parse_trade_inner_instruction),
instruction_parser: Some(parse_buy_instruction),
requires_inner_instruction: false,
},
GenericEventParseConfig {
program_id: PUMPFUN_PROGRAM_ID,
protocol_type: ProtocolType::PumpFun,
inner_instruction_discriminator: discriminators::TRADE_EVENT,
instruction_discriminator: discriminators::SELL_IX,
event_type: EventType::PumpFunSell,
inner_instruction_parser: Some(parse_trade_inner_instruction),
instruction_parser: Some(parse_sell_instruction),
requires_inner_instruction: false,
},
GenericEventParseConfig {
program_id: PUMPFUN_PROGRAM_ID,
protocol_type: ProtocolType::PumpFun,
inner_instruction_discriminator: discriminators::COMPLETE_PUMP_AMM_MIGRATION_EVENT,
instruction_discriminator: discriminators::MIGRATE_IX,
event_type: EventType::PumpFunMigrate,
inner_instruction_parser: Some(parse_migrate_inner_instruction),
instruction_parser: Some(parse_migrate_instruction),
// Failed migrations lack inner instruction data (typically "Bonding curve already migrated")
requires_inner_instruction: true,
},
];
/// 解析 PumpFun instruction data
///
/// 根据判别器路由到具体的 instruction 解析函数
pub fn parse_pumpfun_instruction_data(
discriminator: &[u8],
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
) -> Option<DexEvent> {
match discriminator {
discriminators::CREATE_TOKEN_IX => {
parse_create_token_instruction(data, accounts, metadata)
}
discriminators::BUY_IX => {
parse_buy_instruction(data, accounts, metadata)
}
discriminators::SELL_IX => {
parse_sell_instruction(data, accounts, metadata)
}
discriminators::MIGRATE_IX => {
parse_migrate_instruction(data, accounts, metadata)
}
_ => None,
}
}
/// 解析 PumpFun inner instruction data
///
/// 根据判别器路由到具体的 inner instruction 解析函数
pub fn parse_pumpfun_inner_instruction_data(
discriminator: &[u8],
data: &[u8],
metadata: EventMetadata,
) -> Option<DexEvent> {
match discriminator {
discriminators::CREATE_TOKEN_EVENT => {
parse_create_token_inner_instruction(data, metadata)
}
discriminators::TRADE_EVENT => {
parse_trade_inner_instruction(data, metadata)
}
discriminators::COMPLETE_PUMP_AMM_MIGRATION_EVENT => {
parse_migrate_inner_instruction(data, metadata)
}
_ => None,
}
}
/// 解析 PumpFun 账户数据
///
/// 根据判别器路由到具体的账户解析函数
pub fn parse_pumpfun_account_data(
discriminator: &[u8],
account: &crate::streaming::grpc::AccountPretty,
metadata: crate::streaming::event_parser::common::EventMetadata,
) -> Option<crate::streaming::event_parser::DexEvent> {
match discriminator {
discriminators::BONDING_CURVE_ACCOUNT => {
crate::streaming::event_parser::protocols::pumpfun::types::bonding_curve_parser(account, metadata)
}
discriminators::GLOBAL_ACCOUNT => {
crate::streaming::event_parser::protocols::pumpfun::types::global_parser(account, metadata)
}
_ => None,
}
}
/// 解析迁移事件
fn parse_migrate_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option<DexEvent> {
fn parse_migrate_inner_instruction(data: &[u8], mut metadata: EventMetadata) -> Option<DexEvent> {
metadata.event_type = EventType::PumpFunMigrate;
if let Some(event) = pumpfun_migrate_event_log_decode(data) {
Some(DexEvent::PumpFunMigrateEvent(PumpFunMigrateEvent { metadata, ..event }))
} else {
@@ -71,8 +95,9 @@ fn parse_migrate_inner_instruction(data: &[u8], metadata: EventMetadata) -> Opti
/// 解析创建代币日志事件
fn parse_create_token_inner_instruction(
data: &[u8],
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::PumpFunCreateToken;
if let Some(event) = pumpfun_create_token_event_log_decode(data) {
Some(DexEvent::PumpFunCreateTokenEvent(PumpFunCreateTokenEvent { metadata, ..event }))
} else {
@@ -80,8 +105,10 @@ fn parse_create_token_inner_instruction(
}
}
/// 解析交易事件
/// 解析交易事件 (inner instruction 不设置 event_type，因为不知道是 Buy 还是 Sell)
fn parse_trade_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option<DexEvent> {
// 注意：inner instruction 的 trade event 不设置 event_type
// 因为它会被合并到 instruction event 中，而 instruction event 已经设置了正确的 event_type
if let Some(event) = pumpfun_trade_event_log_decode(data) {
Some(DexEvent::PumpFunTradeEvent(PumpFunTradeEvent { metadata, ..event }))
} else {
@@ -93,8 +120,10 @@ fn parse_trade_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option
fn parse_create_token_instruction(
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::PumpFunCreateToken;
if data.len() < 16 || accounts.len() < 11 {
return None;
}
@@ -154,8 +183,10 @@ fn parse_create_token_instruction(
fn parse_buy_instruction(
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::PumpFunBuy;
if data.len() < 16 || accounts.len() < 13 {
return None;
}
@@ -188,8 +219,10 @@ fn parse_buy_instruction(
fn parse_sell_instruction(
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::PumpFunSell;
if data.len() < 16 || accounts.len() < 11 {
return None;
}
@@ -222,8 +255,10 @@ fn parse_sell_instruction(
fn parse_migrate_instruction(
_data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::PumpFunMigrate;
if accounts.len() < 24 {
return None;
}
src/streaming/event_parser/protocols/pumpfun/types.rs
+7 -3
View File
@@ -4,7 +4,7 @@ use solana_sdk::pubkey::Pubkey;
use crate::streaming::{
event_parser::{
common::EventMetadata,
common::{EventMetadata, EventType},
protocols::pumpfun::{PumpFunBondingCurveAccountEvent, PumpFunGlobalAccountEvent},
DexEvent,
},
@@ -33,8 +33,10 @@ pub fn bonding_curve_decode(data: &[u8]) -> Option<BondingCurve> {
pub fn bonding_curve_parser(
account: &AccountPretty,
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::AccountPumpFunBondingCurve;
if account.data.len() < BONDING_CURVE_SIZE + 8 {
return None;
}
@@ -81,7 +83,9 @@ pub fn global_decode(data: &[u8]) -> Option<Global> {
borsh::from_slice::<Global>(&data[..GLOBAL_SIZE]).ok()
}
pub fn global_parser(account: &AccountPretty, metadata: EventMetadata) -> Option<DexEvent> {
pub fn global_parser(account: &AccountPretty, mut metadata: EventMetadata) -> Option<DexEvent> {
metadata.event_type = EventType::AccountPumpFunGlobal;
if account.data.len() < GLOBAL_SIZE + 8 {
return None;
}
src/streaming/event_parser/protocols/pumpswap/parser.rs
+81 -60
View File
@@ -1,6 +1,5 @@
use crate::streaming::event_parser::{
common::{read_u64_le, EventMetadata, EventType, ProtocolType},
core::GenericEventParseConfig,
common::{read_u64_le, EventMetadata, EventType},
protocols::pumpswap::{
discriminators, pump_swap_buy_event_log_decode, pump_swap_create_pool_event_log_decode,
pump_swap_deposit_event_log_decode, pump_swap_sell_event_log_decode,
@@ -15,62 +14,70 @@ use solana_sdk::pubkey::Pubkey;
pub const PUMPSWAP_PROGRAM_ID: Pubkey =
solana_sdk::pubkey!("pAMMBay6oceH9fJKBRHGP5D4bD4sWpmSwMn52FMfXEA");
// 配置所有事件类型
pub const CONFIGS: &[GenericEventParseConfig] = &[
GenericEventParseConfig {
program_id: PUMPSWAP_PROGRAM_ID,
protocol_type: ProtocolType::PumpSwap,
inner_instruction_discriminator: discriminators::BUY_EVENT,
instruction_discriminator: discriminators::BUY_IX,
event_type: EventType::PumpSwapBuy,
inner_instruction_parser: Some(parse_buy_inner_instruction),
instruction_parser: Some(parse_buy_instruction),
requires_inner_instruction: false,
},
GenericEventParseConfig {
program_id: PUMPSWAP_PROGRAM_ID,
protocol_type: ProtocolType::PumpSwap,
inner_instruction_discriminator: discriminators::SELL_EVENT,
instruction_discriminator: discriminators::SELL_IX,
event_type: EventType::PumpSwapSell,
inner_instruction_parser: Some(parse_sell_inner_instruction),
instruction_parser: Some(parse_sell_instruction),
requires_inner_instruction: false,
},
GenericEventParseConfig {
program_id: PUMPSWAP_PROGRAM_ID,
protocol_type: ProtocolType::PumpSwap,
inner_instruction_discriminator: discriminators::CREATE_POOL_EVENT,
instruction_discriminator: discriminators::CREATE_POOL_IX,
event_type: EventType::PumpSwapCreatePool,
inner_instruction_parser: Some(parse_create_pool_inner_instruction),
instruction_parser: Some(parse_create_pool_instruction),
requires_inner_instruction: false,
},
GenericEventParseConfig {
program_id: PUMPSWAP_PROGRAM_ID,
protocol_type: ProtocolType::PumpSwap,
inner_instruction_discriminator: discriminators::DEPOSIT_EVENT,
instruction_discriminator: discriminators::DEPOSIT_IX,
event_type: EventType::PumpSwapDeposit,
inner_instruction_parser: Some(parse_deposit_inner_instruction),
instruction_parser: Some(parse_deposit_instruction),
requires_inner_instruction: false,
},
GenericEventParseConfig {
program_id: PUMPSWAP_PROGRAM_ID,
protocol_type: ProtocolType::PumpSwap,
inner_instruction_discriminator: discriminators::WITHDRAW_EVENT,
instruction_discriminator: discriminators::WITHDRAW_IX,
event_type: EventType::PumpSwapWithdraw,
inner_instruction_parser: Some(parse_withdraw_inner_instruction),
instruction_parser: Some(parse_withdraw_instruction),
requires_inner_instruction: false,
},
];
/// 解析 PumpSwap instruction data
///
/// 根据判别器路由到具体的 instruction 解析函数
pub fn parse_pumpswap_instruction_data(
discriminator: &[u8],
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
) -> Option<DexEvent> {
match discriminator {
discriminators::BUY_IX => parse_buy_instruction(data, accounts, metadata),
discriminators::SELL_IX => parse_sell_instruction(data, accounts, metadata),
discriminators::CREATE_POOL_IX => {
parse_create_pool_instruction(data, accounts, metadata)
}
discriminators::DEPOSIT_IX => parse_deposit_instruction(data, accounts, metadata),
discriminators::WITHDRAW_IX => parse_withdraw_instruction(data, accounts, metadata),
_ => None,
}
}
/// 解析 PumpSwap inner instruction data
///
/// 根据判别器路由到具体的 inner instruction 解析函数
pub fn parse_pumpswap_inner_instruction_data(
discriminator: &[u8],
data: &[u8],
metadata: EventMetadata,
) -> Option<DexEvent> {
match discriminator {
discriminators::BUY_EVENT => parse_buy_inner_instruction(data, metadata),
discriminators::SELL_EVENT => parse_sell_inner_instruction(data, metadata),
discriminators::CREATE_POOL_EVENT => {
parse_create_pool_inner_instruction(data, metadata)
}
discriminators::DEPOSIT_EVENT => parse_deposit_inner_instruction(data, metadata),
discriminators::WITHDRAW_EVENT => parse_withdraw_inner_instruction(data, metadata),
_ => None,
}
}
/// 解析 PumpSwap 账户数据
///
/// 根据判别器路由到具体的账户解析函数
pub fn parse_pumpswap_account_data(
discriminator: &[u8],
account: &crate::streaming::grpc::AccountPretty,
metadata: crate::streaming::event_parser::common::EventMetadata,
) -> Option<crate::streaming::event_parser::DexEvent> {
match discriminator {
discriminators::GLOBAL_CONFIG_ACCOUNT => {
crate::streaming::event_parser::protocols::pumpswap::types::global_config_parser(account, metadata)
}
discriminators::POOL_ACCOUNT => {
crate::streaming::event_parser::protocols::pumpswap::types::pool_parser(account, metadata)
}
_ => None,
}
}
/// 解析买入日志事件
fn parse_buy_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option<DexEvent> {
// Note: event_type will be set by instruction parser
if let Some(event) = pump_swap_buy_event_log_decode(data) {
Some(DexEvent::PumpSwapBuyEvent(PumpSwapBuyEvent { metadata, ..event }))
} else {
@@ -80,6 +87,7 @@ fn parse_buy_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option<D
/// 解析卖出日志事件
fn parse_sell_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option<DexEvent> {
// Note: event_type will be set by instruction parser
if let Some(event) = pump_swap_sell_event_log_decode(data) {
Some(DexEvent::PumpSwapSellEvent(PumpSwapSellEvent { metadata, ..event }))
} else {
@@ -92,6 +100,7 @@ fn parse_create_pool_inner_instruction(
data: &[u8],
metadata: EventMetadata,
) -> Option<DexEvent> {
// Note: event_type will be set by instruction parser
if let Some(event) = pump_swap_create_pool_event_log_decode(data) {
Some(DexEvent::PumpSwapCreatePoolEvent(PumpSwapCreatePoolEvent { metadata, ..event }))
} else {
@@ -101,6 +110,7 @@ fn parse_create_pool_inner_instruction(
/// 解析存款日志事件
fn parse_deposit_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option<DexEvent> {
// Note: event_type will be set by instruction parser
if let Some(event) = pump_swap_deposit_event_log_decode(data) {
Some(DexEvent::PumpSwapDepositEvent(PumpSwapDepositEvent { metadata, ..event }))
} else {
@@ -110,6 +120,7 @@ fn parse_deposit_inner_instruction(data: &[u8], metadata: EventMetadata) -> Opti
/// 解析提款日志事件
fn parse_withdraw_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option<DexEvent> {
// Note: event_type will be set by instruction parser
if let Some(event) = pump_swap_withdraw_event_log_decode(data) {
Some(DexEvent::PumpSwapWithdrawEvent(PumpSwapWithdrawEvent { metadata, ..event }))
} else {
@@ -121,8 +132,10 @@ fn parse_withdraw_inner_instruction(data: &[u8], metadata: EventMetadata) -> Opt
fn parse_buy_instruction(
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::PumpSwapBuy;
if data.len() < 16 || accounts.len() < 11 {
return None;
}
@@ -156,8 +169,10 @@ fn parse_buy_instruction(
fn parse_sell_instruction(
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::PumpSwapSell;
if data.len() < 16 || accounts.len() < 11 {
return None;
}
@@ -191,8 +206,10 @@ fn parse_sell_instruction(
fn parse_create_pool_instruction(
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::PumpSwapCreatePool;
if data.len() < 18 || accounts.len() < 11 {
return None;
}
@@ -230,8 +247,10 @@ fn parse_create_pool_instruction(
fn parse_deposit_instruction(
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::PumpSwapDeposit;
if data.len() < 24 || accounts.len() < 11 {
return None;
}
@@ -262,8 +281,10 @@ fn parse_deposit_instruction(
fn parse_withdraw_instruction(
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::PumpSwapWithdraw;
if data.len() < 24 || accounts.len() < 11 {
return None;
}
src/streaming/event_parser/protocols/pumpswap/types.rs
+7 -3
View File
@@ -4,7 +4,7 @@ use solana_sdk::pubkey::Pubkey;
use crate::streaming::{
event_parser::{
common::EventMetadata,
common::{EventMetadata, EventType},
protocols::pumpswap::{PumpSwapGlobalConfigAccountEvent, PumpSwapPoolAccountEvent},
DexEvent,
},
@@ -33,8 +33,10 @@ pub fn global_config_decode(data: &[u8]) -> Option<GlobalConfig> {
pub fn global_config_parser(
account: &AccountPretty,
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::AccountPumpSwapGlobalConfig;
if account.data.len() < GLOBAL_CONFIG_SIZE + 8 {
return None;
}
@@ -76,7 +78,9 @@ pub fn pool_decode(data: &[u8]) -> Option<Pool> {
borsh::from_slice::<Pool>(&data[..POOL_SIZE]).ok()
}
pub fn pool_parser(account: &AccountPretty, metadata: EventMetadata) -> Option<DexEvent> {
pub fn pool_parser(account: &AccountPretty, mut metadata: EventMetadata) -> Option<DexEvent> {
metadata.event_type = EventType::AccountPumpSwapPool;
if account.data.len() < POOL_SIZE + 8 {
return None;
}
src/streaming/event_parser/protocols/raydium_amm_v4/parser.rs
+74 -72
View File
@@ -1,6 +1,5 @@
use crate::streaming::event_parser::{
common::{read_u64_le, EventMetadata, EventType, ProtocolType},
core::GenericEventParseConfig,
common::{read_u64_le, EventMetadata, EventType},
protocols::raydium_amm_v4::{
discriminators, RaydiumAmmV4DepositEvent, RaydiumAmmV4Initialize2Event,
RaydiumAmmV4SwapEvent, RaydiumAmmV4WithdrawEvent, RaydiumAmmV4WithdrawPnlEvent,
@@ -9,80 +8,73 @@ use crate::streaming::event_parser::{
};
use solana_sdk::pubkey::Pubkey;
/// Raydium CPMM程序ID
/// Raydium AMM V4程序ID
pub const RAYDIUM_AMM_V4_PROGRAM_ID: Pubkey =
solana_sdk::pubkey!("675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8");
// 配置所有事件类型
pub const CONFIGS: &[GenericEventParseConfig] = &[
GenericEventParseConfig {
program_id: RAYDIUM_AMM_V4_PROGRAM_ID,
protocol_type: ProtocolType::RaydiumAmmV4,
inner_instruction_discriminator: &[],
instruction_discriminator: discriminators::SWAP_BASE_IN,
event_type: EventType::RaydiumAmmV4SwapBaseIn,
inner_instruction_parser: None,
instruction_parser: Some(parse_swap_base_input_instruction),
requires_inner_instruction: false,
},
GenericEventParseConfig {
program_id: RAYDIUM_AMM_V4_PROGRAM_ID,
protocol_type: ProtocolType::RaydiumAmmV4,
inner_instruction_discriminator: &[],
instruction_discriminator: discriminators::SWAP_BASE_OUT,
event_type: EventType::RaydiumAmmV4SwapBaseOut,
inner_instruction_parser: None,
instruction_parser: Some(parse_swap_base_output_instruction),
requires_inner_instruction: false,
},
GenericEventParseConfig {
program_id: RAYDIUM_AMM_V4_PROGRAM_ID,
protocol_type: ProtocolType::RaydiumAmmV4,
inner_instruction_discriminator: &[],
instruction_discriminator: discriminators::DEPOSIT,
event_type: EventType::RaydiumAmmV4Deposit,
inner_instruction_parser: None,
instruction_parser: Some(parse_deposit_instruction),
requires_inner_instruction: false,
},
GenericEventParseConfig {
program_id: RAYDIUM_AMM_V4_PROGRAM_ID,
protocol_type: ProtocolType::RaydiumAmmV4,
inner_instruction_discriminator: &[],
instruction_discriminator: discriminators::INITIALIZE2,
event_type: EventType::RaydiumAmmV4Initialize2,
inner_instruction_parser: None,
instruction_parser: Some(parse_initialize2_instruction),
requires_inner_instruction: false,
},
GenericEventParseConfig {
program_id: RAYDIUM_AMM_V4_PROGRAM_ID,
protocol_type: ProtocolType::RaydiumAmmV4,
inner_instruction_discriminator: &[],
instruction_discriminator: discriminators::WITHDRAW,
event_type: EventType::RaydiumAmmV4Withdraw,
inner_instruction_parser: None,
instruction_parser: Some(parse_withdraw_instruction),
requires_inner_instruction: false,
},
GenericEventParseConfig {
program_id: RAYDIUM_AMM_V4_PROGRAM_ID,
protocol_type: ProtocolType::RaydiumAmmV4,
inner_instruction_discriminator: &[],
instruction_discriminator: discriminators::WITHDRAW_PNL,
event_type: EventType::RaydiumAmmV4WithdrawPnl,
inner_instruction_parser: None,
instruction_parser: Some(parse_withdraw_pnl_instruction),
requires_inner_instruction: false,
},
];
/// 解析 Raydium AMM V4 instruction data
///
/// 根据判别器路由到具体的 instruction 解析函数
pub fn parse_raydium_amm_v4_instruction_data(
discriminator: &[u8],
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
) -> Option<DexEvent> {
match discriminator {
discriminators::SWAP_BASE_IN => {
parse_swap_base_input_instruction(data, accounts, metadata)
}
discriminators::SWAP_BASE_OUT => {
parse_swap_base_output_instruction(data, accounts, metadata)
}
discriminators::DEPOSIT => parse_deposit_instruction(data, accounts, metadata),
discriminators::INITIALIZE2 => parse_initialize2_instruction(data, accounts, metadata),
discriminators::WITHDRAW => parse_withdraw_instruction(data, accounts, metadata),
discriminators::WITHDRAW_PNL => {
parse_withdraw_pnl_instruction(data, accounts, metadata)
}
_ => None,
}
}
/// 解析 Raydium AMM V4 inner instruction data
///
/// Raydium AMM V4 没有 inner instruction 事件
pub fn parse_raydium_amm_v4_inner_instruction_data(
_discriminator: &[u8],
_data: &[u8],
_metadata: EventMetadata,
) -> Option<DexEvent> {
None
}
/// 解析 Raydium AMM V4 账户数据
///
/// 根据判别器路由到具体的账户解析函数
pub fn parse_raydium_amm_v4_account_data(
discriminator: &[u8],
account: &crate::streaming::grpc::AccountPretty,
metadata: crate::streaming::event_parser::common::EventMetadata,
) -> Option<crate::streaming::event_parser::DexEvent> {
match discriminator {
discriminators::AMM_INFO => {
crate::streaming::event_parser::protocols::raydium_amm_v4::types::amm_info_parser(account, metadata)
}
_ => None,
}
}
/// 解析提现指令事件
fn parse_withdraw_pnl_instruction(
_data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::RaydiumAmmV4WithdrawPnl;
if accounts.len() < 17 {
return None;
}
@@ -113,8 +105,10 @@ fn parse_withdraw_pnl_instruction(
fn parse_withdraw_instruction(
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::RaydiumAmmV4Withdraw;
if data.len() < 8 || accounts.len() < 22 {
return None;
}
@@ -153,8 +147,10 @@ fn parse_withdraw_instruction(
fn parse_initialize2_instruction(
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::RaydiumAmmV4Initialize2;
if data.len() < 25 || accounts.len() < 21 {
return None;
}
@@ -198,8 +194,10 @@ fn parse_initialize2_instruction(
fn parse_deposit_instruction(
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::RaydiumAmmV4Deposit;
if data.len() < 24 || accounts.len() < 14 {
return None;
}
@@ -234,8 +232,10 @@ fn parse_deposit_instruction(
fn parse_swap_base_output_instruction(
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::RaydiumAmmV4SwapBaseOut;
if data.len() < 16 || accounts.len() < 17 {
return None;
}
@@ -281,8 +281,10 @@ fn parse_swap_base_output_instruction(
fn parse_swap_base_input_instruction(
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::RaydiumAmmV4SwapBaseIn;
if data.len() < 16 || accounts.len() < 17 {
return None;
}
src/streaming/event_parser/protocols/raydium_amm_v4/types.rs
+5 -2
View File
@@ -4,7 +4,8 @@ use solana_sdk::pubkey::Pubkey;
use crate::streaming::{
event_parser::{
common::EventMetadata, protocols::raydium_amm_v4::RaydiumAmmV4AmmInfoAccountEvent,
common::{EventMetadata, EventType},
protocols::raydium_amm_v4::RaydiumAmmV4AmmInfoAccountEvent,
DexEvent,
},
grpc::AccountPretty,
@@ -86,7 +87,9 @@ pub fn amm_info_decode(data: &[u8]) -> Option<AmmInfo> {
borsh::from_slice::<AmmInfo>(&data[..AMM_INFO_SIZE]).ok()
}
pub fn amm_info_parser(account: &AccountPretty, metadata: EventMetadata) -> Option<DexEvent> {
pub fn amm_info_parser(account: &AccountPretty, mut metadata: EventMetadata) -> Option<DexEvent> {
metadata.event_type = EventType::AccountRaydiumAmmV4AmmInfo;
if account.data.len() < AMM_INFO_SIZE {
return None;
}
src/streaming/event_parser/protocols/raydium_clmm/parser.rs
+90 -93
View File
@@ -1,9 +1,8 @@
use crate::streaming::event_parser::{
common::{
read_i32_le, read_option_bool, read_u128_le, read_u64_le, read_u8_le, EventMetadata,
EventType, ProtocolType,
EventType,
},
core::GenericEventParseConfig,
protocols::raydium_clmm::{
discriminators, RaydiumClmmClosePositionEvent, RaydiumClmmCreatePoolEvent,
RaydiumClmmDecreaseLiquidityV2Event, RaydiumClmmIncreaseLiquidityV2Event,
@@ -18,96 +17,80 @@ use solana_sdk::pubkey::Pubkey;
pub const RAYDIUM_CLMM_PROGRAM_ID: Pubkey =
solana_sdk::pubkey!("CAMMCzo5YL8w4VFF8KVHrK22GGUsp5VTaW7grrKgrWqK");
// 配置所有事件类型
pub const CONFIGS: &[GenericEventParseConfig] = &[
GenericEventParseConfig {
program_id: RAYDIUM_CLMM_PROGRAM_ID,
protocol_type: ProtocolType::RaydiumClmm,
inner_instruction_discriminator: &[],
instruction_discriminator: discriminators::SWAP,
event_type: EventType::RaydiumClmmSwap,
inner_instruction_parser: None,
instruction_parser: Some(parse_swap_instruction),
requires_inner_instruction: false,
},
GenericEventParseConfig {
program_id: RAYDIUM_CLMM_PROGRAM_ID,
protocol_type: ProtocolType::RaydiumClmm,
inner_instruction_discriminator: &[],
instruction_discriminator: discriminators::SWAP_V2,
event_type: EventType::RaydiumClmmSwapV2,
inner_instruction_parser: None,
instruction_parser: Some(parse_swap_v2_instruction),
requires_inner_instruction: false,
},
GenericEventParseConfig {
program_id: RAYDIUM_CLMM_PROGRAM_ID,
protocol_type: ProtocolType::RaydiumClmm,
inner_instruction_discriminator: &[],
instruction_discriminator: discriminators::CLOSE_POSITION,
event_type: EventType::RaydiumClmmClosePosition,
inner_instruction_parser: None,
instruction_parser: Some(parse_close_position_instruction),
requires_inner_instruction: false,
},
GenericEventParseConfig {
program_id: RAYDIUM_CLMM_PROGRAM_ID,
protocol_type: ProtocolType::RaydiumClmm,
inner_instruction_discriminator: &[],
instruction_discriminator: discriminators::DECREASE_LIQUIDITY_V2,
event_type: EventType::RaydiumClmmDecreaseLiquidityV2,
inner_instruction_parser: None,
instruction_parser: Some(parse_decrease_liquidity_v2_instruction),
requires_inner_instruction: false,
},
GenericEventParseConfig {
program_id: RAYDIUM_CLMM_PROGRAM_ID,
protocol_type: ProtocolType::RaydiumClmm,
inner_instruction_discriminator: &[],
instruction_discriminator: discriminators::CREATE_POOL,
event_type: EventType::RaydiumClmmCreatePool,
inner_instruction_parser: None,
instruction_parser: Some(parse_create_pool_instruction),
requires_inner_instruction: false,
},
GenericEventParseConfig {
program_id: RAYDIUM_CLMM_PROGRAM_ID,
protocol_type: ProtocolType::RaydiumClmm,
inner_instruction_discriminator: &[],
instruction_discriminator: discriminators::INCREASE_LIQUIDITY_V2,
event_type: EventType::RaydiumClmmIncreaseLiquidityV2,
inner_instruction_parser: None,
instruction_parser: Some(parse_increase_liquidity_v2_instruction),
requires_inner_instruction: false,
},
GenericEventParseConfig {
program_id: RAYDIUM_CLMM_PROGRAM_ID,
protocol_type: ProtocolType::RaydiumClmm,
inner_instruction_discriminator: &[],
instruction_discriminator: discriminators::OPEN_POSITION_WITH_TOKEN_22_NFT,
event_type: EventType::RaydiumClmmOpenPositionWithToken22Nft,
inner_instruction_parser: None,
instruction_parser: Some(parse_open_position_with_token_22_nft_instruction),
requires_inner_instruction: false,
},
GenericEventParseConfig {
program_id: RAYDIUM_CLMM_PROGRAM_ID,
protocol_type: ProtocolType::RaydiumClmm,
inner_instruction_discriminator: &[],
instruction_discriminator: discriminators::OPEN_POSITION_V2,
event_type: EventType::RaydiumClmmOpenPositionV2,
inner_instruction_parser: None,
instruction_parser: Some(parse_open_position_v2_instruction),
requires_inner_instruction: false,
},
];
/// 解析 Raydium CLMM instruction data
///
/// 根据判别器路由到具体的 instruction 解析函数
pub fn parse_raydium_clmm_instruction_data(
discriminator: &[u8],
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
) -> Option<DexEvent> {
match discriminator {
discriminators::SWAP => parse_swap_instruction(data, accounts, metadata),
discriminators::SWAP_V2 => parse_swap_v2_instruction(data, accounts, metadata),
discriminators::CLOSE_POSITION => {
parse_close_position_instruction(data, accounts, metadata)
}
discriminators::DECREASE_LIQUIDITY_V2 => {
parse_decrease_liquidity_v2_instruction(data, accounts, metadata)
}
discriminators::CREATE_POOL => parse_create_pool_instruction(data, accounts, metadata),
discriminators::INCREASE_LIQUIDITY_V2 => {
parse_increase_liquidity_v2_instruction(data, accounts, metadata)
}
discriminators::OPEN_POSITION_WITH_TOKEN_22_NFT => {
parse_open_position_with_token_22_nft_instruction(data, accounts, metadata)
}
discriminators::OPEN_POSITION_V2 => {
parse_open_position_v2_instruction(data, accounts, metadata)
}
_ => None,
}
}
/// 解析 Raydium CLMM inner instruction data
///
/// Raydium CLMM 没有 inner instruction 事件
pub fn parse_raydium_clmm_inner_instruction_data(
_discriminator: &[u8],
_data: &[u8],
_metadata: EventMetadata,
) -> Option<DexEvent> {
None
}
/// 解析 Raydium CLMM 账户数据
///
/// 根据判别器路由到具体的账户解析函数
pub fn parse_raydium_clmm_account_data(
discriminator: &[u8],
account: &crate::streaming::grpc::AccountPretty,
metadata: crate::streaming::event_parser::common::EventMetadata,
) -> Option<crate::streaming::event_parser::DexEvent> {
match discriminator {
discriminators::AMM_CONFIG => {
crate::streaming::event_parser::protocols::raydium_clmm::types::amm_config_parser(account, metadata)
}
discriminators::POOL_STATE => {
crate::streaming::event_parser::protocols::raydium_clmm::types::pool_state_parser(account, metadata)
}
discriminators::TICK_ARRAY_STATE => {
crate::streaming::event_parser::protocols::raydium_clmm::types::tick_array_state_parser(account, metadata)
}
_ => None,
}
}
/// 解析打开仓位V2指令事件
fn parse_open_position_v2_instruction(
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::RaydiumClmmOpenPositionV2;
if data.len() < 51 || accounts.len() < 22 {
return None;
}
@@ -152,8 +135,10 @@ fn parse_open_position_v2_instruction(
fn parse_open_position_with_token_22_nft_instruction(
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::RaydiumClmmOpenPositionWithToken22Nft;
if data.len() < 51 || accounts.len() < 20 {
return None;
}
@@ -197,8 +182,10 @@ fn parse_open_position_with_token_22_nft_instruction(
fn parse_increase_liquidity_v2_instruction(
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::RaydiumClmmIncreaseLiquidityV2;
if data.len() < 34 || accounts.len() < 15 {
return None;
}
@@ -230,8 +217,10 @@ fn parse_increase_liquidity_v2_instruction(
fn parse_create_pool_instruction(
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::RaydiumClmmCreatePool;
if data.len() < 24 || accounts.len() < 13 {
return None;
}
@@ -259,8 +248,10 @@ fn parse_create_pool_instruction(
fn parse_decrease_liquidity_v2_instruction(
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::RaydiumClmmDecreaseLiquidityV2;
if data.len() < 32 || accounts.len() < 16 {
return None;
}
@@ -293,8 +284,10 @@ fn parse_decrease_liquidity_v2_instruction(
fn parse_close_position_instruction(
_data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::RaydiumClmmClosePosition;
if accounts.len() < 6 {
return None;
}
@@ -313,8 +306,10 @@ fn parse_close_position_instruction(
fn parse_swap_instruction(
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::RaydiumClmmSwap;
if data.len() < 33 || accounts.len() < 10 {
return None;
}
@@ -347,8 +342,10 @@ fn parse_swap_instruction(
fn parse_swap_v2_instruction(
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::RaydiumClmmSwapV2;
if data.len() < 33 || accounts.len() < 13 {
return None;
}
src/streaming/event_parser/protocols/raydium_clmm/types.rs
+10 -4
View File
@@ -4,7 +4,7 @@ use solana_sdk::pubkey::Pubkey;
use crate::streaming::{
event_parser::{
common::EventMetadata,
common::{EventMetadata, EventType},
protocols::raydium_clmm::{
RaydiumClmmAmmConfigAccountEvent, RaydiumClmmPoolStateAccountEvent,
RaydiumClmmTickArrayStateAccountEvent,
@@ -37,7 +37,9 @@ pub fn amm_config_decode(data: &[u8]) -> Option<AmmConfig> {
borsh::from_slice::<AmmConfig>(&data[..AMM_CONFIG_SIZE]).ok()
}
pub fn amm_config_parser(account: &AccountPretty, metadata: EventMetadata) -> Option<DexEvent> {
pub fn amm_config_parser(account: &AccountPretty, mut metadata: EventMetadata) -> Option<DexEvent> {
metadata.event_type = EventType::AccountRaydiumClmmAmmConfig;
if account.data.len() < AMM_CONFIG_SIZE + 8 {
return None;
}
@@ -122,7 +124,9 @@ pub fn pool_state_decode(data: &[u8]) -> Option<PoolState> {
borsh::from_slice::<PoolState>(&data[..POOL_STATE_SIZE]).ok()
}
pub fn pool_state_parser(account: &AccountPretty, metadata: EventMetadata) -> Option<DexEvent> {
pub fn pool_state_parser(account: &AccountPretty, mut metadata: EventMetadata) -> Option<DexEvent> {
metadata.event_type = EventType::AccountRaydiumClmmPoolState;
if account.data.len() < POOL_STATE_SIZE + 8 {
return None;
}
@@ -202,8 +206,10 @@ pub fn tick_array_state_decode(data: &[u8]) -> Option<TickArrayState> {
pub fn tick_array_state_parser(
account: &AccountPretty,
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::AccountRaydiumClmmTickArrayState;
if account.data.len() < TICK_ARRAY_STATE_SIZE + 8 {
return None;
}
src/streaming/event_parser/protocols/raydium_cpmm/parser.rs
+70 -60
View File
@@ -1,8 +1,7 @@
use solana_sdk::pubkey::Pubkey;
use crate::streaming::event_parser::{
common::{read_u64_le, EventMetadata, EventType, ProtocolType},
core::GenericEventParseConfig,
common::{read_u64_le, EventMetadata, EventType},
protocols::raydium_cpmm::{
discriminators, RaydiumCpmmDepositEvent, RaydiumCpmmInitializeEvent, RaydiumCpmmSwapEvent,
RaydiumCpmmWithdrawEvent,
@@ -14,66 +13,69 @@ use crate::streaming::event_parser::{
pub const RAYDIUM_CPMM_PROGRAM_ID: Pubkey =
solana_sdk::pubkey!("CPMMoo8L3F4NbTegBCKVNunggL7H1ZpdTHKxQB5qKP1C");
// 配置所有事件类型
pub const CONFIGS: &[GenericEventParseConfig] = &[
GenericEventParseConfig {
program_id: RAYDIUM_CPMM_PROGRAM_ID,
protocol_type: ProtocolType::RaydiumCpmm,
inner_instruction_discriminator: &[],
instruction_discriminator: discriminators::SWAP_BASE_IN,
event_type: EventType::RaydiumCpmmSwapBaseInput,
inner_instruction_parser: None,
instruction_parser: Some(parse_swap_base_input_instruction),
requires_inner_instruction: false,
},
GenericEventParseConfig {
program_id: RAYDIUM_CPMM_PROGRAM_ID,
protocol_type: ProtocolType::RaydiumCpmm,
inner_instruction_discriminator: &[],
instruction_discriminator: discriminators::SWAP_BASE_OUT,
event_type: EventType::RaydiumCpmmSwapBaseOutput,
inner_instruction_parser: None,
instruction_parser: Some(parse_swap_base_output_instruction),
requires_inner_instruction: false,
},
GenericEventParseConfig {
program_id: RAYDIUM_CPMM_PROGRAM_ID,
protocol_type: ProtocolType::RaydiumCpmm,
inner_instruction_discriminator: &[],
instruction_discriminator: discriminators::DEPOSIT,
event_type: EventType::RaydiumCpmmDeposit,
inner_instruction_parser: None,
instruction_parser: Some(parse_deposit_instruction),
requires_inner_instruction: false,
},
GenericEventParseConfig {
program_id: RAYDIUM_CPMM_PROGRAM_ID,
protocol_type: ProtocolType::RaydiumCpmm,
inner_instruction_discriminator: &[],
instruction_discriminator: discriminators::INITIALIZE,
event_type: EventType::RaydiumCpmmInitialize,
inner_instruction_parser: None,
instruction_parser: Some(parse_initialize_instruction),
requires_inner_instruction: false,
},
GenericEventParseConfig {
program_id: RAYDIUM_CPMM_PROGRAM_ID,
protocol_type: ProtocolType::RaydiumCpmm,
inner_instruction_discriminator: &[],
instruction_discriminator: discriminators::WITHDRAW,
event_type: EventType::RaydiumCpmmWithdraw,
inner_instruction_parser: None,
instruction_parser: Some(parse_withdraw_instruction),
requires_inner_instruction: false,
},
];
/// 解析 Raydium CPMM instruction data
///
/// 根据判别器路由到具体的 instruction 解析函数
pub fn parse_raydium_cpmm_instruction_data(
discriminator: &[u8],
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
) -> Option<DexEvent> {
match discriminator {
discriminators::SWAP_BASE_IN => {
parse_swap_base_input_instruction(data, accounts, metadata)
}
discriminators::SWAP_BASE_OUT => {
parse_swap_base_output_instruction(data, accounts, metadata)
}
discriminators::DEPOSIT => parse_deposit_instruction(data, accounts, metadata),
discriminators::INITIALIZE => parse_initialize_instruction(data, accounts, metadata),
discriminators::WITHDRAW => parse_withdraw_instruction(data, accounts, metadata),
_ => None,
}
}
/// 解析 Raydium CPMM inner instruction data
///
/// Raydium CPMM 没有 inner instruction 事件
pub fn parse_raydium_cpmm_inner_instruction_data(
_discriminator: &[u8],
_data: &[u8],
_metadata: EventMetadata,
) -> Option<DexEvent> {
None
}
/// 解析 Raydium CPMM 账户数据
///
/// 根据判别器路由到具体的账户解析函数
pub fn parse_raydium_cpmm_account_data(
discriminator: &[u8],
account: &crate::streaming::grpc::AccountPretty,
metadata: crate::streaming::event_parser::common::EventMetadata,
) -> Option<crate::streaming::event_parser::DexEvent> {
match discriminator {
discriminators::AMM_CONFIG => {
crate::streaming::event_parser::protocols::raydium_cpmm::types::amm_config_parser(account, metadata)
}
discriminators::POOL_STATE => {
crate::streaming::event_parser::protocols::raydium_cpmm::types::pool_state_parser(account, metadata)
}
_ => None,
}
}
/// 解析提款指令事件
fn parse_withdraw_instruction(
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::RaydiumCpmmWithdraw;
if data.len() < 24 || accounts.len() < 14 {
return None;
}
@@ -103,8 +105,10 @@ fn parse_withdraw_instruction(
fn parse_initialize_instruction(
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::RaydiumCpmmInitialize;
if data.len() < 24 || accounts.len() < 20 {
return None;
}
@@ -140,8 +144,10 @@ fn parse_initialize_instruction(
fn parse_deposit_instruction(
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::RaydiumCpmmDeposit;
if data.len() < 24 || accounts.len() < 13 {
return None;
}
@@ -170,8 +176,10 @@ fn parse_deposit_instruction(
fn parse_swap_base_input_instruction(
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::RaydiumCpmmSwapBaseInput;
if data.len() < 16 || accounts.len() < 13 {
return None;
}
@@ -203,8 +211,10 @@ fn parse_swap_base_input_instruction(
fn parse_swap_base_output_instruction(
data: &[u8],
accounts: &[Pubkey],
metadata: EventMetadata,
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::RaydiumCpmmSwapBaseOutput;
if data.len() < 16 || accounts.len() < 13 {
return None;
}
src/streaming/event_parser/protocols/raydium_cpmm/types.rs
+7 -3
View File
@@ -4,7 +4,7 @@ use solana_sdk::pubkey::Pubkey;
use crate::streaming::{
event_parser::{
common::EventMetadata,
common::{EventMetadata, EventType},
protocols::raydium_cpmm::{
RaydiumCpmmAmmConfigAccountEvent, RaydiumCpmmPoolStateAccountEvent,
},
@@ -36,7 +36,9 @@ pub fn amm_config_decode(data: &[u8]) -> Option<AmmConfig> {
borsh::from_slice::<AmmConfig>(&data[..AMM_CONFIG_SIZE]).ok()
}
pub fn amm_config_parser(account: &AccountPretty, metadata: EventMetadata) -> Option<DexEvent> {
pub fn amm_config_parser(account: &AccountPretty, mut metadata: EventMetadata) -> Option<DexEvent> {
metadata.event_type = EventType::AccountRaydiumCpmmAmmConfig;
if account.data.len() < AMM_CONFIG_SIZE + 8 {
return None;
}
@@ -91,7 +93,9 @@ pub fn pool_state_decode(data: &[u8]) -> Option<PoolState> {
borsh::from_slice::<PoolState>(&data[..POOL_STATE_SIZE]).ok()
}
pub fn pool_state_parser(account: &AccountPretty, metadata: EventMetadata) -> Option<DexEvent> {
pub fn pool_state_parser(account: &AccountPretty, mut metadata: EventMetadata) -> Option<DexEvent> {
metadata.event_type = EventType::AccountRaydiumCpmmPoolState;
if account.data.len() < POOL_STATE_SIZE + 8 {
return None;
}