diff --git a/src/streaming/yellowstone_grpc.rs b/src/streaming/yellowstone_grpc.rs index 5c008bd..769aef6 100755 --- a/src/streaming/yellowstone_grpc.rs +++ b/src/streaming/yellowstone_grpc.rs @@ -1186,90 +1186,5 @@ impl YellowstoneGrpc { } } -// 有序交易处理器,确保交易按顺序处理 -pub struct OrderedTransactionProcessor -where - F: Fn(TransactionPretty) + Send + Sync + 'static, -{ - callback: F, - pending_transactions: std::collections::BTreeMap, // 按 slot 排序 - next_expected_slot: u64, - max_pending_slots: usize, // 最大等待槽位数 -} - -impl OrderedTransactionProcessor -where - F: Fn(TransactionPretty) + Send + Sync + 'static, -{ - pub fn new(callback: F, max_pending_slots: usize) -> Self { - Self { - callback, - pending_transactions: std::collections::BTreeMap::new(), - next_expected_slot: 0, - max_pending_slots, - } - } - - pub fn process_transaction(&mut self, transaction: TransactionPretty) { - let slot = transaction.slot; - - // 如果是第一个交易,设置期望的槽位 - if self.next_expected_slot == 0 { - self.next_expected_slot = slot; - } - - // 如果槽位太旧,直接丢弃 - if slot < self.next_expected_slot.saturating_sub(self.max_pending_slots as u64) { - log::warn!("Dropping old transaction from slot {}", slot); - return; - } - - // 如果槽位太新,先缓存 - if slot > self.next_expected_slot { - self.pending_transactions.insert(slot, transaction); - - // 如果缓存太多,清理旧的 - while self.pending_transactions.len() > self.max_pending_slots { - if let Some((oldest_slot, _)) = self.pending_transactions.iter().next() { - let oldest_slot = *oldest_slot; - self.pending_transactions.remove(&oldest_slot); - log::warn!("Dropping old cached transaction from slot {}", oldest_slot); - } - } - return; - } - - // 处理当前槽位的交易 - if slot == self.next_expected_slot { - (self.callback)(transaction); - self.next_expected_slot += 1; - - // 处理后续连续的槽位 - loop { - let next_slot = self.pending_transactions.keys().next().copied(); - if let Some(slot) = next_slot { - if slot == self.next_expected_slot { - if let Some(transaction) = self.pending_transactions.remove(&slot) { - (self.callback)(transaction); - self.next_expected_slot += 1; - } - } else { - break; - } - } else { - break; - } - } - } else { - // 槽位不匹配,缓存起来 - self.pending_transactions.insert(slot, transaction); - } - } - - pub fn get_stats(&self) -> (u64, usize) { - (self.next_expected_slot, self.pending_transactions.len()) - } -} -