relux_runtime/observe/
progress.rs

1use std::io::Write;
2use std::time::Instant;
3
4use colored::Colorize;
5use tokio::sync::mpsc;
6
7#[derive(Debug, Clone)]
8pub enum ProgressEvent {
9    Send,
10    MatchStart,
11    MatchDone,
12    SleepStart,
13    SleepDone,
14    ShellSwitch(String),
15    FnEnter(String),
16    FnExit,
17    ShellSpawn,
18    EffectSetup(String),
19    EffectTeardown,
20    Cleanup,
21    FailPattern,
22    Timeout,
23    Failure,
24    Error(String),
25    Warning(String),
26    Annotation(String),
27}
28
29pub type ProgressTx = mpsc::UnboundedSender<ProgressEvent>;
30
31pub fn channel() -> (ProgressTx, mpsc::UnboundedReceiver<ProgressEvent>) {
32    mpsc::unbounded_channel()
33}
34
35enum TimedWait {
36    Match,
37    Sleep,
38}
39
40/// Spawns the progress printer task. Returns a JoinHandle that resolves
41/// to the collected progress string once all senders are dropped.
42pub fn spawn_printer(
43    mut rx: mpsc::UnboundedReceiver<ProgressEvent>,
44) -> tokio::task::JoinHandle<String> {
45    tokio::spawn(async move {
46        let mut collected = String::new();
47        let mut timed: Option<(TimedWait, Instant)> = None;
48        let mut timed_tick_count: usize = 0;
49
50        loop {
51            let event = if timed.is_some() {
52                match tokio::time::timeout(std::time::Duration::from_secs(1), rx.recv()).await {
53                    Ok(Some(ev)) => Some(ev),
54                    Ok(None) => None,
55                    Err(_) => {
56                        if let Some((kind, started)) = &timed {
57                            let ch = match kind {
58                                TimedWait::Match => '~',
59                                TimedWait::Sleep => 'z',
60                            };
61                            let elapsed_secs = started.elapsed().as_secs() as usize;
62                            while timed_tick_count < elapsed_secs {
63                                emit(&mut collected, ch);
64                                timed_tick_count += 1;
65                            }
66                        }
67                        continue;
68                    }
69                }
70            } else {
71                rx.recv().await
72            };
73
74            let Some(event) = event else {
75                break;
76            };
77
78            match event {
79                ProgressEvent::Send => {
80                    emit(&mut collected, '.');
81                }
82                ProgressEvent::MatchStart => {
83                    timed = Some((TimedWait::Match, Instant::now()));
84                    timed_tick_count = 0;
85                }
86                ProgressEvent::MatchDone => {
87                    timed = None;
88                    emit(&mut collected, '.');
89                }
90                ProgressEvent::SleepStart => {
91                    timed = Some((TimedWait::Sleep, Instant::now()));
92                    timed_tick_count = 0;
93                }
94                ProgressEvent::SleepDone => {
95                    timed = None;
96                }
97                ProgressEvent::ShellSwitch(_) => {
98                    emit(&mut collected, '|');
99                }
100                ProgressEvent::FnEnter(_) => {
101                    emit(&mut collected, '{');
102                }
103                ProgressEvent::FnExit => {
104                    emit(&mut collected, '}');
105                }
106                ProgressEvent::ShellSpawn => {
107                    emit(&mut collected, 's');
108                }
109                ProgressEvent::EffectSetup(_) => {
110                    emit(&mut collected, '+');
111                }
112                ProgressEvent::EffectTeardown => {
113                    emit(&mut collected, '-');
114                }
115                ProgressEvent::Cleanup => {
116                    emit(&mut collected, 'c');
117                }
118                ProgressEvent::FailPattern => {
119                    timed = None;
120                    emit(&mut collected, '!');
121                }
122                ProgressEvent::Timeout => {
123                    timed = None;
124                    emit(&mut collected, 'T');
125                }
126                ProgressEvent::Failure => {
127                    timed = None;
128                    emit(&mut collected, 'F');
129                }
130                ProgressEvent::Error(_) => {
131                    timed = None;
132                    emit(&mut collected, 'E');
133                }
134                ProgressEvent::Warning(_) => {
135                    emit(&mut collected, 'W');
136                }
137                ProgressEvent::Annotation(text) => {
138                    let s = format!("({text})");
139                    collected.push_str(&s);
140                    eprint!("{}", s.dimmed());
141                    let _ = std::io::stderr().flush();
142                }
143            }
144        }
145
146        collected
147    })
148}
149
150fn emit(collected: &mut String, ch: char) {
151    collected.push(ch);
152    let s = ch.to_string().dimmed();
153    eprint!("{s}");
154    let _ = std::io::stderr().flush();
155}
relux_runtime/observe/progress.rs

relux_runtime/observe/
progress.rs
