fix: route streamer grpc parsing through sol-parser-sdk

Cargo.toml

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

src/streaming/common/event_processor.rs

@@ -1,16 +1,21 @@
 use crate::common::AnyResult;
 use crate::streaming::common::MetricsEventType;
-use crate::streaming::event_parser::common::filter::{passes_event_type_filter, EventTypeFilter};
+use crate::streaming::event_parser::common::filter::{
+    build_sdk_parse_event_filter, passes_event_type_filter, EventTypeFilter,
+};
 use crate::streaming::event_parser::common::high_performance_clock::elapsed_micros_since;
-use crate::streaming::event_parser::core::account_event_parser::AccountEventParser;
 use crate::streaming::event_parser::core::common_event_parser::CommonEventParser;
 use crate::streaming::event_parser::core::event_parser::EventParser;
 use crate::streaming::event_parser::{core::traits::DexEvent, Protocol};
 use crate::streaming::grpc::{EventPretty, MetricsManager};
-use crate::streaming::parser_sdk_bridge::{parse_account_event_for_streamer, AccountParseResult};
+use crate::streaming::parser_sdk_bridge::{
+    adapt_parser_events_list, parse_account_event_for_streamer, AccountParseResult,
+};
 use crate::streaming::shred::TransactionWithSlot;
+use sol_parser_sdk::grpc::parse_subscribe_update_transaction_low_latency;
 use solana_sdk::pubkey::Pubkey;
 use std::sync::Arc;
+use yellowstone_grpc_proto::geyser::SubscribeUpdateTransaction;
 
 /// Wrap the user callback and update transaction metrics after delivery.
 #[inline]
@@ -56,46 +61,45 @@ pub async fn process_grpc_transaction(
                     return Ok(());
                 }
                 AccountParseResult::Filtered => return Ok(()),
-                AccountParseResult::Unsupported => {}
+                AccountParseResult::Unsupported => return Ok(()),
-            }
-
-            let account_event = AccountEventParser::parse_account_event(
-                protocols,
-                account_pretty,
-                event_type_filter,
-            );
-
-            if let Some(event) = account_event {
-                let processing_time_us = event.metadata().handle_us as f64;
-                callback(event);
-                update_metrics(MetricsEventType::Account, 1, processing_time_us);
             }
         }
         EventPretty::Transaction(transaction_pretty) => {
             MetricsManager::global().add_tx_process_count();
 
             let slot = transaction_pretty.slot;
-            let signature = transaction_pretty.signature;
             let block_time = transaction_pretty.block_time;
             let recv_us = transaction_pretty.recv_us;
-            let tx_index = transaction_pretty.tx_index;
             let grpc_tx = transaction_pretty.grpc_tx;
-
             let adapter_callback = create_metrics_callback(callback.clone());
-
+            let block_time_us = block_time.map(|t| t.seconds * 1_000_000 + t.nanos as i64 / 1_000);
-            EventParser::parse_grpc_transaction(
+            let update = SubscribeUpdateTransaction {
+                slot,
+                transaction: Some(grpc_tx),
+                ..Default::default()
+            };
+            let sdk_parse_filter = build_sdk_parse_event_filter(event_type_filter);
+            let sdk_events = parse_subscribe_update_transaction_low_latency(
+                &update,
+                recv_us,
+                block_time_us,
+                sdk_parse_filter.as_ref(),
+            );
+            let events = adapt_parser_events_list(
+                sdk_events,
+                block_time.as_ref(),
+                recv_us,
                 protocols,
                 event_type_filter,
-                grpc_tx,
+            );
-                signature,
+
-                Some(slot),
+            for mut event in events {
-                block_time,
+                event.metadata_mut().handle_us = elapsed_micros_since(recv_us);
-                recv_us,
+                event = crate::streaming::event_parser::core::event_parser::helpers::process_event(
-                bot_wallet,
+                    event, bot_wallet,
-                tx_index,
+                );
-                adapter_callback,
+                adapter_callback(event);
-            )
+            }
-            .await?;
         }
         EventPretty::BlockMeta(block_meta_pretty) => {
             MetricsManager::global().add_block_meta_process_count();

src/streaming/sdk_bridge.rs

@@ -12,8 +12,12 @@ use crate::streaming::grpc::AccountPretty;
 
 pub use sol_parser_sdk as raw;
 
+pub type SdkClientConfig = sol_parser_sdk::grpc::types::ClientConfig;
 pub type SdkDexEvent = sol_parser_sdk::DexEvent;
+pub type SdkEventType = sol_parser_sdk::grpc::types::EventType;
 pub type SdkEventTypeFilter = sol_parser_sdk::grpc::types::EventTypeFilter;
+pub type SdkOrderMode = sol_parser_sdk::grpc::types::OrderMode;
+pub type SdkProtocol = sol_parser_sdk::grpc::types::Protocol;
 
 pub fn event_type_filter_to_sdk(filter: Option<&EventTypeFilter>) -> Option<SdkEventTypeFilter> {
     build_sdk_parse_event_filter(filter)