gavindiaz/solana-streamer
1
0
Fork 0
mirror of https://github.com/0xfnzero/solana-streamer.git synced 2026-08-09 07:07:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

add tokens_subscription

Browse Source
This commit is contained in:
William
2025-01-03 00:38:45 +08:00
parent c68f360944
commit 0439a40ca7
8 changed files with 121 additions and 150 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+1 -41
View File
@@ -13,51 +13,11 @@ This repository is forked from [https://github.com/nhuxhr/pumpfun-rs](https://gi
5. Add `logs_events` to define the event of the logs.
6. Add `logs_parser` to parse the logs.
## Table of Contents
- [Crates](#crates)
- [Features](#features)
- [Installation](#installation)
- [Usage](#usage)
- [API Reference](#api-reference)
- [Contributing](#contributing)
## Crates
| Name | Description | Version |
| ------------------------------------- | ---------------------------------------------------------------------------------- | ------- |
| [`pumpfun`](./crates/pumpfun) | Main client library for interacting with the PumpFun program | 2.2.2 |
| [`pumpfun-cpi`](./crates/pumpfun-cpi) | CPI (Cross-Program Invocation) interfaces for integrating with the PumpFun program | 1.1.0 |
## Features
- **Easy-to-use API**: Simplified interfaces for interacting with the PumpFun Solana program.
- **Cross-Program Invocation**: Seamless integration with other Solana programs.
- **Comprehensive Documentation**: Detailed guides and API references for all functionalities.
## Installation
Add the following to your `Cargo.toml`:
```toml
[dependencies]
mai3-pumpfun-sdk = "2.2.5"
mai3-pumpfun-sdk = "2.3.0"
```
## Usage
For detailed usage instructions, please refer to the documentation of each crate.
## API Reference
For detailed API documentation, run:
```
cargo doc --open
```
This will generate and open the API documentation in your default web browser.
## Contributing
We welcome contributions to the PumpFun Rust SDK! Please see our [Contributing Guide](CONTRIBUTING.md) for more details on how to get started.
crates/pumpfun/Cargo.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "mai3-pumpfun-sdk"
version = "2.2.5"
version = "2.3.0"
edition = "2021"
authors = ["William <william@mai3.io>"]
repository = "https://github.com/MiracleAI-Labs/pumpfun-sdk"
crates/pumpfun/src/instruction/logs_data.rs
+8
View File
@@ -1,4 +1,12 @@
use serde::{Serialize, Deserialize};
#[derive(Debug)]
pub enum DexInstruction {
CreateToken(CreateTokenInfo),
Trade(TradeInfo),
Other,
}
// 添加新的数据结构
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct CreateTokenInfo {
crates/pumpfun/src/instruction/logs_filters.rs
+10 -18
View File
@@ -1,20 +1,13 @@
use crate::instruction::logs_data::{CreateTokenInfo, TradeInfo};
use crate::instruction::logs_parser::{parse_create_token_data, parse_trade_data};
use crate::error::ClientResult;
use crate::instruction::logs_data::DexInstruction;
pub struct LogFilter;
#[derive(Debug)]
pub enum DexInstruction {
CreateToken(CreateTokenInfo),
Trade(TradeInfo),
Other,
}
impl LogFilter {
const PROGRAM_ID: &'static str = "6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P";
/// 解析交易日志并返回具体的指令类型和数据
/// Parse transaction logs and return instruction type and data
pub fn parse_instruction(logs: &[String]) -> ClientResult<Vec<DexInstruction>> {
let mut current_instruction = None;
let mut program_data = String::new();
@@ -22,11 +15,10 @@ impl LogFilter {
let mut last_data_len = 0;
let mut instructions = Vec::new();
for log in logs {
// println!("log: {:?}", log);
// 检查程序调用
// Check program invocation
if log.contains(&format!("Program {} invoke", Self::PROGRAM_ID)) {
invoke_depth += 1;
if invoke_depth == 1 { // 只在顶层调用时重置状态
if invoke_depth == 1 { // Only reset state at top level call
current_instruction = None;
program_data.clear();
last_data_len = 0;
@@ -34,12 +26,12 @@ impl LogFilter {
continue;
}
// 如果不在我们的程序中，跳过
// Skip if not in our program
if invoke_depth == 0 {
continue;
}
// 识别指令类型（只在顶层调用时）
// Identify instruction type (only at top level)
if invoke_depth == 1 && log.contains("Program log: Instruction:") {
if log.contains("Create") {
current_instruction = Some("create");
@@ -49,7 +41,7 @@ impl LogFilter {
continue;
}
// 收集 Program data
// Collect Program data
if log.starts_with("Program data: ") {
let data = log.trim_start_matches("Program data: ");
if data.len() > last_data_len {
@@ -58,10 +50,10 @@ impl LogFilter {
}
}
// 检查程序是否结束
// Check if program ends
if log.contains(&format!("Program {} success", Self::PROGRAM_ID)) {
invoke_depth -= 1;
if invoke_depth == 0 { // 只在顶层程序结束时处理数据
if invoke_depth == 0 { // Only process data when top level program ends
if let Some(instruction_type) = current_instruction {
if !program_data.is_empty() {
match instruction_type {
@@ -85,4 +77,4 @@ impl LogFilter {
Ok(instructions)
}
}
}
crates/pumpfun/src/instruction/logs_parser.rs
+15 -12
View File
@@ -3,7 +3,10 @@ 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}};
use crate::instruction::{
logs_data::{DexInstruction, CreateTokenInfo, TradeInfo},
logs_filters::LogFilter
};
pub async fn process_logs<F>(
signature: &str,
@@ -20,16 +23,16 @@ where
Ok(())
}
// 添加解析函数
// Add parsing function
pub fn parse_create_token_data(data: &str) -> ClientResult<CreateTokenInfo> {
// 首先进行 base64 解码
// First do base64 decoding
let decoded = BASE64.decode(data)
.map_err(|e| ClientError::Other(format!("Failed to decode base64: {}", e)))?;
// 跳过前缀字节（如果有的话）
// Skip prefix bytes (if any)
let mut cursor = if decoded.len() > 8 { 8 } else { 0 };
// 读取名称长度和名称
// Read name length and name
if cursor + 4 > decoded.len() {
return Err(ClientError::Other("Data too short for name length".to_string()));
}
@@ -43,7 +46,7 @@ pub fn parse_create_token_data(data: &str) -> ClientResult<CreateTokenInfo> {
.map_err(|e| ClientError::Other(format!("Invalid UTF-8 in name: {}", e)))?;
cursor += name_len;
// 读取符号长度和符号
// Read symbol length and symbol
if cursor + 4 > decoded.len() {
return Err(ClientError::Other("Data too short for symbol length".to_string()));
}
@@ -57,7 +60,7 @@ pub fn parse_create_token_data(data: &str) -> ClientResult<CreateTokenInfo> {
.map_err(|e| ClientError::Other(format!("Invalid UTF-8 in symbol: {}", e)))?;
cursor += symbol_len;
// 读取 URI 长度和 URI
// Read URI length and URI
if cursor + 4 > decoded.len() {
return Err(ClientError::Other("Data too short for URI length".to_string()));
}
@@ -71,20 +74,20 @@ pub fn parse_create_token_data(data: &str) -> ClientResult<CreateTokenInfo> {
.map_err(|e| ClientError::Other(format!("Invalid UTF-8 in uri: {}", e)))?;
cursor += uri_len;
// 确保还有足够的数据来读取公钥
// Make sure there is enough data to read public keys
if cursor + 32 * 3 > decoded.len() {
return Err(ClientError::Other("Data too short for public keys".to_string()));
}
// 解析 Mint Public Key
// Parse Mint Public Key
let mint = bs58::encode(&decoded[cursor..cursor+32]).into_string();
cursor += 32;
// 解析 Bonding Curve Public Key
// Parse Bonding Curve Public Key
let bonding_curve = bs58::encode(&decoded[cursor..cursor+32]).into_string();
cursor += 32;
// 解析 User Public Key
// Parse User Public Key
let user = bs58::encode(&decoded[cursor..cursor+32]).into_string();
Ok(CreateTokenInfo {
@@ -113,7 +116,7 @@ pub fn parse_trade_data(data: &str) -> ClientResult<TradeInfo> {
)
)?;
let mut cursor = 8; // 跳过前缀
let mut cursor = 8; // Skip prefix
// 1. Mint (32 bytes)
let mint = bs58::encode(&decoded[cursor..cursor + 32]).into_string();
crates/pumpfun/src/instruction/logs_subscribe.rs
+30 -15
View File
@@ -8,9 +8,14 @@ use anchor_client::solana_sdk::commitment_config::CommitmentConfig;
use std::sync::Arc;
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use futures::{future::BoxFuture, Future, StreamExt};
use futures::StreamExt;
use crate::instruction::{
logs_events::DexEvent,
logs_data::DexInstruction,
logs_filters::LogFilter
};
/// 订阅结果，包含订阅任务和取消订阅逻辑
/// Subscription handle containing task and unsubscribe logic
pub struct SubscriptionHandle {
pub task: JoinHandle<()>,
pub unsub_fn: Box<dyn Fn() + Send>,
@@ -28,14 +33,14 @@ pub async fn create_pubsub_client(ws_url: &str) -> PubsubClient {
}
/// 启动订阅
pub async fn start_subscription<F>(
pub async fn tokens_subscription<F>(
ws_url: &str,
program_address: &str,
commitment: CommitmentConfig,
subscription_callback: F,
callback: F,
) -> Result<SubscriptionHandle, Box<dyn std::error::Error>>
where
F: Fn(&str, Vec<String>) + Send + Sync + 'static,
F: Fn(DexEvent) + Send + Sync + 'static,
{
let logs_filter = RpcTransactionLogsFilter::Mentions(vec![program_address.to_string()]);
@@ -43,17 +48,17 @@ where
commitment: Some(commitment),
};
// 创建 PubsubClient
// Create PubsubClient
let sub_client = Arc::new(PubsubClient::new(ws_url).await.unwrap());
let sub_client_clone = Arc::clone(&sub_client);
// 创建一个通道用于取消订阅
let (unsub_tx, mut unsub_rx) = mpsc::channel(1);
// Create channel for unsubscribe
let (unsub_tx, _) = mpsc::channel(1);
// 启动订阅任务
// Start subscription task
let task = tokio::spawn(async move {
let (mut stream, unsub) = sub_client_clone.logs_subscribe(logs_filter, logs_config).await.unwrap();
let (mut stream, _) = sub_client_clone.logs_subscribe(logs_filter, logs_config).await.unwrap();
loop {
let msg = stream.next().await;
@@ -62,22 +67,32 @@ where
if let Some(_err) = msg.value.err {
continue;
}
subscription_callback(&msg.value.signature.as_str(), msg.value.logs);
let instructions = LogFilter::parse_instruction(&msg.value.logs).unwrap();
for instruction in instructions {
match instruction {
DexInstruction::CreateToken(token_info) => {
callback(DexEvent::NewToken(token_info));
}
DexInstruction::Trade(trade_info) => {
callback(DexEvent::NewTrade(trade_info));
}
_ => {}
}
}
}
None => {
println!("Token subscription stream ended");
// break;
}
}
}
});
// 返回订阅句柄和取消逻辑
// Return subscription handle and unsubscribe logic
Ok(SubscriptionHandle {
task,
unsub_fn: Box::new(move || {
let _ = unsub_tx.try_send(()); // 发送取消信号
let _ = unsub_tx.try_send(());
}),
})
}
crates/pumpfun/src/lib.rs
+25 -25
View File
@@ -493,7 +493,7 @@ impl PumpFun {
}
}
use crate::instruction::logs_subscribe::{start_subscription, stop_subscription, SubscriptionHandle};
// use crate::instruction::logs_subscribe::{start_subscription, stop_subscription, SubscriptionHandle};
#[cfg(test)]
@@ -522,33 +522,33 @@ mod tests {
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();
// #[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 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();
// 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;
// // 模拟运行5秒后关闭订阅
// tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
(subscription.unsub_fn)(); // 调用取消逻辑
subscription.task.await.unwrap();
// (subscription.unsub_fn)(); // 调用取消逻辑
// subscription.task.await.unwrap();
println!("Subscription closed.");
}
// println!("Subscription closed.");
// }
}
crates/pumpfun/src/main.rs
+31 -38
View File
@@ -1,26 +1,8 @@
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 mai3_pumpfun_sdk::instruction::logs_subscribe::tokens_subscription;
use mai3_pumpfun_sdk::instruction::logs_events::DexEvent;
use mai3_pumpfun_sdk::instruction::logs_subscribe::stop_subscription;
use anchor_client::solana_sdk::commitment_config::CommitmentConfig;
use std::str::FromStr;
use tokio::signal;
@@ -35,30 +17,41 @@ pub async fn start_token_subscription() -> Result<(), Box<dyn std::error::Error>
println!("Starting token subscription\n");
let ws_url = "wss://api.mainnet-beta.solana.com";
// Program address
let program_address = "6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P";
// Set commitment
let commitment = CommitmentConfig::confirmed();
println!("program_address: {}", PUMPFUN);
let subscription_callback = |signature: &str, logs: Vec<String>| {
println!("=======Signature: {}=============", signature);
for log in logs {
println!("============Log: {}============", log);
// Define callback function
let callback = |event: DexEvent| {
match event {
DexEvent::NewToken(token_info) => {
println!("Received new token event: {:?}", token_info);
},
DexEvent::NewTrade(trade_info) => {
println!("Received new trade event: {:?}", trade_info);
},
DexEvent::Error(err) => {
println!("Received error: {}", err);
}
}
};
let subscription = logs_subscribe::start_subscription(
// Start subscription
let subscription = tokens_subscription(
ws_url,
&PUMPFUN.to_string(),
program_address,
commitment,
subscription_callback,
callback
).await.unwrap();
subscription.task.await.unwrap();
// stop_subscription(subscription);
// Wait for a while to receive events
tokio::time::sleep(tokio::time::Duration::from_secs(60)).await;
tokio::time::sleep(tokio::time::Duration::from_secs(1000)).await;
println!("Subscription closed.");
// Stop subscription
stop_subscription(subscription).await;
Ok(())
}