eva_sdk/
eapi_bus.rs

1//! Helper module for EAPI micro-services
2use crate::service::{self, EventKind};
3use async_trait::async_trait;
4use busrt::QoS;
5use busrt::client::AsyncClient;
6use busrt::rpc::{Rpc, RpcClient, RpcEvent, RpcHandlers};
7use eva_common::acl::OIDMask;
8use eva_common::common_payloads::ParamsId;
9use eva_common::events::{RAW_STATE_TOPIC, RawStateEvent};
10use eva_common::payload::{pack, unpack};
11use eva_common::prelude::*;
12use eva_common::services::Initial;
13use eva_common::services::Registry;
14use log::error;
15use serde::{Deserialize, Deserializer, Serialize, Serializer, ser::SerializeSeq};
16use std::collections::BTreeMap;
17use std::future::Future;
18use std::sync::{Arc, OnceLock};
19use std::time::Duration;
20use tokio::sync::Mutex;
21use uuid::Uuid;
22
23pub const AAA_REPORT_TOPIC: &str = "AAA/REPORT";
24
25static RPC: OnceLock<Arc<RpcClient>> = OnceLock::new();
26static RPC_SECONDARY: OnceLock<Arc<RpcClient>> = OnceLock::new();
27static REGISTRY: OnceLock<Arc<Registry>> = OnceLock::new();
28static CLIENT: OnceLock<Arc<Mutex<dyn AsyncClient>>> = OnceLock::new();
29static TIMEOUT: OnceLock<Duration> = OnceLock::new();
30
31pub enum LvarCommand<'a> {
32    Set {
33        status: ItemStatus,
34        value: &'a Value,
35    },
36    Reset,
37    Clear,
38    Toggle,
39    Increment,
40    Decrement,
41}
42
43impl LvarCommand<'_> {
44    pub fn as_str(&self) -> &'static str {
45        match self {
46            LvarCommand::Set { .. } => "lvar.set",
47            LvarCommand::Reset => "lvar.reset",
48            LvarCommand::Clear => "lvar.clear",
49            LvarCommand::Toggle => "lvar.toggle",
50            LvarCommand::Increment => "lvar.incr",
51            LvarCommand::Decrement => "lvar.decr",
52        }
53    }
54    /// returns new lvar value for increment/decrement, zero for others
55    pub async fn execute(&self, oid: &OID) -> EResult<i64> {
56        #[derive(Serialize)]
57        struct Payload<'a> {
58            i: &'a OID,
59            #[serde(skip_serializing_if = "Option::is_none")]
60            status: Option<ItemStatus>,
61            #[serde(skip_serializing_if = "Option::is_none")]
62            value: Option<&'a Value>,
63        }
64        let payload = Payload {
65            i: oid,
66            status: match self {
67                LvarCommand::Set { status, .. } => Some(*status),
68                LvarCommand::Reset
69                | LvarCommand::Clear
70                | LvarCommand::Toggle
71                | LvarCommand::Increment
72                | LvarCommand::Decrement => None,
73            },
74            value: match self {
75                LvarCommand::Set { value, .. } => Some(value),
76                LvarCommand::Reset
77                | LvarCommand::Clear
78                | LvarCommand::Toggle
79                | LvarCommand::Increment
80                | LvarCommand::Decrement => None,
81            },
82        };
83        let res = call("eva.core", self.as_str(), pack(&payload)?.into()).await?;
84        match self {
85            LvarCommand::Set { .. }
86            | LvarCommand::Reset
87            | LvarCommand::Clear
88            | LvarCommand::Toggle => Ok(0),
89            LvarCommand::Increment | LvarCommand::Decrement => {
90                let value: i64 = unpack(res.payload())?;
91                Ok(value)
92            }
93        }
94    }
95}
96
97pub async fn lvar_set(oid: &OID, status: ItemStatus, value: &Value) -> EResult<()> {
98    LvarCommand::Set { status, value }.execute(oid).await?;
99    Ok(())
100}
101
102pub async fn lvar_reset(oid: &OID) -> EResult<()> {
103    LvarCommand::Reset.execute(oid).await?;
104    Ok(())
105}
106
107pub async fn lvar_clear(oid: &OID) -> EResult<()> {
108    LvarCommand::Clear.execute(oid).await?;
109    Ok(())
110}
111
112pub async fn lvar_toggle(oid: &OID) -> EResult<()> {
113    LvarCommand::Toggle.execute(oid).await?;
114    Ok(())
115}
116
117pub async fn lvar_increment(oid: &OID) -> EResult<i64> {
118    LvarCommand::Increment.execute(oid).await
119}
120
121pub async fn lvar_decrement(oid: &OID) -> EResult<i64> {
122    LvarCommand::Decrement.execute(oid).await
123}
124
125#[async_trait]
126pub trait ClientAccounting {
127    async fn report<'a, T>(&self, event: T) -> EResult<()>
128    where
129        T: TryInto<busrt::borrow::Cow<'a>> + Send;
130}
131
132#[async_trait]
133impl ClientAccounting for Arc<Mutex<dyn AsyncClient>> {
134    /// # Panics
135    ///
136    /// Will panic if RPC not set
137    async fn report<'a, T>(&self, event: T) -> EResult<()>
138    where
139        T: TryInto<busrt::borrow::Cow<'a>> + Send,
140    {
141        let payload: busrt::borrow::Cow = event
142            .try_into()
143            .map_err(|_| Error::invalid_data("Unable to serialize accounting event"))?;
144        self.lock()
145            .await
146            .publish(AAA_REPORT_TOPIC, payload, QoS::Processed)
147            .await?;
148        Ok(())
149    }
150}
151
152#[allow(clippy::ref_option)]
153fn serialize_opt_uuid_as_seq<S>(uuid: &Option<Uuid>, serializer: S) -> Result<S::Ok, S::Error>
154where
155    S: Serializer,
156{
157    if let Some(u) = uuid {
158        let bytes = u.as_bytes();
159        let mut seq = serializer.serialize_seq(Some(bytes.len()))?;
160        for &byte in bytes {
161            seq.serialize_element(&byte)?;
162        }
163        seq.end()
164    } else {
165        serializer.serialize_none()
166    }
167}
168
169fn deserialize_opt_uuid<'de, D>(deserializer: D) -> Result<Option<Uuid>, D::Error>
170where
171    D: Deserializer<'de>,
172{
173    let val: Value = Deserialize::deserialize(deserializer)?;
174    if val == Value::Unit {
175        Ok(None)
176    } else {
177        Ok(Some(
178            Uuid::deserialize(val).map_err(serde::de::Error::custom)?,
179        ))
180    }
181}
182
183#[derive(Serialize, Deserialize, Default)]
184pub struct AccountingEvent<'a> {
185    // the ID is usually assigned by the accounting service and should be None
186    #[serde(
187        default,
188        skip_serializing_if = "Option::is_none",
189        serialize_with = "serialize_opt_uuid_as_seq",
190        deserialize_with = "deserialize_opt_uuid"
191    )]
192    pub id: Option<Uuid>,
193    #[serde(skip_serializing_if = "Option::is_none")]
194    pub u: Option<&'a str>,
195    #[serde(skip_serializing_if = "Option::is_none")]
196    pub src: Option<&'a str>,
197    /// optional, if not filled, the current service id is used
198    #[serde(skip_serializing_if = "Option::is_none")]
199    pub svc: Option<&'a str>,
200    #[serde(skip_serializing_if = "Option::is_none")]
201    pub subj: Option<&'a str>,
202    #[serde(skip_serializing_if = "Option::is_none")]
203    pub oid: Option<OID>,
204    #[serde(default, skip_serializing_if = "Value::is_unit")]
205    pub data: Value,
206    #[serde(skip_serializing_if = "Option::is_none")]
207    pub note: Option<&'a str>,
208    #[serde(default)]
209    /// 0 means OK, negative values mean standard EAPI errors, positive values can be used for
210    /// custom error codes
211    pub code: i16,
212    #[serde(skip_serializing_if = "Option::is_none")]
213    pub err: Option<String>,
214}
215
216impl<'a> TryFrom<AccountingEvent<'a>> for busrt::borrow::Cow<'_> {
217    type Error = Error;
218    #[inline]
219    fn try_from(ev: AccountingEvent<'a>) -> EResult<Self> {
220        Ok(busrt::borrow::Cow::Owned(pack(&ev)?))
221    }
222}
223
224impl<'a> TryFrom<&AccountingEvent<'a>> for busrt::borrow::Cow<'_> {
225    type Error = Error;
226    #[inline]
227    fn try_from(ev: &AccountingEvent<'a>) -> EResult<Self> {
228        Ok(busrt::borrow::Cow::Owned(pack(&ev)?))
229    }
230}
231
232impl<'a> AccountingEvent<'a> {
233    #[inline]
234    pub fn new() -> Self {
235        Self::default()
236    }
237    #[inline]
238    pub fn user(mut self, user: &'a str) -> Self {
239        self.u.replace(user);
240        self
241    }
242    #[inline]
243    pub fn src(mut self, src: &'a str) -> Self {
244        self.src.replace(src);
245        self
246    }
247    #[inline]
248    pub fn svc(mut self, svc: &'a str) -> Self {
249        self.svc.replace(svc);
250        self
251    }
252    #[inline]
253    pub fn subj(mut self, subj: &'a str) -> Self {
254        self.subj.replace(subj);
255        self
256    }
257    #[inline]
258    pub fn data(mut self, data: Value) -> Self {
259        self.data = data;
260        self
261    }
262    #[inline]
263    pub fn note(mut self, note: &'a str) -> Self {
264        self.note.replace(note);
265        self
266    }
267    #[inline]
268    pub fn code(mut self, code: i16) -> Self {
269        self.code = code;
270        self
271    }
272    #[inline]
273    pub fn err(mut self, err: impl Into<String>) -> Self {
274        self.err.replace(err.into());
275        self
276    }
277    /// # Panics
278    ///
279    /// Will panic if module RPC not set
280    pub async fn report(&self) -> EResult<()> {
281        client().report(self).await
282    }
283}
284
285/// Initializes the module
286pub async fn init<H: RpcHandlers + Send + Sync + 'static>(
287    initial: &Initial,
288    handlers: H,
289) -> EResult<Arc<RpcClient>> {
290    let rpc = initial.init_rpc(handlers).await?;
291    set(rpc.clone(), initial.timeout())?;
292    let registry = initial.init_registry(&rpc);
293    set_registry(registry)?;
294    Ok(rpc)
295}
296
297/// Initializes the module in blocking mode with secondary client for calls
298pub async fn init_blocking<H: RpcHandlers + Send + Sync + 'static>(
299    initial: &Initial,
300    handlers: H,
301) -> EResult<(Arc<RpcClient>, Arc<RpcClient>)> {
302    let (rpc, rpc_secondary) = initial.init_rpc_blocking_with_secondary(handlers).await?;
303    set_blocking(rpc.clone(), rpc_secondary.clone(), initial.timeout())?;
304    let registry = initial.init_registry(&rpc);
305    set_registry(registry)?;
306    Ok((rpc, rpc_secondary))
307}
308
309/// Manually initialize the module
310pub fn set(rpc: Arc<RpcClient>, timeout: Duration) -> EResult<()> {
311    CLIENT
312        .set(rpc.client())
313        .map_err(|_| Error::core("Unable to set CLIENT"))?;
314    RPC.set(rpc).map_err(|_| Error::core("Unable to set RPC"))?;
315    TIMEOUT
316        .set(timeout)
317        .map_err(|_| Error::core("Unable to set TIMEOUT"))?;
318    Ok(())
319}
320
321/// Manually initialize the module
322pub fn set_blocking(
323    rpc: Arc<RpcClient>,
324    rpc_secondary: Arc<RpcClient>,
325    timeout: Duration,
326) -> EResult<()> {
327    set(rpc, timeout)?;
328    RPC_SECONDARY
329        .set(rpc_secondary)
330        .map_err(|_| Error::core("Unable to set RPC_SECONDARY"))?;
331    Ok(())
332}
333
334/// Manually initialize registry
335pub fn set_registry(registry: Registry) -> EResult<()> {
336    REGISTRY
337        .set(Arc::new(registry))
338        .map_err(|_| Error::core("Unable to set REGISTRY"))
339}
340
341/// # Panics
342///
343/// Will panic if RPC not set
344pub async fn call0(target: &str, method: &str) -> EResult<RpcEvent> {
345    tokio::time::timeout(
346        timeout(),
347        rpc_secondary().call(target, method, busrt::empty_payload!(), QoS::Processed),
348    )
349    .await?
350    .map_err(Into::into)
351}
352
353/// # Panics
354///
355/// Will panic if RPC not set
356pub async fn call(target: &str, method: &str, params: busrt::borrow::Cow<'_>) -> EResult<RpcEvent> {
357    tokio::time::timeout(
358        timeout(),
359        rpc_secondary().call(target, method, params, QoS::Processed),
360    )
361    .await?
362    .map_err(Into::into)
363}
364
365/// # Panics
366///
367/// Will panic if RPC not set
368pub async fn call_with_timeout(
369    target: &str,
370    method: &str,
371    params: busrt::borrow::Cow<'_>,
372    timeout: Duration,
373) -> EResult<RpcEvent> {
374    tokio::time::timeout(
375        timeout,
376        rpc_secondary().call(target, method, params, QoS::Processed),
377    )
378    .await?
379    .map_err(Into::into)
380}
381
382/// # Panics
383///
384/// Will panic if RPC not set
385#[inline]
386pub fn rpc() -> Arc<RpcClient> {
387    RPC.get().cloned().unwrap()
388}
389
390/// Returns secondary client if initialized, otherwise fallbacks to the primary
391///
392/// Must be used for RPC calls when the primary client works in blocking mode
393///
394/// # Panics
395///
396/// Will panic if RPC not set
397#[inline]
398pub fn rpc_secondary() -> Arc<RpcClient> {
399    if let Some(rpc) = RPC_SECONDARY.get() {
400        rpc.clone()
401    } else {
402        rpc()
403    }
404}
405
406/// # Panics
407///
408/// Will panic if RPC not set
409#[inline]
410pub fn client() -> Arc<Mutex<dyn AsyncClient>> {
411    CLIENT.get().cloned().unwrap()
412}
413
414/// Will return the default EVA ICS timeout (5 sec) if not set
415#[inline]
416pub fn timeout() -> Duration {
417    TIMEOUT
418        .get()
419        .copied()
420        .unwrap_or(eva_common::DEFAULT_TIMEOUT)
421}
422
423/// # Panics
424///
425/// Will panic if REGISTRY not set
426#[inline]
427pub fn registry() -> Arc<Registry> {
428    REGISTRY.get().cloned().unwrap()
429}
430
431///
432/// # Panics
433///
434/// Will panic if RPC not set
435#[inline]
436pub async fn subscribe(topic: &str) -> EResult<()> {
437    tokio::time::timeout(timeout(), subscribe_impl(topic)).await??;
438    Ok(())
439}
440#[inline]
441async fn subscribe_impl(topic: &str) -> EResult<()> {
442    let Some(op) = client()
443        .lock()
444        .await
445        .subscribe(topic, QoS::Processed)
446        .await?
447    else {
448        return Ok(());
449    };
450    op.await??;
451    Ok(())
452}
453
454///
455/// # Panics
456///
457/// Will panic if RPC not set
458#[inline]
459pub async fn subscribe_bulk(topics: &[&str]) -> EResult<()> {
460    tokio::time::timeout(timeout(), subscribe_bulk_impl(topics)).await??;
461    Ok(())
462}
463#[inline]
464async fn subscribe_bulk_impl(topics: &[&str]) -> EResult<()> {
465    let Some(op) = client()
466        .lock()
467        .await
468        .subscribe_bulk(topics, QoS::Processed)
469        .await?
470    else {
471        return Ok(());
472    };
473    op.await??;
474    Ok(())
475}
476
477#[inline]
478pub async fn publish_item_state(
479    oid: &OID,
480    status: ItemStatus,
481    value: Option<&Value>,
482) -> EResult<()> {
483    let ev = value.map_or_else(
484        || RawStateEvent::new0(status),
485        |v| RawStateEvent::new(status, v),
486    );
487    publish_item_state_event(oid, ev).await
488}
489
490#[inline]
491pub async fn publish_item_state_event(oid: &OID, ev: RawStateEvent<'_>) -> EResult<()> {
492    let topic = format!("{}{}", RAW_STATE_TOPIC, oid.as_path());
493    publish(&topic, pack(&ev)?.into()).await?;
494    Ok(())
495}
496
497/// # Panics
498///
499/// Will panic if RPC not set
500#[inline]
501pub async fn publish(topic: &str, payload: busrt::borrow::Cow<'_>) -> EResult<()> {
502    tokio::time::timeout(timeout(), publish_impl(topic, payload)).await??;
503    Ok(())
504}
505#[inline]
506async fn publish_impl(topic: &str, payload: busrt::borrow::Cow<'_>) -> EResult<()> {
507    let Some(op) = client()
508        .lock()
509        .await
510        .publish(topic, payload, QoS::No)
511        .await?
512    else {
513        return Ok(());
514    };
515    op.await??;
516    Ok(())
517}
518
519/// # Panics
520///
521/// Will panic if RPC not set
522#[inline]
523pub async fn publish_confirmed(topic: &str, payload: busrt::borrow::Cow<'_>) -> EResult<()> {
524    tokio::time::timeout(timeout(), publish_confirmed_impl(topic, payload)).await??;
525    Ok(())
526}
527#[inline]
528async fn publish_confirmed_impl(topic: &str, payload: busrt::borrow::Cow<'_>) -> EResult<()> {
529    let Some(op) = client()
530        .lock()
531        .await
532        .publish(topic, payload, QoS::Processed)
533        .await?
534    else {
535        return Ok(());
536    };
537    op.await??;
538    Ok(())
539}
540
541/// # Panics
542///
543/// Will panic if RPC not set
544#[inline]
545pub async fn subscribe_oids<'a, M>(masks: M, kind: EventKind) -> EResult<()>
546where
547    M: IntoIterator<Item = &'a OIDMask>,
548{
549    tokio::time::timeout(timeout(), subscribe_oids_impl(masks, kind)).await??;
550    Ok(())
551}
552#[inline]
553async fn subscribe_oids_impl<'a, M>(masks: M, kind: EventKind) -> EResult<()>
554where
555    M: IntoIterator<Item = &'a OIDMask>,
556{
557    service::subscribe_oids(rpc().as_ref(), masks, kind).await
558}
559
560/// # Panics
561///
562/// Will panic if RPC not set
563#[inline]
564pub async fn unsubscribe_oids<'a, M>(masks: M, kind: EventKind) -> EResult<()>
565where
566    M: IntoIterator<Item = &'a OIDMask>,
567{
568    tokio::time::timeout(timeout(), unsubscribe_oids_impl(masks, kind)).await??;
569    Ok(())
570}
571#[inline]
572async fn unsubscribe_oids_impl<'a, M>(masks: M, kind: EventKind) -> EResult<()>
573where
574    M: IntoIterator<Item = &'a OIDMask>,
575{
576    service::unsubscribe_oids(rpc().as_ref(), masks, kind).await
577}
578
579/// Request announce for the specific items, the core will send states to item state topics,
580/// exclusively for the current service. As the method locks the core inventory for updates, it
581/// should be used with caution
582///
583/// The method will return immediately if the core is inactive as all the states will be
584/// automaticaly announced when the node goes to ready state
585#[inline]
586pub async fn request_announce<'a, M>(masks: M, kind: EventKind) -> EResult<()>
587where
588    M: IntoIterator<Item = &'a OIDMask>,
589{
590    #[derive(Serialize)]
591    struct Payload<'a> {
592        i: Vec<&'a OIDMask>,
593        src: Option<&'a str>,
594        broadcast: bool,
595    }
596    if !service::svc_is_core_active(rpc().as_ref(), timeout()).await {
597        // the core is inactive, no need to announce
598        return Ok(());
599    }
600    let payload = Payload {
601        i: masks.into_iter().collect(),
602        src: match kind {
603            EventKind::Actual | EventKind::Any => None,
604            EventKind::Local => Some(".local"),
605            EventKind::Remote => Some(".remote-any"),
606        },
607        broadcast: false,
608    };
609    call("eva.core", "item.announce", pack(&payload)?.into()).await?;
610    Ok(())
611}
612
613/// # Panics
614///
615/// Will panic if RPC not set
616#[inline]
617pub async fn exclude_oids<'a, M>(masks: M, kind: EventKind) -> EResult<()>
618where
619    M: IntoIterator<Item = &'a OIDMask>,
620{
621    tokio::time::timeout(timeout(), exclude_oids_impl(masks, kind)).await??;
622    Ok(())
623}
624#[inline]
625async fn exclude_oids_impl<'a, M>(masks: M, kind: EventKind) -> EResult<()>
626where
627    M: IntoIterator<Item = &'a OIDMask>,
628{
629    service::exclude_oids(rpc().as_ref(), masks, kind).await
630}
631
632/// Returns true if the core was inactive (not ready) and the service has been waiting for it,
633/// false if the core was already active
634///
635/// # Panics
636///
637/// Will panic if RPC not set
638#[inline]
639pub async fn wait_core(wait_forever: bool) -> EResult<bool> {
640    service::svc_wait_core(rpc().as_ref(), timeout(), wait_forever).await
641}
642
643/// # Panics
644///
645/// Will panic if RPC not set
646#[inline]
647pub fn init_logs(initial: &Initial) -> EResult<()> {
648    service::svc_init_logs(initial, client())
649}
650
651/// calls mark_ready, block and mark_terminating
652///
653/// # Panics
654///
655/// Will panic if RPC not set
656pub async fn run() -> EResult<()> {
657    mark_ready().await?;
658    block().await;
659    mark_terminating().await?;
660    Ok(())
661}
662
663/// # Panics
664///
665/// Will panic if RPC not set
666#[inline]
667pub async fn mark_ready() -> EResult<()> {
668    service::svc_mark_ready(&client()).await
669}
670
671/// # Panics
672///
673/// Will panic if RPC not set
674#[inline]
675pub async fn mark_terminating() -> EResult<()> {
676    service::svc_mark_terminating(&client()).await
677}
678
679/// Must be called once
680pub fn set_bus_error_suicide_timeout(bes_timeout: Duration) -> EResult<()> {
681    service::set_bus_error_suicide_timeout(bes_timeout)
682}
683
684/// Blocks the service while active
685///
686/// In case if the local bus connection is dropped, the service is terminated immediately, as well
687/// as all its subprocesses
688///
689/// This behaviour can be changed by calling set_bus_error_suicide_timeout method and specifying a
690/// proper required shutdown timeout until the service is killed
691///
692/// # Panics
693///
694/// Will panic if RPC not set
695#[inline]
696pub async fn block() {
697    if let Some(secondary) = RPC_SECONDARY.get() {
698        service::svc_block2(rpc().as_ref(), secondary).await;
699    } else {
700        service::svc_block(rpc().as_ref()).await;
701    }
702}
703
704/// Creates items, ignores errors if an item already exists
705///
706/// Must be called after the node core is ready
707///
708/// # Panics
709///
710/// Will panic if RPC not set
711pub async fn create_items<O: AsRef<OID>>(oids: &[O]) -> EResult<()> {
712    for oid in oids {
713        let payload = ParamsId {
714            i: oid.as_ref().as_str(),
715        };
716        if let Err(e) = call("eva.core", "item.create", pack(&payload)?.into()).await
717            && e.kind() != ErrorKind::ResourceAlreadyExists
718        {
719            return Err(e);
720        }
721    }
722    Ok(())
723}
724///
725/// Deploys items
726///
727/// Must be called after the node core is ready
728///
729/// The parameter must contain a list of item deployment payloads is equal to item.deploy
730/// eva.core EAPI call
731/// See also https://info.bma.ai/en/actual/eva4/iac.html#items
732///
733/// The parameter MUST be a collection: either inside the payload, or a list of payloads in
734/// a vector/slice etc.
735///
736/// Example:
737///
738/// ```rust,ignore
739/// let me = initial.id().to_owned();
740/// tokio::spawn(async move {
741///   let _ = eapi_bus::wait_core(true).await;
742///     let x: OID = "lmacro:aaa".parse().unwrap();
743///     let payload = serde_json::json! {[
744///        {
745///          "oid": x,
746///           "action": {"svc": me }
747///        }
748///        ]};
749///        let result = eapi_bus::deploy_items(&payload).await;
750/// });
751/// ```
752///
753/// # Panics
754///
755/// Will panic if RPC not set
756pub async fn deploy_items<T: Serialize>(items: &T) -> EResult<()> {
757    #[derive(Serialize)]
758    struct Payload<'a, T: Serialize> {
759        items: &'a T,
760    }
761    call("eva.core", "item.deploy", pack(&Payload { items })?.into()).await?;
762    Ok(())
763}
764
765pub async fn undeploy_items<T: Serialize>(items: &T) -> EResult<()> {
766    #[derive(Serialize)]
767    struct Payload<'a, T: Serialize> {
768        items: &'a T,
769    }
770    call(
771        "eva.core",
772        "item.undeploy",
773        pack(&Payload { items })?.into(),
774    )
775    .await?;
776    Ok(())
777}
778
779#[derive(Serialize, Debug, Clone)]
780pub struct ParamsRunLmacro {
781    #[serde(skip_serializing_if = "Vec::is_empty")]
782    args: Vec<Value>,
783    #[serde(skip_serializing_if = "BTreeMap::is_empty")]
784    kwargs: BTreeMap<String, Value>,
785    #[serde(skip_serializing)]
786    wait: Duration,
787    #[serde(skip_serializing)]
788    timeout_if_not_finished: bool,
789}
790
791impl Default for ParamsRunLmacro {
792    fn default() -> Self {
793        Self::new()
794    }
795}
796
797impl ParamsRunLmacro {
798    pub fn new() -> Self {
799        Self {
800            args: <_>::default(),
801            kwargs: BTreeMap::new(),
802            wait: eva_common::DEFAULT_TIMEOUT,
803            timeout_if_not_finished: true,
804        }
805    }
806    pub fn arg<V: Serialize>(mut self, arg: V) -> EResult<Self> {
807        self.args.push(to_value(arg)?);
808        Ok(self)
809    }
810    pub fn args<V, I>(mut self, args: I) -> EResult<Self>
811    where
812        V: Serialize,
813        I: IntoIterator<Item = V>,
814    {
815        for arg in args {
816            self.args.push(to_value(arg)?);
817        }
818        Ok(self)
819    }
820    pub fn kwarg<V: Serialize>(mut self, key: impl Into<String>, value: V) -> EResult<Self> {
821        self.kwargs.insert(key.into(), to_value(value)?);
822        Ok(self)
823    }
824    pub fn kwargs<V, I>(mut self, kwargs: I) -> EResult<Self>
825    where
826        V: Serialize,
827        I: IntoIterator<Item = (String, V)>,
828    {
829        for (key, value) in kwargs {
830            self.kwargs.insert(key, to_value(value)?);
831        }
832        Ok(self)
833    }
834    pub fn wait(mut self, wait: Duration) -> Self {
835        self.wait = wait;
836        self
837    }
838    /// Does not return an error if the lmacro action is not finished
839    pub fn allow_unfinished(mut self) -> Self {
840        self.timeout_if_not_finished = false;
841        self
842    }
843}
844
845#[derive(Serialize, Debug, Clone)]
846pub struct ParamsUnitAction {
847    value: Value,
848    #[serde(skip_serializing)]
849    wait: Duration,
850    #[serde(skip_serializing)]
851    timeout_if_not_finished: bool,
852}
853
854impl ParamsUnitAction {
855    pub fn new(value: Value) -> Self {
856        Self {
857            value,
858            wait: eva_common::DEFAULT_TIMEOUT,
859            timeout_if_not_finished: true,
860        }
861    }
862    pub fn wait(mut self, wait: Duration) -> Self {
863        self.wait = wait;
864        self
865    }
866    /// Does not return an error if the unit action is not finished
867    pub fn allow_unfinished(mut self) -> Self {
868        self.timeout_if_not_finished = false;
869        self
870    }
871}
872
873#[derive(Deserialize, Debug, Clone)]
874struct ActionState {
875    exitcode: Option<i16>,
876    #[serde(default)]
877    finished: bool,
878    #[serde(default)]
879    out: Value,
880    #[serde(default)]
881    err: Value,
882}
883
884pub async fn unit_action(i: &OID, params: &ParamsUnitAction) -> EResult<Value> {
885    #[derive(Serialize)]
886    struct Params<'a> {
887        i: &'a OID,
888        #[allow(clippy::struct_field_names)]
889        params: &'a ParamsUnitAction,
890        #[serde(serialize_with = "eva_common::tools::serialize_duration_as_f64")]
891        wait: Duration,
892    }
893    let p = Params {
894        i,
895        params,
896        wait: params.wait,
897    };
898    let payload = pack(&p)?;
899    let recommended_timeout = params.wait + Duration::from_millis(500);
900    let timeout = timeout().max(recommended_timeout);
901    let res: ActionState = unpack(
902        call_with_timeout("eva.core", "action", payload.into(), timeout)
903            .await?
904            .payload(),
905    )?;
906    if (!res.finished || res.exitcode.is_none()) && params.timeout_if_not_finished {
907        return Err(Error::timeout());
908    }
909    if res.exitcode.is_some_and(|code| code != 0) {
910        return Err(Error::failed(res.err.to_string()));
911    }
912    Ok(res.out)
913}
914
915pub async fn run_lmacro(i: &OID, params: &ParamsRunLmacro) -> EResult<Value> {
916    #[derive(Serialize)]
917    struct Params<'a> {
918        i: &'a OID,
919        #[allow(clippy::struct_field_names)]
920        params: &'a ParamsRunLmacro,
921        #[serde(serialize_with = "eva_common::tools::serialize_duration_as_f64")]
922        wait: Duration,
923    }
924    let p = Params {
925        i,
926        params,
927        wait: params.wait,
928    };
929    let payload = pack(&p)?;
930    let recommended_timeout = params.wait + Duration::from_millis(500);
931    let timeout = timeout().max(recommended_timeout);
932    let res: ActionState = unpack(
933        call_with_timeout("eva.core", "run", payload.into(), timeout)
934            .await?
935            .payload(),
936    )?;
937    if (!res.finished || res.exitcode.is_none()) && params.timeout_if_not_finished {
938        return Err(Error::timeout());
939    }
940    if let Some(code) = res.exitcode
941        && code != 0
942    {
943        return Err(Error::failed(res.err.to_string()));
944    }
945    Ok(res.out)
946}
947
948/// Spawns a future which is executed after the node core is ready
949pub fn spawn_when_ready<F>(future: F)
950where
951    F: Future + Send + 'static,
952    F::Output: Send + 'static,
953{
954    tokio::spawn(async move {
955        if let Err(e) = wait_core(true).await {
956            error!("Failed to wait for core: {}", e);
957            return;
958        }
959        future.await;
960    });
961}
eva_sdk/eapi_bus.rs

eva_sdk/
eapi_bus.rs
