sysd_manager_comcontroler/sysdbus/
to_proxy.rs

1#![allow(dead_code)]
2
3use std::sync::OnceLock;
4
5use base::{
6    consts::{MAX_HEART_BEAT_ELAPSE, MIN_HEART_BEAT_ELAPSE},
7    enums::UnitDBusLevel,
8    proxy::{DisEnAbleUnitFiles, DisEnAbleUnitFilesResponse},
9};
10use futures_util::stream::StreamExt;
11use tokio::{
12    task::JoinHandle,
13    time::{self, Duration},
14};
15use tracing::{debug, info, warn};
16use zbus::proxy;
17use zvariant::OwnedObjectPath;
18
19use crate::{
20    errors::SystemdErrors,
21    sysdbus::{get_blocking_connection, get_connection, run_context},
22};
23
24#[proxy(
25    interface = "io.github.plrigaux.SysDManager",
26    default_service = "io.github.plrigaux.SysDManager",
27    default_path = "/io/github/plrigaux/SysDManager"
28)]
29pub trait SysDManagerComLink {
30    fn start_unit(&self, unit_name: &str, mode: &str) -> zbus::fdo::Result<OwnedObjectPath>;
31    fn stop_unit(&self, unit_name: &str, mode: &str) -> zbus::fdo::Result<OwnedObjectPath>;
32    fn restart_unit(&self, unit_name: &str, mode: &str) -> zbus::fdo::Result<OwnedObjectPath>;
33    fn reload_unit(&self, unit_name: &str, mode: &str) -> zbus::fdo::Result<OwnedObjectPath>;
34    fn clean_unit(&self, unit_name: &str, what: &[&str]) -> zbus::Result<()>;
35    fn freeze_unit(&self, unit_name: &str) -> zbus::fdo::Result<()>;
36    fn thaw_unit(&self, unit_name: &str) -> zbus::fdo::Result<()>;
37    fn reload(&self) -> zbus::fdo::Result<()>;
38
39    fn create_drop_in(
40        &mut self,
41        runtime: bool,
42        unit_name: &str,
43        file_name: &str,
44        content: &str,
45    ) -> zbus::fdo::Result<()>;
46    fn save_file(&mut self, file_name: &str, content: &str) -> zbus::fdo::Result<u64>;
47
48    fn revert_unit_files(&self, file_names: &[&str]) -> zbus::fdo::Result<Vec<DisEnAbleUnitFiles>>;
49
50    fn enable_unit_files_with_flags(
51        &mut self,
52        files: &[&str],
53        flags: u64,
54    ) -> zbus::fdo::Result<DisEnAbleUnitFilesResponse>;
55
56    fn disable_unit_files_with_flags(
57        &mut self,
58        files: &[&str],
59        flags: u64,
60    ) -> zbus::fdo::Result<DisEnAbleUnitFilesResponse>;
61
62    #[zbus(signal)]
63    fn hello(msg: String) -> zbus::fdo::Result<()>;
64
65    fn heart_beat(&self) -> zbus::fdo::Result<u64>;
66}
67
68///1 Ensure that the  proxy is up and running
69///2 Tertemine mode
70///2 Connect to the proxy and return a proxy object
71fn ensure_proxy_up() {
72    //TODO ensure_proxy_up
73}
74
75fn get_proxy<'a>() -> Result<SysDManagerComLinkProxyBlocking<'a>, SystemdErrors> {
76    let destination = run_context().destination_address();
77    let connection = get_blocking_connection(UnitDBusLevel::System)?;
78
79    info!("BusName Destination {}", destination);
80    let proxy = SysDManagerComLinkProxyBlocking::builder(&connection)
81        // .path(path)?
82        .destination(destination)?
83        .build()?;
84
85    Ok(proxy)
86}
87
88pub(crate) async fn get_proxy_async<'a>() -> Result<SysDManagerComLinkProxy<'a>, SystemdErrors> {
89    let destination = super::RUN_CONTEXT
90        .get()
91        .expect("Supposed to be init")
92        .destination_address();
93    let connection = get_connection(UnitDBusLevel::System).await?;
94
95    info!("BusName Destination {}", destination);
96    let proxy = SysDManagerComLinkProxy::builder(&connection)
97        //.path(path)?
98        .destination(destination)?
99        .build()
100        .await?;
101
102    Ok(proxy)
103}
104
105pub fn start_unit(unit_name: &str, mode: &str) -> Result<String, SystemdErrors> {
106    let proxy = get_proxy()?;
107
108    let s = proxy.start_unit(unit_name, mode)?;
109    Ok(s.to_string())
110}
111
112pub fn stop_unit(unit_name: &str, mode: &str) -> Result<String, SystemdErrors> {
113    let proxy = get_proxy()?;
114
115    let s = proxy.stop_unit(unit_name, mode)?;
116    Ok(s.to_string())
117}
118
119pub fn restart_unit(unit_name: &str, mode: &str) -> Result<String, SystemdErrors> {
120    let proxy = get_proxy()?;
121
122    let s = proxy.restart_unit(unit_name, mode)?;
123    Ok(s.to_string())
124}
125
126pub fn clean_unit(unit_name: &str, what: &[&str]) -> Result<(), SystemdErrors> {
127    let proxy = get_proxy()?;
128
129    proxy.clean_unit(unit_name, what)?;
130    Ok(())
131}
132
133pub fn freeze_unit(unit_name: &str) -> Result<(), SystemdErrors> {
134    let proxy = get_proxy()?;
135    proxy.freeze_unit(unit_name)?;
136    Ok(())
137}
138
139pub fn thaw_unit(unit_name: &str) -> Result<(), SystemdErrors> {
140    let proxy: SysDManagerComLinkProxyBlocking<'_> = get_proxy()?;
141    proxy.thaw_unit(unit_name)?;
142    Ok(())
143}
144
145pub async fn reload() -> Result<(), SystemdErrors> {
146    let proxy = get_proxy_async().await?;
147    proxy.reload().await?;
148    Ok(())
149}
150
151fn extract_job_id(job: &str) -> Option<u32> {
152    job.rsplit_once('/')
153        .and_then(|(_, id)| id.parse::<u32>().ok())
154}
155
156pub async fn lazy_start_proxy_async() -> Result<(), SystemdErrors> {
157    crate::sysdbus::init_proxy_async2().await?;
158
159    Ok(())
160}
161
162pub(crate) async fn wait_hello(mut hello_stream: HelloStream) -> Result<(), SystemdErrors> {
163    if let Some(msg) = hello_stream.next().await {
164        let args = msg.args()?;
165        info!("Hello Proxy Args {:?}", args);
166    }
167    Ok(())
168}
169
170pub fn lazy_start_proxy_block() -> Result<(), SystemdErrors> {
171    crate::runtime().block_on(async move {
172        warn!("lazy 1");
173        lazy_start_proxy_async().await;
174        warn!("lazy 2");
175    });
176    Ok(())
177}
178
179static HEART_BEAT_HANDLE: OnceLock<JoinHandle<Result<(), SystemdErrors>>> = OnceLock::new();
180pub(crate) fn start_heart_beat() {
181    if let Some(join_handle) = HEART_BEAT_HANDLE.get()
182        && !join_handle.is_finished()
183    {
184        warn!("There is already an heart beat thread running");
185    } else {
186        info!("Starting Heart Beat");
187        let handle = tokio::spawn(send_heart_beat());
188        HEART_BEAT_HANDLE.set(handle);
189    }
190}
191
192async fn send_heart_beat() -> Result<(), SystemdErrors> {
193    let proxy = get_proxy_async().await?;
194    loop {
195        match proxy.heart_beat().await {
196            Ok(delay) => {
197                debug!("Heath Beat delay {delay} millis");
198                let delay = delay.clamp(MIN_HEART_BEAT_ELAPSE, MAX_HEART_BEAT_ELAPSE);
199                time::sleep(Duration::from_millis(delay)).await;
200            }
201            Err(err) => {
202                warn!("Send Heart Beat Error: {err:?}");
203                return Err(err.into());
204            }
205        }
206    }
207}
208
209#[macro_export]
210macro_rules! proxy_call_blocking {
211    ($func:ident, $($param:expr),+) => {
212        match $crate::to_proxy::$func($($param),+) {
213            Ok(ok) => Ok(ok),
214            Err(SystemdErrors::ZFdoServiceUnknowm(msg)) => {
215                warn!("Blocking ServiceUnkown: {:?} Func: {}", msg, stringify!($func));
216                $crate::to_proxy::lazy_start_proxy_block();
217                $crate::to_proxy::$func($($param),+)
218            },
219            Err(err) => Err(err)
220        }
221    }
222}
223
224#[macro_export]
225macro_rules! proxy_call_async {
226    ($func:ident) => {
227        proxy_call_async!($func,)
228    };
229
230    ($func:ident, $($param:expr),*) => {
231        match $crate::to_proxy::$func($($param),*).await {
232            Ok(ok) => Ok(ok),
233            Err(SystemdErrors::ZFdoServiceUnknowm(msg)) => {
234                warn!("Async ServiceUnkown: {:?} Function: {}", msg, stringify!($func));
235                $crate::to_proxy::lazy_start_proxy_async().await;
236                $crate::to_proxy::$func($($param),*).await
237            },
238            Err(err) => Err(err)
239        }
240    }
241}
242
243pub(crate) async fn create_drop_in(
244    runtime: bool,
245    unit_name: &str,
246    file_path: &str,
247    content: &str,
248) -> Result<(), SystemdErrors> {
249    let mut proxy = get_proxy_async().await?;
250    proxy
251        .create_drop_in(runtime, unit_name, file_path, content)
252        .await
253        .map_err(|e| e.into())
254}
255
256pub async fn save_file(file_path: &str, content: &str) -> Result<u64, SystemdErrors> {
257    let mut proxy = get_proxy_async().await?;
258    proxy
259        .save_file(file_path, content)
260        .await
261        .map_err(|e| e.into())
262}
263
264pub async fn revert_unit_files(
265    unit_names: &[&str],
266) -> Result<Vec<DisEnAbleUnitFiles>, SystemdErrors> {
267    let proxy = get_proxy_async().await?;
268    proxy
269        .revert_unit_files(unit_names)
270        .await
271        .map_err(|e| e.into())
272}
273
274pub fn enable_unit_files_with_flags(
275    unit_files: &[&str],
276    flags: u64,
277) -> Result<DisEnAbleUnitFilesResponse, SystemdErrors> {
278    let mut proxy: SysDManagerComLinkProxyBlocking<'_> = get_proxy()?;
279    proxy
280        .enable_unit_files_with_flags(unit_files, flags)
281        .map_err(|err| err.into())
282}
283
284pub fn disable_unit_files_with_flags(
285    unit_files: &[&str],
286    flags: u64,
287) -> Result<DisEnAbleUnitFilesResponse, SystemdErrors> {
288    let mut proxy: SysDManagerComLinkProxyBlocking<'_> = get_proxy()?;
289    proxy
290        .disable_unit_files_with_flags(unit_files, flags)
291        .map_err(|err| err.into())
292}
sysd_manager_comcontroler/sysdbus/to_proxy.rs

sysd_manager_comcontroler/sysdbus/
to_proxy.rs
