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

systemd_transient/
manager.rs
