diff --git a/examples/dynamic_subscription.rs b/examples/dynamic_subscription.rs index 9fe2b6a..01c720a 100644 --- a/examples/dynamic_subscription.rs +++ b/examples/dynamic_subscription.rs @@ -50,15 +50,12 @@ async fn main() -> Result<()> { }; let account_filter = AccountFilter { account: vec![], owner: vec![], filters: vec![] }; - let trade_event_filter = EventTypeFilter { - include: vec![ - EventType::PumpFunBuy, - EventType::PumpFunSell, - EventType::RaydiumCpmmSwapBaseInput, - EventType::RaydiumCpmmSwapBaseOutput, - ], - ..Default::default() - }; + let trade_event_filter = EventTypeFilter::include_only(vec![ + EventType::PumpFunBuy, + EventType::PumpFunSell, + EventType::RaydiumCpmmSwapBaseInput, + EventType::RaydiumCpmmSwapBaseOutput, + ]); if let Err(e) = client .subscribe_events_immediate( diff --git a/examples/grpc_example.rs b/examples/grpc_example.rs index 3dab927..db57d43 100644 --- a/examples/grpc_example.rs +++ b/examples/grpc_example.rs @@ -1,11 +1,17 @@ use solana_streamer_sdk::streaming::{ event_parser::{ + common::{filter::EventTypeFilter, EventType}, protocols::{ - bonk::parser::BONK_PROGRAM_ID, meteora_damm_v2::parser::METEORA_DAMM_V2_PROGRAM_ID, - pumpfun::parser::PUMPFUN_PROGRAM_ID, pumpswap::parser::PUMPSWAP_PROGRAM_ID, + bonk::parser::BONK_PROGRAM_ID, + meteora_damm_v2::parser::METEORA_DAMM_V2_PROGRAM_ID, + pumpfun::parser::PUMPFUN_PROGRAM_ID, + pumpswap::parser::PUMPSWAP_PROGRAM_ID, raydium_amm_v4::parser::RAYDIUM_AMM_V4_PROGRAM_ID, raydium_clmm::parser::RAYDIUM_CLMM_PROGRAM_ID, raydium_cpmm::parser::RAYDIUM_CPMM_PROGRAM_ID, + sol_parser_forward::{ + METEORA_DLMM_PROGRAM_ID, METEORA_POOLS_PROGRAM_ID, ORCA_WHIRLPOOL_PROGRAM_ID, + }, }, DexEvent, Protocol, }, @@ -24,10 +30,10 @@ async fn main() -> Result<(), Box> { async fn test_grpc() -> Result<(), Box> { println!("Subscribing to Yellowstone gRPC events..."); - // Create low-latency configuration + // Create low-latency configuration. let mut config: ClientConfig = ClientConfig::default(); - // Enable performance monitoring, has performance overhead, disabled by default - config.enable_metrics = true; + // Metrics add overhead; enable explicitly with STREAMER_ENABLE_METRICS=1. + config.enable_metrics = std::env::var("STREAMER_ENABLE_METRICS").as_deref() == Ok("1"); let grpc = YellowstoneGrpc::new_with_config( "https://solana-yellowstone-grpc.publicnode.com:443".to_string(), None, @@ -47,6 +53,9 @@ async fn test_grpc() -> Result<(), Box> { Protocol::RaydiumClmm, Protocol::RaydiumAmmV4, Protocol::MeteoraDammV2, + Protocol::OrcaWhirlpool, + Protocol::MeteoraPools, + Protocol::MeteoraDlmm, ]; println!("Protocols to monitor: {:?}", protocols); @@ -60,6 +69,9 @@ async fn test_grpc() -> Result<(), Box> { RAYDIUM_CLMM_PROGRAM_ID.to_string(), // Listen to raydium_clmm program ID RAYDIUM_AMM_V4_PROGRAM_ID.to_string(), // Listen to raydium_amm_v4 program ID METEORA_DAMM_V2_PROGRAM_ID.to_string(), // Listen to meteora_damm_v2 program ID + ORCA_WHIRLPOOL_PROGRAM_ID.to_string(), // Listen to orca_whirlpool program ID + METEORA_POOLS_PROGRAM_ID.to_string(), // Listen to meteora_pools program ID + METEORA_DLMM_PROGRAM_ID.to_string(), // Listen to meteora_dlmm program ID ]; let account_exclude = vec![]; let account_required = vec![]; @@ -75,11 +87,17 @@ async fn test_grpc() -> Result<(), Box> { let account_filter = AccountFilter { account: vec![], owner: account_include.clone(), filters: vec![] }; - // Event filtering - // No event filtering, includes all events - let event_type_filter = None; - // Only include PumpSwapBuy events and PumpSwapSell events - // let event_type_filter = Some(EventTypeFilter { include: vec![EventType::PumpFunTrade] }); + // Event filtering. Set STREAMER_TRADES_ONLY=1 to keep only selected trade events. + let event_type_filter = if std::env::var("STREAMER_TRADES_ONLY").as_deref() == Ok("1") { + Some(EventTypeFilter::include_only(vec![ + EventType::PumpFunBuy, + EventType::PumpFunSell, + EventType::PumpSwapBuy, + EventType::PumpSwapSell, + ])) + } else { + None + }; println!("Starting to listen for events, press Ctrl+C to stop..."); println!("Monitoring programs: {:?}", account_include); @@ -97,7 +115,7 @@ async fn test_grpc() -> Result<(), Box> { ) .await?; - // 支持 stop 方法，测试代码 - 异步1000秒之后停止 + // Demo safety stop: stop automatically after 1000 seconds. let grpc_clone = grpc.clone(); tokio::spawn(async move { tokio::time::sleep(std::time::Duration::from_secs(1000)).await; diff --git a/examples/mint_all_ata_account_listen_example.rs b/examples/mint_all_ata_account_listen_example.rs index cec7850..9c7ac74 100644 --- a/examples/mint_all_ata_account_listen_example.rs +++ b/examples/mint_all_ata_account_listen_example.rs @@ -25,10 +25,10 @@ async fn main() -> Result<(), Box> { async fn test_grpc() -> Result<(), Box> { println!("Subscribing to Yellowstone gRPC events..."); - // Create low-latency configuration + // Create low-latency configuration. let mut config: ClientConfig = ClientConfig::default(); - // Enable performance monitoring, has performance overhead, disabled by default - config.enable_metrics = true; + // Metrics add overhead; enable explicitly with STREAMER_ENABLE_METRICS=1. + config.enable_metrics = std::env::var("STREAMER_ENABLE_METRICS").as_deref() == Ok("1"); let grpc = YellowstoneGrpc::new_with_config( "https://solana-yellowstone-grpc.publicnode.com:443".to_string(), None, @@ -73,8 +73,7 @@ async fn test_grpc() -> Result<(), Box> { }; // Event filtering - let event_type_filter = - Some(EventTypeFilter { include: vec![EventType::TokenAccount], ..Default::default() }); + let event_type_filter = Some(EventTypeFilter::include_only(vec![EventType::TokenAccount])); println!("Starting to listen for events, press Ctrl+C to stop..."); println!("Starting subscription..."); @@ -90,7 +89,7 @@ async fn test_grpc() -> Result<(), Box> { ) .await?; - // 支持 stop 方法，测试代码 - 异步1000秒之后停止 + // Demo safety stop: stop automatically after 1000 seconds. let grpc_clone = grpc.clone(); tokio::spawn(async move { tokio::time::sleep(std::time::Duration::from_secs(1000)).await; diff --git a/examples/nonce_listen_example.rs b/examples/nonce_listen_example.rs index 96a9010..23b56d6 100644 --- a/examples/nonce_listen_example.rs +++ b/examples/nonce_listen_example.rs @@ -17,10 +17,10 @@ async fn main() -> Result<(), Box> { async fn test_grpc() -> Result<(), Box> { println!("Subscribing to Yellowstone gRPC events..."); - // Create low-latency configuration + // Create low-latency configuration. let mut config: ClientConfig = ClientConfig::default(); - // Enable performance monitoring, has performance overhead, disabled by default - config.enable_metrics = true; + // Metrics add overhead; enable explicitly with STREAMER_ENABLE_METRICS=1. + config.enable_metrics = std::env::var("STREAMER_ENABLE_METRICS").as_deref() == Ok("1"); let grpc = YellowstoneGrpc::new_with_config( "https://solana-yellowstone-grpc.publicnode.com:443".to_string(), None, @@ -51,8 +51,7 @@ async fn test_grpc() -> Result<(), Box> { AccountFilter { account: vec![nonce_account.clone()], owner: vec![], filters: vec![] }; // Event filtering - let event_type_filter = - Some(EventTypeFilter { include: vec![EventType::NonceAccount], ..Default::default() }); + let event_type_filter = Some(EventTypeFilter::include_only(vec![EventType::NonceAccount])); println!("Starting to listen for events, press Ctrl+C to stop..."); println!("Starting subscription..."); @@ -68,7 +67,7 @@ async fn test_grpc() -> Result<(), Box> { ) .await?; - // 支持 stop 方法，测试代码 - 异步1000秒之后停止 + // Demo safety stop: stop automatically after 1000 seconds. let grpc_clone = grpc.clone(); tokio::spawn(async move { tokio::time::sleep(std::time::Duration::from_secs(1000)).await; diff --git a/examples/parse_meteora_damm_tx.rs b/examples/parse_meteora_damm_tx.rs index 6af8f2f..efea063 100644 --- a/examples/parse_meteora_damm_tx.rs +++ b/examples/parse_meteora_damm_tx.rs @@ -1,15 +1,18 @@ -//! Parse a Meteora DAMM v2 transaction from RPC using solana-streamer EventParser. +//! Parse a Meteora DAMM v2 transaction from RPC using the SDK-backed streamer RPC helper. //! //! Usage: -//! cargo run --example parse_meteora_damm_tx --release +//! TX_SIGNATURE= 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 solana_sdk::signature::Signature; +use solana_streamer_sdk::fetch_rpc_transaction_as_streamer_events_async; +use solana_streamer_sdk::streaming::event_parser::Protocol; use std::str::FromStr; -use std::sync::Arc; + +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<()> { @@ -20,125 +23,28 @@ async fn main() -> Result<()> { 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); + println!("=== Meteora DAMM v2 RPC Parser (SDK-backed solana-streamer) ===\n"); + println!("Transaction: {}\nRPC: {}\n", tx_sig, rpc_url); - parse_one_tx(&tx_sig, &rpc_url).await?; - println!("\n✓ Done."); - Ok(()) -} + let signature = Signature::from_str(&tx_sig)?; + let rpc_client = solana_client::nonblocking::rpc_client::RpcClient::new(rpc_url); + let protocols = [Protocol::MeteoraDammV2]; -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( + // No event filter: surface every Meteora DAMM v2 event the SDK can parse for this transaction. + let events = fetch_rpc_transaction_as_streamer_events_async( + &rpc_client, + &signature, + now_micros(), &protocols, None, - &versioned_tx, - signature, - Some(slot), - block_time, - recv_us, - &accounts, - &inner_instructions_vec, - None, - None, - callback, ) - .await?; + .await + .map_err(|e| anyhow::anyhow!(e.to_string()))?; + + println!("Parsed {} Meteora DAMM v2 event(s):\n", events.len()); + for event in events { + println!("{:?}\n", event); + } Ok(()) } diff --git a/examples/parse_pump_tx.rs b/examples/parse_pump_tx.rs index 4ec20a2..7bf2921 100644 --- a/examples/parse_pump_tx.rs +++ b/examples/parse_pump_tx.rs @@ -1,4 +1,4 @@ -//! Parse a PumpFun transaction from RPC using solana-streamer EventParser. +//! Parse a PumpFun transaction from RPC using the SDK-backed streamer RPC helper. //! //! Signature: env `TX_SIGNATURE` or first CLI arg. RPC: env `SOLANA_RPC_URL`. //! @@ -8,11 +8,14 @@ //! 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 solana_sdk::signature::Signature; +use solana_streamer_sdk::fetch_rpc_transaction_as_streamer_events_async; +use solana_streamer_sdk::streaming::event_parser::Protocol; use std::str::FromStr; -use std::sync::Arc; + +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<()> { @@ -25,138 +28,28 @@ async fn main() -> Result<()> { 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); + println!("=== PumpFun RPC Parser (SDK-backed solana-streamer) ===\n"); + println!("Transaction: {}\nRPC: {}\n", tx_sig, rpc_url); - parse_one_tx(&tx_sig, &rpc_url).await?; - println!("\n✓ Done."); - Ok(()) -} + let signature = Signature::from_str(&tx_sig)?; + let rpc_client = solana_client::nonblocking::rpc_client::RpcClient::new(rpc_url); + let protocols = [Protocol::PumpFun]; -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( + // No event filter: surface every PumpFun event the SDK can parse for this transaction. + let events = fetch_rpc_transaction_as_streamer_events_async( + &rpc_client, + &signature, + now_micros(), &protocols, None, - &versioned_tx, - signature, - Some(slot), - block_time, - recv_us, - &accounts, - &inner_instructions_vec, - None, - None, - callback, ) - .await?; + .await + .map_err(|e| anyhow::anyhow!(e.to_string()))?; + + println!("Parsed {} PumpFun event(s):\n", events.len()); + for event in events { + println!("{:?}\n", event); + } Ok(()) } diff --git a/examples/parse_pumpswap_tx.rs b/examples/parse_pumpswap_tx.rs index f44326b..915b997 100644 --- a/examples/parse_pumpswap_tx.rs +++ b/examples/parse_pumpswap_tx.rs @@ -1,15 +1,18 @@ -//! Parse a PumpSwap transaction from RPC using solana-streamer EventParser. +//! Parse a PumpSwap transaction from RPC using the SDK-backed streamer RPC helper. //! //! 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 solana_sdk::signature::Signature; +use solana_streamer_sdk::fetch_rpc_transaction_as_streamer_events_async; +use solana_streamer_sdk::streaming::event_parser::Protocol; use std::str::FromStr; -use std::sync::Arc; + +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<()> { @@ -19,126 +22,28 @@ async fn main() -> Result<()> { 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); + println!("=== PumpSwap RPC Parser (SDK-backed solana-streamer) ===\n"); + println!("Transaction: {}\nRPC: {}\n", tx_sig, rpc_url); - parse_one_tx(&tx_sig, &rpc_url).await?; - println!("\n✓ Done."); - Ok(()) -} + let signature = Signature::from_str(&tx_sig)?; + let rpc_client = solana_client::nonblocking::rpc_client::RpcClient::new(rpc_url); + let protocols = [Protocol::PumpSwap]; -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( + // No event filter: surface every PumpSwap event the SDK can parse for this transaction. + let events = fetch_rpc_transaction_as_streamer_events_async( + &rpc_client, + &signature, + now_micros(), &protocols, None, - &versioned_tx, - signature, - Some(slot), - block_time, - recv_us, - &accounts, - &inner_instructions_vec, - None, - None, - callback, ) - .await?; + .await + .map_err(|e| anyhow::anyhow!(e.to_string()))?; + + println!("Parsed {} PumpSwap event(s):\n", events.len()); + for event in events { + println!("{:?}\n", event); + } Ok(()) } diff --git a/examples/parse_tx_events.rs b/examples/parse_tx_events.rs index fe22e83..926a869 100644 --- a/examples/parse_tx_events.rs +++ b/examples/parse_tx_events.rs @@ -1,10 +1,11 @@ -//! RPC 单笔解析示例： -//! - **默认**：本地 ix 路径（便于打印完整日志等）。 -//! - **对齐 sol-parser-sdk**：`SOL_PARSER_SDK_RPC=1` 时对同一笔 RPC 响应使用 `parse_encoded_rpc_transaction_as_streamer_events`。 +//! Single RPC transaction parse example: +//! - Default: SDK-backed parsing through `parse_encoded_rpc_transaction_as_streamer_events`. +//! - Debug compare: set `STREAMER_LOCAL_IX=1` to run the local instruction path. //! -//! 亦可直接使用 crate 根的 `fetch_rpc_transaction_as_streamer_events_async`（单独 RPC 拉取 + 对齐解析）。 +//! You can also call `fetch_rpc_transaction_as_streamer_events_async` from the crate root +//! when you want the helper to fetch and parse by signature. //! -//! 环境变量：`SOLANA_RPC_URL`（可选，默认 mainnet 公共 RPC）。 +//! Environment: `SOLANA_RPC_URL` (optional, defaults to the public mainnet RPC). use anyhow::Result; use solana_commitment_config::CommitmentConfig; @@ -20,9 +21,9 @@ fn rpc_url_from_env() -> String { .unwrap_or_else(|_| "https://api.mainnet-beta.solana.com".to_string()) } -fn use_sol_parser_sdk_rpc_path() -> bool { +fn use_local_ix_path() -> bool { matches!( - std::env::var("SOL_PARSER_SDK_RPC").as_deref(), + std::env::var("STREAMER_LOCAL_IX").as_deref(), Ok("1") | Ok("true") | Ok("yes") | Ok("TRUE") | Ok("YES") ) } @@ -55,7 +56,7 @@ async fn main() -> Result<()> { Ok(()) } -/// 本地版本化消息 + inner ix 路径（与订阅管线中的「无 sdk」解析相近）。 +/// Local versioned-message + inner-instruction path, kept for parser debugging. async fn parse_local_ix_path( transaction: &solana_transaction_status::EncodedConfirmedTransactionWithStatusMeta, signature: solana_sdk::signature::Signature, @@ -66,7 +67,7 @@ async fn parse_local_ix_path( use solana_sdk::{message::compiled_instruction::CompiledInstruction, pubkey::Pubkey}; use solana_transaction_status::{InnerInstruction, InnerInstructions, UiInstruction}; - println!("\n--- Parsed events (本地 ix 路径) ---\n"); + println!("\n--- Parsed events (STREAMER_LOCAL_IX=1, local instruction path) ---\n"); let versioned_tx = match transaction.transaction.transaction.decode() { Some(tx) => tx, @@ -237,8 +238,10 @@ async fn get_single_transaction_details(signature_str: &str) -> Result<()> { Protocol::MeteoraDlmm, ]; - if use_sol_parser_sdk_rpc_path() { - println!("\n--- Parsed events (SOL_PARSER_SDK_RPC=1, sol-parser-sdk 对齐) ---\n"); + if use_local_ix_path() { + parse_local_ix_path(&transaction, signature, recv_us, &protocols).await?; + } else { + println!("\n--- Parsed events (default SDK-backed RPC path) ---\n"); match parse_encoded_rpc_transaction_as_streamer_events( &transaction, recv_us, @@ -251,10 +254,8 @@ async fn get_single_transaction_details(signature_str: &str) -> Result<()> { println!("{:?}\n", ev); } } - Err(e) => println!("SDK-aligned parse error: {}", e), + Err(e) => println!("SDK-backed parse error: {}", e), } - } else { - parse_local_ix_path(&transaction, signature, recv_us, &protocols).await?; } } Err(e) => { @@ -262,8 +263,5 @@ async fn get_single_transaction_details(signature_str: &str) -> Result<()> { } } - println!("Press Ctrl+C to exit example..."); - tokio::signal::ctrl_c().await?; - Ok(()) } diff --git a/examples/pumpfun_quick_test.rs b/examples/pumpfun_quick_test.rs index 7759289..59cde14 100644 --- a/examples/pumpfun_quick_test.rs +++ b/examples/pumpfun_quick_test.rs @@ -19,7 +19,8 @@ async fn main() -> Result<(), Box> { println!("🚀 Quick Test - Subscribing to PumpFun events..."); let mut config = ClientConfig::default(); - config.enable_metrics = true; + // Metrics add overhead; enable explicitly with STREAMER_ENABLE_METRICS=1. + config.enable_metrics = std::env::var("STREAMER_ENABLE_METRICS").as_deref() == Ok("1"); let grpc = YellowstoneGrpc::new_with_config( std::env::var("GRPC_ENDPOINT") diff --git a/examples/pumpfun_trade_filter.rs b/examples/pumpfun_trade_filter.rs index 5091cd7..42e7d98 100644 --- a/examples/pumpfun_trade_filter.rs +++ b/examples/pumpfun_trade_filter.rs @@ -26,7 +26,8 @@ async fn main() -> Result<(), Box> { println!("🚀 PumpFun Trade Event Filter (solana-streamer)\n"); let mut config = ClientConfig::default(); - config.enable_metrics = true; + // Metrics add overhead; enable explicitly with STREAMER_ENABLE_METRICS=1. + config.enable_metrics = std::env::var("STREAMER_ENABLE_METRICS").as_deref() == Ok("1"); let grpc = YellowstoneGrpc::new_with_config( std::env::var("GRPC_ENDPOINT") @@ -45,15 +46,12 @@ async fn main() -> Result<(), Box> { owner: vec![PUMPFUN_PROGRAM_ID.to_string()], filters: vec![], }; - let event_filter = Some(EventTypeFilter { - include: vec![ - EventType::PumpFunBuy, - EventType::PumpFunSell, - EventType::PumpFunCreateToken, - EventType::PumpFunCreateV2Token, - ], - ..Default::default() - }); + let event_filter = Some(EventTypeFilter::include_only(vec![ + EventType::PumpFunBuy, + EventType::PumpFunSell, + EventType::PumpFunCreateToken, + EventType::PumpFunCreateV2Token, + ])); let callback = |event: DexEvent| { let now_us = now_micros(); diff --git a/examples/pumpswap_pool_account_listen_example.rs b/examples/pumpswap_pool_account_listen_example.rs index 5c3d4e8..2a3a0f3 100644 --- a/examples/pumpswap_pool_account_listen_example.rs +++ b/examples/pumpswap_pool_account_listen_example.rs @@ -25,10 +25,10 @@ async fn main() -> Result<(), Box> { async fn test_grpc() -> Result<(), Box> { println!("Subscribing to Yellowstone gRPC events..."); - // Create low-latency configuration + // Create low-latency configuration. let mut config: ClientConfig = ClientConfig::default(); - // Enable performance monitoring, has performance overhead, disabled by default - config.enable_metrics = true; + // Metrics add overhead; enable explicitly with STREAMER_ENABLE_METRICS=1. + config.enable_metrics = std::env::var("STREAMER_ENABLE_METRICS").as_deref() == Ok("1"); let grpc = YellowstoneGrpc::new_with_config( "https://solana-yellowstone-grpc.publicnode.com:443".to_string(), None, @@ -75,8 +75,7 @@ async fn test_grpc() -> Result<(), Box> { }; // Event filtering - let event_type_filter = - Some(EventTypeFilter { include: vec![EventType::TokenAccount], ..Default::default() }); + let event_type_filter = Some(EventTypeFilter::include_only(vec![EventType::TokenAccount])); println!("Starting to listen for events, press Ctrl+C to stop..."); println!("Starting subscription..."); @@ -92,7 +91,7 @@ async fn test_grpc() -> Result<(), Box> { ) .await?; - // 支持 stop 方法，测试代码 - 异步1000秒之后停止 + // Demo safety stop: stop automatically after 1000 seconds. let grpc_clone = grpc.clone(); tokio::spawn(async move { tokio::time::sleep(std::time::Duration::from_secs(1000)).await; diff --git a/examples/sdk_bridge.rs b/examples/sdk_bridge.rs new file mode 100644 index 0000000..9e89790 --- /dev/null +++ b/examples/sdk_bridge.rs @@ -0,0 +1,24 @@ +//! Minimal advanced interop example for the public sol-parser-sdk bridge. +//! +//! Most bots should use the normal streamer subscription callbacks. Use this bridge when you +//! already have raw `sol-parser-sdk::DexEvent` values and want to adapt them into streamer +//! `DexEvent` values, or when you need direct access to raw SDK APIs. + +use solana_streamer_sdk::sdk_bridge; +use solana_streamer_sdk::streaming::event_parser::common::{filter::EventTypeFilter, EventType}; + +fn main() { + let streamer_filter = EventTypeFilter::include_only(vec![ + EventType::PumpFunBuy, + EventType::PumpFunSell, + EventType::TokenAccount, + ]); + let sdk_filter = sdk_bridge::event_type_filter_to_sdk(Some(&streamer_filter)); + + println!("SDK filter built: {}", sdk_filter.is_some()); + println!("Raw SDK event type: {}", std::any::type_name::()); + println!( + "Raw SDK crate re-export: {}", + std::any::type_name::() + ); +} diff --git a/examples/shred_example.rs b/examples/shred_example.rs index bf88644..a1d4961 100644 --- a/examples/shred_example.rs +++ b/examples/shred_example.rs @@ -1,4 +1,5 @@ use solana_streamer_sdk::streaming::{ + event_parser::common::{filter::EventTypeFilter, EventType}, event_parser::{DexEvent, Protocol}, shred::StreamClientConfig, ShredStreamGrpc, @@ -14,34 +15,29 @@ async fn main() -> Result<(), Box> { async fn test_shreds() -> Result<(), Box> { println!("Subscribing to ShredStream events..."); - // Create low-latency configuration + // Create low-latency configuration. let mut config = StreamClientConfig::default(); - // Enable performance monitoring, has performance overhead, disabled by default - config.enable_metrics = true; + // Metrics add overhead; enable explicitly with STREAMER_ENABLE_METRICS=1. + config.enable_metrics = std::env::var("STREAMER_ENABLE_METRICS").as_deref() == Ok("1"); let shred_stream = ShredStreamGrpc::new_with_config("http://127.0.0.1:10800".to_string(), config).await?; let callback = create_event_callback(); - let protocols = vec![ - Protocol::PumpFun, - Protocol::PumpSwap, - Protocol::Bonk, - Protocol::RaydiumCpmm, - Protocol::RaydiumClmm, - Protocol::RaydiumAmmV4, - ]; + let protocols = vec![Protocol::PumpFun]; - // Event filtering - // No event filtering, includes all events - let event_type_filter = None; - // Only include PumpSwapBuy events and PumpSwapSell events - // let event_type_filter = - // EventTypeFilter { include: vec![EventType::PumpSwapBuy, EventType::PumpSwapSell] }; + // ShredStream uses the sol-parser-sdk ShredStream hot path. Keep filters narrow for + // latency-sensitive bots; use None to receive every event the SDK ShredStream path emits. + let event_type_filter = Some(EventTypeFilter::include_only(vec![ + EventType::PumpFunBuy, + EventType::PumpFunSell, + EventType::PumpFunCreateToken, + EventType::PumpFunCreateV2Token, + ])); println!("Listening for events, press Ctrl+C to stop..."); shred_stream.shredstream_subscribe(protocols, None, event_type_filter, callback).await?; - // 支持 stop 方法，测试代码 - 异步1000秒之后停止 + // Demo safety stop: stop automatically after 1000 seconds. let shred_clone = shred_stream.clone(); tokio::spawn(async move { tokio::time::sleep(std::time::Duration::from_secs(1000)).await; diff --git a/examples/token_balance_listen_example.rs b/examples/token_balance_listen_example.rs index 2fa071f..764d3f1 100644 --- a/examples/token_balance_listen_example.rs +++ b/examples/token_balance_listen_example.rs @@ -17,10 +17,10 @@ async fn main() -> Result<(), Box> { async fn test_grpc() -> Result<(), Box> { println!("Subscribing to Yellowstone gRPC events..."); - // Create low-latency configuration + // Create low-latency configuration. let mut config: ClientConfig = ClientConfig::default(); - // Enable performance monitoring, has performance overhead, disabled by default - config.enable_metrics = true; + // Metrics add overhead; enable explicitly with STREAMER_ENABLE_METRICS=1. + config.enable_metrics = std::env::var("STREAMER_ENABLE_METRICS").as_deref() == Ok("1"); let grpc = YellowstoneGrpc::new_with_config( "https://solana-yellowstone-grpc.publicnode.com:443".to_string(), None, @@ -47,8 +47,7 @@ async fn test_grpc() -> Result<(), Box> { AccountFilter { account: vec![account_to_listen], owner: vec![], filters: vec![] }; // Event filtering - let event_type_filter = - Some(EventTypeFilter { include: vec![EventType::TokenAccount], ..Default::default() }); + let event_type_filter = Some(EventTypeFilter::include_only(vec![EventType::TokenAccount])); println!("Starting to listen for events, press Ctrl+C to stop..."); println!("Starting subscription..."); @@ -64,7 +63,7 @@ async fn test_grpc() -> Result<(), Box> { ) .await?; - // 支持 stop 方法，测试代码 - 异步1000秒之后停止 + // Demo safety stop: stop automatically after 1000 seconds. let grpc_clone = grpc.clone(); tokio::spawn(async move { tokio::time::sleep(std::time::Duration::from_secs(1000)).await; diff --git a/examples/token_decimals_listen_example.rs b/examples/token_decimals_listen_example.rs index 10ae281..f306013 100644 --- a/examples/token_decimals_listen_example.rs +++ b/examples/token_decimals_listen_example.rs @@ -17,10 +17,10 @@ async fn main() -> Result<(), Box> { async fn test_grpc() -> Result<(), Box> { println!("Subscribing to Yellowstone gRPC events..."); - // Create low-latency configuration + // Create low-latency configuration. let mut config: ClientConfig = ClientConfig::default(); - // Enable performance monitoring, has performance overhead, disabled by default - config.enable_metrics = true; + // Metrics add overhead; enable explicitly with STREAMER_ENABLE_METRICS=1. + config.enable_metrics = std::env::var("STREAMER_ENABLE_METRICS").as_deref() == Ok("1"); let grpc = YellowstoneGrpc::new_with_config( "https://solana-yellowstone-grpc.publicnode.com:443".to_string(), None, @@ -50,8 +50,7 @@ async fn test_grpc() -> Result<(), Box> { AccountFilter { account: vec![account_to_listen], owner: vec![], filters: vec![] }; // Event filtering - let event_type_filter = - Some(EventTypeFilter { include: vec![EventType::TokenAccount], ..Default::default() }); + let event_type_filter = Some(EventTypeFilter::include_only(vec![EventType::TokenInfo])); println!("Starting to listen for events, press Ctrl+C to stop..."); println!("Starting subscription..."); @@ -67,7 +66,7 @@ async fn test_grpc() -> Result<(), Box> { ) .await?; - // 支持 stop 方法，测试代码 - 异步1000秒之后停止 + // Demo safety stop: stop automatically after 1000 seconds. let grpc_clone = grpc.clone(); tokio::spawn(async move { tokio::time::sleep(std::time::Duration::from_secs(1000)).await;