feat: refactor event parsing architecture and add Raydium Launchpad support
Major changes: - Refactor event parsing system: migrate scattered logs_* modules to unified event_parser architecture - Add unified UnifiedEvent trait and EventParser trait for standardized event parsing interface - Introduce GenericEventParser and EventParserFactory for plugin-style protocol extension - Add match_event! macro to simplify event type matching and handling - Add Raydium Launchpad (Bonk.fun) protocol support: - Complete buy/sell trading functionality - Pool state management and querying - Event parsing and subscription - Update trading system to support new protocol architecture - Refactor constants organization, rename raydium to raydium_launchpad - Update documentation and example code Technical improvements: - Unified event interface design for better code maintainability - Factory pattern implementation for dynamic protocol loading - Generic event parser to reduce code duplication - Improved error handling and type safety Breaking Changes: - Remove old logs_* modules, use new event_parser system - Protocol constants path change: raydium -> raydium_launchpad - Event subscription API updated to unified interface
This commit is contained in:
ysq
+45
-209
@@ -7,19 +7,16 @@ use tonic::transport::Channel;
|
||||
use log::error;
|
||||
use solana_sdk::transaction::VersionedTransaction;
|
||||
|
||||
use crate::common::pumpswap::PumpSwapInstruction;
|
||||
use crate::common::raydium::{RaydiumEvent, RaydiumInstruction};
|
||||
use crate::common::AnyResult;
|
||||
use crate::event_parser::protocols::pumpfun::{PumpFunCreateTokenEvent, PumpFunTradeEvent};
|
||||
use crate::event_parser::protocols::raydium_launchpad::{
|
||||
RaydiumLaunchpadPoolCreateEvent, RaydiumLaunchpadTradeEvent,
|
||||
};
|
||||
use crate::event_parser::{EventParserFactory, Protocol, UnifiedEvent};
|
||||
|
||||
use solana_sdk::pubkey::Pubkey;
|
||||
use crate::common::pumpfun::logs_data::DexInstruction;
|
||||
use crate::common::pumpfun::logs_events::PumpfunEvent;
|
||||
use crate::common::pumpswap::logs_events::PumpSwapEvent;
|
||||
use crate::common::pumpfun::logs_filters::LogFilter;
|
||||
use crate::common::pumpswap::logs_filters::LogFilter as PumpswapLogFilter;
|
||||
use crate::common::raydium::logs_filters::LogFilter as RaydiumLogFilter;
|
||||
use crate::protos::shredstream::shredstream_proxy_client::ShredstreamProxyClient;
|
||||
use crate::protos::shredstream::SubscribeEntriesRequest;
|
||||
use solana_sdk::pubkey::Pubkey;
|
||||
|
||||
const CHANNEL_SIZE: usize = 1000;
|
||||
|
||||
@@ -32,18 +29,22 @@ struct TransactionWithSlot {
|
||||
slot: u64,
|
||||
}
|
||||
|
||||
|
||||
impl ShredStreamGrpc {
|
||||
pub async fn new(endpoint: String) -> AnyResult<Self> {
|
||||
let shredstream_client = ShredstreamProxyClient::connect(endpoint.clone()).await?;
|
||||
Ok(Self {
|
||||
shredstream_client: Arc::new(shredstream_client)
|
||||
Ok(Self {
|
||||
shredstream_client: Arc::new(shredstream_client),
|
||||
})
|
||||
}
|
||||
|
||||
pub async fn shredstream_subscribe<F>(&self, callback: F, bot_wallet: Option<Pubkey>) -> AnyResult<()>
|
||||
pub async fn shredstream_subscribe<F>(
|
||||
&self,
|
||||
protocols: Vec<Protocol>,
|
||||
bot_wallet: Option<Pubkey>,
|
||||
callback: F,
|
||||
) -> AnyResult<()>
|
||||
where
|
||||
F: Fn(PumpfunEvent) + Send + Sync + 'static,
|
||||
F: Fn(Box<dyn UnifiedEvent>) + Send + Sync + 'static,
|
||||
{
|
||||
let request = tonic::Request::new(SubscribeEntriesRequest {});
|
||||
let mut client = (*self.shredstream_client).clone();
|
||||
@@ -74,215 +75,50 @@ impl ShredStreamGrpc {
|
||||
});
|
||||
|
||||
while let Some(transaction_with_slot) = rx.next().await {
|
||||
if let Err(e) = Self::process_pumpfun_transaction(transaction_with_slot, &*callback, bot_wallet).await {
|
||||
if let Err(e) = Self::process_transaction(
|
||||
transaction_with_slot,
|
||||
protocols.clone(),
|
||||
bot_wallet,
|
||||
&*callback,
|
||||
)
|
||||
.await
|
||||
{
|
||||
error!("Error processing transaction: {:?}", e);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn shredstream_subscribe_pumpswap<F>(&self, callback: F) -> AnyResult<()>
|
||||
async fn process_transaction<F>(
|
||||
transaction_with_slot: TransactionWithSlot,
|
||||
protocols: Vec<Protocol>,
|
||||
bot_wallet: Option<Pubkey>,
|
||||
callback: &F,
|
||||
) -> AnyResult<()>
|
||||
where
|
||||
F: Fn(PumpSwapEvent) + Send + Sync + 'static,
|
||||
{
|
||||
let request = tonic::Request::new(SubscribeEntriesRequest {});
|
||||
let mut client = (*self.shredstream_client).clone();
|
||||
let mut stream = client.subscribe_entries(request).await?.into_inner();
|
||||
let (mut tx, mut rx) = mpsc::channel::<TransactionWithSlot>(CHANNEL_SIZE);
|
||||
let callback = Box::new(callback);
|
||||
tokio::spawn(async move {
|
||||
while let Some(message) = stream.next().await {
|
||||
match message {
|
||||
Ok(msg) => {
|
||||
if let Ok(entries) = bincode::deserialize::<Vec<Entry>>(&msg.entries) {
|
||||
for entry in entries {
|
||||
for transaction in entry.transactions {
|
||||
let _ = tx.try_send(TransactionWithSlot {
|
||||
transaction: transaction.clone(),
|
||||
slot: msg.slot,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(error) => {
|
||||
error!("Stream error: {error:?}");
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
while let Some(transaction_with_slot) = rx.next().await {
|
||||
if let Err(e) = Self::process_pumpswap_transaction(transaction_with_slot, &*callback).await {
|
||||
error!("Error processing transaction: {:?}", e);
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn shredstream_subscribe_raydium<F>(&self, callback: F) -> AnyResult<()>
|
||||
where
|
||||
F: Fn(RaydiumEvent) + Send + Sync + 'static,
|
||||
{
|
||||
let request = tonic::Request::new(SubscribeEntriesRequest {});
|
||||
let mut client = (*self.shredstream_client).clone();
|
||||
let mut stream = client.subscribe_entries(request).await?.into_inner();
|
||||
let (mut tx, mut rx) = mpsc::channel::<TransactionWithSlot>(CHANNEL_SIZE);
|
||||
let callback = Box::new(callback);
|
||||
tokio::spawn(async move {
|
||||
while let Some(message) = stream.next().await {
|
||||
match message {
|
||||
Ok(msg) => {
|
||||
if let Ok(entries) = bincode::deserialize::<Vec<Entry>>(&msg.entries) {
|
||||
for entry in entries {
|
||||
for transaction in entry.transactions {
|
||||
let _ = tx.try_send(TransactionWithSlot {
|
||||
transaction: transaction.clone(),
|
||||
slot: msg.slot,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(error) => {
|
||||
error!("Stream error: {error:?}");
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
while let Some(transaction_with_slot) = rx.next().await {
|
||||
if let Err(e) = Self::process_raydium_transaction(transaction_with_slot, &*callback).await {
|
||||
error!("Error processing transaction: {:?}", e);
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn process_pumpfun_transaction<F>(transaction_with_slot: TransactionWithSlot, callback: &F, bot_wallet: Option<Pubkey>) -> AnyResult<()>
|
||||
where
|
||||
F: Fn(PumpfunEvent) + Send + Sync,
|
||||
F: Fn(Box<dyn UnifiedEvent>) + Send + Sync,
|
||||
{
|
||||
let slot = transaction_with_slot.slot;
|
||||
let versioned_tx = transaction_with_slot.transaction;
|
||||
let mut dev_address: Option<Pubkey> = None;
|
||||
// let hash = versioned_tx.signatures[0].to_string();
|
||||
let instructions = LogFilter::parse_compiled_instruction(versioned_tx, bot_wallet).unwrap();
|
||||
for instruction in instructions {
|
||||
// println!("hash: {}\ninstruction: {:?}\n\n", hash, instruction);
|
||||
match instruction {
|
||||
DexInstruction::CreateToken(mut token_info) => {
|
||||
token_info.slot = slot;
|
||||
dev_address = Some(token_info.user);
|
||||
callback(PumpfunEvent::NewToken(token_info));
|
||||
}
|
||||
DexInstruction::UserTrade(mut trade_info) => {
|
||||
trade_info.slot = slot;
|
||||
if Some(trade_info.user) == dev_address {
|
||||
callback(PumpfunEvent::NewDevTrade(trade_info));
|
||||
} else {
|
||||
callback(PumpfunEvent::NewUserTrade(trade_info));
|
||||
}
|
||||
}
|
||||
DexInstruction::BotTrade(mut trade_info) => {
|
||||
trade_info.slot = slot;
|
||||
callback(PumpfunEvent::NewBotTrade(trade_info));
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
let signature = versioned_tx.signatures[0];
|
||||
|
||||
async fn process_pumpswap_transaction<F>(transaction_with_slot: TransactionWithSlot, callback: &F) -> AnyResult<()>
|
||||
where
|
||||
F: Fn(PumpSwapEvent) + Send + Sync,
|
||||
{
|
||||
let slot = transaction_with_slot.slot;
|
||||
let versioned_tx = transaction_with_slot.transaction;
|
||||
let signature = versioned_tx.signatures[0].to_string();
|
||||
let instructions = PumpswapLogFilter::parse_pumpswap_compiled_instruction(versioned_tx).unwrap();
|
||||
for instruction in instructions {
|
||||
match instruction {
|
||||
PumpSwapInstruction::CreatePool(mut create_event) => {
|
||||
create_event.slot = slot;
|
||||
create_event.signature = signature.clone();
|
||||
callback(PumpSwapEvent::CreatePool(create_event));
|
||||
}
|
||||
PumpSwapInstruction::Deposit(mut deposit_event) => {
|
||||
deposit_event.slot = slot;
|
||||
deposit_event.signature = signature.clone();
|
||||
callback(PumpSwapEvent::Deposit(deposit_event));
|
||||
}
|
||||
PumpSwapInstruction::Withdraw(mut withdraw_event) => {
|
||||
withdraw_event.slot = slot;
|
||||
withdraw_event.signature = signature.clone();
|
||||
callback(PumpSwapEvent::Withdraw(withdraw_event));
|
||||
}
|
||||
PumpSwapInstruction::Buy(mut buy_event) => {
|
||||
buy_event.slot = slot;
|
||||
buy_event.signature = signature.clone();
|
||||
callback(PumpSwapEvent::Buy(buy_event));
|
||||
}
|
||||
PumpSwapInstruction::Sell(mut sell_event) => {
|
||||
sell_event.slot = slot;
|
||||
sell_event.signature = signature.clone();
|
||||
callback(PumpSwapEvent::Sell(sell_event));
|
||||
}
|
||||
PumpSwapInstruction::UpdateFeeConfig(mut update_fee_event) => {
|
||||
update_fee_event.slot = slot;
|
||||
update_fee_event.signature = signature.clone();
|
||||
callback(PumpSwapEvent::UpdateFeeConfig(update_fee_event));
|
||||
}
|
||||
PumpSwapInstruction::UpdateAdmin(mut update_admin_event) => {
|
||||
update_admin_event.slot = slot;
|
||||
update_admin_event.signature = signature.clone();
|
||||
callback(PumpSwapEvent::UpdateAdmin(update_admin_event));
|
||||
}
|
||||
PumpSwapInstruction::Disable(mut disable_event) => {
|
||||
disable_event.slot = slot;
|
||||
disable_event.signature = signature.clone();
|
||||
callback(PumpSwapEvent::Disable(disable_event));
|
||||
}
|
||||
_ => {}
|
||||
for protocol in protocols {
|
||||
let parser = EventParserFactory::create_parser(protocol.clone());
|
||||
let events = parser
|
||||
.parse_versioned_transaction(
|
||||
&versioned_tx,
|
||||
&signature.to_string(),
|
||||
Some(slot),
|
||||
bot_wallet.clone(),
|
||||
)
|
||||
.await
|
||||
.unwrap_or_else(|_e| vec![]);
|
||||
for event in events {
|
||||
callback(event);
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn process_raydium_transaction<F>(transaction_with_slot: TransactionWithSlot, callback: &F) -> AnyResult<()>
|
||||
where
|
||||
F: Fn(RaydiumEvent) + Send + Sync,
|
||||
{
|
||||
let slot = transaction_with_slot.slot;
|
||||
let versioned_tx = transaction_with_slot.transaction;
|
||||
let signature = versioned_tx.signatures[0].to_string();
|
||||
let instructions: Vec<RaydiumInstruction> = RaydiumLogFilter::parse_raydium_compiled_instruction(versioned_tx).unwrap();
|
||||
for instruction in instructions {
|
||||
match instruction {
|
||||
RaydiumInstruction::V4Swap(mut v4_swap_event) => {
|
||||
v4_swap_event.slot = slot;
|
||||
v4_swap_event.signature = signature.clone();
|
||||
callback(RaydiumEvent::V4Swap(v4_swap_event));
|
||||
}
|
||||
RaydiumInstruction::SwapBaseInput(mut swap_base_input_event) => {
|
||||
swap_base_input_event.slot = slot;
|
||||
swap_base_input_event.signature = signature.clone();
|
||||
callback(RaydiumEvent::SwapBaseInput(swap_base_input_event));
|
||||
}
|
||||
RaydiumInstruction::SwapBaseOutput(mut swap_base_output_event) => {
|
||||
swap_base_output_event.slot = slot;
|
||||
swap_base_output_event.signature = signature.clone();
|
||||
callback(RaydiumEvent::SwapBaseOutput(swap_base_output_event));
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user