feat: Upgrade SDK to version 0.1.4 and optimize event handling mechanism

1. Version upgrade: SDK version updated from 0.1.3 to 0.1.4
2. Remove BlockMetaEvent: Delete unnecessary block metadata event type and related code
3. Optimize event handling: Add block_time and block_time_ms fields to each event
4. Code optimization: Improve code formatting and structure
5. Example updates: Update example code and usage instructions in README

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "solana-streamer-sdk"
-version = "0.1.3"
+version = "0.1.4"
 edition = "2021"
 authors = ["William <byteblock6@gmail.com>", "sgxiang <sgxiang@gmail.com>", "wei <1415121722@qq.com>"]
 repository = "https://github.com/0xfnzero/solana-streamer"

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<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| {
+                // 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| {

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<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| {
+                // 使用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| {

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<dyn std::error::Error>> {
     test_grpc().await?;
-    // test_shreds().await?;
+    test_shreds().await?;
     Ok(())
 }
 
@@ -36,11 +35,11 @@ async fn test_grpc() -> Result<(), Box<dyn std::error::Error>> {
 
     let callback = create_event_callback();
     let protocols = vec![
-        // Protocol::PumpFun,
+        Protocol::PumpFun,
-        // Protocol::PumpSwap,
+        Protocol::PumpSwap,
         Protocol::Bonk,
-        // Protocol::RaydiumCpmm,
+        Protocol::RaydiumCpmm,
-        // Protocol::RaydiumClmm,
+        Protocol::RaydiumClmm,
     ];
 
     println!("开始监听事件，按 Ctrl+C 停止...");
@@ -74,11 +73,10 @@ 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);
+                // 使用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);

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<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 {
@@ -348,7 +331,8 @@ pub trait EventParser: Send + Sync {
         block_time: Option<Timestamp>,
     ) -> Result<Vec<Box<dyn UnifiedEvent>>> {
         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<Timestamp>,
     ) -> Result<Vec<Box<dyn UnifiedEvent>>> {
         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<Timestamp>,
     ) -> Option<Box<dyn UnifiedEvent>> {
-        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<Timestamp>,
     ) -> Option<Box<dyn UnifiedEvent>> {
-        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) =
+                    if let Some(event) = self
-                        self.parse_inner_instruction_event(config, data, signature, slot, block_time)
+                        .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 {
+impl SDKSystemEventParser {}
-    pub fn parse_block(block: SubscribeUpdateBlockMeta) -> Box<dyn UnifiedEvent> {
-        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
-            },
-        })
-    }
-}

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<String, SubscribeRequestFilterTransactions>;
 
@@ -57,7 +54,12 @@ impl fmt::Debug for TransactionPretty {
 }
 
 impl From<(SubscribeUpdateTransaction, Option<Timestamp>)> for TransactionPretty {
-    fn from((SubscribeUpdateTransaction { transaction, slot }, block_time): (SubscribeUpdateTransaction, Option<Timestamp>)) -> Self {
+    fn from(
+        (SubscribeUpdateTransaction { transaction, slot }, block_time): (
+            SubscribeUpdateTransaction,
+            Option<Timestamp>,
+        ),
+    ) -> 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(
+                        if let Err(e) =
-                            msg,
+                            Self::handle_stream_message(msg, &mut tx, &mut subscribe_tx).await
-                            &mut tx,
-                            &mut subscribe_tx,
-                        )
-                        .await
                         {
                             error!("Error handling message: {:?}", e);
                             break;