diff --git a/src/instruction/dex_events.rs b/src/common/dex_events.rs similarity index 100% rename from src/instruction/dex_events.rs rename to src/common/dex_events.rs diff --git a/src/grpc/event.rs b/src/common/event.rs similarity index 70% rename from src/grpc/event.rs rename to src/common/event.rs index b69d096..5c8aa16 100755 --- a/src/grpc/event.rs +++ b/src/common/event.rs @@ -5,10 +5,28 @@ use borsh::{BorshDeserialize, BorshSerialize}; use regex::Regex; use solana_sdk::pubkey::Pubkey; -use super::myerror::AppError; +use crate::error::AppError; pub const PROGRAM_DATA: &str = "Program data: "; +#[derive(Debug)] +pub enum PumpfunEvent { + NewToken(CreateEvent), + NewUserTrade(TradeEvent), + NewBotTrade(TradeEvent), + Error(String), +} + +#[derive(Clone, Debug, Default, PartialEq, BorshDeserialize, BorshSerialize)] +pub struct CreateEvent { + 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 TradeEvent { pub mint: Pubkey, @@ -51,6 +69,12 @@ pub trait EventTrait: Sized + std::fmt::Debug { fn from_bytes(bytes: &[u8]) -> Result; } +impl EventTrait for CreateEvent { + fn from_bytes(bytes: &[u8]) -> Result { + CreateEvent::try_from_slice(bytes).map_err(|e| AppError::from(anyhow!(e.to_string()))) + } +} + impl EventTrait for TradeEvent { fn from_bytes(bytes: &[u8]) -> Result { TradeEvent::try_from_slice(bytes).map_err(|e| AppError::from(anyhow!(e.to_string()))) @@ -73,8 +97,10 @@ impl EventTrait for SwapBaseInLog { pub struct PumpEvent {} impl PumpEvent { - pub fn parse_logs(logs: &Vec) -> Option { - let mut event: Option = None; + pub fn parse_logs(logs: &Vec) -> (Option, Option) { + let mut create_event: Option = None; + let mut trade_event: Option = None; + if !logs.is_empty() { let logs_iter = logs.iter().peekable(); @@ -83,13 +109,22 @@ impl PumpEvent { let borsh_bytes = general_purpose::STANDARD.decode(log).unwrap(); let slice: &[u8] = &borsh_bytes[8..]; - if let Ok(e) = T::from_bytes(slice) { - event = Some(e); + if create_event.is_none() { + if let Ok(e) = CreateEvent::from_bytes(slice) { + create_event = Some(e); + continue; + } + } + + if trade_event.is_none() { + if let Ok(e) = TradeEvent::from_bytes(slice) { + trade_event = Some(e); + } } } } } - event + (create_event, trade_event) } } diff --git a/src/instruction/logs_data.rs b/src/common/logs_data.rs similarity index 100% rename from src/instruction/logs_data.rs rename to src/common/logs_data.rs diff --git a/src/instruction/logs_events.rs b/src/common/logs_events.rs similarity index 65% rename from src/instruction/logs_events.rs rename to src/common/logs_events.rs index 6c06eb6..b65c1cc 100755 --- a/src/instruction/logs_events.rs +++ b/src/common/logs_events.rs @@ -1,5 +1,6 @@ use serde::{Serialize, Deserialize}; -use crate::instruction::logs_data::{CreateTokenInfo, TradeInfo}; +use crate::common::logs_data::{CreateTokenInfo, TradeInfo}; +use crate::common::event::{CreateEvent, TradeEvent}; #[derive(Debug, Clone, Serialize, Deserialize)] pub enum DexEvent { diff --git a/src/instruction/logs_filters.rs b/src/common/logs_filters.rs similarity index 96% rename from src/instruction/logs_filters.rs rename to src/common/logs_filters.rs index 5ffbd09..cfcc42f 100755 --- a/src/instruction/logs_filters.rs +++ b/src/common/logs_filters.rs @@ -1,5 +1,5 @@ -use crate::instruction::logs_data::DexInstruction; -use crate::instruction::logs_parser::{parse_create_token_data, parse_trade_data}; +use crate::common::logs_data::DexInstruction; +use crate::common::logs_parser::{parse_create_token_data, parse_trade_data}; use crate::error::ClientResult; use solana_sdk::pubkey::Pubkey; pub struct LogFilter; diff --git a/src/instruction/logs_parser.rs b/src/common/logs_parser.rs similarity index 99% rename from src/instruction/logs_parser.rs rename to src/common/logs_parser.rs index fe7e9c0..5ca672c 100755 --- a/src/instruction/logs_parser.rs +++ b/src/common/logs_parser.rs @@ -1,7 +1,7 @@ use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64}; use crate::error::{ClientError, ClientResult}; -use crate::instruction::{ +use crate::common::{ logs_data::{DexInstruction, CreateTokenInfo, TradeInfo}, logs_filters::LogFilter }; diff --git a/src/instruction/logs_subscribe.rs b/src/common/logs_subscribe.rs similarity index 98% rename from src/instruction/logs_subscribe.rs rename to src/common/logs_subscribe.rs index 3a4c25f..aa6c8a6 100755 --- a/src/instruction/logs_subscribe.rs +++ b/src/common/logs_subscribe.rs @@ -8,7 +8,7 @@ use std::sync::Arc; use tokio::sync::mpsc; use tokio::task::JoinHandle; use futures::StreamExt; -use crate::{constants, instruction::{ +use crate::{constants, common::{ logs_data::DexInstruction, logs_events::DexEvent, logs_filters::LogFilter }}; diff --git a/src/common/mod.rs b/src/common/mod.rs new file mode 100644 index 0000000..322728d --- /dev/null +++ b/src/common/mod.rs @@ -0,0 +1,6 @@ +pub mod logs_data; +pub mod logs_parser; +pub mod logs_filters; +pub mod logs_events; +pub mod logs_subscribe; +pub mod event; \ No newline at end of file diff --git a/src/error/mod.rs b/src/error/mod.rs index a531694..ffaab01 100755 --- a/src/error/mod.rs +++ b/src/error/mod.rs @@ -24,6 +24,19 @@ use solana_client::{ }; use solana_sdk::pubkey::ParsePubkeyError; +#[derive(Debug)] +#[allow(dead_code)] +pub struct AppError(anyhow::Error); + +impl From for AppError +where + E: Into, +{ + fn from(err: E) -> Self { + Self(err.into()) + } +} + #[derive(Debug)] pub enum ClientError { /// Bonding curve account was not found diff --git a/src/grpc/mod.rs b/src/grpc/mod.rs index 45e31c0..1199445 100755 --- a/src/grpc/mod.rs +++ b/src/grpc/mod.rs @@ -1,7 +1,306 @@ +use std::{collections::HashMap, fmt, time::Duration}; -pub mod event; -pub mod myerror; -// pub mod subscribe_logs; -// pub mod subscribe_tx; -pub mod yellowstone_grpc; +use futures::{channel::mpsc, sink::Sink, Stream, StreamExt, SinkExt}; +use rustls::crypto::{ring::default_provider, CryptoProvider}; +use tonic::{transport::channel::ClientTlsConfig, Status}; +use yellowstone_grpc_client::{GeyserGrpcClient, GeyserGrpcClientResult}; +use yellowstone_grpc_proto::geyser::{ + CommitmentLevel, SubscribeRequest, SubscribeRequestFilterTransactions, SubscribeUpdate, + SubscribeUpdateTransaction, subscribe_update::UpdateOneof, SubscribeRequestPing, +}; +use log::{error, info}; +use chrono::Local; +use solana_sdk::{pubkey, pubkey::Pubkey, signature::Signature}; +use solana_transaction_status::{ + option_serializer::OptionSerializer, EncodedTransactionWithStatusMeta, UiTransactionEncoding, +}; +use anyhow::anyhow; +use crate::common::event::{PumpEvent, RaydiumEvent, SwapBaseInLog, TradeEvent}; +use crate::error::AppError; +use crate::common::event::PumpfunEvent; + +// 类型别名定义 +type TransactionsFilterMap = HashMap; +type GrpcStreamResult = GeyserGrpcClientResult<( + Box + Unpin + Send>, + Box> + Unpin + Send>, +)>; + +// 常量定义 +const AMM_V4: Pubkey = pubkey!("675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8"); +const PUMP_PROGRAM_ID: Pubkey = pubkey!("6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P"); +const CONNECT_TIMEOUT: u64 = 10; +const REQUEST_TIMEOUT: u64 = 60; +const CHANNEL_SIZE: usize = 1000; + +// 枚举定义 +#[derive(Debug)] +pub enum SwapType { + Pump, + Raydium, +} + +// 结构体定义 +#[allow(dead_code)] +pub struct TransactionPretty { + pub slot: u64, + pub signature: Signature, + pub is_vote: bool, + pub tx: EncodedTransactionWithStatusMeta, +} + +impl fmt::Debug for TransactionPretty { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + struct TxWrap<'a>(&'a EncodedTransactionWithStatusMeta); + impl<'a> fmt::Debug for TxWrap<'a> { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let serialized = serde_json::to_string(self.0).expect("failed to serialize"); + fmt::Display::fmt(&serialized, f) + } + } + + f.debug_struct("TransactionPretty") + .field("slot", &self.slot) + .field("signature", &self.signature) + .field("is_vote", &self.is_vote) + .field("tx", &TxWrap(&self.tx)) + .finish() + } +} + +impl From for TransactionPretty { + fn from(SubscribeUpdateTransaction { transaction, slot }: SubscribeUpdateTransaction) -> Self { + let tx = transaction.expect("should be defined"); + Self { + slot, + 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") + .encode(UiTransactionEncoding::Base64, Some(u8::MAX), true) + .expect("failed to encode"), + } + } +} + +pub struct YellowstoneGrpc { + endpoint: String, +} + +impl YellowstoneGrpc { + pub fn new(endpoint: String) -> Self { + Self { endpoint } + } + + pub async fn connect( + &self, + transactions: TransactionsFilterMap, + ) -> Result< + GeyserGrpcClientResult<( + impl Sink, + impl Stream>, + )>, + AppError, + > { + if CryptoProvider::get_default().is_none() { + default_provider() + .install_default() + .map_err(|e| anyhow::anyhow!("Failed to install crypto provider: {:?}", e))?; + } + + let mut client = GeyserGrpcClient::build_from_shared(self.endpoint.clone())? + .tls_config(ClientTlsConfig::new().with_native_roots())? + .connect_timeout(Duration::from_secs(10)) + .timeout(Duration::from_secs(60)) + .connect() + .await?; + + let subscribe_request = SubscribeRequest { + transactions, + commitment: Some(CommitmentLevel::Processed.into()), + ..Default::default() + }; + + Ok(client.subscribe_with_request(Some(subscribe_request)).await) + } + + pub async fn subscribe_accounts(&self, accounts: Vec) -> Result<(), AppError> { + let transactions = self.get_subscribe_request_filter(accounts, vec![], vec![]); + let (mut subscribe_tx, mut stream) = self.connect(transactions).await??; + let (mut tx, mut rx) = mpsc::channel::(CHANNEL_SIZE); + + tokio::spawn(async move { + while let Some(message) = stream.next().await { + match message { + Ok(msg) => { + if let Err(e) = Self::handle_stream_message(msg, &mut tx, &mut subscribe_tx).await { + error!("Error handling message: {:?}", e); + break; + } + } + Err(error) => { + error!("Stream error: {error:?}"); + break; + } + } + } + }); + + while let Some(transaction_pretty) = rx.next().await { + if let Err(e) = Self::process_transaction(transaction_pretty, &|_| {}, None).await { + error!("Error processing transaction: {:?}", e); + } + } + Ok(()) + } + + pub async fn subscribe_pumpfun(&self, callback: F, bot_wallet: Option) -> Result<(), AppError> + where + F: Fn(PumpfunEvent) + Send + Sync + 'static, + { + let addrs = vec![PUMP_PROGRAM_ID.to_string()]; + let transactions = self.get_subscribe_request_filter(addrs, vec![], vec![]); + let (mut subscribe_tx, mut stream) = self.connect(transactions).await??; + let (mut tx, mut rx) = mpsc::channel::(CHANNEL_SIZE); + + let callback = Box::new(callback); + + tokio::spawn(async move { + while let Some(message) = stream.next().await { + match message { + Ok(msg) => { + if let Err(e) = Self::handle_stream_message(msg, &mut tx, &mut subscribe_tx).await { + error!("Error handling message: {:?}", e); + break; + } + } + Err(error) => { + error!("Stream error: {error:?}"); + break; + } + } + } + }); + + while let Some(transaction_pretty) = rx.next().await { + if let Err(e) = Self::process_transaction(transaction_pretty, &*callback, bot_wallet).await { + error!("Error processing transaction: {:?}", e); + } + } + Ok(()) + } + + pub fn get_subscribe_request_filter( + &self, + account_include: Vec, + account_exclude: Vec, + account_required: Vec, + ) -> TransactionsFilterMap { + let mut transactions = HashMap::new(); + transactions.insert( + "client".to_string(), + SubscribeRequestFilterTransactions { + vote: Some(false), + failed: Some(false), + signature: None, + account_include, + account_exclude, + account_required, + }, + ); + transactions + } + + + async fn handle_stream_message( + msg: SubscribeUpdate, + tx: &mut mpsc::Sender, + subscribe_tx: &mut (impl Sink + Unpin), + ) -> Result<(), AppError> { + match msg.update_oneof { + Some(UpdateOneof::Transaction(sut)) => { + let transaction_pretty = TransactionPretty::from(sut); + tx.try_send(transaction_pretty).map_err(|e| AppError::from(anyhow!("Send error: {:?}", e)))?; + } + Some(UpdateOneof::Ping(_)) => { + subscribe_tx + .send(SubscribeRequest { + ping: Some(SubscribeRequestPing { id: 1 }), + ..Default::default() + }) + .await + .map_err(|e| AppError::from(anyhow!("Ping error: {:?}", e)))?; + info!("service is ping: {}", Local::now()); + } + Some(UpdateOneof::Pong(_)) => { + info!("service is pong: {}", Local::now()); + } + _ => {} + } + Ok(()) + } + + async fn process_transaction(transaction_pretty: TransactionPretty, callback: &F, bot_wallet: Option) -> Result<(), AppError> + where + F: Fn(PumpfunEvent) + Send + Sync, + { + let trade_raw = transaction_pretty.tx; + let meta = trade_raw.meta.as_ref() + .ok_or_else(|| AppError::from(anyhow!("Missing transaction metadata")))?; + + if meta.err.is_some() { + return Ok(()); + } + + let logs = if let OptionSerializer::Some(logs) = &meta.log_messages { + logs + } else { + &vec![] + }; + + if let Ok(swap_type) = Self::get_swap_type(&trade_raw) { + match swap_type { + SwapType::Raydium => { + let event = RaydiumEvent::parse_logs::(logs); + info!("RaydiumEvent {:#?}", event); + } + SwapType::Pump => { + let (create_event, trade_event) = PumpEvent::parse_logs(logs); + if let Some(create_event) = create_event { + callback(PumpfunEvent::NewToken(create_event)); + } + if let Some(trade_event) = trade_event { + if let Some(bot_wallet_pubkey) = bot_wallet { + if trade_event.user == bot_wallet_pubkey { + callback(PumpfunEvent::NewBotTrade(trade_event)); + } else { + callback(PumpfunEvent::NewUserTrade(trade_event)); + } + } else { + callback(PumpfunEvent::NewUserTrade(trade_event)); + } + } + } + } + } + Ok(()) + } + + pub fn get_swap_type(trade_raw: &EncodedTransactionWithStatusMeta) -> Result { + let transaction = trade_raw.transaction.decode() + .ok_or_else(|| AppError::from(anyhow!("Failed to decode transaction")))?; + + let account_keys = transaction.message.static_account_keys(); + let program_index = account_keys + .iter() + .position(|item| item == &AMM_V4 || item == &PUMP_PROGRAM_ID) + .ok_or_else(|| AppError::from(anyhow!("swap type program_id not found")))?; + + match account_keys[program_index] { + AMM_V4 => Ok(SwapType::Raydium), + PUMP_PROGRAM_ID => Ok(SwapType::Pump), + _ => Err(AppError::from(anyhow!("Invalid program_id"))) + } + } +} \ No newline at end of file diff --git a/src/grpc/myerror.rs b/src/grpc/myerror.rs deleted file mode 100755 index e1beb9d..0000000 --- a/src/grpc/myerror.rs +++ /dev/null @@ -1,13 +0,0 @@ -use anyhow::Error; -#[derive(Debug)] -#[allow(dead_code)] -pub struct AppError(Error); - -impl From for AppError -where - E: Into, -{ - fn from(err: E) -> Self { - Self(err.into()) - } -} \ No newline at end of file diff --git a/src/grpc/subscribe_logs.rs b/src/grpc/subscribe_logs.rs deleted file mode 100755 index 5a0fef1..0000000 --- a/src/grpc/subscribe_logs.rs +++ /dev/null @@ -1,136 +0,0 @@ -#[cfg(test)] -mod subscribe_tx_tests { - use crate::common::{ - event::{PumpEvent, RaydiumEvent, SwapBaseInLog, TradeEvent}, - myerror::AppError, - yellowstone_grpc::{TransactionPretty, YellowstoneGrpc}, - }; - use anyhow::anyhow; - use chrono::Local; - use dotenvy::dotenv; - use futures::{channel::mpsc, sink::SinkExt, stream::StreamExt}; - use log::{error, info}; - use solana_sdk::pubkey; - use solana_sdk::pubkey::Pubkey; - use solana_transaction_status::{ - option_serializer::OptionSerializer, EncodedTransactionWithStatusMeta, - }; - use std::env; - use tokio::test; - use yellowstone_grpc_proto::geyser::{ - subscribe_update::UpdateOneof, SubscribeRequest, SubscribeRequestPing, - }; - const AMM_V4: Pubkey = pubkey!("675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8"); - const PUMP_PROGRAM_ID: Pubkey = pubkey!("6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P"); - pub enum SwapType { - Pump, - Raydium, - } - - #[test] - async fn test_subscribe_tx() -> Result<(), AppError> { - dotenv().ok(); - pretty_env_logger::init_custom_env("RUST_LOG"); - let yellowstone_url = env::var("YELLOWSTONE_URL")?; - - info!("::: {:?}", yellowstone_url); - let yellowstone_grpc = YellowstoneGrpc::new(yellowstone_url); - - let addrs = vec![ - "Aa4QWNkS3RLUv7DA9BM1a2Hzm4HDQo5PyRefqDJnpump".to_string(), - "BnDssYyGDF9aj5j2N5BwsJFk9YMneQ8P7LQkoYkrpump".to_string(), - ]; - let transactions = yellowstone_grpc.subscribe_transaction(addrs, vec![], vec![]); - - let (mut subscribe_tx, mut stream) = yellowstone_grpc.connect(transactions).await??; - - let (mut tx, mut rx) = mpsc::channel::(1000); - - tokio::spawn(async move { - while let Some(message) = stream.next().await { - match message { - Ok(msg) => { - match msg.update_oneof { - Some(UpdateOneof::Transaction(sut)) => { - let transaction_pretty: TransactionPretty = sut.into(); - let _ = tx.try_send(transaction_pretty); - } - Some(UpdateOneof::Ping(_)) => { - // This is necessary to keep load balancers that expect client pings alive. If your load balancer doesn't - // require periodic client pings then this is unnecessary - let _ = subscribe_tx - .send(SubscribeRequest { - ping: Some(SubscribeRequestPing { id: 1 }), - ..Default::default() - }) - .await; - info!("service is ping: {}", Local::now()); - } - Some(UpdateOneof::Pong(_)) => { - info!("service is pong: {}", Local::now()); - } - _ => {} - } - continue; - } - Err(error) => { - error!("error: {error:?}"); - break; - } - } - } - }); - - while let Some(transaction_pretty) = rx.next().await { - let trade_raw = transaction_pretty.tx.clone(); - let meta = &trade_raw.meta.clone().unwrap(); - if meta.err.is_some() { - continue; - } - - let logs = if let OptionSerializer::Some(logs) = &meta.log_messages { - logs - } else { - &vec![] - }; - - if let Ok(swap_type) = get_swap_type(&trade_raw) { - match swap_type { - SwapType::Raydium => { - let event = RaydiumEvent::parse_logs::(logs); - info!("RaydiumEvent {:#?}", event); - } - SwapType::Pump => { - let event = PumpEvent::parse_logs::(logs); - info!("PumpEvent {:#?}", event); - } - } - }; - } - Ok(()) - } - - #[allow(dead_code)] - pub fn get_swap_type( - trade_raw: &EncodedTransactionWithStatusMeta, - ) -> Result { - let transaction = &trade_raw.transaction.decode(); - if let Some(transaction) = transaction { - let account_keys = transaction.message.static_account_keys(); - - let program_index = account_keys - .iter() - .position(|item| item == &AMM_V4 || item == &PUMP_PROGRAM_ID) - .ok_or(anyhow!("swap type program_id not found"))?; - - let program_id = account_keys[program_index]; - let _type = match program_id { - AMM_V4 => Ok(SwapType::Raydium), - PUMP_PROGRAM_ID => Ok(SwapType::Pump), - _ => Err(AppError::from(anyhow!("program_id ix not found"))), - }; - return _type; - } - Err(AppError::from(anyhow!("program_id ix not found"))) - } -} \ No newline at end of file diff --git a/src/grpc/subscribe_tx.rs b/src/grpc/subscribe_tx.rs deleted file mode 100755 index 7fb0aa0..0000000 --- a/src/grpc/subscribe_tx.rs +++ /dev/null @@ -1,73 +0,0 @@ -#[cfg(test)] -mod subscribe_tx_tests { - use crate::common::{ - myerror::AppError, - yellowstone_grpc::{TransactionPretty, YellowstoneGrpc}, - }; - use chrono::Local; - use dotenvy::dotenv; - use futures::{channel::mpsc, sink::SinkExt, stream::StreamExt}; - use log::{error, info}; - use std::env; - use tokio::test; - use yellowstone_grpc_proto::geyser::{ - subscribe_update::UpdateOneof, SubscribeRequest, SubscribeRequestPing, - }; - - #[test] - async fn test_subscribe_tx() -> Result<(), AppError> { - dotenv().ok(); - pretty_env_logger::init_custom_env("RUST_LOG"); - let yellowstone_url = env::var("YELLOWSTONE_URL")?; - - info!("::: {:?}", yellowstone_url); - let yellowstone_grpc = YellowstoneGrpc::new(yellowstone_url); - - let addrs = vec!["6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P".to_string()]; - let transactions = yellowstone_grpc.subscribe_transaction(addrs, vec![], vec![]); - - let (mut subscribe_tx, mut stream) = yellowstone_grpc.connect(transactions).await??; - - let (mut tx, mut rx) = mpsc::channel::(1000); - - tokio::spawn(async move { - while let Some(message) = stream.next().await { - match message { - Ok(msg) => { - match msg.update_oneof { - Some(UpdateOneof::Transaction(sut)) => { - let transaction_pretty: TransactionPretty = sut.into(); - let _ = tx.try_send(transaction_pretty); - } - Some(UpdateOneof::Ping(_)) => { - // This is necessary to keep load balancers that expect client pings alive. If your load balancer doesn't - // require periodic client pings then this is unnecessary - let _ = subscribe_tx - .send(SubscribeRequest { - ping: Some(SubscribeRequestPing { id: 1 }), - ..Default::default() - }) - .await; - info!("service is ping: {}", Local::now()); - } - Some(UpdateOneof::Pong(_)) => { - info!("service is pong: {}", Local::now()); - } - _ => {} - } - continue; - } - Err(error) => { - error!("error: {error:?}"); - break; - } - } - } - }); - - while let Some(event) = rx.next().await { - info!("TransactionPretty {:#?}", event); - } - Ok(()) - } -} \ No newline at end of file diff --git a/src/grpc/yellowstone_grpc.rs b/src/grpc/yellowstone_grpc.rs deleted file mode 100755 index ab1d32b..0000000 --- a/src/grpc/yellowstone_grpc.rs +++ /dev/null @@ -1,284 +0,0 @@ -use std::{collections::HashMap, fmt, time::Duration}; - -use futures::{channel::mpsc, sink::Sink, Stream, StreamExt, SinkExt}; -use rustls::crypto::{ring::default_provider, CryptoProvider}; -use tonic::{transport::channel::ClientTlsConfig, Status}; -use yellowstone_grpc_client::{GeyserGrpcClient, GeyserGrpcClientResult}; -use yellowstone_grpc_proto::geyser::{ - CommitmentLevel, SubscribeRequest, SubscribeRequestFilterTransactions, SubscribeUpdate, - SubscribeUpdateTransaction, subscribe_update::UpdateOneof, SubscribeRequestPing, -}; -use log::{error, info}; -use chrono::Local; -use solana_sdk::{pubkey, pubkey::Pubkey, signature::Signature}; -use solana_transaction_status::{ - option_serializer::OptionSerializer, EncodedTransactionWithStatusMeta, UiTransactionEncoding, -}; -use anyhow::anyhow; - -use crate::grpc::event::{PumpEvent, RaydiumEvent, SwapBaseInLog, TradeEvent}; -use crate::grpc::myerror::AppError; - -// 类型别名定义 -type TransactionsFilterMap = HashMap; -type GrpcStreamResult = GeyserGrpcClientResult<( - Box + Unpin + Send>, - Box> + Unpin + Send>, -)>; - -// 常量定义 -const AMM_V4: Pubkey = pubkey!("675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8"); -const PUMP_PROGRAM_ID: Pubkey = pubkey!("6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P"); -const CONNECT_TIMEOUT: u64 = 10; -const REQUEST_TIMEOUT: u64 = 60; -const CHANNEL_SIZE: usize = 1000; - -// 枚举定义 -#[derive(Debug)] -pub enum SwapType { - Pump, - Raydium, -} - -// 结构体定义 -#[allow(dead_code)] -pub struct TransactionPretty { - pub slot: u64, - pub signature: Signature, - pub is_vote: bool, - pub tx: EncodedTransactionWithStatusMeta, -} - -impl fmt::Debug for TransactionPretty { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - struct TxWrap<'a>(&'a EncodedTransactionWithStatusMeta); - impl<'a> fmt::Debug for TxWrap<'a> { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - let serialized = serde_json::to_string(self.0).expect("failed to serialize"); - fmt::Display::fmt(&serialized, f) - } - } - - f.debug_struct("TransactionPretty") - .field("slot", &self.slot) - .field("signature", &self.signature) - .field("is_vote", &self.is_vote) - .field("tx", &TxWrap(&self.tx)) - .finish() - } -} - -impl From for TransactionPretty { - fn from(SubscribeUpdateTransaction { transaction, slot }: SubscribeUpdateTransaction) -> Self { - let tx = transaction.expect("should be defined"); - Self { - slot, - 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") - .encode(UiTransactionEncoding::Base64, Some(u8::MAX), true) - .expect("failed to encode"), - } - } -} - -pub struct YellowstoneGrpc { - endpoint: String, -} - -impl YellowstoneGrpc { - pub fn new(endpoint: String) -> Self { - Self { endpoint } - } - - pub async fn connect( - &self, - transactions: TransactionsFilterMap, - ) -> Result< - GeyserGrpcClientResult<( - impl Sink, - impl Stream>, - )>, - AppError, - > { - if CryptoProvider::get_default().is_none() { - default_provider() - .install_default() - .map_err(|e| anyhow::anyhow!("Failed to install crypto provider: {:?}", e))?; - } - - let mut client = GeyserGrpcClient::build_from_shared(self.endpoint.clone())? - .tls_config(ClientTlsConfig::new().with_native_roots())? - .connect_timeout(Duration::from_secs(10)) - .timeout(Duration::from_secs(60)) - .connect() - .await?; - - let subscribe_request = SubscribeRequest { - transactions, - commitment: Some(CommitmentLevel::Processed.into()), - ..Default::default() - }; - - Ok(client.subscribe_with_request(Some(subscribe_request)).await) - } - - pub async fn subscribe_accounts(&self, accounts: Vec) -> Result<(), AppError> { - let transactions = self.get_subscribe_request_filter(accounts, vec![], vec![]); - let (mut subscribe_tx, mut stream) = self.connect(transactions).await??; - let (mut tx, mut rx) = mpsc::channel::(CHANNEL_SIZE); - - tokio::spawn(async move { - while let Some(message) = stream.next().await { - match message { - Ok(msg) => { - if let Err(e) = Self::handle_stream_message(msg, &mut tx, &mut subscribe_tx).await { - error!("Error handling message: {:?}", e); - break; - } - } - Err(error) => { - error!("Stream error: {error:?}"); - break; - } - } - } - }); - - while let Some(transaction_pretty) = rx.next().await { - if let Err(e) = Self::process_transaction(transaction_pretty).await { - error!("Error processing transaction: {:?}", e); - } - } - Ok(()) - } - - pub async fn subscribe_pumpfun(&self) -> Result<(), AppError> { - let addrs = vec![PUMP_PROGRAM_ID.to_string()]; - let transactions = self.get_subscribe_request_filter(addrs, vec![], vec![]); - let (mut subscribe_tx, mut stream) = self.connect(transactions).await??; - let (mut tx, mut rx) = mpsc::channel::(CHANNEL_SIZE); - - tokio::spawn(async move { - while let Some(message) = stream.next().await { - match message { - Ok(msg) => { - if let Err(e) = Self::handle_stream_message(msg, &mut tx, &mut subscribe_tx).await { - error!("Error handling message: {:?}", e); - break; - } - } - Err(error) => { - error!("Stream error: {error:?}"); - break; - } - } - } - }); - - while let Some(transaction_pretty) = rx.next().await { - if let Err(e) = Self::process_transaction(transaction_pretty).await { - error!("Error processing transaction: {:?}", e); - } - } - Ok(()) - } - - pub fn get_subscribe_request_filter( - &self, - account_include: Vec, - account_exclude: Vec, - account_required: Vec, - ) -> TransactionsFilterMap { - let mut transactions = HashMap::new(); - transactions.insert( - "client".to_string(), - SubscribeRequestFilterTransactions { - vote: Some(false), - failed: Some(false), - signature: None, - account_include, - account_exclude, - account_required, - }, - ); - transactions - } - - - async fn handle_stream_message( - msg: SubscribeUpdate, - tx: &mut mpsc::Sender, - subscribe_tx: &mut (impl Sink + Unpin), - ) -> Result<(), AppError> { - match msg.update_oneof { - Some(UpdateOneof::Transaction(sut)) => { - let transaction_pretty = TransactionPretty::from(sut); - tx.try_send(transaction_pretty).map_err(|e| AppError::from(anyhow!("Send error: {:?}", e)))?; - } - Some(UpdateOneof::Ping(_)) => { - subscribe_tx - .send(SubscribeRequest { - ping: Some(SubscribeRequestPing { id: 1 }), - ..Default::default() - }) - .await - .map_err(|e| AppError::from(anyhow!("Ping error: {:?}", e)))?; - info!("service is ping: {}", Local::now()); - } - Some(UpdateOneof::Pong(_)) => { - info!("service is pong: {}", Local::now()); - } - _ => {} - } - Ok(()) - } - - async fn process_transaction(transaction_pretty: TransactionPretty) -> Result<(), AppError> { - let trade_raw = transaction_pretty.tx; - let meta = trade_raw.meta.as_ref() - .ok_or_else(|| AppError::from(anyhow!("Missing transaction metadata")))?; - - if meta.err.is_some() { - return Ok(()); - } - - let logs = if let OptionSerializer::Some(logs) = &meta.log_messages { - logs - } else { - &vec![] - }; - - if let Ok(swap_type) = Self::get_swap_type(&trade_raw) { - match swap_type { - SwapType::Raydium => { - let event = RaydiumEvent::parse_logs::(logs); - info!("RaydiumEvent {:#?}", event); - } - SwapType::Pump => { - let event = PumpEvent::parse_logs::(logs); - info!("PumpEvent {:#?}", event); - } - } - } - Ok(()) - } - - pub fn get_swap_type(trade_raw: &EncodedTransactionWithStatusMeta) -> Result { - let transaction = trade_raw.transaction.decode() - .ok_or_else(|| AppError::from(anyhow!("Failed to decode transaction")))?; - - let account_keys = transaction.message.static_account_keys(); - let program_index = account_keys - .iter() - .position(|item| item == &AMM_V4 || item == &PUMP_PROGRAM_ID) - .ok_or_else(|| AppError::from(anyhow!("swap type program_id not found")))?; - - match account_keys[program_index] { - AMM_V4 => Ok(SwapType::Raydium), - PUMP_PROGRAM_ID => Ok(SwapType::Pump), - _ => Err(AppError::from(anyhow!("Invalid program_id"))) - } - } -} \ No newline at end of file diff --git a/src/instruction/mod.rs b/src/instruction/mod.rs index f7e97b3..9543774 100755 --- a/src/instruction/mod.rs +++ b/src/instruction/mod.rs @@ -10,18 +10,6 @@ //! - `buy`: Instruction to buy tokens from a bonding curve by providing SOL. //! - `sell`: Instruction to sell tokens back to the bonding curve in exchange for SOL. -pub mod logs_data; -pub mod logs_parser; -pub mod logs_filters; -pub mod logs_events; -pub mod logs_subscribe; - -pub use logs_data::*; -pub use logs_parser::*; -pub use logs_filters::*; -pub use logs_events::*; -pub use logs_subscribe::*; - use crate::{constants, PumpFun}; use spl_associated_token_account::get_associated_token_address; diff --git a/src/lib.rs b/src/lib.rs index 71e8ad7..db5bbcc 100755 --- a/src/lib.rs +++ b/src/lib.rs @@ -5,6 +5,7 @@ pub mod instruction; pub mod utils; pub mod jito; pub mod grpc; +pub mod common; use solana_client::rpc_client::RpcClient; use solana_sdk::{ @@ -22,9 +23,9 @@ use spl_associated_token_account::{ create_associated_token_account, }; -use instruction::logs_subscribe; -use instruction::logs_subscribe::SubscriptionHandle; -use instruction::logs_events::DexEvent; +use common::logs_subscribe; +use common::logs_subscribe::SubscriptionHandle; +use common::logs_events::DexEvent; use std::sync::Arc; use std::time::Instant; diff --git a/src/main.rs b/src/main.rs index b9779b2..f3b520c 100755 --- a/src/main.rs +++ b/src/main.rs @@ -1,4 +1,4 @@ -use mai3_pumpfun_sdk::instruction::{ +use mai3_pumpfun_sdk::common::{ logs_events::DexEvent, logs_subscribe::{tokens_subscription, stop_subscription} };