After fanatids review

[godmodegalactus]

No description provided.

[godmodegalactus]

This logic is no longer needed with new tpu client.

[fanatid]

we can lock the process for an unknown time, better apply timeout + try_recv
something like this

loop {
    let deadline = Instant::now() + INTERVAL_PER_BATCH_IN_MS;
    let mut notifications = vec![];
    loop {
        let maybe_transaction_info = if deadline.elapsed().is_zer() {
            match tokio::time::timeout_at(deadline, recv.recv()).await {
                Ok(Some(tx_info)) => Some(tx_info),
                Ok(None) => bail!("Tx sender service stopped"),
                Err(_error) => None,
            }
        } else {
            match recv.try_recv() {
                tokio::sync::mpsc::error::TryRecvError::Empty => None,
                tokio::sync::mpsc::error::TryRecvError::Disconnected => bail!("Tx sender service stopped")
            }
        };

        if let Some(transaction_info) = maybe_transaction_info {
            self.forward_txs(&transaction_info).await;

            // send transaction notifications in batches
            if let Some(notifier) = &notifier {
                let forwarded_slot = self.data_cache.slot_cache.get_current_slot();
                let forwarded_local_time = Utc::now();
                let tx_notification = TransactionNotification {
                    signature: transaction_info.signature,
                    recent_slot: transaction_info.slot,
                    forwarded_slot,
                    forwarded_local_time,
                    processed_slot: None,
                    cu_consumed: None,
                    cu_requested: None,
                    quic_response: 0,
                };
                notifications.push(tx_notification);
            }
        } else {
            // send transaction notifications in batches
            if let Some(notifier) = &notifier {
                // send notification for sent transactions
                let _ = notifier.send(NotificationMsg::TxNotificationMsg(notifications));
            }
            break;
        }
    }
}

[godmodegalactus]

Have addressed your concerns using tokio select, it seems more readable for me.

[fanatid]

lgtm

[fanatid]

std::mem::take does not need iterator, but maybe resulted generated code would be same