log2src/
progress.rs

1use std::sync::atomic::{AtomicU64, Ordering};
2use std::sync::mpsc::{channel, Receiver, Sender};
3use std::sync::{Arc, OnceLock};
4use std::time::Duration;
5
6static GLOBAL_PROGRESS_TRACKER: OnceLock<Arc<ProgressTracker>> = OnceLock::new();
7
8/// Sets the global progress tracker for the lifetime of the program.
9///
10/// The tracker is used by library operations (such as [`crate::LogMatcher::discover_sources`]
11/// and [`crate::LogMatcher::extract_log_statements`]) to report progress. If no tracker is
12/// registered, those operations run silently.
13///
14/// Call this once at application startup, before invoking any library operations.
15/// A second call is silently ignored — the first registration wins.
16///
17/// # Example
18/// ```no_run
19/// use std::sync::Arc;
20/// use log2src::{ProgressTracker, set_tracker_once};
21///
22/// let tracker = Arc::new(ProgressTracker::new());
23/// set_tracker_once(Arc::clone(&tracker));
24/// ```
25pub fn set_tracker_once(tracker: Arc<ProgressTracker>) {
26    let _ = GLOBAL_PROGRESS_TRACKER.set(tracker);
27}
28
29pub(crate) fn current_global_progress_tracker() -> Arc<ProgressTracker> {
30    GLOBAL_PROGRESS_TRACKER.get().cloned().unwrap_or_default()
31}
32
33pub struct WorkInfo {
34    pub completed: AtomicU64,
35    pub total: u64,
36    pub units: String,
37}
38
39impl WorkInfo {
40    /// Check if the work is still in-progress.
41    pub fn is_in_progress(&self) -> bool {
42        self.completed.load(Ordering::Relaxed) < self.total
43    }
44}
45
46/// A notification of progress for subscribers to a ProgressTracker
47pub enum ProgressUpdate {
48    /// A description of a large amount of work.
49    Step(String),
50    /// The start of a batch of work.
51    BeginStep(String),
52    /// The end of a batch of work.
53    EndStep(String),
54    /// A deterministic amount of work.
55    Work(Arc<WorkInfo>),
56}
57
58pub struct ProgressListener {
59    receiver: Receiver<ProgressUpdate>,
60}
61
62#[derive(Default, Debug)]
63/// A mechanism for tracking progress.
64pub struct ProgressTracker {
65    subscribers: Vec<Sender<ProgressUpdate>>,
66}
67
68pub(crate) struct WorkGuard {
69    info: Arc<WorkInfo>,
70}
71
72impl WorkGuard {
73    /// Increase the amount of deterministic work that has been done.
74    pub fn inc(&self, amount: u64) {
75        self.info.completed.fetch_add(amount, Ordering::Relaxed);
76    }
77}
78
79impl Drop for WorkGuard {
80    fn drop(&mut self) {
81        self.info
82            .completed
83            .store(self.info.total, Ordering::Relaxed);
84    }
85}
86
87impl ProgressTracker {
88    /// Create an empty tracker.
89    pub fn new() -> ProgressTracker {
90        ProgressTracker {
91            subscribers: vec![],
92        }
93    }
94
95    /// Notify subscribers of the beginning of a step in a process.
96    pub(crate) fn begin_step(&self, message: String) {
97        self.subscribers.iter().for_each(|sender| {
98            let _ = sender.send(ProgressUpdate::BeginStep(message.clone()));
99        });
100    }
101
102    pub(crate) fn end_step(&self, message: String) {
103        self.subscribers.iter().for_each(|sender| {
104            let _ = sender.send(ProgressUpdate::EndStep(message.clone()));
105        });
106    }
107
108    /// Notify subscribers that some deterministic amount of work is about to be done.
109    pub(crate) fn doing_work(&self, total: u64, units: String) -> WorkGuard {
110        let info = Arc::new(WorkInfo {
111            completed: AtomicU64::new(0),
112            total,
113            units,
114        });
115
116        self.subscribers.iter().for_each(|sender| {
117            let _ = sender.send(ProgressUpdate::Work(Arc::clone(&info)));
118        });
119
120        WorkGuard {
121            info: Arc::clone(&info),
122        }
123    }
124
125    /// Subscribe to notifications of work for this tracker.
126    pub fn subscribe(&mut self) -> ProgressListener {
127        let (sender, receiver) = channel();
128
129        self.subscribers.push(sender);
130        ProgressListener { receiver }
131    }
132}
133
134impl Iterator for ProgressListener {
135    type Item = ProgressUpdate;
136
137    fn next(&mut self) -> Option<Self::Item> {
138        self.receiver.iter().next()
139    }
140}
141
142impl ProgressListener {
143    pub fn try_next_for(&self, timeout: Duration) -> Option<ProgressUpdate> {
144        self.receiver.recv_timeout(timeout).ok()
145    }
146}
log2src/progress.rs

log2src/
progress.rs
