From f3f046f9269d71fe7bb672f36f3919e8ddf488b0 Mon Sep 17 00:00:00 2001 From: William Date: Thu, 2 Jan 2025 19:44:38 +0800 Subject: [PATCH] remove logs_process --- crates/pumpfun/Cargo.toml | 2 +- crates/pumpfun/src/instruction/logs_parser.rs | 23 ++++++- .../pumpfun/src/instruction/logs_process.rs | 17 ----- .../pumpfun/src/instruction/logs_subscribe.rs | 53 ++++++++------- crates/pumpfun/src/instruction/mod.rs | 2 - crates/pumpfun/src/main.rs | 64 +++++++++++++++++++ 6 files changed, 110 insertions(+), 51 deletions(-) delete mode 100644 crates/pumpfun/src/instruction/logs_process.rs create mode 100755 crates/pumpfun/src/main.rs diff --git a/crates/pumpfun/Cargo.toml b/crates/pumpfun/Cargo.toml index 2db2773..abbf42c 100644 --- a/crates/pumpfun/Cargo.toml +++ b/crates/pumpfun/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "mai3-pumpfun-sdk" -version = "2.2.4" +version = "2.2.5" edition = "2021" authors = ["William "] repository = "https://github.com/MiracleAI-Labs/pumpfun-sdk" diff --git a/crates/pumpfun/src/instruction/logs_parser.rs b/crates/pumpfun/src/instruction/logs_parser.rs index f26461f..13ee41a 100644 --- a/crates/pumpfun/src/instruction/logs_parser.rs +++ b/crates/pumpfun/src/instruction/logs_parser.rs @@ -1,8 +1,25 @@ -use crate::error::{ClientError, ClientResult}; -use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64}; +use futures::{future::BoxFuture, Future, StreamExt}; use serde::{Serialize, Deserialize}; +use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64}; + +use crate::error::{ClientError, ClientResult}; +use crate::instruction::{logs_data::*, logs_filters::{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(()) +} -use crate::instruction::logs_data::*; // 添加解析函数 pub fn parse_create_token_data(data: &str) -> ClientResult { // 首先进行 base64 解码 diff --git a/crates/pumpfun/src/instruction/logs_process.rs b/crates/pumpfun/src/instruction/logs_process.rs deleted file mode 100644 index e89a5a9..0000000 --- a/crates/pumpfun/src/instruction/logs_process.rs +++ /dev/null @@ -1,17 +0,0 @@ -use crate::error::ClientResult; -use crate::instruction::logs_filters::{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 index e80d688..d0b4884 100644 --- a/crates/pumpfun/src/instruction/logs_subscribe.rs +++ b/crates/pumpfun/src/instruction/logs_subscribe.rs @@ -16,6 +16,13 @@ pub struct SubscriptionHandle { pub unsub_fn: Box, } +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() } @@ -25,19 +32,19 @@ pub async fn start_subscription( ws_url: &str, program_address: &str, commitment: CommitmentConfig, - process_logs_callback: F, + subscription_callback: F, ) -> Result> where - F: Fn(String, Vec) + Send + 'static, + F: Fn(&str, Vec) + Send + Sync + 'static, { - // 配置日志订阅 + let logs_filter = RpcTransactionLogsFilter::Mentions(vec![program_address.to_string()]); + 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 = Arc::new(PubsubClient::new(ws_url).await.unwrap()); let sub_client_clone = Arc::clone(&sub_client); @@ -46,33 +53,23 @@ where // 启动订阅任务 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 { - println!("msg: {}", msg); - Some(msg) => { - if let Some(_err) = msg.value.err { - continue; - } - - println!("logs: {}", msg.value.logs); - process_logs_callback(msg.value.signature, msg.value.logs); - } - None => { - println!("Token subscription stream ended"); - break; - } + let msg = stream.next().await; + match msg { + Some(msg) => { + if let Some(_err) = msg.value.err { + continue; } + + subscription_callback(&msg.value.signature.as_str(), msg.value.logs); } - } + None => { + println!("Token subscription stream ended"); + // break; + } + } } }); diff --git a/crates/pumpfun/src/instruction/mod.rs b/crates/pumpfun/src/instruction/mod.rs index e7b7c8f..8c2a1ca 100644 --- a/crates/pumpfun/src/instruction/mod.rs +++ b/crates/pumpfun/src/instruction/mod.rs @@ -14,14 +14,12 @@ pub mod logs_data; pub mod logs_parser; pub mod logs_filters; pub mod logs_events; -pub mod logs_process; pub mod logs_subscribe; pub use logs_data::*; pub use logs_parser::*; pub use logs_filters::*; pub use logs_events::*; -pub use logs_process::*; pub use logs_subscribe::*; use crate::{constants, PumpFun}; diff --git a/crates/pumpfun/src/main.rs b/crates/pumpfun/src/main.rs new file mode 100755 index 0000000..240c10a --- /dev/null +++ b/crates/pumpfun/src/main.rs @@ -0,0 +1,64 @@ +use std::sync::Arc; +use anchor_client::solana_client::nonblocking::rpc_client::RpcClient; +use anchor_spl::associated_token::get_associated_token_address; +use anchor_client::{ + solana_sdk::{ + native_token::LAMPORTS_PER_SOL, + instruction::Instruction, + pubkey::Pubkey, + signature::Keypair, + signer::Signer, + transaction::Transaction, + commitment_config::CommitmentConfig, + compute_budget::ComputeBudgetInstruction, + system_program::ID as SYSTEM_PROGRAM_ID, + sysvar::rent::ID as RENT_ID, + instruction::AccountMeta, + program_error::ProgramError, + program_pack::Pack, + }, + Cluster, +}; +use mai3_pumpfun_sdk::constants::accounts::PUMPFUN; +use mai3_pumpfun_sdk::instruction::logs_subscribe; +use std::str::FromStr; +use tokio::signal; + +#[tokio::main] +async fn main() -> Result<(), Box> { + start_token_subscription().await?; + + Ok(()) +} + +pub async fn start_token_subscription() -> Result<(), Box> { + println!("Starting token subscription\n"); + + let ws_url = "wss://api.mainnet-beta.solana.com"; + let commitment = CommitmentConfig::confirmed(); + + println!("program_address: {}", PUMPFUN); + + let subscription_callback = |signature: &str, logs: Vec| { + println!("=======Signature: {}=============", signature); + for log in logs { + println!("============Log: {}============", log); + } + }; + + let subscription = logs_subscribe::start_subscription( + ws_url, + &PUMPFUN.to_string(), + commitment, + subscription_callback, + ).await.unwrap(); + + subscription.task.await.unwrap(); + // stop_subscription(subscription); + + tokio::time::sleep(tokio::time::Duration::from_secs(1000)).await; + + println!("Subscription closed."); + + Ok(()) +} \ No newline at end of file