diff --git a/src/main.rs b/src/main.rs index 0c7fcc7..9d4f764 100755 --- a/src/main.rs +++ b/src/main.rs @@ -22,7 +22,7 @@ use solana_streamer_sdk::{ #[tokio::main] async fn main() -> Result<(), Box> { test_grpc().await?; - test_shreds().await?; + // test_shreds().await?; Ok(()) } @@ -36,11 +36,11 @@ async fn test_grpc() -> Result<(), Box> { let callback = create_event_callback(); let protocols = vec![ - Protocol::PumpFun, - Protocol::PumpSwap, + // Protocol::PumpFun, + // Protocol::PumpSwap, Protocol::Bonk, - Protocol::RaydiumCpmm, - Protocol::RaydiumClmm, + // Protocol::RaydiumCpmm, + // Protocol::RaydiumClmm, ]; println!("开始监听事件,按 Ctrl+C 停止..."); @@ -75,10 +75,10 @@ fn create_event_callback() -> impl Fn(Box) { |event: Box| { 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); diff --git a/src/streaming/event_parser/common/types.rs b/src/streaming/event_parser/common/types.rs index a43fc2d..28ccfb8 100755 --- a/src/streaming/event_parser/common/types.rs +++ b/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::{ - UiCompiledInstruction, UiInstruction, UiTransactionStatusMeta, UiTransactionTokenBalance, -}; +use solana_transaction_status::UiInstruction; 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, diff --git a/src/streaming/event_parser/core/traits.rs b/src/streaming/event_parser/core/traits.rs index c9f235f..b656c4f 100755 --- a/src/streaming/event_parser/core/traits.rs +++ b/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, ) -> Vec>; /// 从指令中解析事件数据 @@ -87,6 +90,7 @@ pub trait EventParser: Send + Sync { accounts: &[Pubkey], signature: &str, slot: u64, + block_time: Option, ) -> Vec>; /// 从VersionedTransaction中解析指令事件的通用方法 @@ -95,6 +99,7 @@ pub trait EventParser: Send + Sync { versioned_tx: &VersionedTransaction, signature: &str, slot: Option, + block_time: Option, accounts: &[Pubkey], inner_instructions: &[UiInnerInstructions], ) -> Result>> { @@ -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, + block_time: Option, bot_wallet: Option, ) -> Result>> { let accounts: Vec = 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, + block_time: Option, bot_wallet: Option, ) -> Result>> { 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, + block_time: Option, ) -> Result>> { 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, + block_time: Option, ) -> Result>> { 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, ) -> Option> { + 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, ) -> Option> { + 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, ) -> Vec> { 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, ) -> Vec> { 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 { + 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(), diff --git a/src/streaming/event_parser/protocols/bonk/parser.rs b/src/streaming/event_parser/protocols/bonk/parser.rs index 4e20a6f..855b09e 100755 --- a/src/streaming/event_parser/protocols/bonk/parser.rs +++ b/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, ) -> Vec> { 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, ) -> Vec> { 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 { diff --git a/src/streaming/event_parser/protocols/pumpfun/parser.rs b/src/streaming/event_parser/protocols/pumpfun/parser.rs index 4a19b7e..c277029 100755 --- a/src/streaming/event_parser/protocols/pumpfun/parser.rs +++ b/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, ) -> Vec> { 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, ) -> Vec> { 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 { diff --git a/src/streaming/event_parser/protocols/pumpswap/parser.rs b/src/streaming/event_parser/protocols/pumpswap/parser.rs index 52e56a2..64a9c62 100755 --- a/src/streaming/event_parser/protocols/pumpswap/parser.rs +++ b/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, ) -> Vec> { 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, ) -> Vec> { 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 { diff --git a/src/streaming/event_parser/protocols/raydium_clmm/parser.rs b/src/streaming/event_parser/protocols/raydium_clmm/parser.rs index 09bc0a6..501d9b5 100755 --- a/src/streaming/event_parser/protocols/raydium_clmm/parser.rs +++ b/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, ) -> Vec> { 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, ) -> Vec> { 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 { diff --git a/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs b/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs index 8b6984a..f983d09 100755 --- a/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs +++ b/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, ) -> Vec> { 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, ) -> Vec> { 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 { diff --git a/src/streaming/shred_stream.rs b/src/streaming/shred_stream.rs index 471cfdb..c36ed72 100755 --- a/src/streaming/shred_stream.rs +++ b/src/streaming/shred_stream.rs @@ -106,6 +106,7 @@ impl ShredStreamGrpc { &versioned_tx, &signature.to_string(), Some(slot), + None, bot_wallet.clone(), ) .await diff --git a/src/streaming/yellowstone_grpc.rs b/src/streaming/yellowstone_grpc.rs index 9b1eb70..ac91842 100755 --- a/src/streaming/yellowstone_grpc.rs +++ b/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, pub signature: Signature, pub is_vote: bool, pub tx: EncodedTransactionWithStatusMeta, @@ -54,11 +56,12 @@ impl fmt::Debug for TransactionPretty { } } -impl From for TransactionPretty { - fn from(SubscribeUpdateTransaction { transaction, slot }: SubscribeUpdateTransaction) -> Self { +impl From<(SubscribeUpdateTransaction, Option)> for TransactionPretty { + fn from((SubscribeUpdateTransaction { transaction, slot }, block_time): (SubscribeUpdateTransaction, Option)) -> 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, - block: Option<&mut mpsc::Sender>, subscribe_tx: &mut (impl Sink + 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::(CHANNEL_SIZE); - let (mut block, mut rblock) = mpsc::channel::(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> - 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(block: SubscribeUpdateBlockMeta, callback: &F) -> AnyResult<()> - where - F: Fn(Box) + Send + Sync, - { - let event = SDKSystemEventParser::parse_block(block); - callback(event); - Ok(()) - } } diff --git a/src/streaming/yellowstone_sub_system.rs b/src/streaming/yellowstone_sub_system.rs index 3fe6e2e..f8f924e 100755 --- a/src/streaming/yellowstone_sub_system.rs +++ b/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;