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

examples: update for SDK-backed streamer API

Browse Source
This commit is contained in:
0xfnzero
2026-05-15 07:46:29 +08:00
parent ba86cfcd1b
commit 741778a82f
15 changed files with 199 additions and 468 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/dynamic_subscription.rs
+6 -9
View File
@@ -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(
examples/grpc_example.rs
+29 -11
View File
@@ -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<dyn std::error::Error>> {
async fn test_grpc() -> Result<(), Box<dyn std::error::Error>> {
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<dyn std::error::Error>> {
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<dyn std::error::Error>> {
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<dyn std::error::Error>> {
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<dyn std::error::Error>> {
)
.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;
examples/mint_all_ata_account_listen_example.rs
+5 -6
View File
@@ -25,10 +25,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
async fn test_grpc() -> Result<(), Box<dyn std::error::Error>> {
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<dyn std::error::Error>> {
};
// 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<dyn std::error::Error>> {
)
.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;
examples/nonce_listen_example.rs
+5 -6
View File
@@ -17,10 +17,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
async fn test_grpc() -> Result<(), Box<dyn std::error::Error>> {
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<dyn std::error::Error>> {
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<dyn std::error::Error>> {
)
.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;
examples/parse_meteora_damm_tx.rs
+26 -120
View File
@@ -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=<sig> cargo run --example parse_meteora_damm_tx --release
//! TX_SIGNATURE=<sig> SOLANA_RPC_URL=<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<InnerInstructions> = 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<Pubkey> = 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::<Pubkey>() {
address_table_lookups.push(p);
}
}
}
}
let mut accounts: Vec<Pubkey> = 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(())
}
examples/parse_pump_tx.rs
+25 -132
View File
@@ -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<InnerInstructions> = 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<Pubkey> = 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::<Pubkey>().ok()),
);
}
}
let mut accounts: Vec<Pubkey> = 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(())
}
examples/parse_pumpswap_tx.rs
+25 -120
View File
@@ -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=<sig> SOLANA_RPC_URL=<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<InnerInstructions> = 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<Pubkey> = 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::<Pubkey>() {
address_table_lookups.push(p);
}
}
}
}
let mut accounts: Vec<Pubkey> = 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(())
}
examples/parse_tx_events.rs
+15 -17
View File
@@ -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(())
}
examples/pumpfun_quick_test.rs
+2 -1
View File
@@ -19,7 +19,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
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")
examples/pumpfun_trade_filter.rs
+8 -10
View File
@@ -26,7 +26,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
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<dyn std::error::Error>> {
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();
examples/pumpswap_pool_account_listen_example.rs
+5 -6
View File
@@ -25,10 +25,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
async fn test_grpc() -> Result<(), Box<dyn std::error::Error>> {
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<dyn std::error::Error>> {
};
// 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<dyn std::error::Error>> {
)
.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;
examples/sdk_bridge.rs
+24
View File
@@ -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::<sdk_bridge::SdkDexEvent>());
println!(
"Raw SDK crate re-export: {}",
std::any::type_name::<solana_streamer_sdk::parser_sdk::DexEvent>()
);
}
examples/shred_example.rs
+14 -18
View File
@@ -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<dyn std::error::Error>> {
async fn test_shreds() -> Result<(), Box<dyn std::error::Error>> {
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;
examples/token_balance_listen_example.rs
+5 -6
View File
@@ -17,10 +17,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
async fn test_grpc() -> Result<(), Box<dyn std::error::Error>> {
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<dyn std::error::Error>> {
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<dyn std::error::Error>> {
)
.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;
examples/token_decimals_listen_example.rs
+5 -6
View File
@@ -17,10 +17,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
async fn test_grpc() -> Result<(), Box<dyn std::error::Error>> {
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<dyn std::error::Error>> {
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<dyn std::error::Error>> {
)
.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;