diff --git a/Cargo.toml b/Cargo.toml index 7dfb5af..e9bba8f 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "solana-streamer-sdk" -version = "0.1.2" +version = "0.1.3" edition = "2021" authors = ["William ", "sgxiang ", "wei <1415121722@qq.com>"] repository = "https://github.com/0xfnzero/solana-streamer" @@ -60,4 +60,5 @@ hex = "0.4.3" bytemuck = { version = "1.4.0" } arrayref = "0.3.6" borsh-derive = "1.5.5" -indicatif = "0.17.11" \ No newline at end of file +indicatif = "0.17.11" +maplit = "1.0.2" diff --git a/README.md b/README.md index 77fdff7..c01041d 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.2" } +solana-streamer-sdk = { path = "./solana-streamer", version = "0.1.3" } ``` ### Use crates.io ```toml # Add to your Cargo.toml -solana-streamer-sdk = "0.1.2" +solana-streamer-sdk = "0.1.3" ``` ## Usage Examples @@ -87,7 +87,7 @@ async fn test_grpc() -> Result<(), Box> { ]; println!("Listening for events, press Ctrl+C to stop..."); - grpc.subscribe_events(protocols, None, None, None, callback) + grpc.subscribe_events(protocols, None, None, None, None, None, callback) .await?; Ok(()) @@ -117,6 +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| { println!("BonkPoolCreateEvent: {:?}", e.base_mint_param.symbol); }, diff --git a/README_CN.md b/README_CN.md index 93a7336..8959265 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.2" } +solana-streamer-sdk = { path = "./solana-streamer", version = "0.1.3" } ``` ### 使用 crates.io ```toml # 添加到您的 Cargo.toml -solana-streamer-sdk = "0.1.2" +solana-streamer-sdk = "0.1.3" ``` ## 使用示例 @@ -87,7 +87,7 @@ async fn test_grpc() -> Result<(), Box> { ]; println!("开始监听事件，按 Ctrl+C 停止..."); - grpc.subscribe_events(protocols, None, None, None, callback) + grpc.subscribe_events(protocols, None, None, None, None, None, callback) .await?; Ok(()) @@ -117,6 +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| { println!("BonkPoolCreateEvent: {:?}", e.base_mint_param.symbol); }, diff --git a/src/main.rs b/src/main.rs index 2c02bf5..0c7fcc7 100755 --- a/src/main.rs +++ b/src/main.rs @@ -2,6 +2,7 @@ use solana_streamer_sdk::{ match_event, streaming::{ event_parser::{ + core::traits::BlockMetaEvent, protocols::{ bonk::{BonkPoolCreateEvent, BonkTradeEvent}, pumpfun::{PumpFunCreateTokenEvent, PumpFunTradeEvent}, @@ -43,7 +44,7 @@ async fn test_grpc() -> Result<(), Box> { ]; println!("开始监听事件，按 Ctrl+C 停止..."); - grpc.subscribe_events(protocols, None, None, None, callback) + grpc.subscribe_events(protocols, None, None, None, None, None, callback) .await?; Ok(()) @@ -73,6 +74,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| { println!("BonkPoolCreateEvent: {:?}", e.base_mint_param.symbol); }, diff --git a/src/streaming/event_parser/common/types.rs b/src/streaming/event_parser/common/types.rs index d5497b8..a43fc2d 100755 --- a/src/streaming/event_parser/common/types.rs +++ b/src/streaming/event_parser/common/types.rs @@ -18,6 +18,7 @@ pub enum ProtocolType { Bonk, RaydiumCpmm, RaydiumClmm, + SDKSystem, } /// 事件类型枚举 @@ -54,6 +55,7 @@ pub enum EventType { RaydiumClmmSwapV2, // 通用事件 + SDKSystem, Unknown, } @@ -77,6 +79,7 @@ impl EventType { EventType::RaydiumCpmmSwapBaseOutput => "RaydiumCpmmSwapBaseOutput".to_string(), EventType::RaydiumClmmSwap => "RaydiumClmmSwap".to_string(), EventType::RaydiumClmmSwapV2 => "RaydiumClmmSwapV2".to_string(), + EventType::SDKSystem => "SDKSystem".to_string(), EventType::Unknown => "Unknown".to_string(), } } diff --git a/src/streaming/event_parser/core/traits.rs b/src/streaming/event_parser/core/traits.rs index 756d0cc..c9f235f 100755 --- a/src/streaming/event_parser/core/traits.rs +++ b/src/streaming/event_parser/core/traits.rs @@ -1,4 +1,6 @@ use anyhow::Result; +use borsh::BorshDeserialize; +use serde::{Deserialize, Serialize}; use solana_sdk::{ instruction::CompiledInstruction, pubkey::Pubkey, transaction::VersionedTransaction, }; @@ -7,7 +9,9 @@ use solana_transaction_status::{ }; 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, }; @@ -53,6 +57,18 @@ 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 { @@ -534,3 +550,26 @@ impl EventParser for GenericEventParser { vec![self.program_id] } } + +pub struct SDKSystemEventParser {} +impl SDKSystemEventParser { + pub fn parse_block(block: SubscribeUpdateBlockMeta) -> Box { + Box::new(BlockMetaEvent { + metadata: EventMetadata::new( + block.blockhash.to_string(), + "".to_string(), + block.slot, + 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 + }, + }) + } +} diff --git a/src/streaming/yellowstone_grpc.rs b/src/streaming/yellowstone_grpc.rs index 860172a..9b1eb70 100755 --- a/src/streaming/yellowstone_grpc.rs +++ b/src/streaming/yellowstone_grpc.rs @@ -13,9 +13,12 @@ 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; @@ -96,13 +99,22 @@ impl YellowstoneGrpc { pub async fn subscribe_with_request( &self, transactions: TransactionsFilterMap, + commitment: Option, ) -> AnyResult<( impl Sink, impl Stream>, )> { let subscribe_request = SubscribeRequest { transactions, - commitment: Some(CommitmentLevel::Processed.into()), + blocks_meta: hashmap! { + "".to_owned() => SubscribeRequestFilterBlocksMeta { + } + }, + commitment: if let Some(commitment) = commitment { + Some(commitment as i32) + } else { + Some(CommitmentLevel::Processed.into()) + }, ..Default::default() }; @@ -137,11 +149,17 @@ 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<()> { 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); + let transaction_pretty = TransactionPretty::from(sut.clone()); tx.try_send(transaction_pretty)?; } Some(UpdateOneof::Ping(_)) => { @@ -169,6 +187,7 @@ impl YellowstoneGrpc { account_include: Option>, account_exclude: Option>, account_required: Option>, + commitment: Option, callback: F, ) -> AnyResult<()> where @@ -184,28 +203,36 @@ impl YellowstoneGrpc { let mut account_include = account_include.unwrap_or_default(); let account_exclude = account_exclude.unwrap_or_default(); let account_required = account_required.unwrap_or_default(); - + account_include.extend(protocol_accounts.clone()); let transactions = self.get_subscribe_request_filter(account_include, account_exclude, account_required); // 订阅事件 - let (mut subscribe_tx, mut stream) = self.subscribe_with_request(transactions).await?; + let (mut subscribe_tx, mut stream) = self + .subscribe_with_request(transactions, commitment) + .await?; // 创建通道 let (mut tx, mut rx) = mpsc::channel::(CHANNEL_SIZE); + let (mut block, mut rblock) = mpsc::channel::(CHANNEL_SIZE); - // 创建回调函数 - let callback = Box::new(callback); + // 创建回调函数，使用 Arc 包装以便在多个任务中共享 + let callback = std::sync::Arc::new(Box::new(callback)); // 启动处理流的任务 tokio::spawn(async move { 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, + Some(&mut block), + &mut subscribe_tx, + ) + .await { error!("Error handling message: {:?}", e); break; @@ -219,20 +246,34 @@ impl YellowstoneGrpc { } }); - // 处理交易 - while let Some(transaction_pretty) = rx.next().await { - if let Err(e) = Self::process_event_transaction( - transaction_pretty, - &*callback, - bot_wallet, - protocols.clone(), - ) - .await - { - error!("Error processing transaction: {:?}", e); - } - } + // 为交易处理和区块处理克隆 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, + bot_wallet, + protocols.clone(), + ) + .await + { + error!("Error processing transaction: {:?}", e); + } + } + }); + // 处理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(()) } @@ -266,4 +307,14 @@ 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 f673e25..3fe6e2e 100755 --- a/src/streaming/yellowstone_sub_system.rs +++ b/src/streaming/yellowstone_sub_system.rs @@ -1,8 +1,11 @@ -use crate::{common::AnyResult, streaming::yellowstone_grpc::{TransactionPretty, YellowstoneGrpc}}; -use solana_program::pubkey; -use solana_sdk::{pubkey::Pubkey, transaction::VersionedTransaction}; +use crate::{ + common::AnyResult, + streaming::yellowstone_grpc::{TransactionPretty, YellowstoneGrpc}, +}; use futures::{channel::mpsc, StreamExt}; use log::error; +use solana_program::pubkey; +use solana_sdk::{pubkey::Pubkey, transaction::VersionedTransaction}; use solana_transaction_status::EncodedTransactionWithStatusMeta; const SYSTEM_PROGRAM_ID: Pubkey = pubkey!("11111111111111111111111111111111"); @@ -36,7 +39,8 @@ impl YellowstoneGrpc { let account_exclude = account_exclude.unwrap_or_default(); let transactions = self.get_subscribe_request_filter(account_include, account_exclude, addrs); - let (mut subscribe_tx, mut stream) = self.subscribe_with_request(transactions).await?; + let (mut subscribe_tx, mut stream) = + self.subscribe_with_request(transactions, None).await?; let (mut tx, mut rx) = mpsc::channel::(CHANNEL_SIZE); let callback = Box::new(callback); @@ -46,7 +50,7 @@ impl YellowstoneGrpc { match message { Ok(msg) => { if let Err(e) = - Self::handle_stream_message(msg, &mut tx, &mut subscribe_tx).await + Self::handle_stream_message(msg, &mut tx, None, &mut subscribe_tx).await { error!("Error handling message: {:?}", e); break;