1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
|
use std::{process::Stdio, time::Duration};
use serde::{Deserialize, Serialize};
use tokio::io::{AsyncBufReadExt, BufReader};
use crate::status::Sender;
use super::IntoService;
#[derive(Debug, thiserror::Error)]
pub enum Error {
#[error("IO error: {0}")]
Io(#[from] std::io::Error),
#[error("Exited with status code: {code}\n{stderr}")]
ExitCode { code: i32, stderr: String },
#[error("Process terminated by signal")]
Signal,
#[error("Serialization error: {0}")]
Serialization(#[from] serde_json::error::Error),
#[error("{0}")]
Stderr(String),
#[error("{0}")]
Output(String),
#[error("Exited with status code: {0}")]
PersistExitCode(i32),
#[error("Failed to get stderr of child process")]
NoStderr,
#[error("Failed to get stdout of child process")]
NoStdout,
}
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct Command {
pub command: String,
pub args: Vec<String>,
#[serde(default)]
pub persist: bool,
#[serde(default = "super::default_interval")]
pub interval: Duration,
}
impl Command {
async fn persist(&self, tx: Sender) -> Result<(), Error> {
let mut command = tokio::process::Command::new(&self.command);
command.args(&self.args);
let mut child = command
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()?;
let mut lines = BufReader::new(child.stdout.take().ok_or(Error::NoStdout)?).lines();
while let Some(line) = lines.next_line().await? {
let res = ("Ok" == line).then_some(()).ok_or(Error::Output(line));
tx.send_if_modified(|s| s.update(res.into()));
}
match child.wait().await?.code() {
Some(0) => Ok(()),
Some(code) => Err(Error::PersistExitCode(code)),
None => Err(Error::Signal),
}
}
async fn interval(&self) -> Result<(), Error> {
let mut command = tokio::process::Command::new(&self.command);
command.args(&self.args);
let output = command.output().await?;
match output.status.code() {
Some(0) => Ok(()),
Some(code) => {
let stderr = String::from_utf8_lossy(&output.stderr).to_string();
Err(Error::ExitCode { code, stderr })
}
None => Err(Error::Signal),
}
}
}
impl IntoService for Command {
async fn into_service(self, tx: Sender) {
let mut interval = tokio::time::interval(self.interval);
loop {
interval.tick().await;
let res = if self.persist {
self.persist(tx.clone()).await
} else {
self.interval().await
};
tx.send_if_modified(|s| s.update(res.into()));
}
}
}
|
