From c3d599acd52a2362aa405f57bd83637b6fd98468 Mon Sep 17 00:00:00 2001 From: vnxfsc <1047658287@qq.com> Date: Mon, 25 Aug 2025 00:37:31 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=B7=BB=E5=8A=A0=E4=BA=A4=E6=98=93?= =?UTF-8?q?=E7=B4=A2=E5=BC=95=E5=8A=9F=E8=83=BD=E6=94=AF=E6=8C=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 为 TransactionPretty 添加 transaction_index 字段 - 从 Yellowstone gRPC 的 SubscribeUpdateTransaction.transaction.index 提取交易索引 - 为 EventMetadata 添加 transaction_index 字段用于存储交易在 slot 中的索引 - 为 UnifiedEvent trait 添加 transaction_index() 和 set_transaction_index() 方法 - 更新事件处理器以传递交易索引到事件元数据 - 更新主程序日志以显示交易索引信息 交易事件现在包含它们在 slot 中的索引位置，账户事件由于 gRPC 协议限制保持为 None --- src/main.rs | 9 ++++++++- src/streaming/event_parser/common/mod.rs | 8 ++++++++ src/streaming/event_parser/common/types.rs | 9 ++++++++- src/streaming/event_parser/core/traits.rs | 6 ++++++ src/streaming/grpc/event_processor.rs | 7 ++++++- src/streaming/grpc/types.rs | 5 +++++ 6 files changed, 41 insertions(+), 3 deletions(-) diff --git a/src/main.rs b/src/main.rs index dd110cd..6208408 100755 --- a/src/main.rs +++ b/src/main.rs @@ -188,7 +188,14 @@ async fn test_shreds() -> Result<(), Box> { fn create_event_callback() -> impl Fn(Box) { |event: Box| { - println!("🎉 Event received! Type: {:?}, ID: {}", event.event_type(), event.id()); + println!( + "🎉 Event received! Type: {:?}, ID: {}, Slot: {}, Transaction Index: {:?}, Instruction Index: {}", + event.event_type(), + event.id(), + event.slot(), + event.transaction_index(), + event.index() + ); match_event!(event, { // -------------------------- block meta ----------------------- BlockMetaEvent => |e: BlockMetaEvent| { diff --git a/src/streaming/event_parser/common/mod.rs b/src/streaming/event_parser/common/mod.rs index 3ecc161..bc4f9d0 100755 --- a/src/streaming/event_parser/common/mod.rs +++ b/src/streaming/event_parser/common/mod.rs @@ -63,6 +63,14 @@ macro_rules! impl_unified_event { fn index(&self) -> String { self.metadata.index.clone() } + + fn transaction_index(&self) -> Option { + self.metadata.transaction_index + } + + fn set_transaction_index(&mut self, transaction_index: Option) { + self.metadata.set_transaction_index(transaction_index); + } } }; } diff --git a/src/streaming/event_parser/common/types.rs b/src/streaming/event_parser/common/types.rs index 9794cc8..4eaa99f 100755 --- a/src/streaming/event_parser/common/types.rs +++ b/src/streaming/event_parser/common/types.rs @@ -343,6 +343,7 @@ pub struct EventMetadata { pub id: String, pub signature: String, pub slot: u64, + pub transaction_index: Option, // 新增：交易在slot中的索引 pub block_time: i64, pub block_time_ms: i64, pub program_received_time_ms: i64, @@ -352,7 +353,7 @@ pub struct EventMetadata { pub program_id: Pubkey, pub transfer_datas: Vec, pub swap_data: Option, - pub index: String, + pub index: String, // 保留原有的指令索引 } impl EventMetadata { @@ -373,6 +374,7 @@ impl EventMetadata { id, signature, slot, + transaction_index: None, // 默认为None，后续设置 block_time, block_time_ms, program_received_time_ms, @@ -403,6 +405,11 @@ impl EventMetadata { self.swap_data = swap_data; } + /// 设置交易索引 + pub fn set_transaction_index(&mut self, transaction_index: Option) { + self.transaction_index = transaction_index; + } + /// Recycle EventMetadata to object pool pub async fn recycle(self) { EVENT_METADATA_POOL.release(self).await; diff --git a/src/streaming/event_parser/core/traits.rs b/src/streaming/event_parser/core/traits.rs index a6d8574..a23d53c 100755 --- a/src/streaming/event_parser/core/traits.rs +++ b/src/streaming/event_parser/core/traits.rs @@ -67,6 +67,12 @@ pub trait UnifiedEvent: Debug + Send + Sync { /// Get index fn index(&self) -> String; + + /// Get transaction index in slot + fn transaction_index(&self) -> Option; + + /// Set transaction index in slot + fn set_transaction_index(&mut self, transaction_index: Option); } /// 事件解析器trait - 定义了事件解析的核心方法 diff --git a/src/streaming/grpc/event_processor.rs b/src/streaming/grpc/event_processor.rs index 10454fe..18ebfdb 100644 --- a/src/streaming/grpc/event_processor.rs +++ b/src/streaming/grpc/event_processor.rs @@ -90,7 +90,7 @@ impl EventProcessor { // 使用缓存获取解析器 let parser = self.get_or_create_parser(protocols.clone(), event_type_filter); - let all_events = parser + let mut all_events = parser .parse_transaction( transaction_pretty.tx.clone(), &signature, @@ -105,6 +105,11 @@ impl EventProcessor { .await .unwrap_or_else(|_e| vec![]); + // 为所有事件设置交易索引 + for event in &mut all_events { + event.set_transaction_index(transaction_pretty.transaction_index); + } + // 保存事件数量用于日志记录 let event_count = all_events.len(); diff --git a/src/streaming/grpc/types.rs b/src/streaming/grpc/types.rs index 8d80a7c..d75b986 100644 --- a/src/streaming/grpc/types.rs +++ b/src/streaming/grpc/types.rs @@ -66,6 +66,7 @@ impl fmt::Debug for BlockMetaPretty { #[derive(Clone)] pub struct TransactionPretty { pub slot: u64, + pub transaction_index: Option, // 新增：交易在slot中的索引 pub block_hash: String, pub block_time: Option, pub signature: Signature, @@ -85,6 +86,7 @@ impl fmt::Debug for TransactionPretty { f.debug_struct("TransactionPretty") .field("slot", &self.slot) + .field("transaction_index", &self.transaction_index) .field("signature", &self.signature) .field("is_vote", &self.is_vote) .field("tx", &TxWrap(&self.tx)) @@ -127,8 +129,11 @@ impl From<(SubscribeUpdateTransaction, Option)> for TransactionPretty ), ) -> Self { let tx = transaction.expect("should be defined"); + // 根据用户说明，交易索引在 transaction.index 中 + let transaction_index = tx.index; Self { slot, + transaction_index: Some(transaction_index), // 提取交易索引 block_time, block_hash: "".to_string(), signature: Signature::try_from(tx.signature.as_slice()).expect("valid signature"),