bnto_core/
progress.rs

1// =============================================================================
2// Progress Reporting — How Nodes Talk to the UI
3// =============================================================================
4//
5// Target-agnostic progress reporting. Uses a plain Rust closure instead of
6// `js_sys::Function` so bnto-core stays platform-independent. The WASM-specific
7// wrapping lives in each node crate's `wasm_bridge.rs`.
8//
9// Every node gets the same two ways to say it's still working:
10//   - `report(percent, msg)` — it knows how far through it is
11//   - `report_count(count, msg)` — it doesn't know the total (files found
12//     during a walk, rows read from a stream)
13// The executor turns both into `PipelineEvent::NodeProgress`, throttled.
14
15use crate::progress_throttle::Throttle;
16
17/// One live progress update from a running node.
18#[derive(Debug, Clone, Copy, PartialEq, Eq)]
19pub enum LiveProgress {
20    /// How far through a known amount of work, 0-100.
21    Percent(u32),
22    /// Units of work done so far, total unknown.
23    Count(u64),
24}
25
26impl LiveProgress {
27    /// The update as `(done, total)`: a percent is out of 100, a count has
28    /// no total.
29    pub fn done_and_total(self) -> (u64, Option<u64>) {
30        match self {
31            Self::Percent(percent) => (u64::from(percent), Some(100)),
32            Self::Count(count) => (count, None),
33        }
34    }
35}
36
37// =============================================================================
38// ProgressReporter
39// =============================================================================
40
41/// Reports processing progress from a node back to the caller (UI, CLI, etc.).
42///
43/// Every callback is optional: a reporter with none is a no-op (used in tests).
44pub struct ProgressReporter {
45    #[allow(clippy::type_complexity)]
46    live: Option<Box<dyn Fn(LiveProgress, &str)>>,
47    /// Streaming command output lines (stderr from child processes).
48    #[allow(clippy::type_complexity)]
49    output_callback: Option<Box<dyn Fn(&str)>>,
50    /// When set, live updates past the first are let through once per interval.
51    throttle: Option<Throttle>,
52}
53
54impl ProgressReporter {
55    /// Create a reporter that receives percent updates.
56    ///
57    /// The callback receives `(percent 0-100, message)`. Count updates
58    /// (`report_count`) don't reach it; use [`ProgressReporter::with_live`]
59    /// to receive both.
60    ///
61    /// ```rust
62    /// use bnto_core::ProgressReporter;
63    ///
64    /// let reporter = ProgressReporter::new(|percent, message| {
65    ///     println!("{}% — {}", percent, message);
66    /// });
67    /// ```
68    pub fn new(callback: impl Fn(u32, &str) + 'static) -> Self {
69        let live = move |progress: LiveProgress, message: &str| {
70            if let LiveProgress::Percent(percent) = progress {
71                callback(percent, message);
72            }
73        };
74        Self {
75            live: Some(Box::new(live)),
76            output_callback: None,
77            throttle: None,
78        }
79    }
80
81    /// Create a no-op reporter that discards all progress updates.
82    pub fn new_noop() -> Self {
83        Self {
84            live: None,
85            output_callback: None,
86            throttle: None,
87        }
88    }
89
90    /// Create a reporter that receives every live update and command output
91    /// line. The executor uses this to relay both as pipeline events.
92    pub fn with_live(
93        live: impl Fn(LiveProgress, &str) + 'static,
94        output_callback: impl Fn(&str) + 'static,
95    ) -> Self {
96        Self {
97            live: Some(Box::new(live)),
98            output_callback: Some(Box::new(output_callback)),
99            throttle: None,
100        }
101    }
102
103    /// Let at most one live update through per `interval_ms` (the first one
104    /// always gets through). Command output lines are never throttled.
105    pub fn throttled(mut self, clock: impl Fn() -> u64 + 'static, interval_ms: u64) -> Self {
106        self.throttle = Some(Throttle::new(clock, interval_ms));
107        self
108    }
109
110    /// Report how far along the node is (0 to 100), with what it's doing
111    /// ("Compressing image 3/10...").
112    pub fn report(&self, percent: u32, message: &str) {
113        self.send(LiveProgress::Percent(percent), message);
114    }
115
116    /// Report work done so far when the node can't know the total, e.g.
117    /// files found while walking a folder. Call it as often as you like;
118    /// a throttled reporter drops the excess.
119    pub fn report_count(&self, count: u64, message: &str) {
120        self.send(LiveProgress::Count(count), message);
121    }
122
123    /// Report a line of streaming output from a child process.
124    ///
125    /// Called by processors that use `run_command_streaming()` to relay
126    /// stderr lines (e.g. yt-dlp progress) to the pipeline reporter.
127    pub fn report_output(&self, line: &str) {
128        if let Some(cb) = &self.output_callback {
129            cb(line);
130        }
131    }
132
133    fn send(&self, progress: LiveProgress, message: &str) {
134        let Some(cb) = &self.live else { return };
135        if self.throttle.as_ref().is_none_or(Throttle::allow) {
136            cb(progress, message);
137        }
138    }
139}
140
141// =============================================================================
142// Tests
143// =============================================================================
144
145#[cfg(test)]
146mod tests {
147    use super::*;
148
149    use std::cell::Cell;
150    use std::rc::Rc;
151    use std::sync::{Arc, Mutex};
152
153    type Recorded = Arc<Mutex<Vec<(LiveProgress, String)>>>;
154
155    /// A reporter that records every live update it receives.
156    fn recording() -> (Recorded, ProgressReporter) {
157        let calls: Recorded = Arc::new(Mutex::new(Vec::new()));
158        let sink = Arc::clone(&calls);
159        let reporter = ProgressReporter::with_live(
160            move |p, m| sink.lock().unwrap().push((p, m.to_string())),
161            |_| {},
162        );
163        (calls, reporter)
164    }
165
166    #[test]
167    fn test_noop_reporter_doesnt_panic() {
168        let reporter = ProgressReporter::new_noop();
169        reporter.report(0, "Starting...");
170        reporter.report_count(3, "Found 3 files");
171        reporter.report(100, "Done!");
172    }
173
174    #[test]
175    fn test_reporter_calls_callback() {
176        let calls: Arc<Mutex<Vec<(u32, String)>>> = Arc::new(Mutex::new(Vec::new()));
177        let calls_clone = Arc::clone(&calls);
178        let reporter = ProgressReporter::new(move |percent, message| {
179            calls_clone
180                .lock()
181                .unwrap()
182                .push((percent, message.to_string()));
183        });
184
185        reporter.report(0, "Starting...");
186        reporter.report(50, "Halfway there...");
187        reporter.report(100, "Done!");
188
189        let recorded = calls.lock().unwrap();
190        assert_eq!(recorded.len(), 3, "Should have recorded 3 calls");
191        assert_eq!(recorded[0], (0, "Starting...".to_string()));
192        assert_eq!(recorded[1], (50, "Halfway there...".to_string()));
193        assert_eq!(recorded[2], (100, "Done!".to_string()));
194    }
195
196    #[test]
197    fn percent_only_callback_ignores_counts() {
198        let calls = Arc::new(Mutex::new(Vec::new()));
199        let sink = Arc::clone(&calls);
200        let reporter = ProgressReporter::new(move |p, _| sink.lock().unwrap().push(p));
201
202        reporter.report_count(7, "Found 7 files");
203        reporter.report(40, "Working");
204
205        assert_eq!(*calls.lock().unwrap(), vec![40]);
206    }
207
208    #[test]
209    fn live_reporter_receives_percents_and_counts() {
210        let (calls, reporter) = recording();
211
212        reporter.report(25, "Quarter");
213        reporter.report_count(12, "Found 12 files");
214
215        let calls = calls.lock().unwrap();
216        assert_eq!(calls[0], (LiveProgress::Percent(25), "Quarter".to_string()));
217        assert_eq!(
218            calls[1],
219            (LiveProgress::Count(12), "Found 12 files".to_string())
220        );
221    }
222
223    #[test]
224    fn throttled_reporter_lets_one_update_through_per_interval() {
225        let (calls, reporter) = recording();
226        let now = Rc::new(Cell::new(0u64));
227        let clock = Rc::clone(&now);
228        let reporter = reporter.throttled(move || clock.get(), 100);
229
230        for (t, count) in [(0, 1), (10, 2), (60, 3), (100, 4), (150, 5), (210, 6)] {
231            now.set(t);
232            reporter.report_count(count, "walking");
233        }
234
235        let counts: Vec<LiveProgress> = calls.lock().unwrap().iter().map(|c| c.0).collect();
236        assert_eq!(
237            counts,
238            vec![
239                LiveProgress::Count(1),
240                LiveProgress::Count(4),
241                LiveProgress::Count(6)
242            ]
243        );
244    }
245
246    #[test]
247    fn throttle_never_drops_command_output() {
248        let lines = Arc::new(Mutex::new(Vec::new()));
249        let sink = Arc::clone(&lines);
250        let reporter = ProgressReporter::with_live(
251            |_, _| {},
252            move |l| sink.lock().unwrap().push(l.to_string()),
253        )
254        .throttled(|| 0, 100);
255
256        reporter.report_output("one");
257        reporter.report_output("two");
258
259        assert_eq!(lines.lock().unwrap().len(), 2);
260    }
261
262    #[test]
263    fn done_and_total_maps_percent_to_out_of_100() {
264        assert_eq!(LiveProgress::Percent(40).done_and_total(), (40, Some(100)));
265        assert_eq!(LiveProgress::Count(9).done_and_total(), (9, None));
266    }
267
268    #[test]
269    fn report_output_calls_callback() {
270        let received: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
271        let received_clone = Arc::clone(&received);
272
273        let reporter = ProgressReporter::with_live(
274            |_, _| {},
275            move |line| received_clone.lock().unwrap().push(line.to_string()),
276        );
277
278        reporter.report_output("downloading 50%");
279        reporter.report_output("downloading 100%");
280
281        let lines = received.lock().unwrap();
282        assert_eq!(lines.len(), 2);
283        assert_eq!(lines[0], "downloading 50%");
284        assert_eq!(lines[1], "downloading 100%");
285    }
286
287    #[test]
288    fn report_output_noop_doesnt_panic() {
289        let reporter = ProgressReporter::new_noop();
290        reporter.report_output("ignored");
291    }
292}
bnto_core/progress.rs

bnto_core/
progress.rs
