add block_time and block_time_ms

src/main.rs

@@ -22,7 +22,7 @@ use solana_streamer_sdk::{
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     test_grpc().await?;
-    test_shreds().await?;
+    // test_shreds().await?;
     Ok(())
 }
 
@@ -36,11 +36,11 @@ async fn test_grpc() -> Result<(), Box<dyn std::error::Error>> {
 
     let callback = create_event_callback();
     let protocols = vec![
-        Protocol::PumpFun,
+        // Protocol::PumpFun,
-        Protocol::PumpSwap,
+        // Protocol::PumpSwap,
         Protocol::Bonk,
-        Protocol::RaydiumCpmm,
+        // Protocol::RaydiumCpmm,
-        Protocol::RaydiumClmm,
+        // Protocol::RaydiumClmm,
     ];
 
     println!("开始监听事件，按 Ctrl+C 停止...");
@@ -75,10 +75,10 @@ fn create_event_callback() -> impl Fn(Box<dyn UnifiedEvent>) {
     |event: Box<dyn UnifiedEvent>| {
         match_event!(event, {
             BlockMetaEvent => |e: BlockMetaEvent| {
-                println!("BlockMetaEvent: {:?}", e.slot);
+                // println!("BlockMetaEvent: {:?}", e.slot);
             },
             BonkPoolCreateEvent => |e: BonkPoolCreateEvent| {
-                println!("BonkPoolCreateEvent: {:?}", e.base_mint_param.symbol);
+                // println!("BonkPoolCreateEvent: {:?}", e.base_mint_param.symbol);
             },
             BonkTradeEvent => |e: BonkTradeEvent| {
                 println!("BonkTradeEvent: {:?}", e);

src/streaming/event_parser/common/types.rs

@@ -1,10 +1,7 @@
 use borsh::{BorshDeserialize, BorshSerialize};
 use serde::{Deserialize, Serialize};
-use solana_sdk::instruction::CompiledInstruction;
 use solana_sdk::pubkey::Pubkey;
-use solana_transaction_status::{
+use solana_transaction_status::UiInstruction;
-    UiCompiledInstruction, UiInstruction, UiTransactionStatusMeta, UiTransactionTokenBalance,
-};
 use std::collections::hash_map::DefaultHasher;
 use std::hash::{Hash, Hasher};
 
@@ -158,6 +155,8 @@ pub struct EventMetadata {
     pub id: String,
     pub signature: String,
     pub slot: u64,
+    pub block_time: i64,
+    pub block_time_ms: i64,
     pub program_received_time_ms: i64,
     pub protocol: ProtocolType,
     pub event_type: EventType,
@@ -170,6 +169,8 @@ impl EventMetadata {
         id: String,
         signature: String,
         slot: u64,
+        block_time: i64,
+        block_time_ms: i64,
         protocol: ProtocolType,
         event_type: EventType,
         program_id: Pubkey,
@@ -178,6 +179,8 @@ impl EventMetadata {
             id,
             signature,
             slot,
+            block_time,
+            block_time_ms,
             program_received_time_ms: chrono::Utc::now().timestamp_millis(),
             protocol,
             event_type,

src/streaming/event_parser/core/traits.rs

@@ -1,5 +1,6 @@
 use anyhow::Result;
 use borsh::BorshDeserialize;
+use prost_types::Timestamp;
 use serde::{Deserialize, Serialize};
 use solana_sdk::{
     instruction::CompiledInstruction, pubkey::Pubkey, transaction::VersionedTransaction,
@@ -7,6 +8,7 @@ use solana_sdk::{
 use solana_transaction_status::{
     EncodedTransactionWithStatusMeta, UiCompiledInstruction, UiInnerInstructions, UiInstruction,
 };
+use yellowstone_grpc_proto::prelude::UnixTimestamp;
 use std::fmt::Debug;
 use std::{collections::HashMap, str::FromStr};
 use yellowstone_grpc_proto::geyser::SubscribeUpdateBlockMeta;
@@ -78,6 +80,7 @@ pub trait EventParser: Send + Sync {
         instruction: &UiCompiledInstruction,
         signature: &str,
         slot: u64,
+        block_time: Option<Timestamp>,
     ) -> Vec<Box<dyn UnifiedEvent>>;
 
     /// 从指令中解析事件数据
@@ -87,6 +90,7 @@ pub trait EventParser: Send + Sync {
         accounts: &[Pubkey],
         signature: &str,
         slot: u64,
+        block_time: Option<Timestamp>,
     ) -> Vec<Box<dyn UnifiedEvent>>;
 
     /// 从VersionedTransaction中解析指令事件的通用方法
@@ -95,6 +99,7 @@ pub trait EventParser: Send + Sync {
         versioned_tx: &VersionedTransaction,
         signature: &str,
         slot: Option<u64>,
+        block_time: Option<Timestamp>,
         accounts: &[Pubkey],
         inner_instructions: &[UiInnerInstructions],
     ) -> Result<Vec<Box<dyn UnifiedEvent>>> {
@@ -118,7 +123,7 @@ pub trait EventParser: Send + Sync {
                             }
                         }
                         if let Ok(mut events) = self
-                            .parse_instruction(instruction, &accounts, signature, slot)
+                            .parse_instruction(instruction, &accounts, signature, slot, block_time)
                             .await
                         {
                             if events.len() > 0 {
@@ -153,6 +158,7 @@ pub trait EventParser: Send + Sync {
         versioned_tx: &VersionedTransaction,
         signature: &str,
         slot: Option<u64>,
+        block_time: Option<Timestamp>,
         bot_wallet: Option<Pubkey>,
     ) -> Result<Vec<Box<dyn UnifiedEvent>>> {
         let accounts: Vec<Pubkey> = versioned_tx.message.static_account_keys().to_vec();
@@ -161,6 +167,7 @@ pub trait EventParser: Send + Sync {
                 versioned_tx,
                 signature,
                 slot,
+                block_time,
                 &accounts,
                 &vec![],
             )
@@ -174,6 +181,7 @@ pub trait EventParser: Send + Sync {
         tx: EncodedTransactionWithStatusMeta,
         signature: &str,
         slot: Option<u64>,
+        block_time: Option<Timestamp>,
         bot_wallet: Option<Pubkey>,
     ) -> Result<Vec<Box<dyn UnifiedEvent>>> {
         let transaction = tx.transaction;
@@ -209,6 +217,7 @@ pub trait EventParser: Send + Sync {
                     &versioned_tx,
                     signature,
                     slot,
+                    block_time,
                     &accounts,
                     &inner_instructions,
                 )
@@ -238,6 +247,7 @@ pub trait EventParser: Send + Sync {
                                     &accounts,
                                     signature,
                                     slot,
+                                    block_time,
                                 )
                                 .await
                             {
@@ -256,7 +266,7 @@ pub trait EventParser: Send + Sync {
                                 }
                             }
                             if let Ok(mut events) = self
-                                .parse_inner_instruction(compiled, signature, slot)
+                                .parse_inner_instruction(compiled, signature, slot, block_time)
                                 .await
                             {
                                 if events.len() > 0 {
@@ -335,9 +345,10 @@ pub trait EventParser: Send + Sync {
         instruction: &UiCompiledInstruction,
         signature: &str,
         slot: Option<u64>,
+        block_time: Option<Timestamp>,
     ) -> Result<Vec<Box<dyn UnifiedEvent>>> {
         let slot = slot.unwrap_or(0);
-        let events = self.parse_events_from_inner_instruction(instruction, signature, slot);
+        let events = self.parse_events_from_inner_instruction(instruction, signature, slot, block_time);
         Ok(events)
     }
 
@@ -347,9 +358,10 @@ pub trait EventParser: Send + Sync {
         accounts: &[Pubkey],
         signature: &str,
         slot: Option<u64>,
+        block_time: Option<Timestamp>,
     ) -> Result<Vec<Box<dyn UnifiedEvent>>> {
         let slot = slot.unwrap_or(0);
-        let events = self.parse_events_from_instruction(instruction, accounts, signature, slot);
+        let events = self.parse_events_from_instruction(instruction, accounts, signature, slot, block_time);
         Ok(events)
     }
 
@@ -429,11 +441,16 @@ impl GenericEventParser {
         data: &[u8],
         signature: &str,
         slot: u64,
+        block_time: Option<Timestamp>,
     ) -> Option<Box<dyn UnifiedEvent>> {
+        let timestamp = block_time.unwrap_or(Timestamp { seconds: 0, nanos: 0 });
+        let block_time_ms = timestamp.seconds * 1000 + (timestamp.nanos as i64) / 1_000_000;
         let metadata = EventMetadata::new(
             signature.to_string(),
             signature.to_string(),
             slot,
+            timestamp.seconds,
+            block_time_ms,
             self.protocol_type.clone(),
             config.event_type.clone(),
             self.program_id,
@@ -449,11 +466,16 @@ impl GenericEventParser {
         account_pubkeys: &[Pubkey],
         signature: &str,
         slot: u64,
+        block_time: Option<Timestamp>,
     ) -> Option<Box<dyn UnifiedEvent>> {
+        let timestamp = block_time.unwrap_or(Timestamp { seconds: 0, nanos: 0 });
+        let block_time_ms = timestamp.seconds * 1000 + (timestamp.nanos as i64) / 1_000_000;
         let metadata = EventMetadata::new(
             signature.to_string(),
             signature.to_string(),
             slot,
+            timestamp.seconds,
+            block_time_ms,
             self.protocol_type.clone(),
             config.event_type.clone(),
             self.program_id,
@@ -470,6 +492,7 @@ impl EventParser for GenericEventParser {
         inner_instruction: &UiCompiledInstruction,
         signature: &str,
         slot: u64,
+        block_time: Option<Timestamp>,
     ) -> Vec<Box<dyn UnifiedEvent>> {
         let inner_instruction_data = inner_instruction.data.clone();
         let inner_instruction_data_decoded =
@@ -485,7 +508,7 @@ impl EventParser for GenericEventParser {
             if discriminator_matches(&inner_instruction_data_decoded_str, disc) {
                 for config in configs {
                     if let Some(event) =
-                        self.parse_inner_instruction_event(config, data, signature, slot)
+                        self.parse_inner_instruction_event(config, data, signature, slot, block_time)
                     {
                         events.push(event);
                     }
@@ -502,6 +525,7 @@ impl EventParser for GenericEventParser {
         accounts: &[Pubkey],
         signature: &str,
         slot: u64,
+        block_time: Option<Timestamp>,
     ) -> Vec<Box<dyn UnifiedEvent>> {
         let program_id = accounts[instruction.program_id_index as usize];
         if !self.should_handle(&program_id) {
@@ -532,6 +556,7 @@ impl EventParser for GenericEventParser {
                         &account_pubkeys,
                         signature,
                         slot,
+                        block_time,
                     ) {
                         events.push(event);
                     }
@@ -554,11 +579,14 @@ impl EventParser for GenericEventParser {
 pub struct SDKSystemEventParser {}
 impl SDKSystemEventParser {
     pub fn parse_block(block: SubscribeUpdateBlockMeta) -> Box<dyn UnifiedEvent> {
+        let block_time = block.block_time.unwrap_or(UnixTimestamp { timestamp: 0 });
         Box::new(BlockMetaEvent {
             metadata: EventMetadata::new(
                 block.blockhash.to_string(),
                 "".to_string(),
                 block.slot,
+                block_time.timestamp,
+                block_time.timestamp * 1000,
                 ProtocolType::SDKSystem,
                 EventType::SDKSystem,
                 Pubkey::default(),

src/streaming/event_parser/protocols/bonk/parser.rs

@@ -1,3 +1,4 @@
+use prost_types::Timestamp;
 use solana_sdk::{instruction::CompiledInstruction, pubkey::Pubkey};
 use solana_transaction_status::UiCompiledInstruction;
 
@@ -419,9 +420,10 @@ impl EventParser for BonkEventParser {
         inner_instruction: &UiCompiledInstruction,
         signature: &str,
         slot: u64,
+        block_time: Option<Timestamp>,
     ) -> Vec<Box<dyn UnifiedEvent>> {
         self.inner
-            .parse_events_from_inner_instruction(inner_instruction, signature, slot)
+            .parse_events_from_inner_instruction(inner_instruction, signature, slot, block_time)
     }
 
     fn parse_events_from_instruction(
@@ -430,9 +432,10 @@ impl EventParser for BonkEventParser {
         accounts: &[Pubkey],
         signature: &str,
         slot: u64,
+        block_time: Option<Timestamp>,
     ) -> Vec<Box<dyn UnifiedEvent>> {
         self.inner
-            .parse_events_from_instruction(instruction, accounts, signature, slot)
+            .parse_events_from_instruction(instruction, accounts, signature, slot, block_time)
     }
 
     fn should_handle(&self, program_id: &Pubkey) -> bool {

src/streaming/event_parser/protocols/pumpfun/parser.rs

@@ -1,3 +1,4 @@
+use prost_types::Timestamp;
 use solana_sdk::{instruction::CompiledInstruction, pubkey::Pubkey};
 use solana_transaction_status::UiCompiledInstruction;
 
@@ -224,9 +225,10 @@ impl EventParser for PumpFunEventParser {
         inner_instruction: &UiCompiledInstruction,
         signature: &str,
         slot: u64,
+        block_time: Option<Timestamp>,
     ) -> Vec<Box<dyn UnifiedEvent>> {
         self.inner
-            .parse_events_from_inner_instruction(inner_instruction, signature, slot)
+            .parse_events_from_inner_instruction(inner_instruction, signature, slot, block_time)
     }
 
     fn parse_events_from_instruction(
@@ -235,9 +237,10 @@ impl EventParser for PumpFunEventParser {
         accounts: &[Pubkey],
         signature: &str,
         slot: u64,
+        block_time: Option<Timestamp>,
     ) -> Vec<Box<dyn UnifiedEvent>> {
         self.inner
-            .parse_events_from_instruction(instruction, accounts, signature, slot)
+            .parse_events_from_instruction(instruction, accounts, signature, slot, block_time)
     }
 
     fn should_handle(&self, program_id: &Pubkey) -> bool {

src/streaming/event_parser/protocols/pumpswap/parser.rs

@@ -1,3 +1,4 @@
+use prost_types::Timestamp;
 use solana_sdk::{instruction::CompiledInstruction, pubkey::Pubkey};
 use solana_transaction_status::UiCompiledInstruction;
 
@@ -360,9 +361,10 @@ impl EventParser for PumpSwapEventParser {
         inner_instruction: &UiCompiledInstruction,
         signature: &str,
         slot: u64,
+        block_time: Option<Timestamp>,
     ) -> Vec<Box<dyn UnifiedEvent>> {
         self.inner
-            .parse_events_from_inner_instruction(inner_instruction, signature, slot)
+            .parse_events_from_inner_instruction(inner_instruction, signature, slot, block_time)
     }
 
     fn parse_events_from_instruction(
@@ -371,9 +373,10 @@ impl EventParser for PumpSwapEventParser {
         accounts: &[Pubkey],
         signature: &str,
         slot: u64,
+        block_time: Option<Timestamp>,
     ) -> Vec<Box<dyn UnifiedEvent>> {
         self.inner
-            .parse_events_from_instruction(instruction, accounts, signature, slot)
+            .parse_events_from_instruction(instruction, accounts, signature, slot, block_time)
     }
 
     fn should_handle(&self, program_id: &Pubkey) -> bool {

src/streaming/event_parser/protocols/raydium_clmm/parser.rs

@@ -1,3 +1,4 @@
+use prost_types::Timestamp;
 use solana_sdk::{instruction::CompiledInstruction, pubkey::Pubkey};
 use solana_transaction_status::UiCompiledInstruction;
 
@@ -144,9 +145,10 @@ impl EventParser for RaydiumClmmEventParser {
         inner_instruction: &UiCompiledInstruction,
         signature: &str,
         slot: u64,
+        block_time: Option<Timestamp>,
     ) -> Vec<Box<dyn UnifiedEvent>> {
         self.inner
-            .parse_events_from_inner_instruction(inner_instruction, signature, slot)
+            .parse_events_from_inner_instruction(inner_instruction, signature, slot, block_time)
     }
 
     fn parse_events_from_instruction(
@@ -155,9 +157,10 @@ impl EventParser for RaydiumClmmEventParser {
         accounts: &[Pubkey],
         signature: &str,
         slot: u64,
+        block_time: Option<Timestamp>,
     ) -> Vec<Box<dyn UnifiedEvent>> {
         self.inner
-            .parse_events_from_instruction(instruction, accounts, signature, slot)
+            .parse_events_from_instruction(instruction, accounts, signature, slot, block_time)
     }
 
     fn should_handle(&self, program_id: &Pubkey) -> bool {

src/streaming/event_parser/protocols/raydium_cpmm/parser.rs

@@ -1,3 +1,4 @@
+use prost_types::Timestamp;
 use solana_sdk::{instruction::CompiledInstruction, pubkey::Pubkey};
 use solana_transaction_status::UiCompiledInstruction;
 
@@ -133,9 +134,10 @@ impl EventParser for RaydiumCpmmEventParser {
         inner_instruction: &UiCompiledInstruction,
         signature: &str,
         slot: u64,
+        block_time: Option<Timestamp>,
     ) -> Vec<Box<dyn UnifiedEvent>> {
         self.inner
-            .parse_events_from_inner_instruction(inner_instruction, signature, slot)
+            .parse_events_from_inner_instruction(inner_instruction, signature, slot, block_time)
     }
 
     fn parse_events_from_instruction(
@@ -144,9 +146,10 @@ impl EventParser for RaydiumCpmmEventParser {
         accounts: &[Pubkey],
         signature: &str,
         slot: u64,
+        block_time: Option<Timestamp>,
     ) -> Vec<Box<dyn UnifiedEvent>> {
         self.inner
-            .parse_events_from_instruction(instruction, accounts, signature, slot)
+            .parse_events_from_instruction(instruction, accounts, signature, slot, block_time)
     }
 
     fn should_handle(&self, program_id: &Pubkey) -> bool {

src/streaming/shred_stream.rs

@@ -106,6 +106,7 @@ impl ShredStreamGrpc {
                     &versioned_tx,
                     &signature.to_string(),
                     Some(slot),
+                    None,
                     bot_wallet.clone(),
                 )
                 .await

src/streaming/yellowstone_grpc.rs

@@ -3,6 +3,7 @@ use std::{collections::HashMap, fmt, time::Duration};
 use chrono::Local;
 use futures::{channel::mpsc, sink::Sink, SinkExt, Stream, StreamExt};
 use log::{error, info};
+use prost_types::Timestamp;
 use rustls::crypto::{ring::default_provider, CryptoProvider};
 use solana_sdk::{pubkey::Pubkey, signature::Signature};
 use solana_transaction_status::{EncodedTransactionWithStatusMeta, UiTransactionEncoding};
@@ -30,6 +31,7 @@ const MAX_DECODING_MESSAGE_SIZE: usize = 1024 * 1024 * 10;
 #[derive(Clone)]
 pub struct TransactionPretty {
     pub slot: u64,
+    pub block_time: Option<Timestamp>,
     pub signature: Signature,
     pub is_vote: bool,
     pub tx: EncodedTransactionWithStatusMeta,
@@ -54,11 +56,12 @@ impl fmt::Debug for TransactionPretty {
     }
 }
 
-impl From<SubscribeUpdateTransaction> for TransactionPretty {
+impl From<(SubscribeUpdateTransaction, Option<Timestamp>)> for TransactionPretty {
-    fn from(SubscribeUpdateTransaction { transaction, slot }: SubscribeUpdateTransaction) -> Self {
+    fn from((SubscribeUpdateTransaction { transaction, slot }, block_time): (SubscribeUpdateTransaction, Option<Timestamp>)) -> Self {
         let tx = transaction.expect("should be defined");
         Self {
             slot,
+            block_time: block_time,
             signature: Signature::try_from(tx.signature.as_slice()).expect("valid signature"),
             is_vote: tx.is_vote,
             tx: yellowstone_grpc_proto::convert_from::create_tx_with_meta(tx)
@@ -149,17 +152,12 @@ impl YellowstoneGrpc {
     pub async fn handle_stream_message(
         msg: SubscribeUpdate,
         tx: &mut mpsc::Sender<TransactionPretty>,
-        block: Option<&mut mpsc::Sender<SubscribeUpdateBlockMeta>>,
         subscribe_tx: &mut (impl Sink<SubscribeRequest, Error = mpsc::SendError> + Unpin),
     ) -> AnyResult<()> {
+        let created_at = msg.created_at;
         match msg.update_oneof {
-            Some(UpdateOneof::BlockMeta(sut)) => {
-                if let Some(block) = block {
-                    block.try_send(sut)?;
-                }
-            }
             Some(UpdateOneof::Transaction(sut)) => {
-                let transaction_pretty = TransactionPretty::from(sut.clone());
+                let transaction_pretty = TransactionPretty::from((sut, created_at));
                 tx.try_send(transaction_pretty)?;
             }
             Some(UpdateOneof::Ping(_)) => {
@@ -216,7 +214,6 @@ impl YellowstoneGrpc {
 
         // 创建通道
         let (mut tx, mut rx) = mpsc::channel::<TransactionPretty>(CHANNEL_SIZE);
-        let (mut block, mut rblock) = mpsc::channel::<SubscribeUpdateBlockMeta>(CHANNEL_SIZE);
 
         // 创建回调函数，使用 Arc 包装以便在多个任务中共享
         let callback = std::sync::Arc::new(Box::new(callback));
@@ -229,7 +226,6 @@ impl YellowstoneGrpc {
                         if let Err(e) = Self::handle_stream_message(
                             msg,
                             &mut tx,
-                            Some(&mut block),
                             &mut subscribe_tx,
                         )
                         .await
@@ -246,16 +242,12 @@ impl YellowstoneGrpc {
             }
         });
 
-        // 为交易处理和区块处理克隆 Arc<Box<F>>
-        let callback_tx = callback.clone();
-        let callback_block = callback;
-
         // 处理交易
         tokio::spawn(async move {
             while let Some(transaction_pretty) = rx.next().await {
                 if let Err(e) = Self::process_event_transaction(
                     transaction_pretty,
-                    &**callback_tx,
+                    &**callback,
                     bot_wallet,
                     protocols.clone(),
                 )
@@ -265,14 +257,7 @@ impl YellowstoneGrpc {
                 }
             }
         });
-        // 处理block
+        
-        tokio::spawn(async move {
-            while let Some(block) = rblock.next().await {
-                if let Err(e) = Self::process_block(block, &**callback_block).await {
-                    error!("Error processing block: {:?}", e);
-                }
-            }
-        });
         tokio::signal::ctrl_c().await?;
         Ok(())
     }
@@ -296,6 +281,7 @@ impl YellowstoneGrpc {
                     transaction_pretty.tx.clone(),
                     &signature,
                     Some(slot),
+                    transaction_pretty.block_time,
                     bot_wallet.clone(),
                 )
                 .await
@@ -307,14 +293,4 @@ impl YellowstoneGrpc {
 
         Ok(())
     }
-
-    /// 处理区块
-    async fn process_block<F>(block: SubscribeUpdateBlockMeta, callback: &F) -> AnyResult<()>
-    where
-        F: Fn(Box<dyn UnifiedEvent>) + Send + Sync,
-    {
-        let event = SDKSystemEventParser::parse_block(block);
-        callback(event);
-        Ok(())
-    }
 }

src/streaming/yellowstone_sub_system.rs

@@ -50,7 +50,7 @@ impl YellowstoneGrpc {
                 match message {
                     Ok(msg) => {
                         if let Err(e) =
-                            Self::handle_stream_message(msg, &mut tx, None, &mut subscribe_tx).await
+                            Self::handle_stream_message(msg, &mut tx, &mut subscribe_tx).await
                         {
                             error!("Error handling message: {:?}", e);
                             break;