diff --git a/Cargo.toml b/Cargo.toml index 93478f8..560ff26 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -15,7 +15,6 @@ crate-type = ["cdylib", "rlib"] [dependencies] solana-sdk = "3.0.0" solana-client = "3.1.9" -solana-program = "3.0.0" solana-transaction-status = "3.1.9" solana-account-decoder = "3.1.9" solana-entry = { version = "3.1.9", features = ["agave-unstable-api"] } @@ -27,25 +26,16 @@ bincode = "1.3" anyhow = "1.0.102" yellowstone-grpc-client = { version = "10.2.0" } yellowstone-grpc-proto = { version = "10.1.1" } -tokio = { version = "1.49.0", features = ["full", "rt-multi-thread"]} +tokio = { version = "1.50.0", features = ["full", "rt-multi-thread"]} tonic = { version = "0.14.5", features = ["transport"] } -rustls = { version = "0.23.36", features = ["ring"], default-features = false } +rustls = { version = "0.23.37", features = ["ring"], default-features = false } log = "0.4.29" -chrono = "0.4.43" -lazy_static = "1.5.0" -once_cell = "1.21.3" dashmap = "6.1.0" prost = "0.14.3" prost-types = "0.14.3" -maplit = "1.0.2" -env_logger = "0.11.9" crossbeam-queue = "0.3.12" -parking_lot = "0.12.5" wide = "1.1.1" spl-token = { version = "9.0.0", default-features = false, features = ["no-entrypoint"] } spl-token-2022 = { version = "10.0.0", default-features = false, features = ["no-entrypoint"] } solana-commitment-config = { version = "3.1.1", features = ["serde"] } tonic-prost = "0.14.5" - -[dev-dependencies] -criterion = { version = "0.8.2", features = ["html_reports"] } diff --git a/examples/dynamic_subscription.rs b/examples/dynamic_subscription.rs index 38edeb2..b20754f 100644 --- a/examples/dynamic_subscription.rs +++ b/examples/dynamic_subscription.rs @@ -21,8 +21,6 @@ const MONITORING_DURATION_SECS: u64 = 10; /// Demonstrates dynamic subscription updates and filter changes in real-time #[tokio::main] async fn main() -> Result<()> { - env_logger::init(); - println!("Connecting to Yellowstone gRPC at {}", GRPC_ENDPOINT); let client = Arc::new(YellowstoneGrpc::new(GRPC_ENDPOINT.to_string(), API_KEY.map(|s| s.to_string()))?); diff --git a/src/streaming/common/event_processor.rs b/src/streaming/common/event_processor.rs index 894a209..133f02b 100644 --- a/src/streaming/common/event_processor.rs +++ b/src/streaming/common/event_processor.rs @@ -91,7 +91,12 @@ pub async fn process_grpc_transaction( let block_time_ms = block_meta_pretty .block_time .map(|ts| ts.seconds * 1000 + ts.nanos as i64 / 1_000_000) - .unwrap_or_else(|| chrono::Utc::now().timestamp_millis()); + .unwrap_or_else(|| { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_millis() as i64 + }); let block_meta_event = CommonEventParser::generate_block_meta_event( block_meta_pretty.slot, diff --git a/src/streaming/event_parser/common/high_performance_clock.rs b/src/streaming/event_parser/common/high_performance_clock.rs index 3faf27c..6be9e4b 100644 --- a/src/streaming/event_parser/common/high_performance_clock.rs +++ b/src/streaming/event_parser/common/high_performance_clock.rs @@ -1,5 +1,5 @@ use std::fmt::Debug; -use std::time::Instant; +use std::time::{Instant, SystemTime, UNIX_EPOCH}; /// 高性能时钟管理器,减少系统调用开销并最小化延迟 #[derive(Debug)] @@ -25,12 +25,12 @@ impl HighPerformanceClock { // 通过多次采样来减少初始化误差 let mut best_offset = i64::MAX; let mut best_instant = Instant::now(); - let mut best_timestamp = chrono::Utc::now().timestamp_micros(); + let mut best_timestamp = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_micros() as i64; // 进行3次采样,选择延迟最小的 for _ in 0..3 { let instant_before = Instant::now(); - let timestamp = chrono::Utc::now().timestamp_micros(); + let timestamp = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_micros() as i64; let instant_after = Instant::now(); let sample_latency = instant_after.duration_since(instant_before).as_nanos() as i64; @@ -69,7 +69,7 @@ impl HighPerformanceClock { /// 重新校准时钟,减少累积漂移 fn recalibrate(&mut self) { let current_monotonic = Instant::now(); - let current_utc = chrono::Utc::now().timestamp_micros(); + let current_utc = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_micros() as i64; // 计算预期的UTC时间戳(基于单调时钟) let expected_utc = self.base_timestamp_us @@ -113,8 +113,8 @@ impl Default for HighPerformanceClock { } /// 全局高性能时钟实例 -static HIGH_PERF_CLOCK: once_cell::sync::OnceCell = - once_cell::sync::OnceCell::new(); +static HIGH_PERF_CLOCK: std::sync::OnceLock = + std::sync::OnceLock::new(); /// 获取全局高性能时钟实例(最简单的实现) #[inline(always)] diff --git a/src/streaming/event_parser/common/types.rs b/src/streaming/event_parser/common/types.rs index f33d78f..31ba9f7 100755 --- a/src/streaming/event_parser/common/types.rs +++ b/src/streaming/event_parser/common/types.rs @@ -36,9 +36,8 @@ impl EventMetadataPool { } // Global object pool instances -lazy_static::lazy_static! { - pub static ref EVENT_METADATA_POOL: EventMetadataPool = EventMetadataPool::new(); -} +pub static EVENT_METADATA_POOL: std::sync::LazyLock = + std::sync::LazyLock::new(EventMetadataPool::new); #[derive( Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, BorshSerialize, BorshDeserialize, @@ -356,14 +355,13 @@ impl EventMetadata { } } -lazy_static::lazy_static! { - static ref SOL_MINT: Pubkey = Pubkey::from_str("So11111111111111111111111111111111111111111").unwrap(); - static ref SYSTEM_PROGRAMS: [Pubkey; 3] = [ - Pubkey::from_str("TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA").unwrap(), - Pubkey::from_str("TokenzQdBNbLqP5VEhdkAS6EPFLC1PHnBqCXEpPxuEb").unwrap(), - Pubkey::from_str("11111111111111111111111111111111").unwrap(), - ]; -} +static SOL_MINT: std::sync::LazyLock = + std::sync::LazyLock::new(|| Pubkey::from_str("So11111111111111111111111111111111111111111").unwrap()); +static SYSTEM_PROGRAMS: std::sync::LazyLock<[Pubkey; 3]> = std::sync::LazyLock::new(|| [ + Pubkey::from_str("TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA").unwrap(), + Pubkey::from_str("TokenzQdBNbLqP5VEhdkAS6EPFLC1PHnBqCXEpPxuEb").unwrap(), + Pubkey::from_str("11111111111111111111111111111111").unwrap(), +]); /// Parse token transfer data from next instructions pub fn parse_swap_data_from_next_instructions( diff --git a/src/streaming/event_parser/core/global_state.rs b/src/streaming/event_parser/core/global_state.rs index de6dc45..201417c 100644 --- a/src/streaming/event_parser/core/global_state.rs +++ b/src/streaming/event_parser/core/global_state.rs @@ -179,8 +179,8 @@ impl Default for GlobalState { } /// Global state instance -static GLOBAL_STATE: once_cell::sync::Lazy = - once_cell::sync::Lazy::new(GlobalState::new); +static GLOBAL_STATE: std::sync::LazyLock = + std::sync::LazyLock::new(GlobalState::new); /// Get global state instance pub fn get_global_state() -> &'static GlobalState { diff --git a/src/streaming/event_parser/core/parser_cache.rs b/src/streaming/event_parser/core/parser_cache.rs index c1f7908..17d76c6 100644 --- a/src/streaming/event_parser/core/parser_cache.rs +++ b/src/streaming/event_parser/core/parser_cache.rs @@ -54,8 +54,8 @@ impl CacheKey { /// 全局程序ID缓存(使用读写锁保护) static GLOBAL_PROGRAM_IDS_CACHE: LazyLock< - parking_lot::RwLock>>>, -> = LazyLock::new(|| parking_lot::RwLock::new(HashMap::new())); + std::sync::RwLock>>>, +> = LazyLock::new(|| std::sync::RwLock::new(HashMap::new())); /// 获取指定协议的程序ID列表 /// @@ -68,7 +68,7 @@ pub fn get_global_program_ids( // 快速路径:尝试读取缓存 { - let cache = GLOBAL_PROGRAM_IDS_CACHE.read(); + let cache = GLOBAL_PROGRAM_IDS_CACHE.read().unwrap(); if let Some(program_ids) = cache.get(&cache_key) { return program_ids.clone(); } @@ -78,7 +78,7 @@ pub fn get_global_program_ids( let program_ids = Arc::new(EventDispatcher::get_program_ids(protocols)); // 缓存结果(写锁) - GLOBAL_PROGRAM_IDS_CACHE.write().insert(cache_key, program_ids.clone()); + GLOBAL_PROGRAM_IDS_CACHE.write().unwrap().insert(cache_key, program_ids.clone()); program_ids } diff --git a/src/streaming/grpc/pool.rs b/src/streaming/grpc/pool.rs index de671de..d55f5a7 100644 --- a/src/streaming/grpc/pool.rs +++ b/src/streaming/grpc/pool.rs @@ -406,9 +406,8 @@ impl EventPrettyPool { } // 全局池管理器实例 -lazy_static::lazy_static! { - pub static ref GLOBAL_POOL_MANAGER: PoolManager = PoolManager::new(); -} +pub static GLOBAL_POOL_MANAGER: std::sync::LazyLock = + std::sync::LazyLock::new(PoolManager::new); /// 便捷的全局工厂函数 pub mod factory { diff --git a/src/streaming/grpc/subscription.rs b/src/streaming/grpc/subscription.rs index 114f5cb..7144885 100644 --- a/src/streaming/grpc/subscription.rs +++ b/src/streaming/grpc/subscription.rs @@ -1,5 +1,4 @@ use futures::{channel::mpsc, sink::Sink, Stream}; -use maplit::hashmap; use std::{collections::HashMap, time::Duration}; use tonic::{transport::channel::ClientTlsConfig, Status}; use yellowstone_grpc_client::{GeyserGrpcClient, Interceptor}; @@ -55,11 +54,11 @@ impl SubscriptionManager { )> { let blocks_meta = if event_type_filter.is_some() && event_type_filter.unwrap().include_block_event() { - hashmap! { "".to_owned() => SubscribeRequestFilterBlocksMeta {} } + HashMap::from([("".to_owned(), SubscribeRequestFilterBlocksMeta {})]) } else if event_type_filter.is_none() { - hashmap! { "".to_owned() => SubscribeRequestFilterBlocksMeta {} } + HashMap::from([("".to_owned(), SubscribeRequestFilterBlocksMeta {})]) } else { - hashmap! {} + HashMap::new() }; let subscribe_request = SubscribeRequest { accounts: accounts.unwrap_or_default(), diff --git a/src/streaming/shred/pool.rs b/src/streaming/shred/pool.rs index 47805fb..b1c127d 100644 --- a/src/streaming/shred/pool.rs +++ b/src/streaming/shred/pool.rs @@ -137,9 +137,8 @@ impl Default for ShredPoolManager { } // 全局 Shred 池管理器实例 -lazy_static::lazy_static! { - pub static ref GLOBAL_SHRED_POOL_MANAGER: ShredPoolManager = ShredPoolManager::new(); -} +pub static GLOBAL_SHRED_POOL_MANAGER: std::sync::LazyLock = + std::sync::LazyLock::new(ShredPoolManager::new); /// 便捷的全局工厂函数 pub mod factory { diff --git a/src/streaming/yellowstone_grpc.rs b/src/streaming/yellowstone_grpc.rs index 2a6907f..97762d3 100644 --- a/src/streaming/yellowstone_grpc.rs +++ b/src/streaming/yellowstone_grpc.rs @@ -8,7 +8,7 @@ use crate::streaming::event_parser::{Protocol, DexEvent}; use crate::streaming::grpc::pool::factory; use crate::streaming::grpc::{EventPretty, SubscriptionManager}; use anyhow::anyhow; -use chrono::Local; +use std::time::{SystemTime, UNIX_EPOCH}; use futures::channel::mpsc; use futures::{SinkExt, StreamExt}; use log::error; @@ -247,10 +247,12 @@ impl YellowstoneGrpc { }) .await; } - log::debug!("service is ping: {}", Local::now()); + let ts = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs(); + log::debug!("service is ping: {}", ts); } Some(UpdateOneof::Pong(_)) => { - log::debug!("service is pong: {}", Local::now()); + let ts = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs(); + log::debug!("service is pong: {}", ts); } _ => { log::debug!("Received other message type"); diff --git a/src/streaming/yellowstone_sub_system.rs b/src/streaming/yellowstone_sub_system.rs index 74a3696..4ca9877 100755 --- a/src/streaming/yellowstone_sub_system.rs +++ b/src/streaming/yellowstone_sub_system.rs @@ -7,7 +7,7 @@ use crate::{ }; use futures::{SinkExt, StreamExt}; use log::error; -use solana_program::pubkey; +use solana_sdk::pubkey; use solana_sdk::pubkey::Pubkey; use yellowstone_grpc_proto::geyser::SubscribeUpdateTransactionInfo; use yellowstone_grpc_proto::geyser::{