feat: Add block metadata event support

- Implement BlockMetaEvent and SDKSystemEventParser
- Support block metadata subscription and processing
- Optimize event handling with separate tokio tasks
- Add commitment parameter support
- Upgrade version to 0.1.3

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "solana-streamer-sdk"
-version = "0.1.2"
+version = "0.1.3"
 edition = "2021"
 authors = ["William <byteblock6@gmail.com>", "sgxiang <sgxiang@gmail.com>", "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"
+indicatif = "0.17.11"
+maplit = "1.0.2"

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<dyn std::error::Error>> {
     ];
 
     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<dyn std::error::Error>> {
 fn create_event_callback() -> impl Fn(Box<dyn UnifiedEvent>) {
     |event: Box<dyn UnifiedEvent>| {
         match_event!(event, {
+            BlockMetaEvent => |e: BlockMetaEvent| {
+                println!("BlockMetaEvent: {:?}", e.slot);
+            },
             BonkPoolCreateEvent => |e: BonkPoolCreateEvent| {
                 println!("BonkPoolCreateEvent: {:?}", e.base_mint_param.symbol);
             },

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<dyn std::error::Error>> {
     ];
 
     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<dyn std::error::Error>> {
 fn create_event_callback() -> impl Fn(Box<dyn UnifiedEvent>) {
     |event: Box<dyn UnifiedEvent>| {
         match_event!(event, {
+            BlockMetaEvent => |e: BlockMetaEvent| {
+                println!("BlockMetaEvent: {:?}", e.slot);
+            },
             BonkPoolCreateEvent => |e: BonkPoolCreateEvent| {
                 println!("BonkPoolCreateEvent: {:?}", e.base_mint_param.symbol);
             },

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<dyn std::error::Error>> {
     ];
 
     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<dyn std::error::Error>> {
 fn create_event_callback() -> impl Fn(Box<dyn UnifiedEvent>) {
     |event: Box<dyn UnifiedEvent>| {
         match_event!(event, {
+            BlockMetaEvent => |e: BlockMetaEvent| {
+                println!("BlockMetaEvent: {:?}", e.slot);
+            },
             BonkPoolCreateEvent => |e: BonkPoolCreateEvent| {
                 println!("BonkPoolCreateEvent: {:?}", e.base_mint_param.symbol);
             },

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(),
         }
     }

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<TransferData>);
 }
 
+/// 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<dyn UnifiedEvent> {
+        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
+            },
+        })
+    }
+}

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<String, SubscribeRequestFilterTransactions>;
 
@@ -96,13 +99,22 @@ impl YellowstoneGrpc {
     pub async fn subscribe_with_request(
         &self,
         transactions: TransactionsFilterMap,
+        commitment: Option<CommitmentLevel>,
     ) -> AnyResult<(
         impl Sink<SubscribeRequest, Error = mpsc::SendError>,
         impl Stream<Item = Result<SubscribeUpdate, Status>>,
     )> {
         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<TransactionPretty>,
+        block: Option<&mut mpsc::Sender<SubscribeUpdateBlockMeta>>,
         subscribe_tx: &mut (impl Sink<SubscribeRequest, Error = mpsc::SendError> + 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<Vec<String>>,
         account_exclude: Option<Vec<String>>,
         account_required: Option<Vec<String>>,
+        commitment: Option<CommitmentLevel>,
         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::<TransactionPretty>(CHANNEL_SIZE);
+        let (mut block, mut rblock) = mpsc::channel::<SubscribeUpdateBlockMeta>(CHANNEL_SIZE);
 
-        // 创建回调函数
+        // 创建回调函数，使用 Arc 包装以便在多个任务中共享
-        let callback = Box::new(callback);
+        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) =
+                        if let Err(e) = Self::handle_stream_message(
-                            Self::handle_stream_message(msg, &mut tx, &mut subscribe_tx).await
+                            msg,
+                            &mut tx,
+                            Some(&mut block),
+                            &mut subscribe_tx,
+                        )
+                        .await
                         {
                             error!("Error handling message: {:?}", e);
                             break;
@@ -219,20 +246,34 @@ impl YellowstoneGrpc {
             }
         });
 
-        // 处理交易
+        // 为交易处理和区块处理克隆 Arc<Box<F>>
-        while let Some(transaction_pretty) = rx.next().await {
+        let callback_tx = callback.clone();
-            if let Err(e) = Self::process_event_transaction(
+        let callback_block = callback;
-                transaction_pretty,
-                &*callback,
-                bot_wallet,
-                protocols.clone(),
-            )
-            .await
-            {
-                error!("Error processing transaction: {:?}", e);
-            }
-        }
 
+        // 处理交易
+        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<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

@@ -1,8 +1,11 @@
-use crate::{common::AnyResult, streaming::yellowstone_grpc::{TransactionPretty, YellowstoneGrpc}};
+use crate::{
-use solana_program::pubkey;
+    common::AnyResult,
-use solana_sdk::{pubkey::Pubkey, transaction::VersionedTransaction};
+    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::<TransactionPretty>(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;