diff --git a/Cargo.toml b/Cargo.toml index 0e61ef4..d893295 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "solana-streamer-sdk" -version = "0.4.1" +version = "0.4.2" edition = "2021" authors = ["William ", "sgxiang ", "wei <1415121722@qq.com>"] repository = "https://github.com/0xfnzero/solana-streamer" diff --git a/README.md b/README.md index 97d2871..65473f9 100755 --- a/README.md +++ b/README.md @@ -46,14 +46,14 @@ Add the dependency to your `Cargo.toml`: ```toml # Add to your Cargo.toml -solana-streamer-sdk = { path = "./solana-streamer", version = "0.4.1" } +solana-streamer-sdk = { path = "./solana-streamer", version = "0.4.2" } ``` ### Use crates.io ```toml # Add to your Cargo.toml -solana-streamer-sdk = "0.4.1" +solana-streamer-sdk = "0.4.2" ``` ## Configuration System @@ -125,6 +125,8 @@ let config = StreamClientConfig { | ShredStream Stream | `shred_example.rs` | Monitor transaction events using ShredStream | `cargo run --example shred_example` | [examples/shred_example.rs](examples/shred_example.rs) | | Parse Transaction Events | `parse_tx_events` | Parse Solana mainnet transaction data | `cargo run --example parse_tx_events` | [examples/parse_tx_events.rs](examples/parse_tx_events.rs) | | Dynamic Subscription Management | `dynamic_subscription` | Update filters at runtime | `cargo run --example dynamic_subscription` | [examples/dynamic_subscription.rs](examples/dynamic_subscription.rs) | +| Token Balance Monitoring | `token_balance_listen_example` | 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) | +| Nonce Account Monitoring | `nonce_listen_example` | Track nonce account state changes | `cargo run --example nonce_listen_example` | [examples/nonce_listen_example.rs](examples/nonce_listen_example.rs) | ### Event Filtering diff --git a/README_CN.md b/README_CN.md index e287f24..1c56cf8 100644 --- a/README_CN.md +++ b/README_CN.md @@ -46,14 +46,14 @@ git clone https://github.com/0xfnzero/solana-streamer ```toml # 添加到您的 Cargo.toml -solana-streamer-sdk = { path = "./solana-streamer", version = "0.4.1" } +solana-streamer-sdk = { path = "./solana-streamer", version = "0.4.2" } ``` ### 使用 crates.io ```toml # 添加到您的 Cargo.toml -solana-streamer-sdk = "0.4.1" +solana-streamer-sdk = "0.4.2" ``` ## 配置系统 @@ -125,6 +125,8 @@ let config = StreamClientConfig { | ShredStream 流 | `shred_example.rs` | 使用 ShredStream 监控交易事件 | `cargo run --example shred_example` | [examples/shred_example.rs](examples/shred_example.rs) | | 解析交易事件 | `parse_tx_events` | 解析 Solana 主网交易数据 | `cargo run --example parse_tx_events` | [examples/parse_tx_events.rs](examples/parse_tx_events.rs) | | 动态订阅管理 | `dynamic_subscription` | 运行时更新过滤器 | `cargo run --example dynamic_subscription` | [examples/dynamic_subscription.rs](examples/dynamic_subscription.rs) | +| 代币余额监控 | `token_balance_listen_example` | 监控特定代币账户余额变化 | `cargo run --example token_balance_listen_example` | [examples/token_balance_listen_example.rs](examples/token_balance_listen_example.rs) | +| Nonce 账户监控 | `nonce_listen_example` | 跟踪 nonce 账户状态变化 | `cargo run --example nonce_listen_example` | [examples/nonce_listen_example.rs](examples/nonce_listen_example.rs) | ### 事件过滤 diff --git a/examples/nonce_listen_example.rs b/examples/nonce_listen_example.rs new file mode 100644 index 0000000..baa540b --- /dev/null +++ b/examples/nonce_listen_example.rs @@ -0,0 +1,81 @@ +use solana_streamer_sdk::streaming::{ + event_parser::{ + common::{filter::EventTypeFilter, EventType}, + UnifiedEvent, + }, + grpc::ClientConfig, + yellowstone_grpc::{AccountFilter, TransactionFilter}, + YellowstoneGrpc, +}; + +#[tokio::main] +async fn main() -> Result<(), Box> { + println!("Starting Yellowstone gRPC Streamer..."); + test_grpc().await?; + Ok(()) +} + +async fn test_grpc() -> Result<(), Box> { + println!("Subscribing to Yellowstone gRPC events..."); + // Create low-latency configuration + let mut config: ClientConfig = ClientConfig::low_latency(); + // Enable performance monitoring, has performance overhead, disabled by default + config.enable_metrics = true; + let grpc = YellowstoneGrpc::new_with_config( + "https://solana-yellowstone-grpc.publicnode.com:443".to_string(), + None, + config, + )?; + println!("GRPC client created successfully"); + let callback = create_event_callback(); + // Will try to parse corresponding protocol events from transactions + let protocols = vec![]; + println!("Protocols to monitor: {:?}", protocols); + // Filter accounts + let account_include = vec![]; + let account_exclude = vec![]; + let account_required = vec![]; + + // Listen to transaction data + let transaction_filter = + TransactionFilter { account_include, account_exclude, account_required }; + + let nonce_account = "use_your_nonce_account_here".to_string(); + // Listen to account data belonging to owner programs -> account event monitoring + let account_filter = AccountFilter { account: vec![nonce_account], owner: vec![] }; + + // Event filtering + let event_type_filter = Some(EventTypeFilter { include: vec![EventType::AccountNonce] }); + + println!("Starting to listen for events, press Ctrl+C to stop..."); + println!("Starting subscription..."); + + grpc.subscribe_events_immediate( + protocols, + None, + transaction_filter, + account_filter, + event_type_filter, + None, + callback, + ) + .await?; + + // 支持 stop 方法，测试代码 - 异步1000秒之后停止 + let grpc_clone = grpc.clone(); + tokio::spawn(async move { + tokio::time::sleep(std::time::Duration::from_secs(1000)).await; + grpc_clone.stop().await; + }); + + println!("Waiting for Ctrl+C to stop..."); + tokio::signal::ctrl_c().await?; + + Ok(()) +} + +fn create_event_callback() -> impl Fn(Box) { + |event: Box| { + println!("🎉 Event received! {:?}", event); + } +} diff --git a/examples/token_balance_listen_example.rs b/examples/token_balance_listen_example.rs new file mode 100644 index 0000000..bcca67c --- /dev/null +++ b/examples/token_balance_listen_example.rs @@ -0,0 +1,82 @@ +use solana_streamer_sdk::streaming::{ + event_parser::{ + common::{filter::EventTypeFilter, EventType}, + UnifiedEvent, + }, + grpc::ClientConfig, + yellowstone_grpc::{AccountFilter, TransactionFilter}, + YellowstoneGrpc, +}; + +#[tokio::main] +async fn main() -> Result<(), Box> { + println!("Starting Yellowstone gRPC Streamer..."); + test_grpc().await?; + Ok(()) +} + +async fn test_grpc() -> Result<(), Box> { + println!("Subscribing to Yellowstone gRPC events..."); + // Create low-latency configuration + let mut config: ClientConfig = ClientConfig::low_latency(); + // Enable performance monitoring, has performance overhead, disabled by default + config.enable_metrics = true; + let grpc = YellowstoneGrpc::new_with_config( + "https://solana-yellowstone-grpc.publicnode.com:443".to_string(), + None, + config, + )?; + println!("GRPC client created successfully"); + let callback = create_event_callback(); + // Will try to parse corresponding protocol events from transactions + let protocols = vec![]; + println!("Protocols to monitor: {:?}", protocols); + // Filter accounts + let account_include = vec![]; + let account_exclude = vec![]; + let account_required = vec![]; + + // Listen to transaction data + let transaction_filter = + TransactionFilter { account_include, account_exclude, account_required }; + + let account_to_listen = "use_your_token_account_here".to_string(); + + // Listen to account data belonging to owner programs -> account event monitoring + let account_filter = AccountFilter { account: vec![account_to_listen], owner: vec![] }; + + // Event filtering + let event_type_filter = Some(EventTypeFilter { include: vec![EventType::AccountCommon] }); + + println!("Starting to listen for events, press Ctrl+C to stop..."); + println!("Starting subscription..."); + + grpc.subscribe_events_immediate( + protocols.clone(), + None, + transaction_filter.clone(), + account_filter.clone(), + event_type_filter.clone(), + None, + callback, + ) + .await?; + + // 支持 stop 方法，测试代码 - 异步1000秒之后停止 + let grpc_clone = grpc.clone(); + tokio::spawn(async move { + tokio::time::sleep(std::time::Duration::from_secs(1000)).await; + grpc_clone.stop().await; + }); + + println!("Waiting for Ctrl+C to stop..."); + tokio::signal::ctrl_c().await?; + + Ok(()) +} + +fn create_event_callback() -> impl Fn(Box) { + |event: Box| { + println!("🎉 Event received! {:?}", event); + } +} diff --git a/src/streaming/event_parser/common/types.rs b/src/streaming/event_parser/common/types.rs index b935557..0c04cf4 100755 --- a/src/streaming/event_parser/common/types.rs +++ b/src/streaming/event_parser/common/types.rs @@ -141,6 +141,7 @@ pub enum EventType { AccountRaydiumCpmmAmmConfig, AccountRaydiumCpmmPoolState, + AccountNonce, AccountCommon, // Common events @@ -163,6 +164,8 @@ pub const ACCOUNT_EVENT_TYPES: &[EventType] = &[ EventType::AccountRaydiumClmmTickArrayState, EventType::AccountRaydiumCpmmAmmConfig, EventType::AccountRaydiumCpmmPoolState, + EventType::AccountCommon, + EventType::AccountNonce, ]; pub const BLOCK_EVENT_TYPES: &[EventType] = &[EventType::BlockMeta]; @@ -228,6 +231,7 @@ impl fmt::Display for EventType { EventType::AccountRaydiumCpmmAmmConfig => write!(f, "AccountRaydiumCpmmAmmConfig"), EventType::AccountRaydiumCpmmPoolState => write!(f, "AccountRaydiumCpmmPoolState"), EventType::AccountCommon => write!(f, "AccountCommon"), + EventType::AccountNonce => write!(f, "AccountNonce"), EventType::BlockMeta => write!(f, "BlockMeta"), EventType::Unknown => write!(f, "Unknown"), } diff --git a/src/streaming/event_parser/core/account_event_parser.rs b/src/streaming/event_parser/core/account_event_parser.rs index 4ac5c9a..cf10061 100644 --- a/src/streaming/event_parser/core/account_event_parser.rs +++ b/src/streaming/event_parser/core/account_event_parser.rs @@ -2,6 +2,7 @@ use std::collections::HashMap; use std::sync::OnceLock; use serde::{Deserialize, Serialize}; +use solana_account_decoder::parse_nonce::parse_nonce; use solana_sdk::program_pack::Pack; use solana_sdk::pubkey::Pubkey; use spl_token::state::Account; @@ -43,6 +44,19 @@ pub struct CommonAccountEvent { } impl_unified_event!(CommonAccountEvent,); +/// Nonce account event +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct NonceAccountEvent { + pub metadata: EventMetadata, + pub pubkey: Pubkey, + pub executable: bool, + pub lamports: u64, + pub owner: Pubkey, + pub rent_epoch: u64, + pub nonce: String, +} +impl_unified_event!(NonceAccountEvent,); + /// 账户事件解析器 pub type AccountEventParserFn = fn(account: &AccountPretty, metadata: EventMetadata) -> Option>; @@ -52,6 +66,8 @@ static PROTOCOL_CONFIGS_CACHE: OnceLock = OnceLock::new(); +// Nonce account config +static NONCE_CONFIG: OnceLock = OnceLock::new(); pub struct AccountEventParser {} @@ -192,6 +208,19 @@ impl AccountEventParser { } } + if event_type_filter.is_none() + || event_type_filter.unwrap().include.contains(&EventType::AccountNonce) + { + let nonce_config = NONCE_CONFIG.get_or_init(|| AccountEventParseConfig { + program_id: Pubkey::default(), + protocol_type: ProtocolType::Common, + event_type: EventType::AccountNonce, + account_discriminator: &[1, 0, 0, 0, 1, 0, 0, 0], + account_parser: Self::parse_nonce_account_event, + }); + configs.push(nonce_config.clone()); + } + let common_config = COMMON_CONFIG.get_or_init(|| AccountEventParseConfig { program_id: Pubkey::default(), protocol_type: ProtocolType::Common, @@ -256,4 +285,29 @@ impl AccountEventParser { event.set_handle_us(elapsed_micros_since(account.recv_us)); return Some(Box::new(event)); } + + pub fn parse_nonce_account_event( + account: &AccountPretty, + metadata: EventMetadata, + ) -> Option> { + if let Ok(info) = parse_nonce(&account.data) { + match info { + solana_account_decoder::parse_nonce::UiNonceState::Initialized(details) => { + let mut event = NonceAccountEvent { + metadata, + pubkey: account.pubkey, + executable: account.executable, + lamports: account.lamports, + owner: account.owner, + rent_epoch: account.rent_epoch, + nonce: details.blockhash, + }; + event.set_handle_us(elapsed_micros_since(account.recv_us)); + return Some(Box::new(event)); + } + solana_account_decoder::parse_nonce::UiNonceState::Uninitialized => {} + } + } + None + } }