systemd_transient/
manager.rs

1use std::time::Duration;
2
3use futures_util::StreamExt;
4use tracing::warn;
5use zbus::{
6  Connection,
7  zvariant::{ObjectPath, OwnedObjectPath, Value},
8};
9
10use crate::{
11  TransientUnitHandle,
12  dbus::manager::{JobRemovedStream, ManagerProxy},
13  error::SystemdTransientError,
14  props::{UnitProperty, Usec, get_property},
15};
16
17const JOB_MODE_RPELACE: &str = "replace";
18const ADDITIONAL_TIMEOUT_BUFFER: Duration = Duration::from_secs(5);
19
20#[derive(Clone)]
21pub struct SystemdTransientManager {
22  conn: Connection,
23}
24
25impl SystemdTransientManager {
26  pub async fn connect() -> Result<Self, SystemdTransientError> {
27    let conn = Connection::session().await?;
28    Ok(Self { conn })
29  }
30
31  pub async fn start(
32    &self,
33    name: String,
34    props: Vec<UnitProperty>,
35  ) -> Result<TransientUnitHandle, SystemdTransientError> {
36    let proxy = ManagerProxy::new(&self.conn).await?;
37
38    let start_timeout = if let Some(timeout) = get_property(&props, |prop| {
39      if let UnitProperty::TimeoutStartUSec(usec) = prop {
40        Some(Duration::from(*usec))
41      } else {
42        None
43      }
44    }) {
45      timeout
46    } else {
47      Usec(proxy.default_timeout_start_usec().await?).into()
48    } + ADDITIONAL_TIMEOUT_BUFFER;
49
50    let stop_timeout = if let Some(timeout) = get_property(&props, |prop| {
51      if let UnitProperty::TimeoutStopUSec(usec) = prop {
52        Some(Duration::from(*usec))
53      } else {
54        None
55      }
56    }) {
57      timeout
58    } else {
59      Usec(proxy.default_timeout_start_usec().await?).into()
60    } + ADDITIONAL_TIMEOUT_BUFFER;
61
62    let pairs: Vec<_> = props.into_iter().map(UnitProperty::into_pair).collect();
63    let tuples: Vec<(&str, &Value<'_>)> = pairs.iter().map(|(k, v)| (k.as_ref(), v)).collect();
64    let refs: Vec<&(&str, &Value<'_>)> = tuples.iter().collect();
65
66    let action = async |proxy: &ManagerProxy<'_>| {
67      proxy
68        .start_transient_unit(&name, JOB_MODE_RPELACE, &refs, &[])
69        .await
70    };
71
72    run_job(&proxy, action, start_timeout).await?;
73
74    Ok(TransientUnitHandle {
75      name,
76      stop_timeout,
77      manager: self.clone(),
78    })
79  }
80
81  pub(crate) async fn stop(
82    &self,
83    name: &str,
84    timeout: Duration,
85  ) -> Result<(), SystemdTransientError> {
86    let proxy = ManagerProxy::new(&self.conn).await?;
87    let action = async |proxy: &ManagerProxy<'_>| proxy.stop_unit(name, JOB_MODE_RPELACE).await;
88    run_job(&proxy, action, timeout).await
89  }
90}
91
92async fn run_job<F>(
93  proxy: &ManagerProxy<'_>,
94  action: F,
95  timeout: Duration,
96) -> Result<(), SystemdTransientError>
97where
98  F: AsyncFnOnce(&ManagerProxy<'_>) -> zbus::Result<OwnedObjectPath>,
99{
100  let stream = proxy.receive_job_removed().await?;
101  let job = action(&proxy).await?;
102
103  tokio::time::timeout(timeout, wait_for_job(stream, job.as_ref()))
104    .await
105    .map_err(|_| SystemdTransientError::JobTimedOut {
106      job: job.clone(),
107      timeout,
108    })?
109}
110
111async fn wait_for_job(
112  mut stream: JobRemovedStream,
113  job_path: ObjectPath<'_>,
114) -> Result<(), SystemdTransientError> {
115  while let Some(signal) = stream.next().await {
116    let args = match signal.args() {
117      Ok(args) => args,
118      Err(err) => {
119        warn!("unparsable JobRemoved signal: {}", err);
120        continue;
121      }
122    };
123
124    if *args.job() != job_path {
125      continue;
126    }
127
128    return match *args.result() {
129      "done" => Ok(()),
130      r => Err(SystemdTransientError::JobWaitFailed {
131        unit: args.unit().to_string(),
132        reason: r.to_owned(),
133      }),
134    };
135  }
136
137  Err(SystemdTransientError::JobStreamEnded)
138}
systemd_transient/manager.rs

systemd_transient/
manager.rs
