1use std::io::{BufRead, BufReader, Read};
24use std::process::{Command, Stdio};
25use std::sync::atomic::{AtomicBool, Ordering};
26use std::sync::{Arc, Mutex};
27use std::thread;
28use std::time::{Duration, Instant};
29
30use serde::Deserialize;
31
32use crate::config::{BuiltinWidget, WidgetConfig};
33
34const MAX_OUTPUT: u64 = 256 * 1024;
36const STOP_POLL: Duration = Duration::from_millis(100);
38
39#[derive(Debug, Clone, Default, PartialEq, Deserialize)]
41#[serde(default)]
42pub struct WidgetOutput {
43 pub rows: Vec<WidgetRow>,
44}
45
46#[derive(Debug, Clone, Default, PartialEq, Deserialize)]
48#[serde(default)]
49pub struct WidgetRow {
50 pub label: String,
52 #[serde(alias = "text")]
54 pub value: String,
55 pub gauge: Option<f32>,
57 pub gauge_style: GaugeStyle,
59 pub level: Level,
61 pub label_level: Option<Level>,
63 pub heading: bool,
65 pub separator: bool,
68}
69
70#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Deserialize)]
72#[serde(rename_all = "lowercase")]
73pub enum GaugeStyle {
74 #[default]
76 Heat,
77 Level,
79}
80
81#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Deserialize)]
83#[serde(rename_all = "lowercase")]
84pub enum Level {
85 #[default]
86 Normal,
87 Dim,
88 Ok,
89 Warn,
90 Bad,
91}
92
93pub fn parse_output(stdout: &str) -> WidgetOutput {
95 let trimmed = stdout.trim();
96 if trimmed.starts_with('{')
97 && let Ok(output) = serde_json::from_str::<WidgetOutput>(trimmed)
98 {
99 return output;
100 }
101 WidgetOutput {
102 rows: trimmed
103 .lines()
104 .map(|line| WidgetRow {
105 value: line.trim_end().to_owned(),
106 ..Default::default()
107 })
108 .collect(),
109 }
110}
111
112#[derive(Debug, Clone, Default)]
114pub struct WidgetState {
115 pub output: Option<WidgetOutput>,
117 pub error: Option<String>,
119 pub updated: Option<Instant>,
121}
122
123pub struct Widget {
125 pub name: String,
126 pub source: WidgetSource,
127}
128
129pub enum WidgetSource {
131 Clock,
133 Command {
135 runner: CommandRunner,
136 interval: Duration,
137 },
138 Fixed(WidgetOutput),
140 Invalid(String),
142}
143
144impl Widget {
145 pub fn state(&self) -> Option<WidgetState> {
147 match &self.source {
148 WidgetSource::Command { runner, .. } => Some(runner.state()),
149 WidgetSource::Fixed(output) => Some(WidgetState {
150 output: Some(output.clone()),
151 error: None,
152 updated: Some(Instant::now()),
153 }),
154 WidgetSource::Clock | WidgetSource::Invalid(_) => None,
155 }
156 }
157
158 pub fn is_stale(&self, state: &WidgetState) -> bool {
160 match (&self.source, state.updated) {
161 (WidgetSource::Command { interval, .. }, Some(at)) => {
162 at.elapsed() > *interval * 3 + Duration::from_secs(5)
163 }
164 _ => false,
165 }
166 }
167}
168
169pub fn start_widgets<F>(configs: &[WidgetConfig], notify: F) -> Vec<Widget>
172where
173 F: Fn() + Clone + Send + 'static,
174{
175 configs
176 .iter()
177 .map(|config| {
178 let source = match (&config.builtin, &config.command) {
179 (Some(BuiltinWidget::Clock), None) => WidgetSource::Clock,
180 (None, Some(command)) => {
181 let interval = Duration::from_secs(config.interval_secs.max(1));
182 WidgetSource::Command {
183 runner: CommandRunner::spawn(
184 command.clone(),
185 interval,
186 Duration::from_secs(config.timeout_secs.max(1)),
187 notify.clone(),
188 ),
189 interval,
190 }
191 }
192 (Some(_), Some(_)) => {
193 WidgetSource::Invalid("set either `command` or `builtin`, not both".to_owned())
194 }
195 (None, None) => WidgetSource::Invalid("set `command` or `builtin`".to_owned()),
196 };
197 Widget {
198 name: config.name.clone(),
199 source,
200 }
201 })
202 .collect()
203}
204
205pub struct CommandRunner {
208 state: Arc<Mutex<WidgetState>>,
209 stop: Arc<AtomicBool>,
210}
211
212impl CommandRunner {
213 pub fn spawn<F>(command: String, interval: Duration, timeout: Duration, notify: F) -> Self
214 where
215 F: Fn() + Send + 'static,
216 {
217 let state = Arc::new(Mutex::new(WidgetState::default()));
218 let stop = Arc::new(AtomicBool::new(false));
219 let (shared, stopping) = (Arc::clone(&state), Arc::clone(&stop));
220 let spawned = thread::Builder::new()
221 .name("rigtop-widget".to_owned())
222 .spawn(move || {
223 while !stopping.load(Ordering::Relaxed) {
224 let result = run_command(&command, timeout);
225 {
226 let mut state = shared.lock().unwrap_or_else(|e| e.into_inner());
227 match result {
228 Ok(stdout) => {
229 state.output = Some(parse_output(&stdout));
230 state.error = None;
231 }
232 Err(error) => state.error = Some(error),
233 }
234 state.updated = Some(Instant::now());
235 }
236 notify();
237 let wake = Instant::now() + interval;
238 while Instant::now() < wake && !stopping.load(Ordering::Relaxed) {
239 thread::sleep(STOP_POLL);
240 }
241 }
242 });
243 if let Err(err) = spawned {
244 state.lock().unwrap_or_else(|e| e.into_inner()).error =
245 Some(format!("cannot start a thread: {err}"));
246 }
247 Self { state, stop }
248 }
249
250 pub fn state(&self) -> WidgetState {
251 self.state.lock().unwrap_or_else(|e| e.into_inner()).clone()
252 }
253}
254
255impl Drop for CommandRunner {
256 fn drop(&mut self) {
257 self.stop.store(true, Ordering::Relaxed);
258 }
259}
260
261pub fn run_command(command: &str, timeout: Duration) -> Result<String, String> {
264 run_command_with_env(command, &[], timeout)
265}
266
267pub fn run_command_with_env(
269 command: &str,
270 env: &[(&str, &str)],
271 timeout: Duration,
272) -> Result<String, String> {
273 let mut process = if cfg!(windows) {
274 let mut process = Command::new("cmd");
275 process.args(["/C", command]);
276 process
277 } else {
278 let mut process = Command::new("sh");
279 process.args(["-c", command]);
280 process
281 };
282 process
283 .envs(env.iter().copied())
284 .stdin(Stdio::null())
285 .stdout(Stdio::piped())
286 .stderr(Stdio::piped());
287 #[cfg(windows)]
288 {
289 use std::os::windows::process::CommandExt;
290 process.creation_flags(0x0800_0000);
292 }
293 let mut child = process
294 .spawn()
295 .map_err(|err| format!("cannot run: {err}"))?;
296 let stdout = child.stdout.take().expect("piped stdout");
297 let stderr = child.stderr.take().expect("piped stderr");
298 let out = thread::spawn(move || {
300 let mut text = String::new();
301 let _ = stdout.take(MAX_OUTPUT).read_to_string(&mut text);
302 text
303 });
304 let err = thread::spawn(move || {
305 BufReader::new(stderr.take(MAX_OUTPUT))
306 .lines()
307 .map_while(Result::ok)
308 .filter(|l| !l.trim().is_empty())
309 .last()
310 });
311 let deadline = Instant::now() + timeout;
312 let status = loop {
313 match child.try_wait() {
314 Ok(Some(status)) => break status,
315 Ok(None) if Instant::now() >= deadline => {
316 let _ = child.kill();
317 let _ = child.wait();
318 return Err(format!("timed out after {}s", timeout.as_secs()));
319 }
320 Ok(None) => thread::sleep(Duration::from_millis(20)),
321 Err(err) => return Err(err.to_string()),
322 }
323 };
324 let stdout = out.join().unwrap_or_default();
325 let last_error = err.join().ok().flatten();
326 if status.success() {
327 Ok(stdout)
328 } else {
329 Err(last_error.unwrap_or_else(|| format!("exited with {status}")))
330 }
331}
332
333#[cfg(test)]
334mod tests {
335 use super::*;
336
337 #[test]
338 fn parses_json_and_plain_text() {
339 let json = r#"{"rows":[
340 {"label":"5h","value":"42%","gauge":0.42,"level":"ok"},
341 {"text":"rigtop","heading":true},
342 {}
343 ]}"#;
344 let output = parse_output(json);
345 assert_eq!(output.rows.len(), 3);
346 assert_eq!(output.rows[0].gauge, Some(0.42));
347 assert_eq!(output.rows[0].level, Level::Ok);
348 assert_eq!(output.rows[0].gauge_style, GaugeStyle::Heat, "default");
349 let left = parse_output(r#"{"rows":[{"gauge":0.8,"gauge_style":"level"}]}"#);
350 assert_eq!(left.rows[0].gauge_style, GaugeStyle::Level);
351 assert_eq!(output.rows[1].value, "rigtop", "`text` is `value`");
352 assert!(output.rows[1].heading);
353
354 let text = parse_output("USD/JPY 151.2\nEUR/JPY 163.9\n\n");
355 assert_eq!(text.rows.len(), 2);
356 assert_eq!(text.rows[1].value, "EUR/JPY 163.9");
357 assert_eq!(parse_output("{oops").rows[0].value, "{oops");
359 }
360
361 #[cfg(unix)]
362 #[test]
363 fn runs_commands_with_errors_and_timeouts() {
364 let second = Duration::from_secs(1);
365 assert_eq!(run_command("echo hello", second).unwrap().trim(), "hello");
366 assert_eq!(
367 run_command("echo first >&2; echo 'no data' >&2; exit 3", second),
368 Err("no data".to_owned())
369 );
370 assert!(run_command("exit 4", second).unwrap_err().contains("exit"));
371 let started = Instant::now();
372 assert!(
373 run_command("sleep 5", second)
374 .unwrap_err()
375 .contains("timed out")
376 );
377 assert!(started.elapsed() < Duration::from_secs(3));
378 }
379
380 #[cfg(unix)]
381 #[test]
382 fn widgets_from_configuration() {
383 let config =
384 |name: &str, command: Option<&str>, builtin: Option<BuiltinWidget>| WidgetConfig {
385 name: name.to_owned(),
386 command: command.map(str::to_owned),
387 builtin,
388 ..Default::default()
389 };
390 let (tx, rx) = std::sync::mpsc::channel();
391 let notify = move || {
392 let _ = tx.send(());
393 };
394 let widgets = start_widgets(
395 &[
396 config("Clock", None, Some(BuiltinWidget::Clock)),
397 config("Echo", Some(r#"echo '{"rows":[{"value":"hi"}]}'"#), None),
398 config("Empty", None, None),
399 ],
400 notify,
401 );
402 assert!(matches!(widgets[0].source, WidgetSource::Clock));
403 assert!(matches!(widgets[2].source, WidgetSource::Invalid(_)));
404 rx.recv_timeout(Duration::from_secs(5)).expect("first run");
405 let state = widgets[1].state().unwrap();
406 assert_eq!(state.output.unwrap().rows[0].value, "hi");
407 assert!(state.error.is_none());
408 }
409}