mirror of
https://github.com/0xfnzero/solana-streamer.git
synced 2026-08-14 01:18:04 +00:00
feat: refactor to multi-protocol event streaming system
Major architectural refactor, upgrading from simple logging system to comprehensive multi-protocol Solana DEX event streaming system: ✨ New Features: - Support for 5 DEX protocols: PumpFun, PumpSwap, Bonk, Raydium CPMM, Raydium CLMM - Implement unified event interface (UnifiedEvent trait) and event factory pattern - Add dual streaming support: Yellowstone gRPC and ShredStream - Add Chinese documentation (README_CN.md) 🏗️ Architectural Improvements: - Refactor event parsing system with modular design - Implement protocol-specific parsers and event types - Optimize dependency management, update Cargo.toml - Remove legacy logging modules, clean up redundant code 📊 Statistics: - Added 46 files, 4511 lines of code - Removed 1381 lines of legacy code - Net addition of 3130 lines of code Tech Stack: - Rust async/await for asynchronous processing - Protocol Buffers support - Multi-protocol event parsing - High-performance event stream subscription
This commit is contained in:
ysq
@@ -1,97 +0,0 @@
|
||||
use borsh::{BorshDeserialize, BorshSerialize};
|
||||
use solana_sdk::{pubkey::Pubkey, transaction::VersionedTransaction};
|
||||
|
||||
use crate::error::{ClientError, ClientResult};
|
||||
|
||||
#[derive(Debug)]
|
||||
pub enum DexInstruction {
|
||||
CreateToken(CreateTokenInfo),
|
||||
UserTrade(TradeInfo),
|
||||
BotTrade(TradeInfo),
|
||||
Other,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Default, PartialEq, BorshDeserialize, BorshSerialize)]
|
||||
pub struct CreateTokenInfo {
|
||||
pub slot: u64,
|
||||
pub name: String,
|
||||
pub symbol: String,
|
||||
pub uri: String,
|
||||
pub mint: Pubkey,
|
||||
pub bonding_curve: Pubkey,
|
||||
pub user: Pubkey,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Default, PartialEq, BorshDeserialize, BorshSerialize)]
|
||||
pub struct TradeInfo {
|
||||
pub slot: u64,
|
||||
pub mint: Pubkey,
|
||||
pub sol_amount: u64,
|
||||
pub token_amount: u64,
|
||||
pub is_buy: bool,
|
||||
pub user: Pubkey,
|
||||
pub timestamp: i64,
|
||||
pub virtual_sol_reserves: u64,
|
||||
pub virtual_token_reserves: u64,
|
||||
pub real_sol_reserves: u64,
|
||||
pub real_token_reserves: u64,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Default, PartialEq, BorshDeserialize, BorshSerialize)]
|
||||
pub struct CompleteInfo {
|
||||
pub user: Pubkey,
|
||||
pub mint: Pubkey,
|
||||
pub bonding_curve: Pubkey,
|
||||
pub timestamp: u64,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Default, PartialEq, BorshDeserialize, BorshSerialize)]
|
||||
pub struct SwapBaseInLog {
|
||||
pub log_type: u8,
|
||||
// input
|
||||
pub amount_in: u64,
|
||||
pub minimum_out: u64,
|
||||
pub direction: u64,
|
||||
// user info
|
||||
pub user_source: u64,
|
||||
// pool info
|
||||
pub pool_coin: u64,
|
||||
pub pool_pc: u64,
|
||||
// calc result
|
||||
pub out_amount: u64,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Default, PartialEq)]
|
||||
pub struct TransferInfo {
|
||||
pub slot: u64,
|
||||
pub signature: String,
|
||||
pub tx: Option<VersionedTransaction>,
|
||||
}
|
||||
|
||||
pub trait EventTrait: Sized + std::fmt::Debug {
|
||||
fn from_bytes(bytes: &[u8]) -> ClientResult<Self>;
|
||||
}
|
||||
|
||||
impl EventTrait for CreateTokenInfo {
|
||||
fn from_bytes(bytes: &[u8]) -> ClientResult<Self> {
|
||||
CreateTokenInfo::try_from_slice(bytes).map_err(|e| ClientError::Other(e.to_string()))
|
||||
}
|
||||
}
|
||||
|
||||
impl EventTrait for TradeInfo {
|
||||
fn from_bytes(bytes: &[u8]) -> ClientResult<Self> {
|
||||
TradeInfo::try_from_slice(bytes).map_err(|e| ClientError::Other(e.to_string()))
|
||||
}
|
||||
}
|
||||
|
||||
impl EventTrait for CompleteInfo {
|
||||
fn from_bytes(bytes: &[u8]) -> ClientResult<Self> {
|
||||
CompleteInfo::try_from_slice(bytes).map_err(|e| ClientError::Other(e.to_string()))
|
||||
}
|
||||
}
|
||||
|
||||
impl EventTrait for SwapBaseInLog {
|
||||
fn from_bytes(bytes: &[u8]) -> ClientResult<Self> {
|
||||
SwapBaseInLog::try_from_slice(bytes).map_err(|e| ClientError::Other(e.to_string()))
|
||||
}
|
||||
}
|
||||
@@ -1,94 +0,0 @@
|
||||
use base64::engine::general_purpose;
|
||||
use base64::Engine;
|
||||
use regex::Regex;
|
||||
use crate::common::logs_data::{CreateTokenInfo, TradeInfo, EventTrait, TransferInfo};
|
||||
|
||||
pub const PROGRAM_DATA: &str = "Program data: ";
|
||||
|
||||
#[derive(Debug)]
|
||||
pub enum PumpfunEvent {
|
||||
NewToken(CreateTokenInfo),
|
||||
NewDevTrade(TradeInfo),
|
||||
NewUserTrade(TradeInfo),
|
||||
NewBotTrade(TradeInfo),
|
||||
Error(String),
|
||||
}
|
||||
|
||||
|
||||
#[derive(Debug)]
|
||||
pub enum DexEvent {
|
||||
NewToken(CreateTokenInfo),
|
||||
NewUserTrade(TradeInfo),
|
||||
NewBotTrade(TradeInfo),
|
||||
Error(String),
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub enum SystemEvent {
|
||||
NewTransfer(TransferInfo),
|
||||
Error(String),
|
||||
}
|
||||
|
||||
// #[derive(Debug, Clone, Copy)]
|
||||
// pub struct PumpEvent {}
|
||||
|
||||
impl PumpfunEvent {
|
||||
pub fn parse_logs(logs: &Vec<String>) -> (Option<CreateTokenInfo>, Option<TradeInfo>) {
|
||||
let mut create_info: Option<CreateTokenInfo> = None;
|
||||
let mut trade_info: Option<TradeInfo> = None;
|
||||
|
||||
if !logs.is_empty() {
|
||||
let logs_iter = logs.iter().peekable();
|
||||
|
||||
for l in logs_iter.rev() {
|
||||
if let Some(log) = l.strip_prefix(PROGRAM_DATA) {
|
||||
let borsh_bytes = general_purpose::STANDARD.decode(log).unwrap();
|
||||
let slice: &[u8] = &borsh_bytes[8..];
|
||||
|
||||
if create_info.is_none() {
|
||||
if let Ok(e) = CreateTokenInfo::from_bytes(slice) {
|
||||
create_info = Some(e);
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
if trade_info.is_none() {
|
||||
if let Ok(e) = TradeInfo::from_bytes(slice) {
|
||||
trade_info = Some(e);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
(create_info, trade_info)
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy)]
|
||||
pub struct RaydiumEvent {}
|
||||
|
||||
impl RaydiumEvent {
|
||||
pub fn parse_logs<T: EventTrait + Clone>(logs: &Vec<String>) -> Option<T> {
|
||||
let mut event: Option<T> = None;
|
||||
|
||||
if !logs.is_empty() {
|
||||
let logs_iter = logs.iter().peekable();
|
||||
|
||||
for l in logs_iter.rev() {
|
||||
let re = Regex::new(r"ray_log: (?P<base64>[A-Za-z0-9+/=]+)").unwrap();
|
||||
|
||||
if let Some(caps) = re.captures(l) {
|
||||
if let Some(base64) = caps.name("base64") {
|
||||
let bytes = general_purpose::STANDARD.decode(base64.as_str()).unwrap();
|
||||
|
||||
if let Ok(e) = T::from_bytes(&bytes) {
|
||||
event = Some(e);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
event
|
||||
}
|
||||
}
|
||||
@@ -1,152 +0,0 @@
|
||||
use crate::common::logs_data::DexInstruction;
|
||||
use crate::common::logs_parser::{parse_create_token_data, parse_trade_data, parse_instruction_create_token_data, parse_instruction_trade_data};
|
||||
use crate::error::ClientResult;
|
||||
pub struct LogFilter;
|
||||
use solana_sdk::pubkey::Pubkey;
|
||||
use std::str::FromStr;
|
||||
|
||||
use solana_sdk::transaction::VersionedTransaction;
|
||||
|
||||
impl LogFilter {
|
||||
const PROGRAM_ID: &'static str = "6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P";
|
||||
|
||||
/// Parse transaction logs and return instruction type and data
|
||||
pub fn parse_compiled_instruction(
|
||||
versioned_tx: VersionedTransaction,
|
||||
bot_wallet: Option<Pubkey>) -> ClientResult<Vec<DexInstruction>> {
|
||||
let compiled_instructions = versioned_tx.message.instructions();
|
||||
let accounts = versioned_tx.message.static_account_keys();
|
||||
let program_id = Pubkey::from_str(Self::PROGRAM_ID).unwrap_or_default();
|
||||
let pump_index = accounts.iter().position(|key| key == &program_id);
|
||||
let mut instructions: Vec<DexInstruction> = Vec::new();
|
||||
if let Some(index) = pump_index {
|
||||
for instruction in compiled_instructions {
|
||||
if instruction.program_id_index as usize == index {
|
||||
let all_accounts_valid = instruction.accounts.iter()
|
||||
.all(|&acc_idx| (acc_idx as usize) < accounts.len());
|
||||
if !all_accounts_valid {
|
||||
continue;
|
||||
}
|
||||
match instruction.data.first() {
|
||||
// create
|
||||
Some(&24) => {
|
||||
if let Ok(token_info) = parse_instruction_create_token_data(instruction, accounts) {
|
||||
instructions.push(DexInstruction::CreateToken(token_info));
|
||||
};
|
||||
}
|
||||
// buy
|
||||
Some(&102) if instruction.data.len() == 24 && instruction.accounts.len() >= 12 => {
|
||||
if let Ok(trade_info) = parse_instruction_trade_data(instruction, accounts, true) {
|
||||
if let Some(bot_wallet_pubkey) = bot_wallet {
|
||||
if trade_info.user.to_string() == bot_wallet_pubkey.to_string() {
|
||||
instructions.push(DexInstruction::BotTrade(trade_info));
|
||||
} else {
|
||||
instructions.push(DexInstruction::UserTrade(trade_info));
|
||||
}
|
||||
} else {
|
||||
instructions.push(DexInstruction::UserTrade(trade_info));
|
||||
}
|
||||
};
|
||||
}
|
||||
// sell
|
||||
Some(&51) if instruction.data.len() == 24 && instruction.accounts.len() >= 12 => {
|
||||
if let Ok(trade_info) = parse_instruction_trade_data(instruction, accounts, false) {
|
||||
if let Some(bot_wallet_pubkey) = bot_wallet {
|
||||
if trade_info.user.to_string() == bot_wallet_pubkey.to_string() {
|
||||
instructions.push(DexInstruction::BotTrade(trade_info));
|
||||
} else {
|
||||
instructions.push(DexInstruction::UserTrade(trade_info));
|
||||
}
|
||||
} else {
|
||||
instructions.push(DexInstruction::UserTrade(trade_info));
|
||||
}
|
||||
};
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(instructions)
|
||||
}
|
||||
|
||||
|
||||
/// Parse transaction logs and return instruction type and data
|
||||
pub fn parse_instruction(logs: &[String], bot_wallet: Option<Pubkey>) -> ClientResult<Vec<DexInstruction>> {
|
||||
let mut current_instruction = None;
|
||||
let mut program_data = String::new();
|
||||
let mut invoke_depth = 0;
|
||||
let mut last_data_len = 0;
|
||||
let mut instructions = Vec::new();
|
||||
for log in logs {
|
||||
// Check program invocation
|
||||
if log.contains(&format!("Program {} invoke", Self::PROGRAM_ID)) {
|
||||
invoke_depth += 1;
|
||||
if invoke_depth == 1 { // Only reset state at top level call
|
||||
current_instruction = None;
|
||||
program_data.clear();
|
||||
last_data_len = 0;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
// Skip if not in our program
|
||||
if invoke_depth == 0 {
|
||||
continue;
|
||||
}
|
||||
|
||||
// Identify instruction type (only at top level)
|
||||
if invoke_depth == 1 && log.contains("Program log: Instruction:") {
|
||||
if log.contains("Create") {
|
||||
current_instruction = Some("create");
|
||||
} else if log.contains("Buy") || log.contains("Sell") {
|
||||
current_instruction = Some("trade");
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
// Collect Program data
|
||||
if log.starts_with("Program data: ") {
|
||||
let data = log.trim_start_matches("Program data: ");
|
||||
if data.len() > last_data_len {
|
||||
program_data = data.to_string();
|
||||
last_data_len = data.len();
|
||||
}
|
||||
}
|
||||
|
||||
// Check if program ends
|
||||
if log.contains(&format!("Program {} success", Self::PROGRAM_ID)) {
|
||||
invoke_depth -= 1;
|
||||
if invoke_depth == 0 { // Only process data when top level program ends
|
||||
if let Some(instruction_type) = current_instruction {
|
||||
if !program_data.is_empty() {
|
||||
match instruction_type {
|
||||
"create" => {
|
||||
if let Ok(token_info) = parse_create_token_data(&program_data) {
|
||||
instructions.push(DexInstruction::CreateToken(token_info));
|
||||
}
|
||||
},
|
||||
"trade" => {
|
||||
if let Ok(trade_info) = parse_trade_data(&program_data) {
|
||||
if let Some(bot_wallet_pubkey) = bot_wallet {
|
||||
if trade_info.user.to_string() == bot_wallet_pubkey.to_string() {
|
||||
instructions.push(DexInstruction::BotTrade(trade_info));
|
||||
} else {
|
||||
instructions.push(DexInstruction::UserTrade(trade_info));
|
||||
}
|
||||
} else {
|
||||
instructions.push(DexInstruction::UserTrade(trade_info));
|
||||
}
|
||||
}
|
||||
},
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Ok(instructions)
|
||||
}
|
||||
}
|
||||
@@ -1,236 +0,0 @@
|
||||
use std::str::FromStr;
|
||||
|
||||
use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64};
|
||||
|
||||
use crate::error::{ClientError, ClientResult};
|
||||
use crate::common::{
|
||||
logs_data::{DexInstruction, CreateTokenInfo, TradeInfo},
|
||||
logs_filters::LogFilter
|
||||
};
|
||||
|
||||
use solana_sdk::pubkey::Pubkey;
|
||||
use solana_sdk::instruction::CompiledInstruction;
|
||||
use std::time::{SystemTime, UNIX_EPOCH};
|
||||
|
||||
pub async fn process_logs<F>(
|
||||
signature: &str,
|
||||
logs: Vec<String>,
|
||||
callback: F,
|
||||
payer: Option<Pubkey>,
|
||||
) -> ClientResult<()>
|
||||
where
|
||||
F: Fn(&str, DexInstruction) + Send + Sync,
|
||||
{
|
||||
let instructions = LogFilter::parse_instruction(&logs, payer)?;
|
||||
for instruction in instructions {
|
||||
callback(signature, instruction);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// Add parsing function
|
||||
pub fn parse_create_token_data(data: &str) -> ClientResult<CreateTokenInfo> {
|
||||
// First do base64 decoding
|
||||
let decoded = BASE64.decode(data)
|
||||
.map_err(|e| ClientError::Other(format!("Failed to decode base64: {}", e)))?;
|
||||
|
||||
// Skip prefix bytes (if any)
|
||||
let mut cursor = if decoded.len() > 8 { 8 } else { 0 };
|
||||
|
||||
// Read name length and name
|
||||
if cursor + 4 > decoded.len() {
|
||||
return Err(ClientError::Other("Data too short for name length".to_string()));
|
||||
}
|
||||
let name_len = read_u32(&decoded[cursor..]) as usize;
|
||||
cursor += 4;
|
||||
|
||||
if cursor + name_len > decoded.len() {
|
||||
return Err(ClientError::Other(format!("Data too short for name: need {} bytes", name_len)));
|
||||
}
|
||||
let name = String::from_utf8(decoded[cursor..cursor + name_len].to_vec())
|
||||
.map_err(|e| ClientError::Other(format!("Invalid UTF-8 in name: {}", e)))?;
|
||||
cursor += name_len;
|
||||
|
||||
// Read symbol length and symbol
|
||||
if cursor + 4 > decoded.len() {
|
||||
return Err(ClientError::Other("Data too short for symbol length".to_string()));
|
||||
}
|
||||
let symbol_len = read_u32(&decoded[cursor..]) as usize;
|
||||
cursor += 4;
|
||||
|
||||
if cursor + symbol_len > decoded.len() {
|
||||
return Err(ClientError::Other(format!("Data too short for symbol: need {} bytes", symbol_len)));
|
||||
}
|
||||
let symbol = String::from_utf8(decoded[cursor..cursor + symbol_len].to_vec())
|
||||
.map_err(|e| ClientError::Other(format!("Invalid UTF-8 in symbol: {}", e)))?;
|
||||
cursor += symbol_len;
|
||||
|
||||
// Read URI length and URI
|
||||
if cursor + 4 > decoded.len() {
|
||||
return Err(ClientError::Other("Data too short for URI length".to_string()));
|
||||
}
|
||||
let uri_len = read_u32(&decoded[cursor..]) as usize;
|
||||
cursor += 4;
|
||||
|
||||
if cursor + uri_len > decoded.len() {
|
||||
return Err(ClientError::Other(format!("Data too short for URI: need {} bytes", uri_len)));
|
||||
}
|
||||
let uri = String::from_utf8(decoded[cursor..cursor + uri_len].to_vec())
|
||||
.map_err(|e| ClientError::Other(format!("Invalid UTF-8 in uri: {}", e)))?;
|
||||
cursor += uri_len;
|
||||
|
||||
// Make sure there is enough data to read public keys
|
||||
if cursor + 32 * 3 > decoded.len() {
|
||||
return Err(ClientError::Other("Data too short for public keys".to_string()));
|
||||
}
|
||||
|
||||
// Parse Mint Public Key
|
||||
let mint = bs58::encode(&decoded[cursor..cursor+32]).into_string();
|
||||
cursor += 32;
|
||||
|
||||
// Parse Bonding Curve Public Key
|
||||
let bonding_curve = bs58::encode(&decoded[cursor..cursor+32]).into_string();
|
||||
cursor += 32;
|
||||
|
||||
// Parse User Public Key
|
||||
let user = bs58::encode(&decoded[cursor..cursor+32]).into_string();
|
||||
|
||||
Ok(CreateTokenInfo {
|
||||
slot: 0,
|
||||
name,
|
||||
symbol,
|
||||
uri,
|
||||
mint: Pubkey::from_str(&mint).unwrap(),
|
||||
bonding_curve: Pubkey::from_str(&bonding_curve).unwrap(),
|
||||
user: Pubkey::from_str(&user).unwrap(),
|
||||
})
|
||||
}
|
||||
|
||||
fn read_u32(data: &[u8]) -> u32 {
|
||||
let mut bytes = [0u8; 4];
|
||||
bytes.copy_from_slice(&data[..4]);
|
||||
u32::from_le_bytes(bytes)
|
||||
}
|
||||
|
||||
pub fn parse_trade_data(data: &str) -> ClientResult<TradeInfo> {
|
||||
let engine = base64::engine::general_purpose::STANDARD;
|
||||
let decoded = engine.decode(data).map_err(|e|
|
||||
ClientError::Parse(
|
||||
"Failed to decode base64".to_string(),
|
||||
e.to_string()
|
||||
)
|
||||
)?;
|
||||
|
||||
let mut cursor = 8; // Skip prefix
|
||||
|
||||
// 1. Mint (32 bytes)
|
||||
let mint = bs58::encode(&decoded[cursor..cursor + 32]).into_string();
|
||||
cursor += 32;
|
||||
|
||||
// 2. Sol Amount (8 bytes)
|
||||
let sol_amount = u64::from_le_bytes(decoded[cursor..cursor + 8].try_into().unwrap());
|
||||
cursor += 8;
|
||||
|
||||
// 3. Token Amount (8 bytes)
|
||||
let token_amount = u64::from_le_bytes(decoded[cursor..cursor + 8].try_into().unwrap());
|
||||
cursor += 8;
|
||||
|
||||
// 4. Is Buy (1 byte)
|
||||
let is_buy = decoded[cursor] != 0;
|
||||
cursor += 1;
|
||||
|
||||
// 5. User (32 bytes)
|
||||
let user = bs58::encode(&decoded[cursor..cursor + 32]).into_string();
|
||||
cursor += 32;
|
||||
|
||||
// 6. Timestamp (8 bytes)
|
||||
let timestamp = i64::from_le_bytes(decoded[cursor..cursor + 8].try_into().unwrap());
|
||||
cursor += 8;
|
||||
|
||||
// 7. Virtual Sol Reserves (8 bytes)
|
||||
let virtual_sol_reserves = u64::from_le_bytes(decoded[cursor..cursor + 8].try_into().unwrap());
|
||||
cursor += 8;
|
||||
|
||||
// 8. Virtual Token Reserves (8 bytes)
|
||||
let virtual_token_reserves = u64::from_le_bytes(decoded[cursor..cursor + 8].try_into().unwrap());
|
||||
cursor += 8;
|
||||
|
||||
let real_sol_reserves = u64::from_le_bytes(decoded[cursor..cursor + 8].try_into().unwrap());
|
||||
cursor += 8;
|
||||
|
||||
let real_token_reserves = u64::from_le_bytes(decoded[cursor..cursor + 8].try_into().unwrap());
|
||||
|
||||
Ok(TradeInfo {
|
||||
slot: 0,
|
||||
mint: Pubkey::from_str(&mint).unwrap(),
|
||||
sol_amount,
|
||||
token_amount,
|
||||
is_buy,
|
||||
user: Pubkey::from_str(&user).unwrap(),
|
||||
timestamp,
|
||||
virtual_sol_reserves,
|
||||
virtual_token_reserves,
|
||||
real_sol_reserves,
|
||||
real_token_reserves,
|
||||
})
|
||||
}
|
||||
|
||||
fn current_timestamp_millis() -> i64 {
|
||||
let duration = SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.expect("Time went backwards");
|
||||
|
||||
duration.as_millis() as i64
|
||||
}
|
||||
|
||||
pub fn parse_instruction_create_token_data(instruction: &CompiledInstruction, accounts: &[Pubkey]) -> ClientResult<CreateTokenInfo> {
|
||||
let data = instruction.data.clone();
|
||||
let mut offset = 0;
|
||||
offset += 8;
|
||||
let len1 = u32::from_le_bytes(data[offset..offset + 4].try_into().unwrap()) as usize;
|
||||
offset += 4;
|
||||
let name = String::from_utf8_lossy(&data[offset..offset + len1]);
|
||||
offset += len1;
|
||||
let len2 = u32::from_le_bytes(data[offset..offset + 4].try_into().unwrap()) as usize;
|
||||
offset += 4;
|
||||
let symbol = String::from_utf8_lossy(&data[offset..offset + len2]);
|
||||
offset += len2;
|
||||
let _flag = u32::from_le_bytes(data[offset..offset + 4].try_into().unwrap());
|
||||
offset += 4;
|
||||
let hash_start = data.len() - 32;
|
||||
let ipfs_bytes = &data[offset..hash_start];
|
||||
let uri = String::from_utf8_lossy(ipfs_bytes);
|
||||
let mint = accounts[instruction.accounts[0] as usize];
|
||||
let user = accounts[instruction.accounts[7] as usize];
|
||||
let bonding_curve= accounts[instruction.accounts[2] as usize];
|
||||
Ok(CreateTokenInfo {
|
||||
slot: 0,
|
||||
name: name.to_string(),
|
||||
symbol: symbol.to_string(),
|
||||
uri: uri.to_string(),
|
||||
mint,
|
||||
bonding_curve,
|
||||
user,
|
||||
})
|
||||
}
|
||||
|
||||
pub fn parse_instruction_trade_data(instruction: &CompiledInstruction, accounts: &[Pubkey], is_buy: bool) -> ClientResult<TradeInfo> {
|
||||
let data = instruction.data.clone();
|
||||
let amount = u64::from_le_bytes(data[8..16].try_into().unwrap());
|
||||
let max_sol_cost_or_min_sol_output = u64::from_le_bytes(data[16..24].try_into().unwrap());
|
||||
let user = accounts[instruction.accounts[6] as usize];
|
||||
let mint = accounts[instruction.accounts[2] as usize];
|
||||
Ok(TradeInfo {
|
||||
slot: 0,
|
||||
mint,
|
||||
sol_amount: max_sol_cost_or_min_sol_output,
|
||||
token_amount: amount,
|
||||
is_buy,
|
||||
user,
|
||||
timestamp: current_timestamp_millis(),
|
||||
virtual_sol_reserves: 0,
|
||||
virtual_token_reserves: 0,
|
||||
real_sol_reserves: 0,
|
||||
real_token_reserves: 0,
|
||||
})
|
||||
}
|
||||
@@ -1,105 +0,0 @@
|
||||
use solana_client::{
|
||||
nonblocking::pubsub_client::PubsubClient,
|
||||
rpc_config::{RpcTransactionLogsConfig, RpcTransactionLogsFilter}
|
||||
};
|
||||
|
||||
use solana_sdk::{commitment_config::CommitmentConfig, pubkey::Pubkey};
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::mpsc;
|
||||
use tokio::task::JoinHandle;
|
||||
use futures::StreamExt;
|
||||
use crate::common::{
|
||||
logs_data::DexInstruction, logs_filters::LogFilter
|
||||
};
|
||||
|
||||
use super::logs_events::PumpfunEvent;
|
||||
|
||||
/// Subscription handle containing task and unsubscribe logic
|
||||
pub struct SubscriptionHandle {
|
||||
pub task: JoinHandle<()>,
|
||||
pub unsub_fn: Box<dyn Fn() + Send>,
|
||||
}
|
||||
|
||||
impl SubscriptionHandle {
|
||||
pub async fn shutdown(self) {
|
||||
(self.unsub_fn)();
|
||||
self.task.abort();
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn create_pubsub_client(ws_url: &str) -> PubsubClient {
|
||||
PubsubClient::new(ws_url).await.unwrap()
|
||||
}
|
||||
|
||||
/// 启动订阅
|
||||
pub async fn tokens_subscription<F>(
|
||||
ws_url: &str,
|
||||
commitment: CommitmentConfig,
|
||||
callback: F,
|
||||
bot_wallet: Option<Pubkey>,
|
||||
) -> Result<SubscriptionHandle, Box<dyn std::error::Error>>
|
||||
where
|
||||
F: Fn(PumpfunEvent) + Send + Sync + 'static,
|
||||
{
|
||||
let program_address = "6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P".to_string();
|
||||
let logs_filter = RpcTransactionLogsFilter::Mentions(vec![program_address]);
|
||||
|
||||
let logs_config = RpcTransactionLogsConfig {
|
||||
commitment: Some(commitment),
|
||||
};
|
||||
|
||||
// Create PubsubClient
|
||||
let sub_client = Arc::new(PubsubClient::new(ws_url).await.unwrap());
|
||||
|
||||
let sub_client_clone = Arc::clone(&sub_client);
|
||||
|
||||
// Create channel for unsubscribe
|
||||
let (unsub_tx, _) = mpsc::channel(1);
|
||||
|
||||
// Start subscription task
|
||||
let task = tokio::spawn(async move {
|
||||
let (mut stream, _) = sub_client_clone.logs_subscribe(logs_filter, logs_config).await.unwrap();
|
||||
|
||||
loop {
|
||||
let msg = stream.next().await;
|
||||
match msg {
|
||||
Some(msg) => {
|
||||
if let Some(_err) = msg.value.err {
|
||||
continue;
|
||||
}
|
||||
|
||||
let instructions = LogFilter::parse_instruction(&msg.value.logs, bot_wallet).unwrap();
|
||||
for instruction in instructions {
|
||||
match instruction {
|
||||
DexInstruction::CreateToken(token_info) => {
|
||||
callback(PumpfunEvent::NewToken(token_info));
|
||||
}
|
||||
DexInstruction::UserTrade(trade_info) => {
|
||||
callback(PumpfunEvent::NewUserTrade(trade_info));
|
||||
}
|
||||
DexInstruction::BotTrade(trade_info) => {
|
||||
callback(PumpfunEvent::NewBotTrade(trade_info));
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
}
|
||||
None => {
|
||||
println!("Token subscription stream ended");
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
// Return subscription handle and unsubscribe logic
|
||||
Ok(SubscriptionHandle {
|
||||
task,
|
||||
unsub_fn: Box::new(move || {
|
||||
let _ = unsub_tx.try_send(());
|
||||
}),
|
||||
})
|
||||
}
|
||||
|
||||
pub async fn stop_subscription(handle: SubscriptionHandle) {
|
||||
handle.shutdown().await;
|
||||
}
|
||||
Executable → Regular
+2
-6
@@ -1,6 +1,2 @@
|
||||
pub mod logs_data;
|
||||
pub mod logs_parser;
|
||||
pub mod logs_filters;
|
||||
pub mod logs_subscribe;
|
||||
pub mod logs_events;
|
||||
|
||||
pub mod types;
|
||||
pub use types::*;
|
||||
|
||||
@@ -0,0 +1,2 @@
|
||||
pub type SolanaRpcClient = solana_client::nonblocking::rpc_client::RpcClient;
|
||||
pub type AnyResult<T> = anyhow::Result<T>;
|
||||
Reference in New Issue
Block a user