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

Merge PR #65: refactor streaming core cleanup and dependency simplification

Browse Source
This commit is contained in:
Wood
2026-03-09 10:10:24 +08:00
parent bdf984f2c1 dad231a26b
commit adb4cfc52d
11 changed files with 309 additions and 940 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.toml
-2
View File
@@ -24,7 +24,6 @@ serde-big-array = "0.5.1"
futures = "0.3.32" futures = "0.3.32"
bincode = "1.3" bincode = "1.3"
anyhow = "1.0.102" anyhow = "1.0.102"
bs58 = "0.5.1"
yellowstone-grpc-client = { version = "10.2.0" } yellowstone-grpc-client = { version = "10.2.0" }
yellowstone-grpc-proto = { version = "10.1.1" } yellowstone-grpc-proto = { version = "10.1.1" }
tokio = { version = "1.50.0", features = ["full", "rt-multi-thread"]} tokio = { version = "1.50.0", features = ["full", "rt-multi-thread"]}
@@ -35,7 +34,6 @@ dashmap = "6.1.0"
prost = "0.14.3" prost = "0.14.3"
prost-types = "0.14.3" prost-types = "0.14.3"
crossbeam-queue = "0.3.12" crossbeam-queue = "0.3.12"
wide = "1.1.1"
spl-token = { version = "9.0.0", default-features = false, features = ["no-entrypoint"] } spl-token = { version = "9.0.0", default-features = false, features = ["no-entrypoint"] }
spl-token-2022 = { version = "10.0.0", default-features = false, features = ["no-entrypoint"] } spl-token-2022 = { version = "10.0.0", default-features = false, features = ["no-entrypoint"] }
solana-commitment-config = { version = "3.1.1", features = ["serde"] } solana-commitment-config = { version = "3.1.1", features = ["serde"] }
src/streaming/common/mod.rs
+1 -3
View File
@@ -4,12 +4,10 @@ pub mod metrics;
pub mod constants; pub mod constants;
pub mod subscription; pub mod subscription;
pub mod event_processor; pub mod event_processor;
pub mod simd_utils;
// 重新导出主要类型 // 重新导出主要类型
pub use config::*; pub use config::*;
pub use metrics::*; pub use metrics::*;
pub use constants::*; pub use constants::*;
pub use subscription::*; pub use subscription::*;
pub use event_processor::*; pub use event_processor::*;
pub use simd_utils::*;
src/streaming/common/simd_utils.rs
-295
View File
@@ -1,295 +0,0 @@
use wide::*;
/// SIMD-accelerated data parsing utilities
pub struct SimdUtils;
impl SimdUtils {
/// SIMD-accelerated byte array comparison
/// For arrays with length >= 16, uses SIMD instructions for fast comparison
#[inline(always)]
pub fn fast_bytes_equal(a: &[u8], b: &[u8]) -> bool {
if a.len() != b.len() {
return false;
}
let len = a.len();
// For small arrays, use standard comparison directly
if len < 16 {
return a == b;
}
// Use SIMD to process 16-byte chunks
let chunks = len / 16;
let remainder = len % 16;
// Process complete 16-byte chunks
for i in 0..chunks {
let offset = i * 16;
let chunk_a = u8x16::from(&a[offset..offset + 16]);
let chunk_b = u8x16::from(&b[offset..offset + 16]);
if !chunk_a.simd_eq(chunk_b).all() {
return false;
}
}
// Process remaining bytes
if remainder > 0 {
let start = chunks * 16;
return &a[start..] == &b[start..];
}
true
}
/// Fast discriminator matching, specifically for instruction discriminator comparison
#[inline(always)]
pub fn fast_discriminator_match(data: &[u8], discriminator: &[u8]) -> bool {
if data.len() < discriminator.len() {
return false;
}
let disc_len = discriminator.len();
// Optimize for common discriminator lengths
match disc_len {
1 => data[0] == discriminator[0],
2 => {
let data_u16 = u16::from_le_bytes([data[0], data[1]]);
let disc_u16 = u16::from_le_bytes([discriminator[0], discriminator[1]]);
data_u16 == disc_u16
}
4 => {
let data_u32 = u32::from_le_bytes([data[0], data[1], data[2], data[3]]);
let disc_u32 = u32::from_le_bytes([
discriminator[0],
discriminator[1],
discriminator[2],
discriminator[3],
]);
data_u32 == disc_u32
}
8 => {
let data_u64 = u64::from_le_bytes([
data[0], data[1], data[2], data[3], data[4], data[5], data[6], data[7],
]);
let disc_u64 = u64::from_le_bytes([
discriminator[0],
discriminator[1],
discriminator[2],
discriminator[3],
discriminator[4],
discriminator[5],
discriminator[6],
discriminator[7],
]);
data_u64 == disc_u64
}
16 => {
// Use SIMD to process 16-byte discriminators
let data_chunk = u8x16::from(&data[..16]);
let disc_chunk = u8x16::from(discriminator);
data_chunk.simd_eq(disc_chunk).all()
}
_ => {
// For other lengths, use generic SIMD comparison
Self::fast_bytes_equal(&data[..disc_len], discriminator)
}
}
}
/// SIMD-accelerated memory search to find specific patterns in data
#[inline(always)]
pub fn find_pattern_simd(haystack: &[u8], needle: &[u8]) -> Option<usize> {
if needle.is_empty() || haystack.len() < needle.len() {
return None;
}
let needle_len = needle.len();
let haystack_len = haystack.len();
// For single-byte search, use optimized method
if needle_len == 1 {
let target = needle[0];
return haystack.iter().position(|&b| b == target);
}
// For multi-byte search, use SIMD acceleration
if needle_len <= 16 && haystack_len >= 16 {
let first_byte = needle[0];
let chunks = (haystack_len - needle_len + 1) / 16;
for chunk_idx in 0..chunks {
let start = chunk_idx * 16;
let end = std::cmp::min(start + 16, haystack_len - needle_len + 1);
// Use SIMD to find first byte matches
let chunk = &haystack[start..start + 16];
let target_vec = u8x16::splat(first_byte);
let chunk_vec = u8x16::from(chunk);
let matches = chunk_vec.simd_eq(target_vec);
// Check each match position
let matches_array: [u8; 16] = matches.into();
for i in 0..16 {
if start + i >= end {
break;
}
if matches_array[i] != 0 && start + i + needle_len <= haystack_len {
if Self::fast_bytes_equal(
&haystack[start + i..start + i + needle_len],
needle,
) {
return Some(start + i);
}
}
}
}
// Process remaining part
let remaining_start = chunks * 16;
for i in remaining_start..=(haystack_len - needle_len) {
if Self::fast_bytes_equal(&haystack[i..i + needle_len], needle) {
return Some(i);
}
}
} else {
// Fallback to standard search
for i in 0..=(haystack_len - needle_len) {
if Self::fast_bytes_equal(&haystack[i..i + needle_len], needle) {
return Some(i);
}
}
}
None
}
/// SIMD-accelerated data validation to check if data conforms to specific format
#[inline(always)]
pub fn validate_data_format(data: &[u8], min_length: usize) -> bool {
if data.len() < min_length {
return false;
}
true
}
/// Fast checksum calculation (maintains API consistency)
#[inline(always)]
pub fn fast_checksum(data: &[u8]) -> u32 {
// Simplified implementation, directly sum all bytes
data.iter().map(|&b| b as u32).sum()
}
/// SIMD-accelerated data copy (for large data blocks)
#[inline(always)]
pub fn fast_copy(src: &[u8], dst: &mut [u8]) {
if src.len() != dst.len() {
panic!("Source and destination must have the same length");
}
let len = src.len();
if len >= 32 {
// Use 32-byte SIMD copy
let chunks = len / 32;
for i in 0..chunks {
let start = i * 32;
let src_chunk1 = u8x16::from(&src[start..start + 16]);
let src_chunk2 = u8x16::from(&src[start + 16..start + 32]);
let chunk1_array: [u8; 16] = src_chunk1.into();
let chunk2_array: [u8; 16] = src_chunk2.into();
dst[start..start + 16].copy_from_slice(&chunk1_array);
dst[start + 16..start + 32].copy_from_slice(&chunk2_array);
}
// Process remaining bytes
let remaining_start = chunks * 32;
dst[remaining_start..].copy_from_slice(&src[remaining_start..]);
} else {
// For small data, use standard copy
dst.copy_from_slice(src);
}
}
/// SIMD-accelerated account indices validation
/// Validates that all indices in the account index array are less than the total account count
#[inline(always)]
pub fn validate_account_indices_simd(indices: &[u8], account_count: usize) -> bool {
if indices.is_empty() {
return true;
}
let max_valid_index = account_count as u8;
// For small arrays, use standard comparison directly
if indices.len() < 16 {
return indices.iter().all(|&idx| idx < max_valid_index);
}
// Use SIMD for batch loading and comparison
let chunks = indices.len() / 16;
let remainder = indices.len() % 16;
// Process complete 16-byte chunks
for i in 0..chunks {
let start = i * 16;
let indices_chunk = u8x16::from(&indices[start..start + 16]);
// Convert SIMD vector to array for fast batch checking
let indices_array: [u8; 16] = indices_chunk.into();
// Use unrolled loop for fast comparison, compiler will optimize this
if indices_array[0] >= max_valid_index
|| indices_array[1] >= max_valid_index
|| indices_array[2] >= max_valid_index
|| indices_array[3] >= max_valid_index
|| indices_array[4] >= max_valid_index
|| indices_array[5] >= max_valid_index
|| indices_array[6] >= max_valid_index
|| indices_array[7] >= max_valid_index
|| indices_array[8] >= max_valid_index
|| indices_array[9] >= max_valid_index
|| indices_array[10] >= max_valid_index
|| indices_array[11] >= max_valid_index
|| indices_array[12] >= max_valid_index
|| indices_array[13] >= max_valid_index
|| indices_array[14] >= max_valid_index
|| indices_array[15] >= max_valid_index
{
return false;
}
}
// Process remaining bytes
if remainder > 0 {
let remaining_start = chunks * 16;
return indices[remaining_start..].iter().all(|&idx| idx < max_valid_index);
}
true
}
/// SIMD-accelerated instruction data validation
/// Validates basic format and length requirements of instruction data
#[inline(always)]
pub fn validate_instruction_data_simd(
data: &[u8],
min_length: usize,
discriminator_length: usize,
) -> bool {
// Basic length check
if data.len() < min_length || data.len() < discriminator_length {
return false;
}
// Use existing data format validation
Self::validate_data_format(data, min_length)
}
}
src/streaming/event_parser/common/types.rs
+219 -450
View File
@@ -2,9 +2,9 @@ use borsh::{BorshDeserialize, BorshSerialize};
use crossbeam_queue::ArrayQueue; use crossbeam_queue::ArrayQueue;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use solana_sdk::{pubkey::Pubkey, signature::Signature}; use solana_sdk::{pubkey::Pubkey, signature::Signature};
use std::{borrow::Cow, fmt, str::FromStr, sync::Arc}; use std::{borrow::Cow, fmt, sync::Arc};
use crate::streaming::{common::SimdUtils, event_parser::DexEvent}; use crate::streaming::event_parser::DexEvent;
// Object pool size configuration // Object pool size configuration
const EVENT_METADATA_POOL_SIZE: usize = 1000; const EVENT_METADATA_POOL_SIZE: usize = 1000;
@@ -165,122 +165,7 @@ pub const BLOCK_EVENT_TYPES: &[EventType] = &[EventType::BlockMeta];
impl fmt::Display for EventType { impl fmt::Display for EventType {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self { write!(f, "{:?}", self)
EventType::PumpSwapBuy => write!(f, "PumpSwapBuy"),
EventType::PumpSwapSell => write!(f, "PumpSwapSell"),
EventType::PumpSwapCreatePool => write!(f, "PumpSwapCreatePool"),
EventType::PumpSwapDeposit => write!(f, "PumpSwapDeposit"),
EventType::PumpSwapWithdraw => write!(f, "PumpSwapWithdraw"),
EventType::PumpFunCreateToken => write!(f, "PumpFunCreateToken"),
EventType::PumpFunCreateV2Token => write!(f, "PumpFunCreateV2Token"),
EventType::PumpFunBuy => write!(f, "PumpFunBuy"),
EventType::PumpFunSell => write!(f, "PumpFunSell"),
EventType::PumpFunMigrate => write!(f, "PumpFunMigrate"),
EventType::BonkBuyExactIn => write!(f, "BonkBuyExactIn"),
EventType::BonkBuyExactOut => write!(f, "BonkBuyExactOut"),
EventType::BonkSellExactIn => write!(f, "BonkSellExactIn"),
EventType::BonkSellExactOut => write!(f, "BonkSellExactOut"),
EventType::BonkInitialize => write!(f, "BonkInitialize"),
EventType::BonkInitializeV2 => write!(f, "BonkInitializeV2"),
EventType::BonkInitializeWithToken2022 => write!(f, "BonkInitializeWithToken2022"),
EventType::BonkMigrateToAmm => write!(f, "BonkMigrateToAmm"),
EventType::BonkMigrateToCpswap => write!(f, "BonkMigrateToCpswap"),
EventType::RaydiumCpmmSwapBaseInput => write!(f, "RaydiumCpmmSwapBaseInput"),
EventType::RaydiumCpmmSwapBaseOutput => write!(f, "RaydiumCpmmSwapBaseOutput"),
EventType::RaydiumCpmmDeposit => write!(f, "RaydiumCpmmDeposit"),
EventType::RaydiumCpmmInitialize => write!(f, "RaydiumCpmmInitialize"),
EventType::RaydiumCpmmWithdraw => write!(f, "RaydiumCpmmWithdraw"),
EventType::RaydiumClmmSwap => write!(f, "RaydiumClmmSwap"),
EventType::RaydiumClmmSwapV2 => write!(f, "RaydiumClmmSwapV2"),
EventType::RaydiumClmmClosePosition => write!(f, "RaydiumClmmClosePosition"),
EventType::RaydiumClmmDecreaseLiquidityV2 => {
write!(f, "RaydiumClmmDecreaseLiquidityV2")
}
EventType::RaydiumClmmCreatePool => write!(f, "RaydiumClmmCreatePool"),
EventType::RaydiumClmmIncreaseLiquidityV2 => {
write!(f, "RaydiumClmmIncreaseLiquidityV2")
}
EventType::RaydiumClmmOpenPositionWithToken22Nft => {
write!(f, "RaydiumClmmOpenPositionWithToken22Nft")
}
EventType::RaydiumClmmOpenPositionV2 => write!(f, "RaydiumClmmOpenPositionV2"),
EventType::RaydiumAmmV4SwapBaseIn => write!(f, "RaydiumAmmV4SwapBaseIn"),
EventType::RaydiumAmmV4SwapBaseOut => write!(f, "RaydiumAmmV4SwapBaseOut"),
EventType::RaydiumAmmV4Deposit => write!(f, "RaydiumAmmV4Deposit"),
EventType::RaydiumAmmV4Initialize2 => write!(f, "RaydiumAmmV4Initialize2"),
EventType::RaydiumAmmV4Withdraw => write!(f, "RaydiumAmmV4Withdraw"),
EventType::RaydiumAmmV4WithdrawPnl => write!(f, "RaydiumAmmV4WithdrawPnl"),
EventType::MeteoraDammV2Swap => write!(f, "MeteoraDammV2Swap"),
EventType::MeteoraDammV2Swap2 => write!(f, "MeteoraDammV2Swap2"),
EventType::MeteoraDammV2InitializePool => write!(f, "MeteoraDammV2InitializePool"),
EventType::MeteoraDammV2InitializeCustomizablePool => write!(f, "MeteoraDammV2InitializeCustomizablePool"),
EventType::MeteoraDammV2InitializePoolWithDynamicConfig => write!(f, "MeteoraDammV2InitializePoolWithDynamicConfig"),
EventType::AccountRaydiumAmmV4AmmInfo => write!(f, "AccountRaydiumAmmV4AmmInfo"),
EventType::AccountPumpSwapGlobalConfig => write!(f, "AccountPumpSwapGlobalConfig"),
EventType::AccountPumpSwapPool => write!(f, "AccountPumpSwapPool"),
EventType::AccountBonkPoolState => write!(f, "AccountBonkPoolState"),
EventType::AccountBonkGlobalConfig => write!(f, "AccountBonkGlobalConfig"),
EventType::AccountBonkPlatformConfig => write!(f, "AccountBonkPlatformConfig"),
EventType::AccountBonkVestingRecord => write!(f, "AccountBonkVestingRecord"),
EventType::AccountPumpFunBondingCurve => write!(f, "AccountPumpFunBondingCurve"),
EventType::AccountPumpFunGlobal => write!(f, "AccountPumpFunGlobal"),
EventType::AccountRaydiumClmmAmmConfig => write!(f, "AccountRaydiumClmmAmmConfig"),
EventType::AccountRaydiumClmmPoolState => write!(f, "AccountRaydiumClmmPoolState"),
EventType::AccountRaydiumClmmTickArrayState => {
write!(f, "AccountRaydiumClmmTickArrayState")
}
EventType::AccountRaydiumCpmmAmmConfig => write!(f, "AccountRaydiumCpmmAmmConfig"),
EventType::AccountRaydiumCpmmPoolState => write!(f, "AccountRaydiumCpmmPoolState"),
EventType::TokenAccount => write!(f, "TokenAccount"),
EventType::NonceAccount => write!(f, "NonceAccount"),
EventType::BlockMeta => write!(f, "BlockMeta"),
EventType::SetComputeUnitLimit => write!(f, "SetComputeUnitLimit"),
EventType::SetComputeUnitPrice => write!(f, "SetComputeUnitPrice"),
EventType::Unknown => write!(f, "Unknown"),
}
}
}
/// Parse result
#[derive(Debug, Clone)]
pub struct ParseResult<T> {
pub success: bool,
pub data: Option<T>,
pub error: Option<String>,
}
impl<T> ParseResult<T> {
pub fn success(data: T) -> Self {
Self { success: true, data: Some(data), error: None }
}
pub fn failure(error: String) -> Self {
Self { success: false, data: None, error: Some(error) }
}
pub fn is_success(&self) -> bool {
self.success
}
pub fn is_failure(&self) -> bool {
!self.success
}
}
/// Protocol information
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ProtocolInfo {
pub name: String,
pub program_ids: Vec<Pubkey>,
}
impl ProtocolInfo {
pub fn new(name: String, program_ids: Vec<Pubkey>) -> Self {
Self { name, program_ids }
}
pub fn supports_program(&self, program_id: &Pubkey) -> bool {
self.program_ids.contains(program_id)
} }
} }
@@ -353,356 +238,240 @@ impl EventMetadata {
pub fn set_swap_data(&mut self, swap_data: SwapData) { pub fn set_swap_data(&mut self, swap_data: SwapData) {
self.swap_data = Some(swap_data); self.swap_data = Some(swap_data);
} }
/// Recycle EventMetadata to object pool
pub fn recycle(self) {
EVENT_METADATA_POOL.release(self);
}
} }
static SOL_MINT: std::sync::LazyLock<Pubkey> = static SOL_MINT: std::sync::LazyLock<Pubkey> =
std::sync::LazyLock::new(|| Pubkey::from_str("So11111111111111111111111111111111111111111").unwrap()); std::sync::LazyLock::new(spl_token::native_mint::id);
static SYSTEM_PROGRAMS: std::sync::LazyLock<[Pubkey; 3]> = std::sync::LazyLock::new(|| [ static SYSTEM_PROGRAMS: std::sync::LazyLock<[Pubkey; 3]> = std::sync::LazyLock::new(|| [
Pubkey::from_str("TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA").unwrap(), spl_token::id(),
Pubkey::from_str("TokenzQdBNbLqP5VEhdkAS6EPFLC1PHnBqCXEpPxuEb").unwrap(), spl_token_2022::id(),
Pubkey::from_str("11111111111111111111111111111111").unwrap(), solana_sdk::pubkey!("11111111111111111111111111111111"),
]); ]);
/// Parse token transfer data from next instructions /// Trait abstracting over different inner-instruction types for swap data extraction
pub trait InnerInstructionLike {
fn program_id_index(&self) -> usize;
fn accounts(&self) -> &[u8];
fn data(&self) -> &[u8];
}
/// Adapter for standard Solana compiled instructions
impl InnerInstructionLike for solana_sdk::message::compiled_instruction::CompiledInstruction {
fn program_id_index(&self) -> usize {
self.program_id_index as usize
}
fn accounts(&self) -> &[u8] {
&self.accounts
}
fn data(&self) -> &[u8] {
&self.data
}
}
/// Adapter for gRPC inner instructions (yellowstone)
impl InnerInstructionLike for yellowstone_grpc_proto::prelude::InnerInstruction {
fn program_id_index(&self) -> usize {
self.program_id_index as usize
}
fn accounts(&self) -> &[u8] {
&self.accounts
}
fn data(&self) -> &[u8] {
&self.data
}
}
/// Extract event context (mint/token account/vault info) from a DexEvent
fn extract_swap_context(event: &DexEvent) -> (
SwapData,
Option<Pubkey>, Option<Pubkey>,
Option<Pubkey>, Option<Pubkey>,
Option<Pubkey>, Option<Pubkey>,
) {
let mut swap_data = SwapData::default();
let mut from_mint: Option<Pubkey> = None;
let mut to_mint: Option<Pubkey> = None;
let mut user_from_token: Option<Pubkey> = None;
let mut user_to_token: Option<Pubkey> = None;
let mut from_vault: Option<Pubkey> = None;
let mut to_vault: Option<Pubkey> = None;
match event {
DexEvent::BonkTradeEvent(e) => {
from_mint = Some(e.base_token_mint);
to_mint = Some(e.quote_token_mint);
user_from_token = Some(e.user_base_token);
user_to_token = Some(e.user_quote_token);
from_vault = Some(e.base_vault);
to_vault = Some(e.quote_vault);
}
DexEvent::PumpFunTradeEvent(e) => {
swap_data.from_mint = if e.is_buy { *SOL_MINT } else { e.mint };
swap_data.to_mint = if e.is_buy { e.mint } else { *SOL_MINT };
}
DexEvent::PumpSwapBuyEvent(e) => {
swap_data.from_mint = e.quote_mint;
swap_data.to_mint = e.base_mint;
}
DexEvent::PumpSwapSellEvent(e) => {
swap_data.from_mint = e.base_mint;
swap_data.to_mint = e.quote_mint;
}
DexEvent::RaydiumCpmmSwapEvent(e) => {
from_mint = Some(e.input_token_mint);
to_mint = Some(e.output_token_mint);
user_from_token = Some(e.input_token_account);
user_to_token = Some(e.output_token_account);
from_vault = Some(e.input_vault);
to_vault = Some(e.output_vault);
}
DexEvent::RaydiumClmmSwapEvent(e) => {
swap_data.description =
Some("Unable to get from_mint and to_mint from RaydiumClmmSwapEvent".into());
user_from_token = Some(e.input_token_account);
user_to_token = Some(e.output_token_account);
from_vault = Some(e.input_vault);
to_vault = Some(e.output_vault);
}
DexEvent::RaydiumClmmSwapV2Event(e) => {
from_mint = Some(e.input_vault_mint);
to_mint = Some(e.output_vault_mint);
user_from_token = Some(e.input_token_account);
user_to_token = Some(e.output_token_account);
from_vault = Some(e.input_vault);
to_vault = Some(e.output_vault);
}
DexEvent::RaydiumAmmV4SwapEvent(e) => {
swap_data.description =
Some("Unable to get from_mint and to_mint from RaydiumAmmV4SwapEvent".into());
user_from_token = Some(e.user_source_token_account);
user_to_token = Some(e.user_destination_token_account);
from_vault = Some(e.pool_pc_token_account);
to_vault = Some(e.pool_coin_token_account);
}
_ => {}
}
(swap_data, from_mint, to_mint, user_from_token, user_to_token, from_vault, to_vault)
}
/// Generic swap data extraction that works with any instruction type implementing InnerInstructionLike
fn extract_swap_data_from_instructions<I: InnerInstructionLike>(
event: &DexEvent,
instructions: impl Iterator<Item = I>,
current_index: i8,
accounts: &[Pubkey],
) -> Option<SwapData> {
let (mut swap_data, fm, tm, uft, utt, fv, tv) = extract_swap_context(event);
let user_to_token = utt.unwrap_or_default();
let user_from_token = uft.unwrap_or_default();
let to_vault = tv.unwrap_or_default();
let from_vault = fv.unwrap_or_default();
let to_mint = tm.unwrap_or_default();
let from_mint = fm.unwrap_or_default();
for instruction in instructions.skip((current_index + 1) as usize) {
let program_id = accounts[instruction.program_id_index()];
if !SYSTEM_PROGRAMS.contains(&program_id) {
break;
}
let data = instruction.data();
let accs = instruction.accounts();
if data.len() < 8 {
continue;
}
let get_pubkey = |i: usize| accounts[accs[i] as usize];
let (source, destination, amount) = match data[0] {
12 if accs.len() >= 4 => {
let amt = u64::from_le_bytes(data[1..9].try_into().unwrap());
(get_pubkey(0), get_pubkey(2), amt)
}
3 if accs.len() >= 3 => {
let amt = u64::from_le_bytes(data[1..9].try_into().unwrap());
(get_pubkey(0), get_pubkey(1), amt)
}
2 if accs.len() >= 2 => {
let amt = u64::from_le_bytes(data[4..12].try_into().unwrap());
(get_pubkey(0), get_pubkey(1), amt)
}
_ => continue,
};
match (source, destination) {
(s, d) if s == user_to_token && d == to_vault => {
swap_data.from_mint = to_mint;
swap_data.from_amount = amount;
}
(s, d) if s == from_vault && d == user_from_token => {
swap_data.to_mint = from_mint;
swap_data.to_amount = amount;
}
(s, d) if s == user_from_token && d == from_vault => {
swap_data.from_mint = from_mint;
swap_data.from_amount = amount;
}
(s, d) if s == to_vault && d == user_to_token => {
swap_data.to_mint = to_mint;
swap_data.to_amount = amount;
}
(s, d) if s == user_from_token && d == to_vault => {
swap_data.from_mint = from_mint;
swap_data.from_amount = amount;
}
(s, d) if s == from_vault && d == user_to_token => {
swap_data.to_mint = to_mint;
swap_data.to_amount = amount;
}
_ => {}
}
if swap_data.from_mint != Pubkey::default() && swap_data.to_mint != Pubkey::default() {
break;
}
if swap_data.from_amount != 0 && swap_data.to_amount != 0 {
break;
}
}
if swap_data.from_mint != Pubkey::default()
|| swap_data.to_mint != Pubkey::default()
|| swap_data.from_amount != 0
|| swap_data.to_amount != 0
{
Some(swap_data)
} else {
None
}
}
/// Parse token transfer data from standard Solana inner instructions
pub fn parse_swap_data_from_next_instructions( pub fn parse_swap_data_from_next_instructions(
event: &DexEvent, event: &DexEvent,
inner_instruction: &solana_transaction_status::InnerInstructions, inner_instruction: &solana_transaction_status::InnerInstructions,
current_index: i8, current_index: i8,
accounts: &[Pubkey], accounts: &[Pubkey],
) -> Option<SwapData> { ) -> Option<SwapData> {
let mut swap_data = SwapData { extract_swap_data_from_instructions(
from_mint: Pubkey::default(), event,
to_mint: Pubkey::default(), inner_instruction.instructions.iter().map(|ix| ix.instruction.clone()),
from_amount: 0, current_index,
to_amount: 0, accounts,
description: None, )
};
// 先根据 event 取出关键信息
// let mut user: Option<Pubkey> = None;
let mut from_mint: Option<Pubkey> = None;
let mut to_mint: Option<Pubkey> = None;
let mut user_from_token: Option<Pubkey> = None;
let mut user_to_token: Option<Pubkey> = None;
let mut from_vault: Option<Pubkey> = None;
let mut to_vault: Option<Pubkey> = None;
match event {
DexEvent::BonkTradeEvent(e) => {
// user = Some(e.payer);
from_mint = Some(e.base_token_mint);
to_mint = Some(e.quote_token_mint);
user_from_token = Some(e.user_base_token);
user_to_token = Some(e.user_quote_token);
from_vault = Some(e.base_vault);
to_vault = Some(e.quote_vault);
}
DexEvent::PumpFunTradeEvent(e) => {
swap_data.from_mint = if e.is_buy { *SOL_MINT } else { e.mint };
swap_data.to_mint = if e.is_buy { e.mint } else { *SOL_MINT };
}
DexEvent::PumpSwapBuyEvent(e) => {
swap_data.from_mint = e.quote_mint;
swap_data.to_mint = e.base_mint;
}
DexEvent::PumpSwapSellEvent(e) => {
swap_data.from_mint = e.base_mint;
swap_data.to_mint = e.quote_mint;
}
DexEvent::RaydiumCpmmSwapEvent(e) => {
// user = Some(e.payer);
from_mint = Some(e.input_token_mint);
to_mint = Some(e.output_token_mint);
user_from_token = Some(e.input_token_account);
user_to_token = Some(e.output_token_account);
from_vault = Some(e.input_vault);
to_vault = Some(e.output_vault);
}
DexEvent::RaydiumClmmSwapEvent(e) => {
// user = Some(e.payer);
swap_data.description =
Some("Unable to get from_mint and to_mint from RaydiumClmmSwapEvent".into());
user_from_token = Some(e.input_token_account);
user_to_token = Some(e.output_token_account);
from_vault = Some(e.input_vault);
to_vault = Some(e.output_vault);
}
DexEvent::RaydiumClmmSwapV2Event(e) => {
// user = Some(e.payer);
from_mint = Some(e.input_vault_mint);
to_mint = Some(e.output_vault_mint);
user_from_token = Some(e.input_token_account);
user_to_token = Some(e.output_token_account);
from_vault = Some(e.input_vault);
to_vault = Some(e.output_vault);
}
DexEvent::RaydiumAmmV4SwapEvent(e) => {
// user = Some(e.user_source_owner);
swap_data.description =
Some("Unable to get from_mint and to_mint from RaydiumAmmV4SwapEvent".into());
user_from_token = Some(e.user_source_token_account);
user_to_token = Some(e.user_destination_token_account);
from_vault = Some(e.pool_pc_token_account);
to_vault = Some(e.pool_coin_token_account);
}
_ => {}
}
let user_to_token = user_to_token.unwrap_or_default();
let user_from_token = user_from_token.unwrap_or_default();
let to_vault = to_vault.unwrap_or_default();
let from_vault = from_vault.unwrap_or_default();
let to_mint = to_mint.unwrap_or_default();
let from_mint = from_mint.unwrap_or_default();
// 单次循环完成提取和判断
for instruction in inner_instruction.instructions.iter().skip((current_index + 1) as usize) {
let compiled = &instruction.instruction;
let program_id = accounts[compiled.program_id_index as usize];
if !SYSTEM_PROGRAMS.contains(&program_id) {
break;
}
let data = &compiled.data;
// 使用 SIMD 验证数据格式
if !SimdUtils::validate_data_format(data, 8) {
continue;
}
let get_pubkey = |i: usize| accounts[compiled.accounts[i] as usize];
let (source, destination, amount) = match data[0] {
12 if compiled.accounts.len() >= 4 => {
let amt = u64::from_le_bytes(data[1..9].try_into().unwrap());
(get_pubkey(0), get_pubkey(2), amt)
}
3 if compiled.accounts.len() >= 3 => {
let amt = u64::from_le_bytes(data[1..9].try_into().unwrap());
(get_pubkey(0), get_pubkey(1), amt)
}
2 if compiled.accounts.len() >= 2 => {
let amt = u64::from_le_bytes(data[4..12].try_into().unwrap());
(get_pubkey(0), get_pubkey(1), amt)
}
_ => continue,
};
match (source, destination) {
(s, d) if s == user_to_token && d == to_vault => {
swap_data.from_mint = to_mint;
swap_data.from_amount = amount;
}
(s, d) if s == from_vault && d == user_from_token => {
swap_data.to_mint = from_mint;
swap_data.to_amount = amount;
}
(s, d) if s == user_from_token && d == from_vault => {
swap_data.from_mint = from_mint;
swap_data.from_amount = amount;
}
(s, d) if s == to_vault && d == user_to_token => {
swap_data.to_mint = to_mint;
swap_data.to_amount = amount;
}
(s, d) if s == user_from_token && d == to_vault => {
swap_data.from_mint = from_mint;
swap_data.from_amount = amount;
}
(s, d) if s == from_vault && d == user_to_token => {
swap_data.to_mint = to_mint;
swap_data.to_amount = amount;
}
_ => {}
}
if swap_data.from_mint != Pubkey::default() && swap_data.to_mint != Pubkey::default() {
break;
}
if swap_data.from_amount != 0 && swap_data.to_amount != 0 {
break;
}
}
if swap_data.from_mint != Pubkey::default()
|| swap_data.to_mint != Pubkey::default()
|| swap_data.from_amount != 0
|| swap_data.to_amount != 0
{
Some(swap_data)
} else {
None
}
} }
/// Parse token transfer data from next instructions /// Parse token transfer data from gRPC inner instructions
/// TODO: - wait refactor
pub fn parse_swap_data_from_next_grpc_instructions( pub fn parse_swap_data_from_next_grpc_instructions(
event: &DexEvent, event: &DexEvent,
inner_instruction: &yellowstone_grpc_proto::prelude::InnerInstructions, inner_instruction: &yellowstone_grpc_proto::prelude::InnerInstructions,
current_index: i8, current_index: i8,
accounts: &[Pubkey], accounts: &[Pubkey],
) -> Option<SwapData> { ) -> Option<SwapData> {
let mut swap_data = SwapData { extract_swap_data_from_instructions(
from_mint: Pubkey::default(), event,
to_mint: Pubkey::default(), inner_instruction.instructions.iter().cloned(),
from_amount: 0, current_index,
to_amount: 0, accounts,
description: None, )
};
// 先根据 event 取出关键信息
// let mut user: Option<Pubkey> = None;
let mut from_mint: Option<Pubkey> = None;
let mut to_mint: Option<Pubkey> = None;
let mut user_from_token: Option<Pubkey> = None;
let mut user_to_token: Option<Pubkey> = None;
let mut from_vault: Option<Pubkey> = None;
let mut to_vault: Option<Pubkey> = None;
match event {
DexEvent::BonkTradeEvent(e) => {
// user = Some(e.payer);
from_mint = Some(e.base_token_mint);
to_mint = Some(e.quote_token_mint);
user_from_token = Some(e.user_base_token);
user_to_token = Some(e.user_quote_token);
from_vault = Some(e.base_vault);
to_vault = Some(e.quote_vault);
}
DexEvent::PumpFunTradeEvent(e) => {
swap_data.from_mint = if e.is_buy { *SOL_MINT } else { e.mint };
swap_data.to_mint = if e.is_buy { e.mint } else { *SOL_MINT };
}
DexEvent::PumpSwapBuyEvent(e) => {
swap_data.from_mint = e.quote_mint;
swap_data.to_mint = e.base_mint;
}
DexEvent::PumpSwapSellEvent(e) => {
swap_data.from_mint = e.base_mint;
swap_data.to_mint = e.quote_mint;
}
DexEvent::RaydiumCpmmSwapEvent(e) => {
// user = Some(e.payer);
from_mint = Some(e.input_token_mint);
to_mint = Some(e.output_token_mint);
user_from_token = Some(e.input_token_account);
user_to_token = Some(e.output_token_account);
from_vault = Some(e.input_vault);
to_vault = Some(e.output_vault);
}
DexEvent::RaydiumClmmSwapEvent(e) => {
// user = Some(e.payer);
swap_data.description =
Some("Unable to get from_mint and to_mint from RaydiumClmmSwapEvent".into());
user_from_token = Some(e.input_token_account);
user_to_token = Some(e.output_token_account);
from_vault = Some(e.input_vault);
to_vault = Some(e.output_vault);
}
DexEvent::RaydiumClmmSwapV2Event(e) => {
// user = Some(e.payer);
from_mint = Some(e.input_vault_mint);
to_mint = Some(e.output_vault_mint);
user_from_token = Some(e.input_token_account);
user_to_token = Some(e.output_token_account);
from_vault = Some(e.input_vault);
to_vault = Some(e.output_vault);
}
DexEvent::RaydiumAmmV4SwapEvent(e) => {
// user = Some(e.user_source_owner);
swap_data.description =
Some("Unable to get from_mint and to_mint from RaydiumAmmV4SwapEvent".into());
user_from_token = Some(e.user_source_token_account);
user_to_token = Some(e.user_destination_token_account);
from_vault = Some(e.pool_pc_token_account);
to_vault = Some(e.pool_coin_token_account);
}
_ => {}
}
let user_to_token = user_to_token.unwrap_or_default();
let user_from_token = user_from_token.unwrap_or_default();
let to_vault = to_vault.unwrap_or_default();
let from_vault = from_vault.unwrap_or_default();
let to_mint = to_mint.unwrap_or_default();
let from_mint = from_mint.unwrap_or_default();
// 单次循环完成提取和判断
for instruction in inner_instruction.instructions.iter().skip((current_index + 1) as usize) {
let compiled = &instruction;
let program_id = accounts[compiled.program_id_index as usize];
if !SYSTEM_PROGRAMS.contains(&program_id) {
break;
}
let data = &compiled.data;
// 使用 SIMD 验证数据格式
if !SimdUtils::validate_data_format(data, 8) {
continue;
}
let get_pubkey = |i: usize| accounts[compiled.accounts[i] as usize];
let (source, destination, amount) = match data[0] {
12 if compiled.accounts.len() >= 4 => {
let amt = u64::from_le_bytes(data[1..9].try_into().unwrap());
(get_pubkey(0), get_pubkey(2), amt)
}
3 if compiled.accounts.len() >= 3 => {
let amt = u64::from_le_bytes(data[1..9].try_into().unwrap());
(get_pubkey(0), get_pubkey(1), amt)
}
2 if compiled.accounts.len() >= 2 => {
let amt = u64::from_le_bytes(data[4..12].try_into().unwrap());
(get_pubkey(0), get_pubkey(1), amt)
}
_ => continue,
};
match (source, destination) {
(s, d) if s == user_to_token && d == to_vault => {
swap_data.from_mint = to_mint;
swap_data.from_amount = amount;
}
(s, d) if s == from_vault && d == user_from_token => {
swap_data.to_mint = from_mint;
swap_data.to_amount = amount;
}
(s, d) if s == user_from_token && d == from_vault => {
swap_data.from_mint = from_mint;
swap_data.from_amount = amount;
}
(s, d) if s == to_vault && d == user_to_token => {
swap_data.to_mint = to_mint;
swap_data.to_amount = amount;
}
(s, d) if s == user_from_token && d == to_vault => {
swap_data.from_mint = from_mint;
swap_data.from_amount = amount;
}
(s, d) if s == from_vault && d == user_to_token => {
swap_data.to_mint = to_mint;
swap_data.to_amount = amount;
}
_ => {}
}
if swap_data.from_mint != Pubkey::default() && swap_data.to_mint != Pubkey::default() {
break;
}
if swap_data.from_amount != 0 && swap_data.to_amount != 0 {
break;
}
}
if swap_data.from_mint != Pubkey::default()
|| swap_data.to_mint != Pubkey::default()
|| swap_data.from_amount != 0
|| swap_data.to_amount != 0
{
Some(swap_data)
} else {
None
}
} }
src/streaming/event_parser/common/utils.rs
+2 -2
View File
@@ -58,14 +58,14 @@ pub fn read_u8_le(data: &[u8], offset: usize) -> Option<u8> {
} }
pub fn read_option_bool(data: &[u8], offset: &mut usize) -> Option<Option<bool>> { pub fn read_option_bool(data: &[u8], offset: &mut usize) -> Option<Option<bool>> {
let has_value = data.get(*offset)?.clone(); let has_value = data.get(*offset).copied()?;
*offset += 1; *offset += 1;
if has_value == 0 { if has_value == 0 {
return Some(None); return Some(None);
} }
let value = data.get(*offset)?.clone(); let value = data.get(*offset).copied()?;
*offset += 1; *offset += 1;
Some(Some(value != 0)) Some(Some(value != 0))
src/streaming/event_parser/core/event_parser.rs
+1 -1
View File
@@ -85,7 +85,7 @@ impl EventParser {
let recent_blockhash = if message.recent_blockhash.is_empty() { let recent_blockhash = if message.recent_blockhash.is_empty() {
None None
} else { } else {
Some(bs58::encode(&message.recent_blockhash).into_string()) Some(solana_sdk::bs58::encode(&message.recent_blockhash).into_string())
}; };
Self::parse_instruction_events_from_grpc_transaction( Self::parse_instruction_events_from_grpc_transaction(
protocols, protocols,
src/streaming/event_parser/core/global_state.rs
+3 -8
View File
@@ -49,16 +49,11 @@ impl GlobalState {
return; // Another thread is cleaning up return; // Another thread is cleaning up
} }
// Collect signatures to remove (random selection for simplicity) // Collect only the batch we need to remove (avoid allocating full list)
let mut signatures_to_remove: Vec<Signature> = self.signature_data.iter() let signatures_to_remove: Vec<Signature> = self.signature_data.iter()
.take(CLEANUP_BATCH_SIZE)
.map(|entry| *entry.key()) .map(|entry| *entry.key())
.collect(); .collect();
if signatures_to_remove.len() <= MAX_SIGNATURES {
return; // Race condition, already cleaned up
}
signatures_to_remove.truncate(CLEANUP_BATCH_SIZE);
// Remove old signatures atomically // Remove old signatures atomically
for signature in signatures_to_remove { for signature in signatures_to_remove {
src/streaming/event_parser/core/traits.rs
+80 -116
View File
@@ -90,122 +90,86 @@ pub enum DexEvent {
SetComputeUnitPriceEvent(SetComputeUnitPriceEvent), SetComputeUnitPriceEvent(SetComputeUnitPriceEvent),
} }
impl DexEvent { /// Macro to generate metadata accessors for all DexEvent variants
pub fn metadata(&self) -> &EventMetadata { macro_rules! impl_dex_event_metadata {
match self { ($($variant:ident),* $(,)?) => {
DexEvent::BonkTradeEvent(e) => &e.metadata, impl DexEvent {
DexEvent::BonkPoolCreateEvent(e) => &e.metadata, pub fn metadata(&self) -> &EventMetadata {
DexEvent::BonkMigrateToAmmEvent(e) => &e.metadata, match self {
DexEvent::BonkMigrateToCpswapEvent(e) => &e.metadata, $(DexEvent::$variant(e) => &e.metadata,)*
DexEvent::BonkPoolStateAccountEvent(e) => &e.metadata, }
DexEvent::BonkGlobalConfigAccountEvent(e) => &e.metadata, }
DexEvent::BonkPlatformConfigAccountEvent(e) => &e.metadata,
DexEvent::PumpFunCreateTokenEvent(e) => &e.metadata,
DexEvent::PumpFunCreateV2TokenEvent(e) => &e.metadata,
DexEvent::PumpFunTradeEvent(e) => &e.metadata,
DexEvent::PumpFunMigrateEvent(e) => &e.metadata,
DexEvent::PumpFunBondingCurveAccountEvent(e) => &e.metadata,
DexEvent::PumpFunGlobalAccountEvent(e) => &e.metadata,
DexEvent::PumpSwapBuyEvent(e) => &e.metadata,
DexEvent::PumpSwapSellEvent(e) => &e.metadata,
DexEvent::PumpSwapCreatePoolEvent(e) => &e.metadata,
DexEvent::PumpSwapDepositEvent(e) => &e.metadata,
DexEvent::PumpSwapWithdrawEvent(e) => &e.metadata,
DexEvent::PumpSwapGlobalConfigAccountEvent(e) => &e.metadata,
DexEvent::PumpSwapPoolAccountEvent(e) => &e.metadata,
DexEvent::RaydiumAmmV4SwapEvent(e) => &e.metadata,
DexEvent::RaydiumAmmV4DepositEvent(e) => &e.metadata,
DexEvent::RaydiumAmmV4WithdrawEvent(e) => &e.metadata,
DexEvent::RaydiumAmmV4WithdrawPnlEvent(e) => &e.metadata,
DexEvent::RaydiumAmmV4Initialize2Event(e) => &e.metadata,
DexEvent::RaydiumAmmV4AmmInfoAccountEvent(e) => &e.metadata,
DexEvent::RaydiumClmmSwapEvent(e) => &e.metadata,
DexEvent::RaydiumClmmSwapV2Event(e) => &e.metadata,
DexEvent::RaydiumClmmClosePositionEvent(e) => &e.metadata,
DexEvent::RaydiumClmmIncreaseLiquidityV2Event(e) => &e.metadata,
DexEvent::RaydiumClmmDecreaseLiquidityV2Event(e) => &e.metadata,
DexEvent::RaydiumClmmCreatePoolEvent(e) => &e.metadata,
DexEvent::RaydiumClmmOpenPositionWithToken22NftEvent(e) => &e.metadata,
DexEvent::RaydiumClmmOpenPositionV2Event(e) => &e.metadata,
DexEvent::RaydiumClmmAmmConfigAccountEvent(e) => &e.metadata,
DexEvent::RaydiumClmmPoolStateAccountEvent(e) => &e.metadata,
DexEvent::RaydiumClmmTickArrayStateAccountEvent(e) => &e.metadata,
DexEvent::RaydiumCpmmSwapEvent(e) => &e.metadata,
DexEvent::RaydiumCpmmDepositEvent(e) => &e.metadata,
DexEvent::RaydiumCpmmWithdrawEvent(e) => &e.metadata,
DexEvent::RaydiumCpmmInitializeEvent(e) => &e.metadata,
DexEvent::RaydiumCpmmAmmConfigAccountEvent(e) => &e.metadata,
DexEvent::RaydiumCpmmPoolStateAccountEvent(e) => &e.metadata,
DexEvent::MeteoraDammV2SwapEvent(e) => &e.metadata,
DexEvent::MeteoraDammV2Swap2Event(e) => &e.metadata,
DexEvent::MeteoraDammV2InitializePoolEvent(e) => &e.metadata,
DexEvent::MeteoraDammV2InitializeCustomizablePoolEvent(e) => &e.metadata,
DexEvent::MeteoraDammV2InitializePoolWithDynamicConfigEvent(e) => &e.metadata,
DexEvent::TokenAccountEvent(e) => &e.metadata,
DexEvent::NonceAccountEvent(e) => &e.metadata,
DexEvent::TokenInfoEvent(e) => &e.metadata,
DexEvent::BlockMetaEvent(e) => &e.metadata,
DexEvent::SetComputeUnitLimitEvent(e) => &e.metadata,
DexEvent::SetComputeUnitPriceEvent(e) => &e.metadata,
}
}
pub fn metadata_mut(&mut self) -> &mut EventMetadata { pub fn metadata_mut(&mut self) -> &mut EventMetadata {
match self { match self {
DexEvent::BonkTradeEvent(e) => &mut e.metadata, $(DexEvent::$variant(e) => &mut e.metadata,)*
DexEvent::BonkPoolCreateEvent(e) => &mut e.metadata, }
DexEvent::BonkMigrateToAmmEvent(e) => &mut e.metadata, }
DexEvent::BonkMigrateToCpswapEvent(e) => &mut e.metadata,
DexEvent::BonkPoolStateAccountEvent(e) => &mut e.metadata,
DexEvent::BonkGlobalConfigAccountEvent(e) => &mut e.metadata,
DexEvent::BonkPlatformConfigAccountEvent(e) => &mut e.metadata,
DexEvent::PumpFunCreateTokenEvent(e) => &mut e.metadata,
DexEvent::PumpFunCreateV2TokenEvent(e) => &mut e.metadata,
DexEvent::PumpFunTradeEvent(e) => &mut e.metadata,
DexEvent::PumpFunMigrateEvent(e) => &mut e.metadata,
DexEvent::PumpFunBondingCurveAccountEvent(e) => &mut e.metadata,
DexEvent::PumpFunGlobalAccountEvent(e) => &mut e.metadata,
DexEvent::PumpSwapBuyEvent(e) => &mut e.metadata,
DexEvent::PumpSwapSellEvent(e) => &mut e.metadata,
DexEvent::PumpSwapCreatePoolEvent(e) => &mut e.metadata,
DexEvent::PumpSwapDepositEvent(e) => &mut e.metadata,
DexEvent::PumpSwapWithdrawEvent(e) => &mut e.metadata,
DexEvent::PumpSwapGlobalConfigAccountEvent(e) => &mut e.metadata,
DexEvent::PumpSwapPoolAccountEvent(e) => &mut e.metadata,
DexEvent::RaydiumAmmV4SwapEvent(e) => &mut e.metadata,
DexEvent::RaydiumAmmV4DepositEvent(e) => &mut e.metadata,
DexEvent::RaydiumAmmV4WithdrawEvent(e) => &mut e.metadata,
DexEvent::RaydiumAmmV4WithdrawPnlEvent(e) => &mut e.metadata,
DexEvent::RaydiumAmmV4Initialize2Event(e) => &mut e.metadata,
DexEvent::RaydiumAmmV4AmmInfoAccountEvent(e) => &mut e.metadata,
DexEvent::RaydiumClmmSwapEvent(e) => &mut e.metadata,
DexEvent::RaydiumClmmSwapV2Event(e) => &mut e.metadata,
DexEvent::RaydiumClmmClosePositionEvent(e) => &mut e.metadata,
DexEvent::RaydiumClmmIncreaseLiquidityV2Event(e) => &mut e.metadata,
DexEvent::RaydiumClmmDecreaseLiquidityV2Event(e) => &mut e.metadata,
DexEvent::RaydiumClmmCreatePoolEvent(e) => &mut e.metadata,
DexEvent::RaydiumClmmOpenPositionWithToken22NftEvent(e) => &mut e.metadata,
DexEvent::RaydiumClmmOpenPositionV2Event(e) => &mut e.metadata,
DexEvent::RaydiumClmmAmmConfigAccountEvent(e) => &mut e.metadata,
DexEvent::RaydiumClmmPoolStateAccountEvent(e) => &mut e.metadata,
DexEvent::RaydiumClmmTickArrayStateAccountEvent(e) => &mut e.metadata,
DexEvent::RaydiumCpmmSwapEvent(e) => &mut e.metadata,
DexEvent::RaydiumCpmmDepositEvent(e) => &mut e.metadata,
DexEvent::RaydiumCpmmWithdrawEvent(e) => &mut e.metadata,
DexEvent::RaydiumCpmmInitializeEvent(e) => &mut e.metadata,
DexEvent::RaydiumCpmmAmmConfigAccountEvent(e) => &mut e.metadata,
DexEvent::RaydiumCpmmPoolStateAccountEvent(e) => &mut e.metadata,
DexEvent::MeteoraDammV2SwapEvent(e) => &mut e.metadata,
DexEvent::MeteoraDammV2Swap2Event(e) => &mut e.metadata,
DexEvent::MeteoraDammV2InitializePoolEvent(e) => &mut e.metadata,
DexEvent::MeteoraDammV2InitializeCustomizablePoolEvent(e) => &mut e.metadata,
DexEvent::MeteoraDammV2InitializePoolWithDynamicConfigEvent(e) => &mut e.metadata,
DexEvent::TokenAccountEvent(e) => &mut e.metadata,
DexEvent::NonceAccountEvent(e) => &mut e.metadata,
DexEvent::TokenInfoEvent(e) => &mut e.metadata,
DexEvent::BlockMetaEvent(e) => &mut e.metadata,
DexEvent::SetComputeUnitLimitEvent(e) => &mut e.metadata,
DexEvent::SetComputeUnitPriceEvent(e) => &mut e.metadata,
} }
} };
} }
impl_dex_event_metadata!(
// Bonk events
BonkTradeEvent,
BonkPoolCreateEvent,
BonkMigrateToAmmEvent,
BonkMigrateToCpswapEvent,
BonkPoolStateAccountEvent,
BonkGlobalConfigAccountEvent,
BonkPlatformConfigAccountEvent,
// PumpFun events
PumpFunCreateTokenEvent,
PumpFunCreateV2TokenEvent,
PumpFunTradeEvent,
PumpFunMigrateEvent,
PumpFunBondingCurveAccountEvent,
PumpFunGlobalAccountEvent,
// PumpSwap events
PumpSwapBuyEvent,
PumpSwapSellEvent,
PumpSwapCreatePoolEvent,
PumpSwapDepositEvent,
PumpSwapWithdrawEvent,
PumpSwapGlobalConfigAccountEvent,
PumpSwapPoolAccountEvent,
// Raydium AMM V4 events
RaydiumAmmV4SwapEvent,
RaydiumAmmV4DepositEvent,
RaydiumAmmV4WithdrawEvent,
RaydiumAmmV4WithdrawPnlEvent,
RaydiumAmmV4Initialize2Event,
RaydiumAmmV4AmmInfoAccountEvent,
// Raydium CLMM events
RaydiumClmmSwapEvent,
RaydiumClmmSwapV2Event,
RaydiumClmmClosePositionEvent,
RaydiumClmmIncreaseLiquidityV2Event,
RaydiumClmmDecreaseLiquidityV2Event,
RaydiumClmmCreatePoolEvent,
RaydiumClmmOpenPositionWithToken22NftEvent,
RaydiumClmmOpenPositionV2Event,
RaydiumClmmAmmConfigAccountEvent,
RaydiumClmmPoolStateAccountEvent,
RaydiumClmmTickArrayStateAccountEvent,
// Raydium CPMM events
RaydiumCpmmSwapEvent,
RaydiumCpmmDepositEvent,
RaydiumCpmmWithdrawEvent,
RaydiumCpmmInitializeEvent,
RaydiumCpmmAmmConfigAccountEvent,
RaydiumCpmmPoolStateAccountEvent,
// Meteora DAMM v2 events
MeteoraDammV2SwapEvent,
MeteoraDammV2Swap2Event,
MeteoraDammV2InitializePoolEvent,
MeteoraDammV2InitializeCustomizablePoolEvent,
MeteoraDammV2InitializePoolWithDynamicConfigEvent,
// Common events
TokenAccountEvent,
NonceAccountEvent,
TokenInfoEvent,
BlockMetaEvent,
SetComputeUnitLimitEvent,
SetComputeUnitPriceEvent,
);
src/streaming/grpc/subscription.rs
+1 -1
View File
@@ -85,7 +85,7 @@ impl SubscriptionManager {
if event_type_filter.is_some() && !event_type_filter.unwrap().include_account_event() { if event_type_filter.is_some() && !event_type_filter.unwrap().include_account_event() {
return None; return None;
} }
if account_filter.len() == 0 { if account_filter.is_empty() {
return None; return None;
} }
let mut accounts = HashMap::new(); let mut accounts = HashMap::new();
src/streaming/grpc/types.rs
-60
View File
@@ -104,63 +104,3 @@ impl Default for TransactionPretty {
} }
} }
// impl From<SubscribeUpdateAccount> for AccountPretty {
// fn from(account: SubscribeUpdateAccount) -> Self {
// let account_info = account.account.unwrap();
// Self {
// slot: account.slot,
// signature: if let Some(txn_signature) = account_info.txn_signature {
// Signature::try_from(txn_signature.as_slice()).expect("valid signature")
// } else {
// Signature::default()
// },
// pubkey: Pubkey::try_from(account_info.pubkey.as_slice()).expect("valid pubkey"),
// executable: account_info.executable,
// lamports: account_info.lamports,
// owner: Pubkey::try_from(account_info.owner.as_slice()).expect("valid pubkey"),
// rent_epoch: account_info.rent_epoch,
// data: account_info.data,
// recv_us: get_high_perf_clock(),
// }
// }
// }
// impl From<(SubscribeUpdateBlockMeta, Option<Timestamp>)> for BlockMetaPretty {
// fn from(
// (SubscribeUpdateBlockMeta { slot, blockhash, .. }, block_time): (
// SubscribeUpdateBlockMeta,
// Option<Timestamp>,
// ),
// ) -> Self {
// Self {
// block_hash: blockhash,
// block_time,
// slot,
// recv_us: get_high_perf_clock(),
// }
// }
// }
// impl From<(SubscribeUpdateTransaction, Option<Timestamp>)> for TransactionPretty {
// fn from(
// (SubscribeUpdateTransaction { transaction, slot }, block_time): (
// SubscribeUpdateTransaction,
// Option<Timestamp>,
// ),
// ) -> Self {
// let tx = transaction.expect("should be defined");
// // 根据用户说明，交易索引在 transaction.index 中
// let tx_index = tx.index;
// Self {
// slot,
// tx_index: Some(tx_index), // 提取交易索引
// block_time,
// block_hash: String::new(),
// signature: Signature::try_from(tx.signature.as_slice()).expect("valid signature"),
// is_vote: tx.is_vote,
// tx: yellowstone_grpc_proto::convert_from::create_tx_with_meta(tx)
// .expect("valid tx with meta"),
// recv_us: get_high_perf_clock(),
// }
// }
// }
src/streaming/shred_stream.rs
+2 -2
View File
@@ -12,7 +12,7 @@ use crate::streaming::event_parser::{Protocol, DexEvent};
use crate::streaming::grpc::MetricsManager; use crate::streaming::grpc::MetricsManager;
use crate::streaming::shred::pool::factory; use crate::streaming::shred::pool::factory;
use log::error; use log::error;
use solana_entry::entry::Entry; use solana_entry::entry::Entry as SolanaEntry;
use super::ShredStreamGrpc; use super::ShredStreamGrpc;
@@ -49,7 +49,7 @@ impl ShredStreamGrpc {
while let Some(message) = stream.next().await { while let Some(message) = stream.next().await {
match message { match message {
Ok(msg) => { Ok(msg) => {
if let Ok(entries) = bincode::deserialize::<Vec<Entry>>(&msg.entries) { if let Ok(entries) = bincode::deserialize::<Vec<SolanaEntry>>(&msg.entries) {
for entry in entries { for entry in entries {
for (tx_index, transaction) in entry.transactions.iter().enumerate() { for (tx_index, transaction) in entry.transactions.iter().enumerate() {
let transaction_with_slot = let transaction_with_slot =