diff --git a/crates/pumpfun/Cargo.toml b/crates/pumpfun/Cargo.toml index ad48d0d..90daa1b 100644 --- a/crates/pumpfun/Cargo.toml +++ b/crates/pumpfun/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "mai3-pumpfun-sdk" -version = "2.1.0" +version = "2.2.1" edition = "2021" authors = ["William "] repository = "https://github.com/MiracleAI-Labs/pumpfun-sdk" @@ -23,5 +23,6 @@ serde = { version = "1.0.215", features = ["derive"] } serde_json = "1.0.132" solana-sdk = "1.18.26" tokio = "1.42.0" +futures-util = "0.3.29" base64 = "0.22.1" bs58 = "0.5.1" diff --git a/crates/pumpfun/README.md b/crates/pumpfun/README.md index f23dc89..93949c7 100644 --- a/crates/pumpfun/README.md +++ b/crates/pumpfun/README.md @@ -27,7 +27,7 @@ use anchor_client::{ }, Cluster, }; -use pumpfun::{accounts::BondingCurveAccount, utils::CreateTokenMetadata, PriorityFee, PumpFun}; +use mai3_pumpfun_sdk::{accounts::BondingCurveAccount, utils::CreateTokenMetadata, PriorityFee, PumpFun}; // Create a new PumpFun client let payer: Keypair = Keypair::new(); diff --git a/crates/pumpfun/RUSTDOC.md b/crates/pumpfun/RUSTDOC.md index f64cfcf..9b6dde3 100644 --- a/crates/pumpfun/RUSTDOC.md +++ b/crates/pumpfun/RUSTDOC.md @@ -26,7 +26,7 @@ use anchor_client::{ }, Cluster, }; -use pumpfun::{accounts::BondingCurveAccount, utils::CreateTokenMetadata, PriorityFee, PumpFun}; +use mai3_pumpfun_sdk::{accounts::BondingCurveAccount, utils::CreateTokenMetadata, PriorityFee, PumpFun}; // Create a new PumpFun client let payer: Keypair = Keypair::new(); diff --git a/crates/pumpfun/src/instruction/logs_event.rs b/crates/pumpfun/src/instruction/logs_event.rs new file mode 100644 index 0000000..ae43d80 --- /dev/null +++ b/crates/pumpfun/src/instruction/logs_event.rs @@ -0,0 +1,9 @@ +use serde::{Serialize, Deserialize}; +use crate::instruction::logs_data::{CreateTokenInfo, TradeInfo}; + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub enum DexEvent { + NewToken(CreateTokenInfo), + NewTrade(TradeInfo), + Error(String), +} \ No newline at end of file diff --git a/crates/pumpfun/src/instruction/logs_filter.rs b/crates/pumpfun/src/instruction/logs_filter.rs index e5a19a5..edbd0cf 100644 --- a/crates/pumpfun/src/instruction/logs_filter.rs +++ b/crates/pumpfun/src/instruction/logs_filter.rs @@ -1,5 +1,5 @@ use crate::instruction::logs_data::{CreateTokenInfo, TradeInfo}; -use crate::instruction::logs_paser::{parse_create_token_data, parse_trade_data}; +use crate::instruction::logs_parser::{parse_create_token_data, parse_trade_data}; use crate::error::ClientResult; pub struct LogFilter; diff --git a/crates/pumpfun/src/instruction/logs_paser.rs b/crates/pumpfun/src/instruction/logs_parser.rs similarity index 100% rename from crates/pumpfun/src/instruction/logs_paser.rs rename to crates/pumpfun/src/instruction/logs_parser.rs diff --git a/crates/pumpfun/src/instruction/logs_process.rs b/crates/pumpfun/src/instruction/logs_process.rs new file mode 100644 index 0000000..f30d9c7 --- /dev/null +++ b/crates/pumpfun/src/instruction/logs_process.rs @@ -0,0 +1,17 @@ +use crate::error::ClientResult; +use crate::instruction::logs_filter::{LogFilter, DexInstruction}; + +pub async fn process_logs( + signature: &str, + logs: Vec, + callback: F, +) -> ClientResult<()> +where + F: Fn(&str, DexInstruction) + Send + Sync, +{ + let instructions = LogFilter::parse_instruction(&logs)?; + for instruction in instructions { + callback(signature, instruction); + } + Ok(()) +} \ No newline at end of file diff --git a/crates/pumpfun/src/instruction/logs_subscribe.rs b/crates/pumpfun/src/instruction/logs_subscribe.rs new file mode 100644 index 0000000..e74078b --- /dev/null +++ b/crates/pumpfun/src/instruction/logs_subscribe.rs @@ -0,0 +1,89 @@ +use anchor_client::solana_client::{ + nonblocking::pubsub_client::PubsubClient, + rpc_config::{RpcTransactionLogsConfig, RpcTransactionLogsFilter} +}; + +use anchor_client::solana_sdk::commitment_config::CommitmentConfig; + +use std::sync::Arc; +use tokio::sync::mpsc; +use tokio::task::JoinHandle; +use futures_util::StreamExt; + +/// 订阅结果，包含订阅任务和取消订阅逻辑 +pub struct SubscriptionHandle { + pub task: JoinHandle<()>, + pub unsub_fn: Box, +} + +pub async fn create_pubsub_client(ws_url: &str) -> PubsubClient { + PubsubClient::new(ws_url).await.unwrap() +} + +/// 启动订阅 +pub async fn start_subscription( + ws_url: &str, + program_address: &str, + commitment: CommitmentConfig, + process_logs_callback: F, +) -> Result> +where + F: Fn(String, Vec) + Send + 'static, +{ + // 配置日志订阅 + let logs_config = RpcTransactionLogsConfig { + commitment: Some(commitment), + }; + let logs_filter = RpcTransactionLogsFilter::Mentions(vec![program_address.to_string()]); + + // 创建 PubsubClient + let sub_client = Arc::new(create_pubsub_client(ws_url).await); + + let sub_client_clone = Arc::clone(&sub_client); + + // 创建一个通道用于取消订阅 + let (unsub_tx, mut unsub_rx) = mpsc::channel(1); + + // 启动订阅任务 + let task = tokio::spawn(async move { + // let subs_client = Arc::clone(&sub_client); + let (mut stream, unsub) = sub_client_clone.logs_subscribe(logs_filter, logs_config).await.unwrap(); + loop { + tokio::select! { + _ = unsub_rx.recv() => { + eprintln!("Received shutdown signal. Unsubscribing..."); + unsub().await; + break; + } + msg = stream.next() => { + match msg { + Some(msg) => { + if let Some(_err) = msg.value.err { + continue; + } + + process_logs_callback(msg.value.signature, msg.value.logs); + } + None => { + println!("Token subscription stream ended"); + break; + } + } + } + } + } + }); + + // 返回订阅句柄和取消逻辑 + Ok(SubscriptionHandle { + task, + unsub_fn: Box::new(move || { + let _ = unsub_tx.try_send(()); // 发送取消信号 + }), + }) +} + +pub async fn stop_subscription(handle: SubscriptionHandle) { + (handle.unsub_fn)(); + handle.task.abort(); +} diff --git a/crates/pumpfun/src/instruction/mod.rs b/crates/pumpfun/src/instruction/mod.rs index 0e1f695..af0ca95 100644 --- a/crates/pumpfun/src/instruction/mod.rs +++ b/crates/pumpfun/src/instruction/mod.rs @@ -11,12 +11,18 @@ //! - `sell`: Instruction to sell tokens back to the bonding curve in exchange for SOL. pub mod logs_data; -pub mod logs_paser; +pub mod logs_parser; pub mod logs_filter; +pub mod logs_event; +pub mod logs_process; +pub mod logs_subscribe; pub use logs_data::*; -pub use logs_paser::*; +pub use logs_parser::*; pub use logs_filter::*; +pub use logs_event::*; +pub use logs_process::*; +pub use logs_subscribe::*; use crate::{constants, PumpFun}; use anchor_client::anchor_lang::InstructionData; diff --git a/crates/pumpfun/src/lib.rs b/crates/pumpfun/src/lib.rs index a179cd6..e634477 100644 --- a/crates/pumpfun/src/lib.rs +++ b/crates/pumpfun/src/lib.rs @@ -493,6 +493,9 @@ impl PumpFun { } } +use crate::instruction::logs_subscribe::{start_subscription, stop_subscription, SubscriptionHandle}; + + #[cfg(test)] mod tests { use super::*; @@ -518,4 +521,34 @@ mod tests { assert!(bonding_curve_pda.is_some()); assert!(metadata_pda != Pubkey::default()); } + + #[tokio::test] + async fn test_logs_subscription() { + let ws_url = "wss://api.mainnet-beta.solana.com"; + let program_address = "6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P"; + let commitment = CommitmentConfig::confirmed(); + + let process_logs_callback = |signature: String, logs: Vec| { + println!("Signature: {}", signature); + for log in logs { + println!("Log: {}", log); + } + }; + + let subscription = start_subscription( + ws_url, + program_address, + commitment, + process_logs_callback, + ) + .await.unwrap(); + + // 模拟运行5秒后关闭订阅 + tokio::time::sleep(tokio::time::Duration::from_secs(5)).await; + + (subscription.unsub_fn)(); // 调用取消逻辑 + subscription.task.await.unwrap(); + + println!("Subscription closed."); + } } diff --git a/crates/pumpfun/src/utils/mod.rs b/crates/pumpfun/src/utils/mod.rs index f8af304..01720b5 100644 --- a/crates/pumpfun/src/utils/mod.rs +++ b/crates/pumpfun/src/utils/mod.rs @@ -78,7 +78,7 @@ pub struct CreateTokenMetadata { /// # Examples /// /// ```rust,no_run -/// use pumpfun::utils::{CreateTokenMetadata, create_token_metadata}; +/// use mai3_pumpfun_sdk::utils::{CreateTokenMetadata, create_token_metadata}; /// /// # async fn run() -> Result<(), Box> { /// let metadata = CreateTokenMetadata { @@ -177,7 +177,7 @@ pub async fn create_token_metadata( /// /// # Example /// ```rust -/// use pumpfun::utils; +/// use mai3_pumpfun_sdk::utils; /// /// let amount = 1_000_000_000; // 1 SOL in lamports /// let slippage = 100; // 1% slippage tolerance @@ -200,7 +200,7 @@ pub fn calculate_with_slippage_buy(amount: u64, basis_points: u64) -> u64 { /// /// # Example /// ```rust -/// use pumpfun::utils; +/// use mai3_pumpfun_sdk::utils; /// /// let amount = 1_000_000_000; // 1 SOL in lamports /// let slippage = 100; // 1% slippage tolerance