Merge branch 'main' of github.com:0xfnzero/solana-streamer into main

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "solana-streamer-sdk"
-version = "0.1.0"
+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

@@ -20,6 +20,8 @@ A lightweight Rust library for real-time event streaming from Solana DEX trading
 
 ## Installation
 
+### Direct Clone
+
 Clone this project to your project directory:
 
 ```bash
@@ -31,7 +33,14 @@ Add the dependency to your `Cargo.toml`:
 
 ```toml
 # Add to your Cargo.toml
-solana-streamer-sdk = { path = "./solana-streamer", version = "0.1.0" }
+solana-streamer-sdk = { path = "./solana-streamer", version = "0.1.3" }
+```
+
+### Use crates.io
+
+```toml
+# Add to your Cargo.toml
+solana-streamer-sdk = "0.1.3"
 ```
 
 ## Usage Examples
@@ -78,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(())
@@ -108,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

@@ -20,6 +20,8 @@
 
 ## 安装
 
+### 直接克隆
+
 将项目克隆到您的项目目录：
 
 ```bash
@@ -31,7 +33,14 @@ git clone https://github.com/0xfnzero/solana-streamer
 
 ```toml
 # 添加到您的 Cargo.toml
-solana-streamer-sdk = { path = "./solana-streamer", version = "0.1.0" }
+solana-streamer-sdk = { path = "./solana-streamer", version = "0.1.3" }
+```
+
+### 使用 crates.io
+
+```toml
+# 添加到您的 Cargo.toml
+solana-streamer-sdk = "0.1.3"
 ```
 
 ## 使用示例
@@ -78,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(())
@@ -108,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/mod.rs

@@ -46,6 +46,10 @@ macro_rules! impl_unified_event {
                     )*
                 }
             }
+
+            fn set_transfer_datas(&mut self, transfer_datas: Vec<$crate::streaming::event_parser::common::types::TransferData>) {
+                self.metadata.transfer_datas = transfer_datas;
+            }
         }
     };
 }

src/streaming/event_parser/common/types.rs

@@ -1,6 +1,10 @@
 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 std::collections::hash_map::DefaultHasher;
 use std::hash::{Hash, Hasher};
 
@@ -14,6 +18,7 @@ pub enum ProtocolType {
     Bonk,
     RaydiumCpmm,
     RaydiumClmm,
+    SDKSystem,
 }
 
 /// 事件类型枚举
@@ -50,6 +55,7 @@ pub enum EventType {
     RaydiumClmmSwapV2,
 
     // 通用事件
+    SDKSystem,
     Unknown,
 }
 
@@ -73,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(),
         }
     }
@@ -129,6 +136,20 @@ impl ProtocolInfo {
     }
 }
 
+/// 交易数据
+#[derive(
+    Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, BorshSerialize, BorshDeserialize,
+)]
+pub struct TransferData {
+    pub token_program: Pubkey,
+    pub source: Pubkey,
+    pub destination: Pubkey,
+    pub authority: Option<Pubkey>,
+    pub amount: u64,
+    pub decimals: Option<u8>,
+    pub mint: Option<Pubkey>,
+}
+
 /// 事件元数据
 #[derive(
     Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, BorshSerialize, BorshDeserialize,
@@ -141,6 +162,7 @@ pub struct EventMetadata {
     pub protocol: ProtocolType,
     pub event_type: EventType,
     pub program_id: Pubkey,
+    pub transfer_datas: Vec<TransferData>,
 }
 
 impl EventMetadata {
@@ -160,6 +182,7 @@ impl EventMetadata {
             protocol,
             event_type,
             program_id,
+            transfer_datas: vec![],
         }
     }
     pub fn set_id(&mut self, id: String) {
@@ -171,3 +194,123 @@ impl EventMetadata {
         self.id = format!("{:x}", hash_value);
     }
 }
+
+/// 解析接下来指令中的token转账数据
+pub fn parse_transfer_datas_from_next_instructions(
+    inner_instruction: &solana_transaction_status::UiInnerInstructions,
+    current_index: i8,
+    accounts: &[Pubkey],
+    event_type: EventType,
+) -> Vec<TransferData> {
+    let take = match event_type {
+        EventType::PumpFunBuy => 4,
+        EventType::PumpFunSell => 1,
+        EventType::PumpSwapBuy => 3,
+        EventType::PumpSwapSell => 3,
+        EventType::BonkBuyExactIn
+        | EventType::BonkBuyExactOut
+        | EventType::BonkSellExactIn
+        | EventType::BonkSellExactOut => 3,
+        EventType::RaydiumCpmmSwapBaseInput
+        | EventType::RaydiumCpmmSwapBaseOutput
+        | EventType::RaydiumClmmSwap
+        | EventType::RaydiumClmmSwapV2 => 2,
+        _ => 0,
+    };
+    if take == 0 {
+        return vec![];
+    }
+    let mut transfer_datas = vec![];
+    // 获取当前指令之后的两个指令
+    let next_instructions: Vec<&UiInstruction> = inner_instruction
+        .instructions
+        .iter()
+        .skip((current_index + 1) as usize)
+        .take(take)
+        .collect();
+
+    for instruction in next_instructions {
+        if let UiInstruction::Compiled(compiled) = instruction {
+            if let Ok(data) = bs58::decode(compiled.data.clone()).into_vec() {
+                // Token Program: transferChecked
+                // Token 2022 Program: transferChecked
+                if data[0] == 12 {
+                    let account_pubkeys: Vec<Pubkey> = compiled
+                        .accounts
+                        .iter()
+                        .map(|a| accounts[*a as usize])
+                        .collect();
+                    if account_pubkeys.len() < 4 {
+                        continue;
+                    }
+                    let (source, mint, destination, authority) = (
+                        account_pubkeys[0],
+                        account_pubkeys[1],
+                        account_pubkeys[2],
+                        account_pubkeys[3],
+                    );
+                    let amount = u64::from_le_bytes(data[1..9].try_into().unwrap());
+                    let decimals = data[9];
+                    let token_program = accounts[compiled.program_id_index as usize];
+                    transfer_datas.push(TransferData {
+                        amount,
+                        decimals: Some(decimals),
+                        mint: Some(mint),
+                        source,
+                        destination,
+                        authority: Some(authority),
+                        token_program,
+                    });
+                }
+                // Token Program: transfer
+                else if data[0] == 3 {
+                    let account_pubkeys: Vec<Pubkey> = compiled
+                        .accounts
+                        .iter()
+                        .map(|a| accounts[*a as usize])
+                        .collect();
+                    if account_pubkeys.len() < 3 {
+                        continue;
+                    }
+                    let (source, destination, authority) =
+                        (account_pubkeys[0], account_pubkeys[1], account_pubkeys[2]);
+                    let amount = u64::from_le_bytes(data[1..9].try_into().unwrap());
+                    let token_program = accounts[compiled.program_id_index as usize];
+                    transfer_datas.push(TransferData {
+                        amount,
+                        decimals: None,
+                        mint: None,
+                        source,
+                        destination,
+                        authority: Some(authority),
+                        token_program,
+                    });
+                }
+                //System Program: transfer
+                else if data[0] == 2 {
+                    let account_pubkeys: Vec<Pubkey> = compiled
+                        .accounts
+                        .iter()
+                        .map(|a| accounts[*a as usize])
+                        .collect();
+                    if account_pubkeys.len() < 2 {
+                        continue;
+                    }
+                    let (source, destination) = (account_pubkeys[0], account_pubkeys[1]);
+                    let amount = u64::from_le_bytes(data[4..12].try_into().unwrap());
+                    let token_program = accounts[compiled.program_id_index as usize];
+                    transfer_datas.push(TransferData {
+                        amount,
+                        decimals: None,
+                        mint: None,
+                        source,
+                        destination,
+                        authority: None,
+                        token_program,
+                    });
+                }
+            }
+        }
+    }
+    transfer_datas
+}

src/streaming/event_parser/core/traits.rs

@@ -1,13 +1,20 @@
 use anyhow::Result;
+use borsh::BorshDeserialize;
+use serde::{Deserialize, Serialize};
 use solana_sdk::{
     instruction::CompiledInstruction, pubkey::Pubkey, transaction::VersionedTransaction,
 };
 use solana_transaction_status::{
-    EncodedTransactionWithStatusMeta, UiCompiledInstruction, UiInstruction,
+    EncodedTransactionWithStatusMeta, UiCompiledInstruction, UiInnerInstructions, UiInstruction,
 };
 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,
+};
 use crate::streaming::event_parser::{
     common::{utils::*, EventMetadata, EventType, ProtocolType},
     protocols::{
@@ -46,8 +53,22 @@ pub trait UnifiedEvent: Debug + Send + Sync {
     fn merge(&mut self, _other: Box<dyn UnifiedEvent>) {
         // 默认实现：不进行任何合并操作
     }
+
+    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 {
@@ -75,6 +96,7 @@ pub trait EventParser: Send + Sync {
         signature: &str,
         slot: Option<u64>,
         accounts: &[Pubkey],
+        inner_instructions: &[UiInnerInstructions],
     ) -> Result<Vec<Box<dyn UnifiedEvent>>> {
         let mut instruction_events = Vec::new();
         // 获取交易的指令和账户
@@ -85,7 +107,7 @@ pub trait EventParser: Send + Sync {
         let has_program = accounts.iter().any(|account| self.should_handle(account));
         if has_program {
             // 解析每个指令
-            for instruction in compiled_instructions {
+            for (index, instruction) in compiled_instructions.iter().enumerate() {
                 if let Some(program_id) = accounts.get(instruction.program_id_index as usize) {
                     if self.should_handle(program_id) {
                         let max_idx = instruction.accounts.iter().max().unwrap_or(&0);
@@ -95,11 +117,29 @@ pub trait EventParser: Send + Sync {
                                 accounts.push(Pubkey::default());
                             }
                         }
-                        if let Ok(events) = self
+                        if let Ok(mut events) = self
                             .parse_instruction(instruction, &accounts, signature, slot)
                             .await
                         {
-                            instruction_events.extend(events);
+                            if events.len() > 0 {
+                                if let Some(inn) =
+                                    inner_instructions.iter().find(|inner_instruction| {
+                                        inner_instruction.index == index as u8
+                                    })
+                                {
+                                    events.iter_mut().for_each(|event| {
+                                        let transfer_datas =
+                                            parse_transfer_datas_from_next_instructions(
+                                                &inn,
+                                                -1 as i8,
+                                                &accounts,
+                                                event.event_type(),
+                                            );
+                                        event.set_transfer_datas(transfer_datas.clone());
+                                    });
+                                }
+                                instruction_events.extend(events);
+                            }
                         }
                     }
                 }
@@ -122,6 +162,7 @@ pub trait EventParser: Send + Sync {
                 signature,
                 slot,
                 &accounts,
+                &vec![],
             )
             .await
             .unwrap_or_else(|_e| vec![]);
@@ -143,7 +184,9 @@ pub trait EventParser: Send + Sync {
             .ok_or_else(|| anyhow::anyhow!("Missing transaction metadata"))?;
 
         let mut address_table_lookups: Vec<Pubkey> = vec![];
+        let mut inner_instructions: Vec<UiInnerInstructions> = vec![];
         if meta.err.is_none() {
+            inner_instructions = meta.inner_instructions.as_ref().unwrap().clone();
             let loaded_addresses = meta.loaded_addresses.as_ref().unwrap();
             for lookup in &loaded_addresses.writable {
                 address_table_lookups.push(Pubkey::from_str(lookup).unwrap());
@@ -167,6 +210,7 @@ pub trait EventParser: Send + Sync {
                     signature,
                     slot,
                     &accounts,
+                    &inner_instructions,
                 )
                 .await
                 .unwrap_or_else(|_e| vec![]);
@@ -178,9 +222,8 @@ pub trait EventParser: Send + Sync {
         let mut inner_instruction_events = Vec::new();
         // 检查交易是否成功
         if meta.err.is_none() {
-            let inner_instructions = meta.inner_instructions.as_ref().unwrap();
             for inner_instruction in inner_instructions {
-                for instruction in &inner_instruction.instructions {
+                for (index, instruction) in inner_instruction.instructions.iter().enumerate() {
                     match instruction {
                         UiInstruction::Compiled(compiled) => {
                             // 解析嵌套指令
@@ -189,7 +232,7 @@ pub trait EventParser: Send + Sync {
                                 accounts: compiled.accounts.clone(),
                                 data: bs58::decode(compiled.data.clone()).into_vec().unwrap(),
                             };
-                            if let Ok(events) = self
+                            if let Ok(mut events) = self
                                 .parse_instruction(
                                     &compiled_instruction,
                                     &accounts,
@@ -198,13 +241,37 @@ pub trait EventParser: Send + Sync {
                                 )
                                 .await
                             {
-                                instruction_events.extend(events);
+                                if events.len() > 0 {
+                                    events.iter_mut().for_each(|event| {
+                                        let transfer_datas =
+                                            parse_transfer_datas_from_next_instructions(
+                                                &inner_instruction,
+                                                index as i8,
+                                                &accounts,
+                                                event.event_type(),
+                                            );
+                                        event.set_transfer_datas(transfer_datas.clone());
+                                    });
+                                    instruction_events.extend(events);
+                                }
                             }
-                            if let Ok(events) = self
+                            if let Ok(mut events) = self
                                 .parse_inner_instruction(compiled, signature, slot)
                                 .await
                             {
-                                inner_instruction_events.extend(events);
+                                if events.len() > 0 {
+                                    events.iter_mut().for_each(|event| {
+                                        let transfer_datas =
+                                            parse_transfer_datas_from_next_instructions(
+                                                &inner_instruction,
+                                                index as i8,
+                                                &accounts,
+                                                event.event_type(),
+                                            );
+                                        event.set_transfer_datas(transfer_datas.clone());
+                                    });
+                                    inner_instruction_events.extend(events);
+                                }
                             }
                         }
                         _ => {}
@@ -483,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(_)) => {
@@ -168,6 +186,8 @@ impl YellowstoneGrpc {
         bot_wallet: Option<Pubkey>,
         account_include: Option<Vec<String>>,
         account_exclude: Option<Vec<String>>,
+        account_required: Option<Vec<String>>,
+        commitment: Option<CommitmentLevel>,
         callback: F,
     ) -> AnyResult<()>
     where
@@ -182,27 +202,37 @@ impl YellowstoneGrpc {
             .collect::<Vec<String>>();
         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, vec![]);
+            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;
@@ -216,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(())
     }
 
@@ -263,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;