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

systemd_transient/
manager.rs
