add logs_process and logs_subscribe

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 <william@mai3.io>"]
 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"

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();

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();

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),
+}

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;

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<F>(
+    signature: &str,
+    logs: Vec<String>,
+    callback: F,
+) -> ClientResult<()>
+where
+    F: Fn(&str, DexInstruction) + Send + Sync,
+{
+    let instructions = LogFilter::parse_instruction(&logs)?;
+    for instruction in instructions {
+        callback(signature, instruction);
+    }
+    Ok(())
+}

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<dyn Fn() + Send>,
+}
+
+pub async fn create_pubsub_client(ws_url: &str) -> PubsubClient {
+    PubsubClient::new(ws_url).await.unwrap()
+}
+
+/// 启动订阅
+pub async fn start_subscription<F>(
+    ws_url: &str,
+    program_address: &str,
+    commitment: CommitmentConfig,
+    process_logs_callback: F,
+) -> Result<SubscriptionHandle, Box<dyn std::error::Error>>
+where
+    F: Fn(String, Vec<String>) + 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();
+}

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;

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<String>| {
+            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.");
+    }
 }

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<dyn std::error::Error>> {
 /// 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