diff --git a/README.md b/README.md index d52156a..7dfa77a 100755 --- a/README.md +++ b/README.md @@ -182,8 +182,19 @@ let grpc = YellowstoneGrpc::new_with_config(endpoint, token, config)?; | Monitor transaction events using Yellowstone gRPC | `cargo run --example grpc_example` | [examples/grpc_example.rs](examples/grpc_example.rs) | | Monitor transaction events using ShredStream | `cargo run --example shred_example` | [examples/shred_example.rs](examples/shred_example.rs) | | Parse Solana mainnet transaction data | `cargo run --example parse_tx_events` | [examples/parse_tx_events.rs](examples/parse_tx_events.rs) | +| Parse PumpFun transaction from RPC (signature: `TX_SIGNATURE` or CLI arg) | `cargo run --example parse_pump_tx --release` | [examples/parse_pump_tx.rs](examples/parse_pump_tx.rs) | +| Parse PumpSwap transaction from RPC | `cargo run --example parse_pumpswap_tx --release` | [examples/parse_pumpswap_tx.rs](examples/parse_pumpswap_tx.rs) | +| Parse Meteora DAMM v2 transaction from RPC | `TX_SIGNATURE= cargo run --example parse_meteora_damm_tx --release` | [examples/parse_meteora_damm_tx.rs](examples/parse_meteora_damm_tx.rs) | +| Debug PumpFun transaction (fetch, print meta/logs, parse) | `TX_SIGNATURE= cargo run --example debug_pump_tx --release` | [examples/debug_pump_tx.rs](examples/debug_pump_tx.rs) | +| Debug PumpSwap transaction (fetch, print meta, parse) | `TX_SIGNATURE= cargo run --example debug_pumpswap_tx --release` | [examples/debug_pumpswap_tx.rs](examples/debug_pumpswap_tx.rs) | | Update filters at runtime | `cargo run --example dynamic_subscription` | [examples/dynamic_subscription.rs](examples/dynamic_subscription.rs) | +| Quick test: subscribe to PumpFun, print first 10 or run 60s | `cargo run --example pumpfun_quick_test --release` | [examples/pumpfun_quick_test.rs](examples/pumpfun_quick_test.rs) | +| PumpFun trade filter: Buy/Sell/Create with latency | `cargo run --example pumpfun_trade_filter --release` | [examples/pumpfun_trade_filter.rs](examples/pumpfun_trade_filter.rs) | +| PumpFun gRPC subscription with metrics | `cargo run --example pumpfun_with_metrics --release` | [examples/pumpfun_with_metrics.rs](examples/pumpfun_with_metrics.rs) | +| PumpSwap gRPC subscription with metrics | `cargo run --example pumpswap_with_metrics --release` | [examples/pumpswap_with_metrics.rs](examples/pumpswap_with_metrics.rs) | +| Meteora DAMM v2 gRPC subscription | `cargo run --example meteora_damm_grpc --release` | [examples/meteora_damm_grpc.rs](examples/meteora_damm_grpc.rs) | | Monitor specific token account balance changes | `cargo run --example token_balance_listen_example` | [examples/token_balance_listen_example.rs](examples/token_balance_listen_example.rs) | +| Monitor token decimals via account subscription | `cargo run --example token_decimals_listen_example` | [examples/token_decimals_listen_example.rs](examples/token_decimals_listen_example.rs) | | Track nonce account state changes | `cargo run --example nonce_listen_example` | [examples/nonce_listen_example.rs](examples/nonce_listen_example.rs) | | Monitor PumpSwap pool accounts using memcmp filters | `cargo run --example pumpswap_pool_account_listen_example` | [examples/pumpswap_pool_account_listen_example.rs](examples/pumpswap_pool_account_listen_example.rs) | | Monitor all associated token accounts for specific mints using memcmp filters | `cargo run --example mint_all_ata_account_listen_example` | [examples/mint_all_ata_account_listen_example.rs](examples/mint_all_ata_account_listen_example.rs) | diff --git a/README_CN.md b/README_CN.md index 9743784..1e23d4d 100644 --- a/README_CN.md +++ b/README_CN.md @@ -181,8 +181,19 @@ let grpc = YellowstoneGrpc::new_with_config(endpoint, token, config)?; | 使用 Yellowstone gRPC 监控交易事件 | `cargo run --example grpc_example` | [examples/grpc_example.rs](examples/grpc_example.rs) | | 使用 ShredStream 监控交易事件 | `cargo run --example shred_example` | [examples/shred_example.rs](examples/shred_example.rs) | | 解析 Solana 主网交易数据 | `cargo run --example parse_tx_events` | [examples/parse_tx_events.rs](examples/parse_tx_events.rs) | +| 从 RPC 解析 PumpFun 交易(签名:环境变量 `TX_SIGNATURE` 或 CLI 参数) | `cargo run --example parse_pump_tx --release` | [examples/parse_pump_tx.rs](examples/parse_pump_tx.rs) | +| 从 RPC 解析 PumpSwap 交易 | `cargo run --example parse_pumpswap_tx --release` | [examples/parse_pumpswap_tx.rs](examples/parse_pumpswap_tx.rs) | +| 从 RPC 解析 Meteora DAMM v2 交易 | `TX_SIGNATURE= cargo run --example parse_meteora_damm_tx --release` | [examples/parse_meteora_damm_tx.rs](examples/parse_meteora_damm_tx.rs) | +| 调试 PumpFun 交易(拉取、打印 meta/logs、解析) | `TX_SIGNATURE= cargo run --example debug_pump_tx --release` | [examples/debug_pump_tx.rs](examples/debug_pump_tx.rs) | +| 调试 PumpSwap 交易(拉取、打印 meta、解析) | `TX_SIGNATURE= cargo run --example debug_pumpswap_tx --release` | [examples/debug_pumpswap_tx.rs](examples/debug_pumpswap_tx.rs) | | 运行时更新过滤器 | `cargo run --example dynamic_subscription` | [examples/dynamic_subscription.rs](examples/dynamic_subscription.rs) | +| 快速测试:订阅 PumpFun,打印前 10 条或运行 60 秒 | `cargo run --example pumpfun_quick_test --release` | [examples/pumpfun_quick_test.rs](examples/pumpfun_quick_test.rs) | +| PumpFun 交易过滤:买入/卖出/创建及延迟统计 | `cargo run --example pumpfun_trade_filter --release` | [examples/pumpfun_trade_filter.rs](examples/pumpfun_trade_filter.rs) | +| PumpFun gRPC 订阅(含指标) | `cargo run --example pumpfun_with_metrics --release` | [examples/pumpfun_with_metrics.rs](examples/pumpfun_with_metrics.rs) | +| PumpSwap gRPC 订阅(含指标) | `cargo run --example pumpswap_with_metrics --release` | [examples/pumpswap_with_metrics.rs](examples/pumpswap_with_metrics.rs) | +| Meteora DAMM v2 gRPC 订阅 | `cargo run --example meteora_damm_grpc --release` | [examples/meteora_damm_grpc.rs](examples/meteora_damm_grpc.rs) | | 监控特定代币账户余额变化 | `cargo run --example token_balance_listen_example` | [examples/token_balance_listen_example.rs](examples/token_balance_listen_example.rs) | +| 通过账户订阅监控代币精度 | `cargo run --example token_decimals_listen_example` | [examples/token_decimals_listen_example.rs](examples/token_decimals_listen_example.rs) | | 跟踪 nonce 账户状态变化 | `cargo run --example nonce_listen_example` | [examples/nonce_listen_example.rs](examples/nonce_listen_example.rs) | | 使用 memcmp 过滤器监控 PumpSwap 池账户 | `cargo run --example pumpswap_pool_account_listen_example` | [examples/pumpswap_pool_account_listen_example.rs](examples/pumpswap_pool_account_listen_example.rs) | | 使用 memcmp 过滤器监控特定代币的所有关联代币账户 | `cargo run --example mint_all_ata_account_listen_example` | [examples/mint_all_ata_account_listen_example.rs](examples/mint_all_ata_account_listen_example.rs) | diff --git a/examples/debug_pump_tx.rs b/examples/debug_pump_tx.rs new file mode 100644 index 0000000..7877044 --- /dev/null +++ b/examples/debug_pump_tx.rs @@ -0,0 +1,163 @@ +//! Debug PumpFun transaction: fetch from RPC, print meta/logs, then parse with EventParser. +//! +//! Usage: cargo run --example debug_pump_tx --release +//! TX_SIGNATURE= SOLANA_RPC_URL= cargo run --example debug_pump_tx --release + +use anyhow::Result; +use solana_commitment_config::CommitmentConfig; +use solana_streamer_sdk::streaming::event_parser::core::event_parser::EventParser; +use solana_streamer_sdk::streaming::event_parser::{DexEvent, Protocol}; +use std::str::FromStr; +use std::sync::Arc; + +#[tokio::main] +async fn main() -> Result<()> { + let default_sig = "5curEt85cQhAK6R9pntSJ4fmYCiPEG22NjZyGrnGSbNwAkHJMN25T9Efp1n9Tf9vGXhnDXMQYrCNpoRHQTMcZ1s9"; + let tx_sig = std::env::var("TX_SIGNATURE").unwrap_or_else(|_| default_sig.to_string()); + let rpc_url = std::env::var("SOLANA_RPC_URL") + .unwrap_or_else(|_| "https://api.mainnet-beta.solana.com".to_string()); + + println!("=== Debug PumpFun Transaction ===\n"); + println!("Signature: {}\n", tx_sig); + println!("RPC: {}\n", rpc_url); + + let signature = solana_sdk::signature::Signature::from_str(&tx_sig)?; + let client = solana_client::nonblocking::rpc_client::RpcClient::new(rpc_url.clone()); + + let transaction = match client + .get_transaction_with_config( + &signature, + solana_client::rpc_config::RpcTransactionConfig { + encoding: Some(solana_transaction_status::UiTransactionEncoding::Base64), + commitment: Some(CommitmentConfig::confirmed()), + max_supported_transaction_version: Some(0), + }, + ) + .await + { + Ok(tx) => tx, + Err(e) => { + eprintln!("Failed to fetch: {}", e); + return Ok(()); + } + }; + + println!("Slot: {}", transaction.slot); + println!("Block time: {:?}", transaction.block_time); + + if let Some(ref meta) = transaction.transaction.meta { + println!("\n=== Transaction Meta ==="); + println!("Fee: {}", meta.fee); + if let solana_transaction_status::option_serializer::OptionSerializer::Some(units) = &meta.compute_units_consumed { + println!("Compute units: {}", units); + } + if let solana_transaction_status::option_serializer::OptionSerializer::Some(logs) = &meta.log_messages { + println!("\n=== Logs ({} lines) ===", logs.len()); + for (i, log) in logs.iter().enumerate() { + if log.contains("Program 6EF8") || log.contains("invoke") || log.contains("success") { + println!(" {}: {}", i, log); + } + } + } + } + + println!("\n=== Parsing with EventParser ==="); + let versioned_tx = match transaction.transaction.transaction.decode() { + Some(tx) => tx, + None => { + println!("Failed to decode transaction"); + return Ok(()); + } + }; + + use prost_types::Timestamp; + use solana_sdk::{message::compiled_instruction::CompiledInstruction, pubkey::Pubkey}; + use solana_transaction_status::{InnerInstruction, InnerInstructions, UiInstruction}; + + let mut inner_instructions_vec: Vec = Vec::new(); + if let Some(meta) = &transaction.transaction.meta { + if let solana_transaction_status::option_serializer::OptionSerializer::Some(ui_inner_insts) = + &meta.inner_instructions + { + for ui_inner in ui_inner_insts { + let mut converted = Vec::new(); + for ui_instruction in &ui_inner.instructions { + if let UiInstruction::Compiled(ui_compiled) = ui_instruction { + if let Ok(data) = solana_sdk::bs58::decode(&ui_compiled.data).into_vec() { + converted.push(InnerInstruction { + instruction: CompiledInstruction { + program_id_index: ui_compiled.program_id_index, + accounts: ui_compiled.accounts.to_vec(), + data, + }, + stack_height: ui_compiled.stack_height, + }); + } + } + } + inner_instructions_vec.push(InnerInstructions { + index: ui_inner.index, + instructions: converted, + }); + } + } + } + + let mut address_table_lookups: Vec = vec![]; + if let Some(meta) = &transaction.transaction.meta { + if let solana_transaction_status::option_serializer::OptionSerializer::Some(loaded) = + &meta.loaded_addresses + { + for s in loaded.writable.iter().chain(loaded.readonly.iter()) { + if let Ok(p) = s.parse::() { + address_table_lookups.push(p); + } + } + } + } + + let mut accounts: Vec = versioned_tx.message.static_account_keys().to_vec(); + accounts.extend(address_table_lookups); + + let slot = transaction.slot; + let block_time = transaction + .block_time + .map(|t| Timestamp { seconds: t as i64, nanos: 0 }); + let recv_us = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_micros() as i64; + + let protocols = vec![ + Protocol::PumpFun, + Protocol::PumpSwap, + Protocol::Bonk, + Protocol::RaydiumClmm, + Protocol::RaydiumCpmm, + Protocol::RaydiumAmmV4, + Protocol::MeteoraDammV2, + ]; + + let callback = Arc::new(|event: DexEvent| { + println!("Parsed event: {:?}\n", event); + }); + + EventParser::parse_instruction_events_from_versioned_transaction( + &protocols, + None, + &versioned_tx, + signature, + Some(slot), + block_time, + recv_us, + &accounts, + &inner_instructions_vec, + None, + None, + callback, + ) + .await?; + + println!("\n✓ Done."); + Ok(()) +} diff --git a/examples/debug_pumpswap_tx.rs b/examples/debug_pumpswap_tx.rs new file mode 100644 index 0000000..dc85a3c --- /dev/null +++ b/examples/debug_pumpswap_tx.rs @@ -0,0 +1,148 @@ +//! Debug PumpSwap transaction: fetch from RPC, print meta, then parse with EventParser. +//! +//! Usage: TX_SIGNATURE= cargo run --example debug_pumpswap_tx --release + +use anyhow::Result; +use solana_commitment_config::CommitmentConfig; +use solana_streamer_sdk::streaming::event_parser::core::event_parser::EventParser; +use solana_streamer_sdk::streaming::event_parser::{DexEvent, Protocol}; +use std::str::FromStr; +use std::sync::Arc; + +#[tokio::main] +async fn main() -> Result<()> { + let tx_sig = std::env::var("TX_SIGNATURE").unwrap_or_else(|_| { + eprintln!("Usage: TX_SIGNATURE= cargo run --example debug_pumpswap_tx --release"); + std::process::exit(1); + }); + let rpc_url = std::env::var("SOLANA_RPC_URL") + .unwrap_or_else(|_| "https://api.mainnet-beta.solana.com".to_string()); + + println!("=== Debug PumpSwap Transaction ===\n"); + println!("Signature: {}\n", tx_sig); + + let signature = solana_sdk::signature::Signature::from_str(&tx_sig)?; + let client = solana_client::nonblocking::rpc_client::RpcClient::new(rpc_url.clone()); + + let transaction = match client + .get_transaction_with_config( + &signature, + solana_client::rpc_config::RpcTransactionConfig { + encoding: Some(solana_transaction_status::UiTransactionEncoding::Base64), + commitment: Some(CommitmentConfig::confirmed()), + max_supported_transaction_version: Some(0), + }, + ) + .await + { + Ok(tx) => tx, + Err(e) => { + eprintln!("Failed to fetch: {}", e); + return Ok(()); + } + }; + + println!("Slot: {}", transaction.slot); + if let Some(ref meta) = transaction.transaction.meta { + println!("Fee: {}", meta.fee); + } + + let versioned_tx = match transaction.transaction.transaction.decode() { + Some(tx) => tx, + None => { + println!("Failed to decode"); + return Ok(()); + } + }; + + use prost_types::Timestamp; + use solana_sdk::{message::compiled_instruction::CompiledInstruction, pubkey::Pubkey}; + use solana_transaction_status::{InnerInstruction, InnerInstructions, UiInstruction}; + + let mut inner_instructions_vec: Vec = Vec::new(); + if let Some(meta) = &transaction.transaction.meta { + if let solana_transaction_status::option_serializer::OptionSerializer::Some(ui_inner_insts) = + &meta.inner_instructions + { + for ui_inner in ui_inner_insts { + let mut converted = Vec::new(); + for ui_instruction in &ui_inner.instructions { + if let UiInstruction::Compiled(ui_compiled) = ui_instruction { + if let Ok(data) = solana_sdk::bs58::decode(&ui_compiled.data).into_vec() { + converted.push(InnerInstruction { + instruction: CompiledInstruction { + program_id_index: ui_compiled.program_id_index, + accounts: ui_compiled.accounts.to_vec(), + data, + }, + stack_height: ui_compiled.stack_height, + }); + } + } + } + inner_instructions_vec.push(InnerInstructions { + index: ui_inner.index, + instructions: converted, + }); + } + } + } + + let mut address_table_lookups: Vec = vec![]; + if let Some(meta) = &transaction.transaction.meta { + if let solana_transaction_status::option_serializer::OptionSerializer::Some(loaded) = + &meta.loaded_addresses + { + for s in loaded.writable.iter().chain(loaded.readonly.iter()) { + if let Ok(p) = s.parse::() { + address_table_lookups.push(p); + } + } + } + } + + let mut accounts: Vec = versioned_tx.message.static_account_keys().to_vec(); + accounts.extend(address_table_lookups); + + let slot = transaction.slot; + let block_time = transaction + .block_time + .map(|t| Timestamp { seconds: t as i64, nanos: 0 }); + let recv_us = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_micros() as i64; + + let protocols = vec![ + Protocol::PumpFun, + Protocol::PumpSwap, + Protocol::Bonk, + Protocol::RaydiumClmm, + Protocol::RaydiumCpmm, + Protocol::RaydiumAmmV4, + Protocol::MeteoraDammV2, + ]; + + let callback = Arc::new(|event: DexEvent| { + println!("Event: {:?}\n", event); + }); + + EventParser::parse_instruction_events_from_versioned_transaction( + &protocols, + None, + &versioned_tx, + signature, + Some(slot), + block_time, + recv_us, + &accounts, + &inner_instructions_vec, + None, + None, + callback, + ) + .await?; + + println!("\n✓ Done."); + Ok(()) +} diff --git a/examples/meteora_damm_grpc.rs b/examples/meteora_damm_grpc.rs new file mode 100644 index 0000000..5cd6eaa --- /dev/null +++ b/examples/meteora_damm_grpc.rs @@ -0,0 +1,52 @@ +//! Meteora DAMM v2 subscription via gRPC. +//! +//! Usage: cargo run --example meteora_damm_grpc --release + +use solana_streamer_sdk::streaming::event_parser::{DexEvent, Protocol}; +use solana_streamer_sdk::streaming::event_parser::protocols::meteora_damm_v2::parser::METEORA_DAMM_V2_PROGRAM_ID; +use solana_streamer_sdk::streaming::grpc::ClientConfig; +use solana_streamer_sdk::streaming::yellowstone_grpc::{AccountFilter, TransactionFilter, YellowstoneGrpc}; + +#[tokio::main] +async fn main() -> Result<(), Box> { + let _ = rustls::crypto::ring::default_provider().install_default(); + + println!("Meteora DAMM v2 gRPC (solana-streamer)\n"); + + let grpc = YellowstoneGrpc::new_with_config( + std::env::var("GRPC_ENDPOINT").unwrap_or_else(|_| "https://solana-yellowstone-grpc.publicnode.com:443".to_string()), + std::env::var("GRPC_AUTH_TOKEN").ok(), + ClientConfig::default(), + )?; + + let transaction_filter = TransactionFilter { + account_include: vec![METEORA_DAMM_V2_PROGRAM_ID.to_string()], + account_exclude: vec![], + account_required: vec![], + }; + let account_filter = AccountFilter { + account: vec![], + owner: vec![METEORA_DAMM_V2_PROGRAM_ID.to_string()], + filters: vec![], + }; + + let callback = |event: DexEvent| { + println!("Event: {:?}", event.metadata().event_type); + }; + + grpc.subscribe_events_immediate( + vec![Protocol::MeteoraDammV2], + None, + vec![transaction_filter], + vec![account_filter], + None, + None, + callback, + ) + .await?; + + println!("Press Ctrl+C to stop...\n"); + tokio::signal::ctrl_c().await?; + grpc.stop().await; + Ok(()) +} diff --git a/examples/nonce_listen_example.rs b/examples/nonce_listen_example.rs index 32e2a4e..1933f5b 100644 --- a/examples/nonce_listen_example.rs +++ b/examples/nonce_listen_example.rs @@ -40,10 +40,13 @@ async fn test_grpc() -> Result<(), Box> { let transaction_filter = TransactionFilter { account_include, account_exclude, account_required }; - let nonce_account = "use_your_nonce_account_here".to_string(); + let nonce_account = std::env::var("NONCE_ACCOUNT").unwrap_or_else(|_| { + eprintln!("Usage: NONCE_ACCOUNT= cargo run --example nonce_listen_example --release"); + std::process::exit(1); + }); // Listen to account data belonging to owner programs -> account event monitoring let account_filter = - AccountFilter { account: vec![nonce_account], owner: vec![], filters: vec![] }; + AccountFilter { account: vec![nonce_account.clone()], owner: vec![], filters: vec![] }; // Event filtering let event_type_filter = Some(EventTypeFilter { include: vec![EventType::NonceAccount] }); diff --git a/examples/parse_meteora_damm_tx.rs b/examples/parse_meteora_damm_tx.rs new file mode 100644 index 0000000..c3bb9f0 --- /dev/null +++ b/examples/parse_meteora_damm_tx.rs @@ -0,0 +1,148 @@ +//! Parse a Meteora DAMM v2 transaction from RPC using solana-streamer EventParser. +//! +//! Usage: +//! cargo run --example parse_meteora_damm_tx --release +//! TX_SIGNATURE= SOLANA_RPC_URL= cargo run --example parse_meteora_damm_tx --release + +use anyhow::Result; +use solana_commitment_config::CommitmentConfig; +use solana_streamer_sdk::streaming::event_parser::core::event_parser::EventParser; +use solana_streamer_sdk::streaming::event_parser::{DexEvent, Protocol}; +use std::str::FromStr; +use std::sync::Arc; + +#[tokio::main] +async fn main() -> Result<()> { + let tx_sig = std::env::var("TX_SIGNATURE").unwrap_or_else(|_| { + eprintln!("Usage: TX_SIGNATURE= cargo run --example parse_meteora_damm_tx --release"); + std::process::exit(1); + }); + let rpc_url = std::env::var("SOLANA_RPC_URL") + .unwrap_or_else(|_| "https://api.mainnet-beta.solana.com".to_string()); + + println!("=== Meteora DAMM v2 Transaction Parser (solana-streamer) ===\n"); + println!("Transaction: {}\n", tx_sig); + + parse_one_tx(&tx_sig, &rpc_url).await?; + println!("\n✓ Done."); + Ok(()) +} + +async fn parse_one_tx(signature_str: &str, rpc_url: &str) -> Result<()> { + use prost_types::Timestamp; + use solana_sdk::{ + message::compiled_instruction::CompiledInstruction, + pubkey::Pubkey, + signature::Signature, + }; + use solana_transaction_status::{InnerInstruction, InnerInstructions, UiInstruction, UiTransactionEncoding}; + + let signature = Signature::from_str(signature_str)?; + let client = solana_client::nonblocking::rpc_client::RpcClient::new(rpc_url.to_string()); + + let transaction = match client + .get_transaction_with_config( + &signature, + solana_client::rpc_config::RpcTransactionConfig { + encoding: Some(UiTransactionEncoding::Base64), + commitment: Some(CommitmentConfig::confirmed()), + max_supported_transaction_version: Some(0), + }, + ) + .await + { + Ok(tx) => tx, + Err(e) => anyhow::bail!("Failed to fetch transaction: {}", e), + }; + + let versioned_tx = match transaction.transaction.transaction.decode() { + Some(tx) => tx, + None => anyhow::bail!("Failed to decode transaction"), + }; + + let mut inner_instructions_vec: Vec = Vec::new(); + if let Some(meta) = &transaction.transaction.meta { + if let solana_transaction_status::option_serializer::OptionSerializer::Some(ui_inner_insts) = + &meta.inner_instructions + { + for ui_inner in ui_inner_insts { + let mut converted = Vec::new(); + for ui_instruction in &ui_inner.instructions { + if let UiInstruction::Compiled(ui_compiled) = ui_instruction { + if let Ok(data) = solana_sdk::bs58::decode(&ui_compiled.data).into_vec() { + converted.push(InnerInstruction { + instruction: CompiledInstruction { + program_id_index: ui_compiled.program_id_index, + accounts: ui_compiled.accounts.to_vec(), + data, + }, + stack_height: ui_compiled.stack_height, + }); + } + } + } + inner_instructions_vec.push(InnerInstructions { + index: ui_inner.index, + instructions: converted, + }); + } + } + } + + let mut address_table_lookups: Vec = vec![]; + if let Some(meta) = &transaction.transaction.meta { + if let solana_transaction_status::option_serializer::OptionSerializer::Some(loaded) = + &meta.loaded_addresses + { + for s in loaded.writable.iter().chain(loaded.readonly.iter()) { + if let Ok(p) = s.parse::() { + address_table_lookups.push(p); + } + } + } + } + + let mut accounts: Vec = versioned_tx.message.static_account_keys().to_vec(); + accounts.extend(address_table_lookups); + + let slot = transaction.slot; + let block_time = transaction + .block_time + .map(|t| Timestamp { seconds: t as i64, nanos: 0 }); + let recv_us = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_micros() as i64; + + let protocols = vec![ + Protocol::PumpFun, + Protocol::PumpSwap, + Protocol::Bonk, + Protocol::RaydiumClmm, + Protocol::RaydiumCpmm, + Protocol::RaydiumAmmV4, + Protocol::MeteoraDammV2, + ]; + + let callback = Arc::new(|event: DexEvent| { + println!("Event: {:?}\n", event); + }); + + EventParser::parse_instruction_events_from_versioned_transaction( + &protocols, + None, + &versioned_tx, + signature, + Some(slot), + block_time, + recv_us, + &accounts, + &inner_instructions_vec, + None, + None, + callback, + ) + .await?; + + Ok(()) +} diff --git a/examples/parse_pump_tx.rs b/examples/parse_pump_tx.rs new file mode 100644 index 0000000..4c99b0a --- /dev/null +++ b/examples/parse_pump_tx.rs @@ -0,0 +1,165 @@ +//! Parse a PumpFun transaction from RPC using solana-streamer EventParser. +//! +//! Signature: env `TX_SIGNATURE` or first CLI arg. RPC: env `SOLANA_RPC_URL`. +//! +//! cargo run --example parse_pump_tx --release +//! TX_SIGNATURE=your_sig cargo run --example parse_pump_tx --release +//! cargo run --example parse_pump_tx --release -- your_sig +//! SOLANA_RPC_URL=https://... cargo run --example parse_pump_tx --release + +use anyhow::Result; +use solana_commitment_config::CommitmentConfig; +use solana_streamer_sdk::streaming::event_parser::core::event_parser::EventParser; +use solana_streamer_sdk::streaming::event_parser::{DexEvent, Protocol}; +use std::str::FromStr; +use std::sync::Arc; + +#[tokio::main] +async fn main() -> Result<()> { + let default_sig = "64srGF8CnTz9zPbdayWYmzs5aVRFBcfjDcidFVvBgAD25VMh52wr88vma7ytSbAZT3C5Giu5BPyGfNfLexLSrKhP"; + let tx_sig = std::env::var("TX_SIGNATURE") + .ok() + .or_else(|| std::env::args().nth(1)) + .unwrap_or_else(|| default_sig.to_string()); + let rpc_url = std::env::var("SOLANA_RPC_URL") + .unwrap_or_else(|_| "https://api.mainnet-beta.solana.com".to_string()); + + println!("=== PumpFun Transaction Parser (solana-streamer) ===\n"); + println!("Transaction: {}\n", tx_sig); + println!("RPC: {}\n", rpc_url); + + parse_one_tx(&tx_sig, &rpc_url).await?; + println!("\n✓ Done."); + Ok(()) +} + +async fn parse_one_tx(signature_str: &str, rpc_url: &str) -> Result<()> { + use prost_types::Timestamp; + use solana_sdk::{ + message::compiled_instruction::CompiledInstruction, + pubkey::Pubkey, + signature::Signature, + }; + use solana_transaction_status::{InnerInstruction, InnerInstructions, UiInstruction, UiTransactionEncoding}; + + let signature = Signature::from_str(signature_str)?; + let client = solana_client::nonblocking::rpc_client::RpcClient::new(rpc_url.to_string()); + + let transaction = match client + .get_transaction_with_config( + &signature, + solana_client::rpc_config::RpcTransactionConfig { + encoding: Some(UiTransactionEncoding::Base64), + commitment: Some(CommitmentConfig::confirmed()), + max_supported_transaction_version: Some(0), + }, + ) + .await + { + Ok(tx) => tx, + Err(e) => { + anyhow::bail!( + "Failed to fetch transaction: {}. If RPC returned null, try an archive RPC (SOLANA_RPC_URL).", + e + ); + } + }; + + println!("Slot: {}", transaction.slot); + if let Some(bt) = transaction.block_time { + println!("Block time: {}", bt); + } + + let versioned_tx = match transaction.transaction.transaction.decode() { + Some(tx) => tx, + None => anyhow::bail!("Failed to decode transaction"), + }; + + let mut inner_instructions_vec: Vec = Vec::new(); + if let Some(meta) = &transaction.transaction.meta { + if let solana_transaction_status::option_serializer::OptionSerializer::Some(ui_inner_insts) = + &meta.inner_instructions + { + for ui_inner in ui_inner_insts { + let mut converted = Vec::new(); + for ui_instruction in &ui_inner.instructions { + if let UiInstruction::Compiled(ui_compiled) = ui_instruction { + if let Ok(data) = solana_sdk::bs58::decode(&ui_compiled.data).into_vec() { + converted.push(InnerInstruction { + instruction: CompiledInstruction { + program_id_index: ui_compiled.program_id_index, + accounts: ui_compiled.accounts.to_vec(), + data, + }, + stack_height: ui_compiled.stack_height, + }); + } + } + } + inner_instructions_vec.push(InnerInstructions { + index: ui_inner.index, + instructions: converted, + }); + } + } + } + + let mut address_table_lookups: Vec = vec![]; + if let Some(meta) = &transaction.transaction.meta { + if let solana_transaction_status::option_serializer::OptionSerializer::Some(loaded) = + &meta.loaded_addresses + { + address_table_lookups.extend( + loaded + .writable + .iter() + .chain(loaded.readonly.iter()) + .filter_map(|s| s.parse::().ok()), + ); + } + } + + let mut accounts: Vec = versioned_tx.message.static_account_keys().to_vec(); + accounts.extend(address_table_lookups); + + let slot = transaction.slot; + let block_time = transaction + .block_time + .map(|t| Timestamp { seconds: t as i64, nanos: 0 }); + let recv_us = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_micros() as i64; + + let protocols = vec![ + Protocol::PumpFun, + Protocol::PumpSwap, + Protocol::Bonk, + Protocol::RaydiumClmm, + Protocol::RaydiumCpmm, + Protocol::RaydiumAmmV4, + Protocol::MeteoraDammV2, + ]; + + let callback = Arc::new(|event: DexEvent| { + println!("Event: {:?}\n", event); + }); + + EventParser::parse_instruction_events_from_versioned_transaction( + &protocols, + None, + &versioned_tx, + signature, + Some(slot), + block_time, + recv_us, + &accounts, + &inner_instructions_vec, + None, + None, + callback, + ) + .await?; + + Ok(()) +} diff --git a/examples/parse_pumpswap_tx.rs b/examples/parse_pumpswap_tx.rs new file mode 100644 index 0000000..9e39766 --- /dev/null +++ b/examples/parse_pumpswap_tx.rs @@ -0,0 +1,147 @@ +//! Parse a PumpSwap transaction from RPC using solana-streamer EventParser. +//! +//! Usage: +//! cargo run --example parse_pumpswap_tx --release +//! TX_SIGNATURE= SOLANA_RPC_URL= cargo run --example parse_pumpswap_tx --release + +use anyhow::Result; +use solana_commitment_config::CommitmentConfig; +use solana_streamer_sdk::streaming::event_parser::core::event_parser::EventParser; +use solana_streamer_sdk::streaming::event_parser::{DexEvent, Protocol}; +use std::str::FromStr; +use std::sync::Arc; + +#[tokio::main] +async fn main() -> Result<()> { + let default_sig = "3zsihbygW7hoKGtduAyDDFzp4E1eis8gaBzEzzNKr8ma39baffpFcphok9wHFgR3EauDe9vYYsVf4Puh5pZ6UJiS"; + let tx_sig = std::env::var("TX_SIGNATURE").unwrap_or_else(|_| default_sig.to_string()); + let rpc_url = std::env::var("SOLANA_RPC_URL") + .unwrap_or_else(|_| "https://api.mainnet-beta.solana.com".to_string()); + + println!("=== PumpSwap Transaction Parser (solana-streamer) ===\n"); + println!("Transaction: {}\n", tx_sig); + println!("RPC: {}\n", rpc_url); + + parse_one_tx(&tx_sig, &rpc_url).await?; + println!("\n✓ Done."); + Ok(()) +} + +async fn parse_one_tx(signature_str: &str, rpc_url: &str) -> Result<()> { + use prost_types::Timestamp; + use solana_sdk::{ + message::compiled_instruction::CompiledInstruction, + pubkey::Pubkey, + signature::Signature, + }; + use solana_transaction_status::{InnerInstruction, InnerInstructions, UiInstruction, UiTransactionEncoding}; + + let signature = Signature::from_str(signature_str)?; + let client = solana_client::nonblocking::rpc_client::RpcClient::new(rpc_url.to_string()); + + let transaction = match client + .get_transaction_with_config( + &signature, + solana_client::rpc_config::RpcTransactionConfig { + encoding: Some(UiTransactionEncoding::Base64), + commitment: Some(CommitmentConfig::confirmed()), + max_supported_transaction_version: Some(0), + }, + ) + .await + { + Ok(tx) => tx, + Err(e) => anyhow::bail!("Failed to fetch transaction: {}", e), + }; + + let versioned_tx = match transaction.transaction.transaction.decode() { + Some(tx) => tx, + None => anyhow::bail!("Failed to decode transaction"), + }; + + let mut inner_instructions_vec: Vec = Vec::new(); + if let Some(meta) = &transaction.transaction.meta { + if let solana_transaction_status::option_serializer::OptionSerializer::Some(ui_inner_insts) = + &meta.inner_instructions + { + for ui_inner in ui_inner_insts { + let mut converted = Vec::new(); + for ui_instruction in &ui_inner.instructions { + if let UiInstruction::Compiled(ui_compiled) = ui_instruction { + if let Ok(data) = solana_sdk::bs58::decode(&ui_compiled.data).into_vec() { + converted.push(InnerInstruction { + instruction: CompiledInstruction { + program_id_index: ui_compiled.program_id_index, + accounts: ui_compiled.accounts.to_vec(), + data, + }, + stack_height: ui_compiled.stack_height, + }); + } + } + } + inner_instructions_vec.push(InnerInstructions { + index: ui_inner.index, + instructions: converted, + }); + } + } + } + + let mut address_table_lookups: Vec = vec![]; + if let Some(meta) = &transaction.transaction.meta { + if let solana_transaction_status::option_serializer::OptionSerializer::Some(loaded) = + &meta.loaded_addresses + { + for s in loaded.writable.iter().chain(loaded.readonly.iter()) { + if let Ok(p) = s.parse::() { + address_table_lookups.push(p); + } + } + } + } + + let mut accounts: Vec = versioned_tx.message.static_account_keys().to_vec(); + accounts.extend(address_table_lookups); + + let slot = transaction.slot; + let block_time = transaction + .block_time + .map(|t| Timestamp { seconds: t as i64, nanos: 0 }); + let recv_us = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_micros() as i64; + + let protocols = vec![ + Protocol::PumpFun, + Protocol::PumpSwap, + Protocol::Bonk, + Protocol::RaydiumClmm, + Protocol::RaydiumCpmm, + Protocol::RaydiumAmmV4, + Protocol::MeteoraDammV2, + ]; + + let callback = Arc::new(|event: DexEvent| { + println!("Event: {:?}\n", event); + }); + + EventParser::parse_instruction_events_from_versioned_transaction( + &protocols, + None, + &versioned_tx, + signature, + Some(slot), + block_time, + recv_us, + &accounts, + &inner_instructions_vec, + None, + None, + callback, + ) + .await?; + + Ok(()) +} diff --git a/examples/pumpfun_quick_test.rs b/examples/pumpfun_quick_test.rs new file mode 100644 index 0000000..fdcbfb4 --- /dev/null +++ b/examples/pumpfun_quick_test.rs @@ -0,0 +1,69 @@ +//! Quick test: subscribe to PumpFun events and print the first 10 (or run 60s). +//! +//! Usage: cargo run --example pumpfun_quick_test --release + +use solana_streamer_sdk::streaming::event_parser::common::types::EventType; +use solana_streamer_sdk::streaming::event_parser::DexEvent; +use solana_streamer_sdk::streaming::grpc::ClientConfig; +use solana_streamer_sdk::streaming::yellowstone_grpc::{AccountFilter, TransactionFilter, YellowstoneGrpc}; +use solana_streamer_sdk::streaming::event_parser::Protocol; +use solana_streamer_sdk::streaming::event_parser::protocols::pumpfun::parser::PUMPFUN_PROGRAM_ID; +use std::sync::atomic::{AtomicU64, Ordering}; + +#[tokio::main] +async fn main() -> Result<(), Box> { + let _ = rustls::crypto::ring::default_provider().install_default(); + + println!("🚀 Quick Test - Subscribing to PumpFun events..."); + + let mut config = ClientConfig::default(); + config.enable_metrics = true; + + let grpc = YellowstoneGrpc::new_with_config( + std::env::var("GRPC_ENDPOINT").unwrap_or_else(|_| "https://solana-yellowstone-grpc.publicnode.com:443".to_string()), + std::env::var("GRPC_AUTH_TOKEN").ok(), + config, + )?; + + let transaction_filter = TransactionFilter { + account_include: vec![PUMPFUN_PROGRAM_ID.to_string()], + account_exclude: vec![], + account_required: vec![], + }; + let account_filter = AccountFilter { + account: vec![], + owner: vec![PUMPFUN_PROGRAM_ID.to_string()], + filters: vec![], + }; + + let event_count = std::sync::Arc::new(AtomicU64::new(0)); + let count = event_count.clone(); + let callback = move |event: DexEvent| { + let n = count.fetch_add(1, Ordering::Relaxed); + let event_type = match event.metadata().event_type { + EventType::PumpFunCreateToken | EventType::PumpFunCreateV2Token => "PumpFunCreate", + EventType::PumpFunBuy => "PumpFunBuy", + EventType::PumpFunSell => "PumpFunSell", + EventType::PumpFunMigrate => "PumpFunMigrate", + _ => "Other", + }; + println!("✅ Event #{}: {}", n + 1, event_type); + }; + + grpc.subscribe_events_immediate( + vec![Protocol::PumpFun], + None, + vec![transaction_filter], + vec![account_filter], + None, + None, + callback, + ) + .await?; + + println!("🎧 Listening (up to 60s or Ctrl+C)...\n"); + tokio::time::sleep(tokio::time::Duration::from_secs(60)).await; + grpc.stop().await; + println!("✓ Stopped."); + Ok(()) +} diff --git a/examples/pumpfun_trade_filter.rs b/examples/pumpfun_trade_filter.rs new file mode 100644 index 0000000..8620597 --- /dev/null +++ b/examples/pumpfun_trade_filter.rs @@ -0,0 +1,119 @@ +//! PumpFun trade event filter: subscribe to Buy/Sell/Create and print details with latency. +//! +//! Usage: cargo run --example pumpfun_trade_filter --release + +use solana_streamer_sdk::streaming::event_parser::common::types::EventType; +use solana_streamer_sdk::streaming::event_parser::protocols::pumpfun::events::{PumpFunCreateTokenEvent, PumpFunCreateV2TokenEvent, PumpFunTradeEvent}; +use solana_streamer_sdk::streaming::event_parser::DexEvent; +use solana_streamer_sdk::streaming::grpc::ClientConfig; +use solana_streamer_sdk::streaming::yellowstone_grpc::{AccountFilter, TransactionFilter, YellowstoneGrpc}; +use solana_streamer_sdk::streaming::event_parser::Protocol; +use solana_streamer_sdk::streaming::event_parser::common::filter::EventTypeFilter; +use solana_streamer_sdk::streaming::event_parser::protocols::pumpfun::parser::PUMPFUN_PROGRAM_ID; + +fn now_micros() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_micros() as i64 +} + +#[tokio::main] +async fn main() -> Result<(), Box> { + let _ = rustls::crypto::ring::default_provider().install_default(); + + println!("🚀 PumpFun Trade Event Filter (solana-streamer)\n"); + + let mut config = ClientConfig::default(); + config.enable_metrics = true; + + let grpc = YellowstoneGrpc::new_with_config( + std::env::var("GRPC_ENDPOINT").unwrap_or_else(|_| "https://solana-yellowstone-grpc.publicnode.com:443".to_string()), + std::env::var("GRPC_AUTH_TOKEN").ok(), + config, + )?; + + let transaction_filter = TransactionFilter { + account_include: vec![PUMPFUN_PROGRAM_ID.to_string()], + account_exclude: vec![], + account_required: vec![], + }; + let account_filter = AccountFilter { + account: vec![], + owner: vec![PUMPFUN_PROGRAM_ID.to_string()], + filters: vec![], + }; + let event_filter = Some(EventTypeFilter { + include: vec![ + EventType::PumpFunBuy, + EventType::PumpFunSell, + EventType::PumpFunCreateToken, + EventType::PumpFunCreateV2Token, + ], + }); + + let callback = |event: DexEvent| { + let now_us = now_micros(); + match &event { + DexEvent::PumpFunTradeEvent(e) => print_trade(e, now_us), + DexEvent::PumpFunCreateTokenEvent(e) => print_create_legacy(e, now_us), + DexEvent::PumpFunCreateV2TokenEvent(e) => print_create_v2(e, now_us), + _ => {} + } + }; + + grpc.subscribe_events_immediate( + vec![Protocol::PumpFun], + None, + vec![transaction_filter], + vec![account_filter], + event_filter, + None, + callback, + ) + .await?; + + println!("🛑 Press Ctrl+C to stop...\n"); + tokio::signal::ctrl_c().await?; + grpc.stop().await; + Ok(()) +} + +fn print_trade(e: &PumpFunTradeEvent, now_us: i64) { + let latency_us = now_us - e.metadata.recv_us; + let kind = if e.is_buy { "BUY" } else { "SELL" }; + println!( + "│ {} | sig={:.8}.. | mint={:.8}.. | sol={} tok={} | user={:.8}.. | latency={} μs", + kind, + e.metadata.signature, + e.mint, + e.sol_amount, + e.token_amount, + e.user, + latency_us + ); +} + +fn print_create_legacy(e: &PumpFunCreateTokenEvent, now_us: i64) { + let latency_us = now_us - e.metadata.recv_us; + println!( + "│ CREATE | sig={:.8}.. | name={} symbol={} | mint={:.8}.. | latency={} μs", + e.metadata.signature, + e.name, + e.symbol, + e.mint, + latency_us + ); +} + +fn print_create_v2(e: &PumpFunCreateV2TokenEvent, now_us: i64) { + let latency_us = now_us - e.metadata.recv_us; + println!( + "│ CREATE_V2 | sig={:.8}.. | name={} symbol={} | mint={:.8}.. | latency={} μs", + e.metadata.signature, + e.name, + e.symbol, + e.mint, + latency_us + ); +} diff --git a/examples/pumpfun_with_metrics.rs b/examples/pumpfun_with_metrics.rs new file mode 100644 index 0000000..dcdc580 --- /dev/null +++ b/examples/pumpfun_with_metrics.rs @@ -0,0 +1,55 @@ +//! PumpFun subscription with metrics enabled. +//! +//! Usage: cargo run --example pumpfun_with_metrics --release + +use solana_streamer_sdk::streaming::event_parser::{DexEvent, Protocol}; +use solana_streamer_sdk::streaming::event_parser::protocols::pumpfun::parser::PUMPFUN_PROGRAM_ID; +use solana_streamer_sdk::streaming::grpc::ClientConfig; +use solana_streamer_sdk::streaming::yellowstone_grpc::{AccountFilter, TransactionFilter, YellowstoneGrpc}; + +#[tokio::main] +async fn main() -> Result<(), Box> { + let _ = rustls::crypto::ring::default_provider().install_default(); + + println!("PumpFun with metrics (solana-streamer)\n"); + + let mut config = ClientConfig::default(); + config.enable_metrics = true; + + let grpc = YellowstoneGrpc::new_with_config( + std::env::var("GRPC_ENDPOINT").unwrap_or_else(|_| "https://solana-yellowstone-grpc.publicnode.com:443".to_string()), + std::env::var("GRPC_AUTH_TOKEN").ok(), + config, + )?; + + let transaction_filter = TransactionFilter { + account_include: vec![PUMPFUN_PROGRAM_ID.to_string()], + account_exclude: vec![], + account_required: vec![], + }; + let account_filter = AccountFilter { + account: vec![], + owner: vec![PUMPFUN_PROGRAM_ID.to_string()], + filters: vec![], + }; + + let callback = |event: DexEvent| { + println!("Event: {:?}", event.metadata().event_type); + }; + + grpc.subscribe_events_immediate( + vec![Protocol::PumpFun], + None, + vec![transaction_filter], + vec![account_filter], + None, + None, + callback, + ) + .await?; + + println!("Press Ctrl+C to stop (metrics printed periodically)...\n"); + tokio::signal::ctrl_c().await?; + grpc.stop().await; + Ok(()) +} diff --git a/examples/pumpswap_with_metrics.rs b/examples/pumpswap_with_metrics.rs new file mode 100644 index 0000000..6234cee --- /dev/null +++ b/examples/pumpswap_with_metrics.rs @@ -0,0 +1,55 @@ +//! PumpSwap subscription with metrics enabled. +//! +//! Usage: cargo run --example pumpswap_with_metrics --release + +use solana_streamer_sdk::streaming::event_parser::{DexEvent, Protocol}; +use solana_streamer_sdk::streaming::event_parser::protocols::pumpswap::parser::PUMPSWAP_PROGRAM_ID; +use solana_streamer_sdk::streaming::grpc::ClientConfig; +use solana_streamer_sdk::streaming::yellowstone_grpc::{AccountFilter, TransactionFilter, YellowstoneGrpc}; + +#[tokio::main] +async fn main() -> Result<(), Box> { + let _ = rustls::crypto::ring::default_provider().install_default(); + + println!("PumpSwap with metrics (solana-streamer)\n"); + + let mut config = ClientConfig::default(); + config.enable_metrics = true; + + let grpc = YellowstoneGrpc::new_with_config( + std::env::var("GRPC_ENDPOINT").unwrap_or_else(|_| "https://solana-yellowstone-grpc.publicnode.com:443".to_string()), + std::env::var("GRPC_AUTH_TOKEN").ok(), + config, + )?; + + let transaction_filter = TransactionFilter { + account_include: vec![PUMPSWAP_PROGRAM_ID.to_string()], + account_exclude: vec![], + account_required: vec![], + }; + let account_filter = AccountFilter { + account: vec![], + owner: vec![PUMPSWAP_PROGRAM_ID.to_string()], + filters: vec![], + }; + + let callback = |event: DexEvent| { + println!("Event: {:?}", event.metadata().event_type); + }; + + grpc.subscribe_events_immediate( + vec![Protocol::PumpSwap], + None, + vec![transaction_filter], + vec![account_filter], + None, + None, + callback, + ) + .await?; + + println!("Press Ctrl+C to stop...\n"); + tokio::signal::ctrl_c().await?; + grpc.stop().await; + Ok(()) +} diff --git a/examples/token_decimals_listen_example.rs b/examples/token_decimals_listen_example.rs index a192d7a..dbd276f 100644 --- a/examples/token_decimals_listen_example.rs +++ b/examples/token_decimals_listen_example.rs @@ -40,7 +40,10 @@ async fn test_grpc() -> Result<(), Box> { let transaction_filter = TransactionFilter { account_include, account_exclude, account_required }; - let account_to_listen = "EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v".to_string(); + let account_to_listen = std::env::var("MINT_ACCOUNT").unwrap_or_else(|_| { + eprintln!("Usage: MINT_ACCOUNT= cargo run --example token_decimals_listen_example --release"); + std::process::exit(1); + }); // Listen to account data belonging to owner programs -> account event monitoring let account_filter = @@ -79,7 +82,7 @@ async fn test_grpc() -> Result<(), Box> { fn create_event_callback() -> impl Fn(DexEvent) { |event: DexEvent| match event { DexEvent::TokenInfoEvent(e) => { - println!("TokenInfoEvent: {:?}", e.decimals); + println!("TokenInfo pubkey={} decimals={} supply={}", e.pubkey, e.decimals, e.supply); } _ => {} }