clone_solana_ledger/
entry_notifier_service.rs

1use {
2    crate::entry_notifier_interface::EntryNotifierArc,
3    crossbeam_channel::{unbounded, Receiver, RecvTimeoutError, Sender},
4    clone_solana_entry::entry::EntrySummary,
5    clone_solana_sdk::clock::Slot,
6    std::{
7        sync::{
8            atomic::{AtomicBool, Ordering},
9            Arc,
10        },
11        thread::{self, Builder, JoinHandle},
12        time::Duration,
13    },
14};
15
16pub struct EntryNotification {
17    pub slot: Slot,
18    pub index: usize,
19    pub entry: EntrySummary,
20    pub starting_transaction_index: usize,
21}
22
23pub type EntryNotifierSender = Sender<EntryNotification>;
24pub type EntryNotifierReceiver = Receiver<EntryNotification>;
25
26pub struct EntryNotifierService {
27    sender: EntryNotifierSender,
28    thread_hdl: JoinHandle<()>,
29}
30
31impl EntryNotifierService {
32    pub fn new(entry_notifier: EntryNotifierArc, exit: Arc<AtomicBool>) -> Self {
33        let (entry_notification_sender, entry_notification_receiver) = unbounded();
34        let thread_hdl = Builder::new()
35            .name("solEntryNotif".to_string())
36            .spawn(move || loop {
37                if exit.load(Ordering::Relaxed) {
38                    break;
39                }
40
41                if let Err(RecvTimeoutError::Disconnected) =
42                    Self::notify_entry(&entry_notification_receiver, entry_notifier.clone())
43                {
44                    break;
45                }
46            })
47            .unwrap();
48        Self {
49            sender: entry_notification_sender,
50            thread_hdl,
51        }
52    }
53
54    fn notify_entry(
55        entry_notification_receiver: &EntryNotifierReceiver,
56        entry_notifier: EntryNotifierArc,
57    ) -> Result<(), RecvTimeoutError> {
58        let EntryNotification {
59            slot,
60            index,
61            entry,
62            starting_transaction_index,
63        } = entry_notification_receiver.recv_timeout(Duration::from_secs(1))?;
64        entry_notifier.notify_entry(slot, index, &entry, starting_transaction_index);
65        Ok(())
66    }
67
68    pub fn sender(&self) -> &EntryNotifierSender {
69        &self.sender
70    }
71
72    pub fn sender_cloned(&self) -> EntryNotifierSender {
73        self.sender.clone()
74    }
75
76    pub fn join(self) -> thread::Result<()> {
77        self.thread_hdl.join()
78    }
79}
clone_solana_ledger/entry_notifier_service.rs

clone_solana_ledger/
entry_notifier_service.rs
