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

refactor: simplify event API method names for better ergonomics

Browse Source 
- Rename verbose method names to shorter, clearer alternatives
- program_received_time_us() → recv_us()
- program_handle_time_consuming_us() → handle_us()
- instruction_outer_index() → outer_index()
- instruction_inner_index() → inner_index()
- Update all implementations across event parsers and processors

Improves developer experience while maintaining semantic clarity.
This commit is contained in:
ysq
2025-09-03 15:49:50 +08:00
parent 7601eb370c
commit 1060b04b12
16 changed files with 122 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[package] [package]
name = "solana-streamer-sdk" name = "solana-streamer-sdk"
version = "0.4.0" version = "0.4.1"
edition = "2021" edition = "2021"
authors = ["William <byteblock6@gmail.com>", "sgxiang <sgxiang@gmail.com>", "wei <1415121722@qq.com>"] authors = ["William <byteblock6@gmail.com>", "sgxiang <sgxiang@gmail.com>", "wei <1415121722@qq.com>"]
repository = "https://github.com/0xfnzero/solana-streamer" repository = "https://github.com/0xfnzero/solana-streamer"
README.md
+2 -2
View File
@@ -46,14 +46,14 @@ Add the dependency to your `Cargo.toml`:
```toml ```toml
# Add to your Cargo.toml # Add to your Cargo.toml
solana-streamer-sdk = { path = "./solana-streamer", version = "0.4.0" } solana-streamer-sdk = { path = "./solana-streamer", version = "0.4.1" }
``` ```
### Use crates.io ### Use crates.io
```toml ```toml
# Add to your Cargo.toml # Add to your Cargo.toml
solana-streamer-sdk = "0.4.0" solana-streamer-sdk = "0.4.1"
``` ```
## Configuration System ## Configuration System
README_CN.md
+2 -2
View File
@@ -45,14 +45,14 @@ git clone https://github.com/0xfnzero/solana-streamer
```toml ```toml
# 添加到您的 Cargo.toml # 添加到您的 Cargo.toml
solana-streamer-sdk = { path = "./solana-streamer", version = "0.4.0" } solana-streamer-sdk = { path = "./solana-streamer", version = "0.4.1" }
``` ```
### 使用 crates.io ### 使用 crates.io
```toml ```toml
# 添加到您的 Cargo.toml # 添加到您的 Cargo.toml
solana-streamer-sdk = "0.4.0" solana-streamer-sdk = "0.4.1"
``` ```
## 配置系统 ## 配置系统
src/main.rs
+1 -1
View File
@@ -196,7 +196,7 @@ fn create_event_callback() -> impl Fn(Box<dyn UnifiedEvent>) {
match_event!(event, { match_event!(event, {
// -------------------------- block meta ----------------------- // -------------------------- block meta -----------------------
BlockMetaEvent => |e: BlockMetaEvent| { BlockMetaEvent => |e: BlockMetaEvent| {
println!("BlockMetaEvent: {:?}", e.metadata.program_handle_time_consuming_us); println!("BlockMetaEvent: {:?}", e.metadata.handle_us);
}, },
// -------------------------- bonk ----------------------- // -------------------------- bonk -----------------------
BonkPoolCreateEvent => |e: BonkPoolCreateEvent| { BonkPoolCreateEvent => |e: BonkPoolCreateEvent| {
src/streaming/common/event_processor.rs
+8 -8
View File
@@ -94,7 +94,7 @@ impl EventProcessor {
let metrics_manager = self.metrics_manager.clone(); let metrics_manager = self.metrics_manager.clone();
Arc::new(move |event: Box<dyn UnifiedEvent>| { Arc::new(move |event: Box<dyn UnifiedEvent>| {
let processing_time_us = event.program_handle_time_consuming_us() as f64; let processing_time_us = event.handle_us() as f64;
callback(event); callback(event);
metrics_manager.update_metrics(MetricsEventType::Transaction, 1, processing_time_us); metrics_manager.update_metrics(MetricsEventType::Transaction, 1, processing_time_us);
}) })
@@ -171,7 +171,7 @@ impl EventProcessor {
self.event_type_filter.as_ref(), self.event_type_filter.as_ref(),
); );
if let Some(event) = account_event { if let Some(event) = account_event {
let processing_time_us = event.program_handle_time_consuming_us() as f64; let processing_time_us = event.handle_us() as f64;
self.invoke_callback(event); self.invoke_callback(event);
self.update_metrics(MetricsEventType::Account, 1, processing_time_us); self.update_metrics(MetricsEventType::Account, 1, processing_time_us);
} }
@@ -181,7 +181,7 @@ impl EventProcessor {
let slot = transaction_pretty.slot; let slot = transaction_pretty.slot;
let signature = transaction_pretty.signature; let signature = transaction_pretty.signature;
let block_time = transaction_pretty.block_time; let block_time = transaction_pretty.block_time;
let program_received_time_us = transaction_pretty.program_received_time_us; let recv_us = transaction_pretty.recv_us;
let transaction_index = transaction_pretty.transaction_index; let transaction_index = transaction_pretty.transaction_index;
let grpc_tx = transaction_pretty.grpc_tx; let grpc_tx = transaction_pretty.grpc_tx;
@@ -193,7 +193,7 @@ impl EventProcessor {
signature, signature,
Some(slot), Some(slot),
block_time, block_time,
program_received_time_us, recv_us,
bot_wallet, bot_wallet,
transaction_index, transaction_index,
adapter_callback, adapter_callback,
@@ -210,9 +210,9 @@ impl EventProcessor {
block_meta_pretty.slot, block_meta_pretty.slot,
block_meta_pretty.block_hash, block_meta_pretty.block_hash,
block_time_ms, block_time_ms,
block_meta_pretty.program_received_time_us, block_meta_pretty.recv_us,
); );
let processing_time_us = block_meta_event.program_handle_time_consuming_us() as f64; let processing_time_us = block_meta_event.handle_us() as f64;
self.invoke_callback(block_meta_event); self.invoke_callback(block_meta_event);
self.update_metrics(MetricsEventType::BlockMeta, 1, processing_time_us); self.update_metrics(MetricsEventType::BlockMeta, 1, processing_time_us);
} }
@@ -302,7 +302,7 @@ impl EventProcessor {
let slot = transaction_with_slot.slot; let slot = transaction_with_slot.slot;
let signature = tx.signatures[0]; let signature = tx.signatures[0];
let program_received_time_us = transaction_with_slot.program_received_time_us; let recv_us = transaction_with_slot.recv_us;
let parser = self.get_parser(); let parser = self.get_parser();
let adapter_callback = self.create_adapter_callback(); let adapter_callback = self.create_adapter_callback();
@@ -312,7 +312,7 @@ impl EventProcessor {
signature, signature,
Some(slot), Some(slot),
None, None,
program_received_time_us, recv_us,
bot_wallet, bot_wallet,
None, None,
&[], &[],
src/streaming/event_parser/common/mod.rs
+10 -10
View File
@@ -20,16 +20,16 @@ macro_rules! impl_unified_event {
self.metadata.slot self.metadata.slot
} }
fn program_received_time_us(&self) -> i64 { fn recv_us(&self) -> i64 {
self.metadata.program_received_time_us self.metadata.recv_us
} }
fn program_handle_time_consuming_us(&self) -> i64 { fn handle_us(&self) -> i64 {
self.metadata.program_handle_time_consuming_us self.metadata.handle_us
} }
fn set_program_handle_time_consuming_us(&mut self, program_handle_time_consuming_us: i64) { fn set_handle_us(&mut self, handle_us: i64) {
self.metadata.program_handle_time_consuming_us = program_handle_time_consuming_us; self.metadata.handle_us = handle_us;
} }
fn as_any(&self) -> &dyn std::any::Any { fn as_any(&self) -> &dyn std::any::Any {
@@ -60,12 +60,12 @@ macro_rules! impl_unified_event {
self.metadata.swap_data.is_some() self.metadata.swap_data.is_some()
} }
fn instruction_outer_index(&self) -> i64 { fn outer_index(&self) -> i64 {
self.metadata.instruction_outer_index self.metadata.outer_index
} }
fn instruction_inner_index(&self) -> Option<i64> { fn inner_index(&self) -> Option<i64> {
self.metadata.instruction_inner_index self.metadata.inner_index
} }
fn transaction_index(&self) -> Option<u64> { fn transaction_index(&self) -> Option<u64> {
self.metadata.transaction_index self.metadata.transaction_index
src/streaming/event_parser/common/types.rs
+11 -11
View File
@@ -293,14 +293,14 @@ pub struct EventMetadata {
pub transaction_index: Option<u64>, // 新增：交易在slot中的索引 pub transaction_index: Option<u64>, // 新增：交易在slot中的索引
pub block_time: i64, pub block_time: i64,
pub block_time_ms: i64, pub block_time_ms: i64,
pub program_received_time_us: i64, pub recv_us: i64,
pub program_handle_time_consuming_us: i64, pub handle_us: i64,
pub protocol: ProtocolType, pub protocol: ProtocolType,
pub event_type: EventType, pub event_type: EventType,
pub program_id: Pubkey, pub program_id: Pubkey,
pub swap_data: Option<SwapData>, pub swap_data: Option<SwapData>,
pub instruction_outer_index: i64, pub outer_index: i64,
pub instruction_inner_index: Option<i64>, pub inner_index: Option<i64>,
} }
impl EventMetadata { impl EventMetadata {
@@ -313,9 +313,9 @@ impl EventMetadata {
protocol: ProtocolType, protocol: ProtocolType,
event_type: EventType, event_type: EventType,
program_id: Pubkey, program_id: Pubkey,
instruction_outer_index: i64, outer_index: i64,
instruction_inner_index: Option<i64>, inner_index: Option<i64>,
program_received_time_us: i64, recv_us: i64,
transaction_index: Option<u64>, transaction_index: Option<u64>,
) -> Self { ) -> Self {
Self { Self {
@@ -323,14 +323,14 @@ impl EventMetadata {
slot, slot,
block_time, block_time,
block_time_ms, block_time_ms,
program_received_time_us, recv_us,
program_handle_time_consuming_us: 0, handle_us: 0,
protocol, protocol,
event_type, event_type,
program_id, program_id,
swap_data: None, swap_data: None,
instruction_outer_index, outer_index,
instruction_inner_index, inner_index,
transaction_index, transaction_index,
} }
} }
src/streaming/event_parser/core/account_event_parser.rs
+3 -3
View File
@@ -184,13 +184,13 @@ impl AccountEventParser {
protocol: config.protocol_type, protocol: config.protocol_type,
event_type: config.event_type, event_type: config.event_type,
program_id: config.program_id, program_id: config.program_id,
program_received_time_us: account.program_received_time_us, recv_us: account.recv_us,
..Default::default() ..Default::default()
}, },
); );
if let Some(mut event) = event { if let Some(mut event) = event {
event.set_program_handle_time_consuming_us(elapsed_micros_since( event.set_handle_us(elapsed_micros_since(
account.program_received_time_us, account.recv_us,
)); ));
return Some(event); return Some(event);
} }
src/streaming/event_parser/core/common_event_parser.rs
+3 -3
View File
@@ -8,12 +8,12 @@ impl CommonEventParser {
slot: u64, slot: u64,
block_hash: String, block_hash: String,
block_time_ms: i64, block_time_ms: i64,
program_received_time_us: i64, recv_us: i64,
) -> Box<dyn UnifiedEvent> { ) -> Box<dyn UnifiedEvent> {
let mut block_meta_event = let mut block_meta_event =
BlockMetaEvent::new(slot, block_hash, block_time_ms, program_received_time_us); BlockMetaEvent::new(slot, block_hash, block_time_ms, recv_us);
block_meta_event block_meta_event
.set_program_handle_time_consuming_us(elapsed_micros_since(program_received_time_us)); .set_handle_us(elapsed_micros_since(recv_us));
Box::new(block_meta_event) Box::new(block_meta_event)
} }
} }
src/streaming/event_parser/core/macros.rs
+8 -8
View File
@@ -30,7 +30,7 @@ macro_rules! impl_event_parser_delegate {
signature: solana_sdk::signature::Signature, signature: solana_sdk::signature::Signature,
slot: u64, slot: u64,
block_time: Option<prost_types::Timestamp>, block_time: Option<prost_types::Timestamp>,
program_received_time_us: i64, recv_us: i64,
outer_index: i64, outer_index: i64,
inner_index: Option<i64>, inner_index: Option<i64>,
transaction_index: Option<u64>, transaction_index: Option<u64>,
@@ -41,7 +41,7 @@ macro_rules! impl_event_parser_delegate {
signature, signature,
slot, slot,
block_time, block_time,
program_received_time_us, recv_us,
outer_index, outer_index,
inner_index, inner_index,
transaction_index, transaction_index,
@@ -55,13 +55,13 @@ macro_rules! impl_event_parser_delegate {
signature: solana_sdk::signature::Signature, signature: solana_sdk::signature::Signature,
slot: u64, slot: u64,
block_time: Option<prost_types::Timestamp>, block_time: Option<prost_types::Timestamp>,
program_received_time_us: i64, recv_us: i64,
outer_index: i64, outer_index: i64,
inner_index: Option<i64>, inner_index: Option<i64>,
transaction_index: Option<u64>, transaction_index: Option<u64>,
config: &GenericEventParseConfig, config: &GenericEventParseConfig,
) -> Vec<Box<dyn $crate::streaming::event_parser::core::traits::UnifiedEvent>> { ) -> Vec<Box<dyn $crate::streaming::event_parser::core::traits::UnifiedEvent>> {
self.inner.parse_events_from_grpc_inner_instruction(inner_instruction, signature, slot, block_time, program_received_time_us, outer_index, inner_index, transaction_index, config) self.inner.parse_events_from_grpc_inner_instruction(inner_instruction, signature, slot, block_time, recv_us, outer_index, inner_index, transaction_index, config)
} }
fn parse_events_from_instruction( fn parse_events_from_instruction(
@@ -71,7 +71,7 @@ macro_rules! impl_event_parser_delegate {
signature: solana_sdk::signature::Signature, signature: solana_sdk::signature::Signature,
slot: u64, slot: u64,
block_time: Option<prost_types::Timestamp>, block_time: Option<prost_types::Timestamp>,
program_received_time_us: i64, recv_us: i64,
outer_index: i64, outer_index: i64,
inner_index: Option<i64>, inner_index: Option<i64>,
bot_wallet: Option<solana_sdk::pubkey::Pubkey>, bot_wallet: Option<solana_sdk::pubkey::Pubkey>,
@@ -89,7 +89,7 @@ macro_rules! impl_event_parser_delegate {
signature, signature,
slot, slot,
block_time, block_time,
program_received_time_us, recv_us,
outer_index, outer_index,
inner_index, inner_index,
bot_wallet, bot_wallet,
@@ -106,7 +106,7 @@ macro_rules! impl_event_parser_delegate {
signature: solana_sdk::signature::Signature, signature: solana_sdk::signature::Signature,
slot: u64, slot: u64,
block_time: Option<prost_types::Timestamp>, block_time: Option<prost_types::Timestamp>,
program_received_time_us: i64, recv_us: i64,
outer_index: i64, outer_index: i64,
inner_index: Option<i64>, inner_index: Option<i64>,
bot_wallet: Option<solana_sdk::pubkey::Pubkey>, bot_wallet: Option<solana_sdk::pubkey::Pubkey>,
@@ -118,7 +118,7 @@ macro_rules! impl_event_parser_delegate {
+ Sync, + Sync,
>, >,
) -> anyhow::Result<()> { ) -> anyhow::Result<()> {
self.inner.parse_events_from_grpc_instruction(instruction, accounts, signature, slot, block_time, program_received_time_us, outer_index, inner_index, bot_wallet, transaction_index, inner_instructions, callback) self.inner.parse_events_from_grpc_instruction(instruction, accounts, signature, slot, block_time, recv_us, outer_index, inner_index, bot_wallet, transaction_index, inner_instructions, callback)
} }
fn should_handle(&self, program_id: &solana_sdk::pubkey::Pubkey) -> bool { fn should_handle(&self, program_id: &solana_sdk::pubkey::Pubkey) -> bool {
src/streaming/event_parser/core/traits.rs
+49 -49
View File
@@ -236,13 +236,13 @@ pub trait UnifiedEvent: Debug + Send + Sync {
fn slot(&self) -> u64; fn slot(&self) -> u64;
/// Get program received timestamp (milliseconds) /// Get program received timestamp (milliseconds)
fn program_received_time_us(&self) -> i64; fn recv_us(&self) -> i64;
/// Processing time consumption (milliseconds) /// Processing time consumption (milliseconds)
fn program_handle_time_consuming_us(&self) -> i64; fn handle_us(&self) -> i64;
/// Set processing time consumption (milliseconds) /// Set processing time consumption (milliseconds)
fn set_program_handle_time_consuming_us(&mut self, program_handle_time_consuming_us: i64); fn set_handle_us(&mut self, handle_us: i64);
/// Convert event to Any for downcasting /// Convert event to Any for downcasting
fn as_any(&self) -> &dyn std::any::Any; fn as_any(&self) -> &dyn std::any::Any;
@@ -265,8 +265,8 @@ pub trait UnifiedEvent: Debug + Send + Sync {
fn swap_data_is_parsed(&self) -> bool; fn swap_data_is_parsed(&self) -> bool;
/// Get index /// Get index
fn instruction_outer_index(&self) -> i64; fn outer_index(&self) -> i64;
fn instruction_inner_index(&self) -> Option<i64>; fn inner_index(&self) -> Option<i64>;
/// Get transaction index in slot /// Get transaction index in slot
fn transaction_index(&self) -> Option<u64>; fn transaction_index(&self) -> Option<u64>;
@@ -285,7 +285,7 @@ pub trait EventParser: Send + Sync {
signature: Signature, signature: Signature,
slot: u64, slot: u64,
block_time: Option<Timestamp>, block_time: Option<Timestamp>,
program_received_time_us: i64, recv_us: i64,
outer_index: i64, outer_index: i64,
inner_index: Option<i64>, inner_index: Option<i64>,
transaction_index: Option<u64>, transaction_index: Option<u64>,
@@ -300,7 +300,7 @@ pub trait EventParser: Send + Sync {
signature: Signature, signature: Signature,
slot: u64, slot: u64,
block_time: Option<Timestamp>, block_time: Option<Timestamp>,
program_received_time_us: i64, recv_us: i64,
outer_index: i64, outer_index: i64,
inner_index: Option<i64>, inner_index: Option<i64>,
transaction_index: Option<u64>, transaction_index: Option<u64>,
@@ -316,7 +316,7 @@ pub trait EventParser: Send + Sync {
signature: Signature, signature: Signature,
slot: u64, slot: u64,
block_time: Option<Timestamp>, block_time: Option<Timestamp>,
program_received_time_us: i64, recv_us: i64,
outer_index: i64, outer_index: i64,
inner_index: Option<i64>, inner_index: Option<i64>,
bot_wallet: Option<Pubkey>, bot_wallet: Option<Pubkey>,
@@ -335,7 +335,7 @@ pub trait EventParser: Send + Sync {
signature: Signature, signature: Signature,
slot: u64, slot: u64,
block_time: Option<Timestamp>, block_time: Option<Timestamp>,
program_received_time_us: i64, recv_us: i64,
outer_index: i64, outer_index: i64,
inner_index: Option<i64>, inner_index: Option<i64>,
bot_wallet: Option<Pubkey>, bot_wallet: Option<Pubkey>,
@@ -351,7 +351,7 @@ pub trait EventParser: Send + Sync {
signature: Signature, signature: Signature,
slot: Option<u64>, slot: Option<u64>,
block_time: Option<Timestamp>, block_time: Option<Timestamp>,
program_received_time_us: i64, recv_us: i64,
accounts: &[Pubkey], accounts: &[Pubkey],
inner_instructions: &[yellowstone_grpc_proto::prelude::InnerInstructions], inner_instructions: &[yellowstone_grpc_proto::prelude::InnerInstructions],
bot_wallet: Option<Pubkey>, bot_wallet: Option<Pubkey>,
@@ -383,7 +383,7 @@ pub trait EventParser: Send + Sync {
signature, signature,
slot, slot,
block_time, block_time,
program_received_time_us, recv_us,
index as i64, index as i64,
None, None,
bot_wallet, bot_wallet,
@@ -407,7 +407,7 @@ pub trait EventParser: Send + Sync {
signature: Signature, signature: Signature,
slot: Option<u64>, slot: Option<u64>,
block_time: Option<Timestamp>, block_time: Option<Timestamp>,
program_received_time_us: i64, recv_us: i64,
accounts: &[Pubkey], accounts: &[Pubkey],
inner_instructions: &[InnerInstructions], inner_instructions: &[InnerInstructions],
bot_wallet: Option<Pubkey>, bot_wallet: Option<Pubkey>,
@@ -440,7 +440,7 @@ pub trait EventParser: Send + Sync {
signature, signature,
slot, slot,
block_time, block_time,
program_received_time_us, recv_us,
index as i64, index as i64,
None, None,
bot_wallet, bot_wallet,
@@ -462,7 +462,7 @@ pub trait EventParser: Send + Sync {
signature: Signature, signature: Signature,
slot: Option<u64>, slot: Option<u64>,
block_time: Option<Timestamp>, block_time: Option<Timestamp>,
program_received_time_us: i64, recv_us: i64,
bot_wallet: Option<Pubkey>, bot_wallet: Option<Pubkey>,
transaction_index: Option<u64>, transaction_index: Option<u64>,
inner_instructions: &[InnerInstructions], inner_instructions: &[InnerInstructions],
@@ -477,7 +477,7 @@ pub trait EventParser: Send + Sync {
signature, signature,
slot, slot,
block_time, block_time,
program_received_time_us, recv_us,
bot_wallet, bot_wallet,
transaction_index, transaction_index,
inner_instructions, inner_instructions,
@@ -493,7 +493,7 @@ pub trait EventParser: Send + Sync {
signature: Signature, signature: Signature,
slot: Option<u64>, slot: Option<u64>,
block_time: Option<Timestamp>, block_time: Option<Timestamp>,
program_received_time_us: i64, recv_us: i64,
bot_wallet: Option<Pubkey>, bot_wallet: Option<Pubkey>,
transaction_index: Option<u64>, transaction_index: Option<u64>,
inner_instructions: &[InnerInstructions], inner_instructions: &[InnerInstructions],
@@ -505,7 +505,7 @@ pub trait EventParser: Send + Sync {
signature, signature,
slot, slot,
block_time, block_time,
program_received_time_us, recv_us,
&accounts, &accounts,
inner_instructions, inner_instructions,
bot_wallet, bot_wallet,
@@ -522,7 +522,7 @@ pub trait EventParser: Send + Sync {
signature: Signature, signature: Signature,
slot: Option<u64>, slot: Option<u64>,
block_time: Option<Timestamp>, block_time: Option<Timestamp>,
program_received_time_us: i64, recv_us: i64,
bot_wallet: Option<Pubkey>, bot_wallet: Option<Pubkey>,
transaction_index: Option<u64>, transaction_index: Option<u64>,
callback: Arc<dyn Fn(Box<dyn UnifiedEvent>) + Send + Sync>, callback: Arc<dyn Fn(Box<dyn UnifiedEvent>) + Send + Sync>,
@@ -537,7 +537,7 @@ pub trait EventParser: Send + Sync {
signature, signature,
slot, slot,
block_time, block_time,
program_received_time_us, recv_us,
bot_wallet, bot_wallet,
transaction_index, transaction_index,
adapter_callback, adapter_callback,
@@ -551,7 +551,7 @@ pub trait EventParser: Send + Sync {
signature: Signature, signature: Signature,
slot: Option<u64>, slot: Option<u64>,
block_time: Option<Timestamp>, block_time: Option<Timestamp>,
program_received_time_us: i64, recv_us: i64,
bot_wallet: Option<Pubkey>, bot_wallet: Option<Pubkey>,
transaction_index: Option<u64>, transaction_index: Option<u64>,
callback: Arc<dyn for<'a> Fn(&'a Box<dyn UnifiedEvent>) + Send + Sync>, callback: Arc<dyn for<'a> Fn(&'a Box<dyn UnifiedEvent>) + Send + Sync>,
@@ -600,7 +600,7 @@ pub trait EventParser: Send + Sync {
signature, signature,
slot, slot,
block_time, block_time,
program_received_time_us, recv_us,
&accounts_arc, &accounts_arc,
&inner_instructions_arc, &inner_instructions_arc,
bot_wallet, bot_wallet,
@@ -625,7 +625,7 @@ pub trait EventParser: Send + Sync {
signature, signature,
slot, slot,
block_time, block_time,
program_received_time_us, recv_us,
inner_instruction.index as i64, inner_instruction.index as i64,
Some(index as i64), Some(index as i64),
bot_wallet, bot_wallet,
@@ -732,7 +732,7 @@ pub trait EventParser: Send + Sync {
let slot = transaction.slot; let slot = transaction.slot;
let block_time = transaction.block_time.map(|t| Timestamp { seconds: t as i64, nanos: 0 }); let block_time = transaction.block_time.map(|t| Timestamp { seconds: t as i64, nanos: 0 });
let program_received_time_us = get_high_perf_clock(); let recv_us = get_high_perf_clock();
let bot_wallet = None; let bot_wallet = None;
let transaction_index = None; let transaction_index = None;
// 解析指令事件 // 解析指令事件
@@ -741,7 +741,7 @@ pub trait EventParser: Send + Sync {
signature, signature,
Some(slot), Some(slot),
block_time, block_time,
program_received_time_us, recv_us,
&accounts_arc, &accounts_arc,
&inner_instructions_arc, &inner_instructions_arc,
bot_wallet, bot_wallet,
@@ -759,7 +759,7 @@ pub trait EventParser: Send + Sync {
signature, signature,
Some(slot), Some(slot),
block_time, block_time,
program_received_time_us, recv_us,
inner_instruction.index as i64, inner_instruction.index as i64,
Some(index as i64), Some(index as i64),
bot_wallet, bot_wallet,
@@ -780,7 +780,7 @@ pub trait EventParser: Send + Sync {
signature: Signature, signature: Signature,
slot: Option<u64>, slot: Option<u64>,
block_time: Option<Timestamp>, block_time: Option<Timestamp>,
program_received_time_us: i64, recv_us: i64,
outer_index: i64, outer_index: i64,
inner_index: Option<i64>, inner_index: Option<i64>,
transaction_index: Option<u64>, transaction_index: Option<u64>,
@@ -792,7 +792,7 @@ pub trait EventParser: Send + Sync {
signature, signature,
slot, slot,
block_time, block_time,
program_received_time_us, recv_us,
outer_index, outer_index,
inner_index, inner_index,
transaction_index, transaction_index,
@@ -809,7 +809,7 @@ pub trait EventParser: Send + Sync {
signature: Signature, signature: Signature,
slot: Option<u64>, slot: Option<u64>,
block_time: Option<Timestamp>, block_time: Option<Timestamp>,
program_received_time_us: i64, recv_us: i64,
outer_index: i64, outer_index: i64,
inner_index: Option<i64>, inner_index: Option<i64>,
bot_wallet: Option<Pubkey>, bot_wallet: Option<Pubkey>,
@@ -824,7 +824,7 @@ pub trait EventParser: Send + Sync {
signature, signature,
slot, slot,
block_time, block_time,
program_received_time_us, recv_us,
outer_index, outer_index,
inner_index, inner_index,
bot_wallet, bot_wallet,
@@ -842,7 +842,7 @@ pub trait EventParser: Send + Sync {
signature: Signature, signature: Signature,
slot: Option<u64>, slot: Option<u64>,
block_time: Option<Timestamp>, block_time: Option<Timestamp>,
program_received_time_us: i64, recv_us: i64,
outer_index: i64, outer_index: i64,
inner_index: Option<i64>, inner_index: Option<i64>,
bot_wallet: Option<Pubkey>, bot_wallet: Option<Pubkey>,
@@ -857,7 +857,7 @@ pub trait EventParser: Send + Sync {
signature, signature,
slot, slot,
block_time, block_time,
program_received_time_us, recv_us,
outer_index, outer_index,
inner_index, inner_index,
bot_wallet, bot_wallet,
@@ -938,7 +938,7 @@ impl GenericEventParser {
signature: Signature, signature: Signature,
slot: u64, slot: u64,
block_time: Option<Timestamp>, block_time: Option<Timestamp>,
program_received_time_us: i64, recv_us: i64,
outer_index: i64, outer_index: i64,
inner_index: Option<i64>, inner_index: Option<i64>,
transaction_index: Option<u64>, transaction_index: Option<u64>,
@@ -956,7 +956,7 @@ impl GenericEventParser {
config.program_id, config.program_id,
outer_index, outer_index,
inner_index, inner_index,
program_received_time_us, recv_us,
transaction_index, transaction_index,
); );
parser(data, metadata) parser(data, metadata)
@@ -975,7 +975,7 @@ impl GenericEventParser {
signature: Signature, signature: Signature,
slot: u64, slot: u64,
block_time: Option<Timestamp>, block_time: Option<Timestamp>,
program_received_time_us: i64, recv_us: i64,
outer_index: i64, outer_index: i64,
inner_index: Option<i64>, inner_index: Option<i64>,
transaction_index: Option<u64>, transaction_index: Option<u64>,
@@ -993,7 +993,7 @@ impl GenericEventParser {
config.program_id, config.program_id,
outer_index, outer_index,
inner_index, inner_index,
program_received_time_us, recv_us,
transaction_index, transaction_index,
); );
parser(data, account_pubkeys, metadata) parser(data, account_pubkeys, metadata)
@@ -1016,7 +1016,7 @@ impl EventParser for GenericEventParser {
signature: Signature, signature: Signature,
slot: u64, slot: u64,
block_time: Option<Timestamp>, block_time: Option<Timestamp>,
program_received_time_us: i64, recv_us: i64,
outer_index: i64, outer_index: i64,
inner_index: Option<i64>, inner_index: Option<i64>,
transaction_index: Option<u64>, transaction_index: Option<u64>,
@@ -1034,7 +1034,7 @@ impl EventParser for GenericEventParser {
signature, signature,
slot, slot,
block_time, block_time,
program_received_time_us, recv_us,
outer_index, outer_index,
inner_index, inner_index,
transaction_index, transaction_index,
@@ -1052,7 +1052,7 @@ impl EventParser for GenericEventParser {
signature: Signature, signature: Signature,
slot: u64, slot: u64,
block_time: Option<Timestamp>, block_time: Option<Timestamp>,
program_received_time_us: i64, recv_us: i64,
outer_index: i64, outer_index: i64,
inner_index: Option<i64>, inner_index: Option<i64>,
transaction_index: Option<u64>, transaction_index: Option<u64>,
@@ -1070,7 +1070,7 @@ impl EventParser for GenericEventParser {
signature, signature,
slot, slot,
block_time, block_time,
program_received_time_us, recv_us,
outer_index, outer_index,
inner_index, inner_index,
transaction_index, transaction_index,
@@ -1089,7 +1089,7 @@ impl EventParser for GenericEventParser {
signature: Signature, signature: Signature,
slot: u64, slot: u64,
block_time: Option<Timestamp>, block_time: Option<Timestamp>,
program_received_time_us: i64, recv_us: i64,
outer_index: i64, outer_index: i64,
inner_index: Option<i64>, inner_index: Option<i64>,
bot_wallet: Option<Pubkey>, bot_wallet: Option<Pubkey>,
@@ -1141,7 +1141,7 @@ impl EventParser for GenericEventParser {
signature, signature,
slot, slot,
block_time, block_time,
program_received_time_us, recv_us,
outer_index, outer_index,
inner_index, inner_index,
transaction_index, transaction_index,
@@ -1165,7 +1165,7 @@ impl EventParser for GenericEventParser {
signature, signature,
slot, slot,
block_time, block_time,
program_received_time_us, recv_us,
outer_index, outer_index,
inner_index, inner_index,
transaction_index, transaction_index,
@@ -1205,8 +1205,8 @@ impl EventParser for GenericEventParser {
event.merge(&*inner_instruction_event); event.merge(&*inner_instruction_event);
} }
// 设置处理时间（使用高性能时钟） // 设置处理时间（使用高性能时钟）
event.set_program_handle_time_consuming_us(elapsed_micros_since( event.set_handle_us(elapsed_micros_since(
program_received_time_us, recv_us,
)); ));
event = process_event(event, bot_wallet); event = process_event(event, bot_wallet);
callback(&event); callback(&event);
@@ -1224,7 +1224,7 @@ impl EventParser for GenericEventParser {
signature: Signature, signature: Signature,
slot: u64, slot: u64,
block_time: Option<Timestamp>, block_time: Option<Timestamp>,
program_received_time_us: i64, recv_us: i64,
outer_index: i64, outer_index: i64,
inner_index: Option<i64>, inner_index: Option<i64>,
bot_wallet: Option<Pubkey>, bot_wallet: Option<Pubkey>,
@@ -1276,7 +1276,7 @@ impl EventParser for GenericEventParser {
signature, signature,
slot, slot,
block_time, block_time,
program_received_time_us, recv_us,
outer_index, outer_index,
inner_index, inner_index,
transaction_index, transaction_index,
@@ -1300,7 +1300,7 @@ impl EventParser for GenericEventParser {
signature, signature,
slot, slot,
block_time, block_time,
program_received_time_us, recv_us,
outer_index, outer_index,
inner_index, inner_index,
transaction_index, transaction_index,
@@ -1340,8 +1340,8 @@ impl EventParser for GenericEventParser {
event.merge(&*inner_instruction_event); event.merge(&*inner_instruction_event);
} }
// 设置处理时间（使用高性能时钟） // 设置处理时间（使用高性能时钟）
event.set_program_handle_time_consuming_us(elapsed_micros_since( event.set_handle_us(elapsed_micros_since(
program_received_time_us, recv_us,
)); ));
event = process_event(event, bot_wallet); event = process_event(event, bot_wallet);
callback(&event); callback(&event);
src/streaming/event_parser/protocols/block/block_meta_event.rs
+2 -2
View File
@@ -18,7 +18,7 @@ impl BlockMetaEvent {
slot: u64, slot: u64,
block_hash: String, block_hash: String,
block_time_ms: i64, block_time_ms: i64,
program_received_time_us: i64, recv_us: i64,
) -> Self { ) -> Self {
let metadata = EventMetadata::new( let metadata = EventMetadata::new(
Signature::default(), Signature::default(),
@@ -30,7 +30,7 @@ impl BlockMetaEvent {
solana_sdk::pubkey::Pubkey::default(), solana_sdk::pubkey::Pubkey::default(),
0, 0,
None, None,
program_received_time_us, recv_us,
None, None,
); );
Self { metadata, slot, block_hash } Self { metadata, slot, block_hash }
src/streaming/grpc/pool.rs
+3 -3
View File
@@ -118,7 +118,7 @@ impl PooledAccountPretty {
self.account.data = new_data; self.account.data = new_data;
} }
self.account.program_received_time_us = get_high_perf_clock(); self.account.recv_us = get_high_perf_clock();
} }
} }
@@ -196,7 +196,7 @@ impl PooledBlockMetaPretty {
self.block_meta.slot = block_update.slot; self.block_meta.slot = block_update.slot;
self.block_meta.block_hash = block_update.blockhash; self.block_meta.block_hash = block_update.blockhash;
self.block_meta.block_time = block_time; self.block_meta.block_time = block_time;
self.block_meta.program_received_time_us = get_high_perf_clock(); self.block_meta.recv_us = get_high_perf_clock();
} }
} }
@@ -282,7 +282,7 @@ impl PooledTransactionPretty {
self.transaction.signature = self.transaction.signature =
Signature::try_from(tx.signature.as_slice()).expect("valid signature"); Signature::try_from(tx.signature.as_slice()).expect("valid signature");
self.transaction.is_vote = tx.is_vote; self.transaction.is_vote = tx.is_vote;
self.transaction.program_received_time_us = get_high_perf_clock(); self.transaction.recv_us = get_high_perf_clock();
self.transaction.grpc_tx = tx; self.transaction.grpc_tx = tx;
} }
} }
src/streaming/grpc/types.rs
+9 -9
View File
@@ -29,7 +29,7 @@ pub struct AccountPretty {
pub owner: Pubkey, pub owner: Pubkey,
pub rent_epoch: u64, pub rent_epoch: u64,
pub data: Vec<u8>, pub data: Vec<u8>,
pub program_received_time_us: i64, pub recv_us: i64,
} }
impl fmt::Debug for AccountPretty { impl fmt::Debug for AccountPretty {
@@ -52,7 +52,7 @@ pub struct BlockMetaPretty {
pub slot: u64, pub slot: u64,
pub block_hash: String, pub block_hash: String,
pub block_time: Option<Timestamp>, pub block_time: Option<Timestamp>,
pub program_received_time_us: i64, pub recv_us: i64,
} }
impl fmt::Debug for BlockMetaPretty { impl fmt::Debug for BlockMetaPretty {
@@ -61,7 +61,7 @@ impl fmt::Debug for BlockMetaPretty {
.field("slot", &self.slot) .field("slot", &self.slot)
.field("block_hash", &self.block_hash) .field("block_hash", &self.block_hash)
.field("block_time", &self.block_time) .field("block_time", &self.block_time)
.field("program_received_time_us", &self.program_received_time_us) .field("recv_us", &self.recv_us)
.finish() .finish()
} }
} }
@@ -74,7 +74,7 @@ pub struct TransactionPretty {
pub block_time: Option<Timestamp>, pub block_time: Option<Timestamp>,
pub signature: Signature, pub signature: Signature,
pub is_vote: bool, pub is_vote: bool,
pub program_received_time_us: i64, pub recv_us: i64,
pub grpc_tx: SubscribeUpdateTransactionInfo, pub grpc_tx: SubscribeUpdateTransactionInfo,
} }
@@ -85,7 +85,7 @@ impl fmt::Debug for TransactionPretty {
.field("transaction_index", &self.transaction_index) .field("transaction_index", &self.transaction_index)
.field("signature", &self.signature) .field("signature", &self.signature)
.field("is_vote", &self.is_vote) .field("is_vote", &self.is_vote)
.field("program_received_time_us", &self.program_received_time_us) .field("recv_us", &self.recv_us)
.finish() .finish()
} }
} }
@@ -100,7 +100,7 @@ impl Default for TransactionPretty {
signature: Signature::default(), signature: Signature::default(),
is_vote: false, is_vote: false,
grpc_tx: SubscribeUpdateTransactionInfo::default(), grpc_tx: SubscribeUpdateTransactionInfo::default(),
program_received_time_us: 0, recv_us: 0,
} }
} }
} }
@@ -121,7 +121,7 @@ impl Default for TransactionPretty {
// owner: Pubkey::try_from(account_info.owner.as_slice()).expect("valid pubkey"), // owner: Pubkey::try_from(account_info.owner.as_slice()).expect("valid pubkey"),
// rent_epoch: account_info.rent_epoch, // rent_epoch: account_info.rent_epoch,
// data: account_info.data, // data: account_info.data,
// program_received_time_us: get_high_perf_clock(), // recv_us: get_high_perf_clock(),
// } // }
// } // }
// } // }
@@ -137,7 +137,7 @@ impl Default for TransactionPretty {
// block_hash: blockhash, // block_hash: blockhash,
// block_time, // block_time,
// slot, // slot,
// program_received_time_us: get_high_perf_clock(), // recv_us: get_high_perf_clock(),
// } // }
// } // }
// } // }
@@ -161,7 +161,7 @@ impl Default for TransactionPretty {
// is_vote: tx.is_vote, // is_vote: tx.is_vote,
// tx: yellowstone_grpc_proto::convert_from::create_tx_with_meta(tx) // tx: yellowstone_grpc_proto::convert_from::create_tx_with_meta(tx)
// .expect("valid tx with meta"), // .expect("valid tx with meta"),
// program_received_time_us: get_high_perf_clock(), // recv_us: get_high_perf_clock(),
// } // }
// } // }
// } // }
src/streaming/shred/pool.rs
+7 -7
View File
@@ -52,11 +52,11 @@ impl PooledTransactionWithSlot {
&mut self, &mut self,
transaction: VersionedTransaction, transaction: VersionedTransaction,
slot: u64, slot: u64,
program_received_time_us: i64 recv_us: i64
) { ) {
self.transaction.transaction = transaction; self.transaction.transaction = transaction;
self.transaction.slot = slot; self.transaction.slot = slot;
self.transaction.program_received_time_us = program_received_time_us; self.transaction.recv_us = recv_us;
} }
/// 使用优化的工厂方法创建 TransactionWithSlot（移动数据而不是克隆） /// 使用优化的工厂方法创建 TransactionWithSlot（移动数据而不是克隆）
@@ -72,7 +72,7 @@ impl Drop for PooledTransactionWithSlot {
if pool.len() < self.max_size { if pool.len() < self.max_size {
// 清理敏感数据 // 清理敏感数据
self.transaction.slot = 0; self.transaction.slot = 0;
self.transaction.program_received_time_us = 0; self.transaction.recv_us = 0;
// 重置交易为默认值以清理敏感数据 // 重置交易为默认值以清理敏感数据
self.transaction.transaction = VersionedTransaction::default(); self.transaction.transaction = VersionedTransaction::default();
pool.push_back(std::mem::take(&mut self.transaction)); pool.push_back(std::mem::take(&mut self.transaction));
@@ -118,10 +118,10 @@ impl ShredPoolManager {
&self, &self,
transaction: VersionedTransaction, transaction: VersionedTransaction,
slot: u64, slot: u64,
program_received_time_us: i64, recv_us: i64,
) -> TransactionWithSlot { ) -> TransactionWithSlot {
let mut pooled_tx = self.transaction_pool.acquire(); let mut pooled_tx = self.transaction_pool.acquire();
pooled_tx.reset_from_data(transaction, slot, program_received_time_us); pooled_tx.reset_from_data(transaction, slot, recv_us);
pooled_tx.into_transaction_with_slot() pooled_tx.into_transaction_with_slot()
} }
} }
@@ -145,12 +145,12 @@ pub mod factory {
pub fn create_transaction_with_slot_pooled( pub fn create_transaction_with_slot_pooled(
transaction: VersionedTransaction, transaction: VersionedTransaction,
slot: u64, slot: u64,
program_received_time_us: i64, recv_us: i64,
) -> TransactionWithSlot { ) -> TransactionWithSlot {
GLOBAL_SHRED_POOL_MANAGER.create_transaction_with_slot_optimized( GLOBAL_SHRED_POOL_MANAGER.create_transaction_with_slot_optimized(
transaction, transaction,
slot, slot,
program_received_time_us recv_us
) )
} }
} }
src/streaming/shred/types.rs
+3 -3
View File
@@ -5,7 +5,7 @@ use solana_sdk::transaction::VersionedTransaction;
pub struct TransactionWithSlot { pub struct TransactionWithSlot {
pub transaction: VersionedTransaction, pub transaction: VersionedTransaction,
pub slot: u64, pub slot: u64,
pub program_received_time_us: i64, pub recv_us: i64,
} }
impl TransactionWithSlot { impl TransactionWithSlot {
@@ -13,8 +13,8 @@ impl TransactionWithSlot {
pub fn new( pub fn new(
transaction: VersionedTransaction, transaction: VersionedTransaction,
slot: u64, slot: u64,
program_received_time_us: i64, recv_us: i64,
) -> Self { ) -> Self {
Self { transaction, slot, program_received_time_us } Self { transaction, slot, recv_us }
} }
} }