diff --git a/Cargo.toml b/Cargo.toml index 8b26654..3bc410e 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "solana-streamer-sdk" -version = "0.1.0" +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 7671249..c01041d 100755 --- a/README.md +++ b/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> { ]; 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> { 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 42633b2..8959265 100644 --- a/README_CN.md +++ b/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> { ]; 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> { 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/mod.rs b/src/streaming/event_parser/common/mod.rs index a070106..d25d73e 100755 --- a/src/streaming/event_parser/common/mod.rs +++ b/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; + } } }; } diff --git a/src/streaming/event_parser/common/types.rs b/src/streaming/event_parser/common/types.rs index 0ad95f9..a43fc2d 100755 --- a/src/streaming/event_parser/common/types.rs +++ b/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, + pub amount: u64, + pub decimals: Option, + pub mint: Option, +} + /// 事件元数据 #[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, } 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 { + 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 = 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 = 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 = 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 +} diff --git a/src/streaming/event_parser/core/traits.rs b/src/streaming/event_parser/core/traits.rs index 331fe43..c9f235f 100755 --- a/src/streaming/event_parser/core/traits.rs +++ b/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) { // 默认实现：不进行任何合并操作 } + + 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 { @@ -75,6 +96,7 @@ pub trait EventParser: Send + Sync { signature: &str, slot: Option, accounts: &[Pubkey], + inner_instructions: &[UiInnerInstructions], ) -> Result>> { 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 = vec![]; + let mut inner_instructions: Vec = 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 { + 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 909aebc..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(_)) => { @@ -168,6 +186,8 @@ impl YellowstoneGrpc { bot_wallet: Option, account_include: Option>, account_exclude: Option>, + account_required: Option>, + commitment: Option, callback: F, ) -> AnyResult<()> where @@ -182,27 +202,37 @@ impl YellowstoneGrpc { .collect::>(); 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::(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; @@ -216,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(()) } @@ -263,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;