diff --git a/Cargo.toml b/Cargo.toml index 3bc410e..bb4ce0e 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "solana-streamer-sdk" -version = "0.1.3" +version = "0.1.4" edition = "2021" authors = ["William ", "sgxiang ", "wei <1415121722@qq.com>"] repository = "https://github.com/0xfnzero/solana-streamer" diff --git a/README.md b/README.md index c01041d..be22147 100755 --- a/README.md +++ b/README.md @@ -33,14 +33,14 @@ Add the dependency to your `Cargo.toml`: ```toml # Add to your Cargo.toml -solana-streamer-sdk = { path = "./solana-streamer", version = "0.1.3" } +solana-streamer-sdk = { path = "./solana-streamer", version = "0.1.4" } ``` ### Use crates.io ```toml # Add to your Cargo.toml -solana-streamer-sdk = "0.1.3" +solana-streamer-sdk = "0.1.4" ``` ## Usage Examples @@ -117,10 +117,9 @@ async fn test_shreds() -> Result<(), Box> { fn create_event_callback() -> impl Fn(Box) { |event: Box| { match_event!(event, { - BlockMetaEvent => |e: BlockMetaEvent| { - println!("BlockMetaEvent: {:?}", e.slot); - }, BonkPoolCreateEvent => |e: BonkPoolCreateEvent| { + // When using grpc, you can get block_time from each event + println!("block_time: {:?}, block_time_ms: {:?}", e.metadata.block_time, e.metadata.block_time_ms); println!("BonkPoolCreateEvent: {:?}", e.base_mint_param.symbol); }, BonkTradeEvent => |e: BonkTradeEvent| { diff --git a/README_CN.md b/README_CN.md index 8959265..0fceed8 100644 --- a/README_CN.md +++ b/README_CN.md @@ -33,14 +33,14 @@ git clone https://github.com/0xfnzero/solana-streamer ```toml # 添加到您的 Cargo.toml -solana-streamer-sdk = { path = "./solana-streamer", version = "0.1.3" } +solana-streamer-sdk = { path = "./solana-streamer", version = "0.1.4" } ``` ### 使用 crates.io ```toml # 添加到您的 Cargo.toml -solana-streamer-sdk = "0.1.3" +solana-streamer-sdk = "0.1.4" ``` ## 使用示例 @@ -117,10 +117,9 @@ async fn test_shreds() -> Result<(), Box> { fn create_event_callback() -> impl Fn(Box) { |event: Box| { match_event!(event, { - BlockMetaEvent => |e: BlockMetaEvent| { - println!("BlockMetaEvent: {:?}", e.slot); - }, BonkPoolCreateEvent => |e: BonkPoolCreateEvent| { + // 使用grpc的时候，可以从每个事件中获取到block_time + println!("block_time: {:?}, block_time_ms: {:?}", e.metadata.block_time, e.metadata.block_time_ms); println!("BonkPoolCreateEvent: {:?}", e.base_mint_param.symbol); }, BonkTradeEvent => |e: BonkTradeEvent| { diff --git a/src/main.rs b/src/main.rs index 9d4f764..cb674c4 100755 --- a/src/main.rs +++ b/src/main.rs @@ -2,7 +2,6 @@ use solana_streamer_sdk::{ match_event, streaming::{ event_parser::{ - core::traits::BlockMetaEvent, protocols::{ bonk::{BonkPoolCreateEvent, BonkTradeEvent}, pumpfun::{PumpFunCreateTokenEvent, PumpFunTradeEvent}, @@ -22,7 +21,7 @@ use solana_streamer_sdk::{ #[tokio::main] async fn main() -> Result<(), Box> { test_grpc().await?; - // test_shreds().await?; + test_shreds().await?; Ok(()) } @@ -36,11 +35,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 停止..."); @@ -74,11 +73,10 @@ async fn test_shreds() -> Result<(), Box> { fn create_event_callback() -> impl Fn(Box) { |event: Box| { match_event!(event, { - BlockMetaEvent => |e: BlockMetaEvent| { - // println!("BlockMetaEvent: {:?}", e.slot); - }, BonkPoolCreateEvent => |e: BonkPoolCreateEvent| { - // println!("BonkPoolCreateEvent: {:?}", e.base_mint_param.symbol); + // 使用grpc的时候，可以从每个事件中获取到block_time + println!("block_time: {:?}, block_time_ms: {:?}", e.metadata.block_time, e.metadata.block_time_ms); + println!("BonkPoolCreateEvent: {:?}", e.base_mint_param.symbol); }, BonkTradeEvent => |e: BonkTradeEvent| { println!("BonkTradeEvent: {:?}", e); diff --git a/src/streaming/event_parser/core/traits.rs b/src/streaming/event_parser/core/traits.rs index b656c4f..2bee766 100755 --- a/src/streaming/event_parser/core/traits.rs +++ b/src/streaming/event_parser/core/traits.rs @@ -1,19 +1,14 @@ use anyhow::Result; -use borsh::BorshDeserialize; use prost_types::Timestamp; -use serde::{Deserialize, Serialize}; use solana_sdk::{ instruction::CompiledInstruction, pubkey::Pubkey, transaction::VersionedTransaction, }; 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; -use crate::impl_unified_event; use crate::streaming::event_parser::common::{ parse_transfer_datas_from_next_instructions, TransferData, }; @@ -59,18 +54,6 @@ pub trait UnifiedEvent: Debug + Send + Sync { fn set_transfer_datas(&mut self, transfer_datas: Vec); } -/// block meta 事件 -#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] -pub struct BlockMetaEvent { - pub metadata: EventMetadata, - pub slot: u64, - pub blockhash: String, - pub block_time: i64, -} - -// 使用宏生成UnifiedEvent实现，指定需要合并的字段 -impl_unified_event!(BlockMetaEvent,); - /// 事件解析器trait - 定义了事件解析的核心方法 #[async_trait::async_trait] pub trait EventParser: Send + Sync { @@ -348,7 +331,8 @@ pub trait EventParser: Send + Sync { block_time: Option, ) -> Result>> { let slot = slot.unwrap_or(0); - let events = self.parse_events_from_inner_instruction(instruction, signature, slot, block_time); + let events = + self.parse_events_from_inner_instruction(instruction, signature, slot, block_time); Ok(events) } @@ -361,7 +345,8 @@ pub trait EventParser: Send + Sync { block_time: Option, ) -> Result>> { let slot = slot.unwrap_or(0); - let events = self.parse_events_from_instruction(instruction, accounts, signature, slot, block_time); + let events = + self.parse_events_from_instruction(instruction, accounts, signature, slot, block_time); Ok(events) } @@ -443,7 +428,10 @@ impl GenericEventParser { slot: u64, block_time: Option, ) -> Option> { - let timestamp = block_time.unwrap_or(Timestamp { seconds: 0, nanos: 0 }); + 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(), @@ -468,7 +456,10 @@ impl GenericEventParser { slot: u64, block_time: Option, ) -> Option> { - let timestamp = block_time.unwrap_or(Timestamp { seconds: 0, nanos: 0 }); + 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(), @@ -507,8 +498,8 @@ impl EventParser for GenericEventParser { for (disc, configs) in &self.inner_instruction_configs { 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, block_time) + if let Some(event) = self + .parse_inner_instruction_event(config, data, signature, slot, block_time) { events.push(event); } @@ -577,27 +568,4 @@ 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(), - ), - slot: block.slot, - blockhash: block.blockhash.to_string(), - block_time: if let Some(block_time) = block.block_time { - block_time.timestamp - } else { - 0 - }, - }) - } -} +impl SDKSystemEventParser {} diff --git a/src/streaming/yellowstone_grpc.rs b/src/streaming/yellowstone_grpc.rs index ac91842..f6004d1 100755 --- a/src/streaming/yellowstone_grpc.rs +++ b/src/streaming/yellowstone_grpc.rs @@ -14,12 +14,9 @@ use yellowstone_grpc_proto::geyser::{ SubscribeRequestFilterTransactions, SubscribeRequestPing, SubscribeUpdate, SubscribeUpdateTransaction, }; -use yellowstone_grpc_proto::geyser::{SubscribeRequestFilterBlocksMeta, SubscribeUpdateBlockMeta}; use crate::common::AnyResult; -use crate::streaming::event_parser::core::traits::SDKSystemEventParser; use crate::streaming::event_parser::{EventParserFactory, Protocol, UnifiedEvent}; -use maplit::hashmap; type TransactionsFilterMap = HashMap; @@ -57,7 +54,12 @@ impl fmt::Debug for TransactionPretty { } impl From<(SubscribeUpdateTransaction, Option)> for TransactionPretty { - fn from((SubscribeUpdateTransaction { transaction, slot }, block_time): (SubscribeUpdateTransaction, Option)) -> Self { + fn from( + (SubscribeUpdateTransaction { transaction, slot }, block_time): ( + SubscribeUpdateTransaction, + Option, + ), + ) -> Self { let tx = transaction.expect("should be defined"); Self { slot, @@ -109,10 +111,6 @@ impl YellowstoneGrpc { )> { let subscribe_request = SubscribeRequest { transactions, - blocks_meta: hashmap! { - "".to_owned() => SubscribeRequestFilterBlocksMeta { - } - }, commitment: if let Some(commitment) = commitment { Some(commitment as i32) } else { @@ -223,12 +221,8 @@ impl YellowstoneGrpc { while let Some(message) = stream.next().await { match message { Ok(msg) => { - if let Err(e) = Self::handle_stream_message( - msg, - &mut tx, - &mut subscribe_tx, - ) - .await + if let Err(e) = + Self::handle_stream_message(msg, &mut tx, &mut subscribe_tx).await { error!("Error handling message: {:?}", e); break; @@ -257,7 +251,7 @@ impl YellowstoneGrpc { } } }); - + tokio::signal::ctrl_c().await?; Ok(()) }