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

systemd_transient/
manager.rs
