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
|
use std::{process::Stdio, time::Duration};
use serde::Deserialize;
use tokio::{
io::{AsyncBufReadExt, BufReader},
sync::watch::Sender,
};
use crate::{Error, Status};
use super::ServiceSpawner;
#[derive(Debug, Clone, Deserialize)]
pub struct Command {
pub command: String,
pub args: Vec<String>,
pub interval: Option<Duration>,
}
impl Command {
async fn spawn_interval(
mut command: tokio::process::Command,
period: Duration,
tx: Sender<Status>,
) -> Result<(), Error> {
let mut interval = tokio::time::interval(period);
loop {
interval.tick().await;
let status = command.output().await.map_or_else(Into::into, |o| {
if o.status.success() {
Status::Ok
} else {
let stdout = String::from_utf8_lossy(&o.stdout).trim().to_string();
Status::Error(Some(format!("Service state: {}", stdout)))
}
});
tx.send_if_modified(|s| s.update(status));
}
}
async fn spawn_persist(
mut command: tokio::process::Command,
tx: Sender<Status>,
) -> Result<(), Error> {
let mut child = command.stdout(Stdio::piped()).spawn()?;
let mut stdout = BufReader::new(child.stdout.take().unwrap()).lines();
while let Some(line) = stdout.next_line().await? {
let status: Status = serde_json::from_str(&line)
.unwrap_or_else(|err| Status::Error(Some(format!("Serialization error: {err}"))));
tx.send_if_modified(|s| s.update(status));
}
let exit_status = child.wait().await?;
let status = match exit_status.code() {
Some(0) => Status::Ok,
Some(code) => Status::Error(Some(format!("Exited with status code: {code}"))),
None => Status::Error(Some("Process terminated by signal".to_string())),
};
tx.send_if_modified(|s| s.update(status));
Ok(())
}
}
impl ServiceSpawner for Command {
async fn spawn(self, tx: Sender<Status>) -> Result<(), Error> {
let mut command = tokio::process::Command::new(self.command);
command.args(self.args);
if let Some(period) = self.interval {
Self::spawn_interval(command, period, tx).await
} else {
Self::spawn_persist(command, tx).await
}
}
}
|
