pamoja_profile/
rules.rs

1//! Rules between nodes, as data: what one node's reading calls for at another node's
2//! actuator, written as a file, and the engine that runs it off a link.
3
4use std::collections::{BTreeMap, BTreeSet};
5use std::future::Future;
6use std::pin::Pin;
7
8use pamoja_codec::Codec;
9use pamoja_core::{Actuator, Error, Receive, Result, Transport};
10use pamoja_kit::{Edge, Trigger};
11use serde::{Deserialize, Serialize};
12
13/// Which side of a line a reading must be on for a condition to hold.
14#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
15#[serde(rename_all = "snake_case")]
16pub enum Compare {
17    /// The condition holds while the reading is above the threshold.
18    Above,
19    /// The condition holds while the reading is below the threshold.
20    Below,
21}
22
23/// A line drawn through one topic's readings, with the release band that stops it
24/// firing over and over.
25///
26/// In a rule file it is the `when` object, which names the line with `below` or `above`:
27///
28/// ```json
29/// { "topic": "garden/bed-1/moisture", "below": 30.0, "hysteresis": 5.0 }
30/// ```
31///
32/// The condition becomes true the moment a reading crosses the threshold in the named
33/// direction, and stops holding once a reading has come back past the threshold by the
34/// hysteresis; readings in between leave it as it was. That is a
35/// [`Trigger`](pamoja_kit::Trigger), and [`trigger`](Condition::trigger) builds it.
36#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
37#[serde(try_from = "ConditionFile", into = "ConditionFile")]
38pub struct Condition {
39    /// The topic whose readings are watched, exactly as the node publishes to it.
40    pub topic: String,
41    /// Which side of the threshold the condition holds on.
42    pub compare: Compare,
43    /// The line a reading crosses.
44    pub threshold: f32,
45    /// How far back past the line a reading must come for the condition to clear.
46    pub hysteresis: f32,
47}
48
49/// A condition as a rule file writes it, with the line named by the side it holds on.
50#[derive(Serialize, Deserialize)]
51#[serde(deny_unknown_fields)]
52struct ConditionFile {
53    topic: String,
54    #[serde(default, skip_serializing_if = "Option::is_none")]
55    above: Option<f32>,
56    #[serde(default, skip_serializing_if = "Option::is_none")]
57    below: Option<f32>,
58    #[serde(default)]
59    hysteresis: f32,
60}
61
62impl TryFrom<ConditionFile> for Condition {
63    type Error = String;
64
65    fn try_from(file: ConditionFile) -> core::result::Result<Self, String> {
66        let (compare, threshold) = match (file.above, file.below) {
67            (Some(line), None) => (Compare::Above, line),
68            (None, Some(line)) => (Compare::Below, line),
69            (Some(_), Some(_)) => {
70                return Err(format!(
71                    "the condition on `{}` names both `above` and `below`; a rule draws one line, so write a second rule for the other",
72                    file.topic
73                ))
74            }
75            (None, None) => {
76                return Err(format!(
77                    "the condition on `{}` needs a line to cross: `above` or `below`, such as \"below\": 30.0",
78                    file.topic
79                ))
80            }
81        };
82        Ok(Condition {
83            topic: file.topic,
84            compare,
85            threshold,
86            hysteresis: file.hysteresis,
87        })
88    }
89}
90
91impl From<Condition> for ConditionFile {
92    fn from(condition: Condition) -> Self {
93        let (above, below) = match condition.compare {
94            Compare::Above => (Some(condition.threshold), None),
95            Compare::Below => (None, Some(condition.threshold)),
96        };
97        ConditionFile {
98            topic: condition.topic,
99            above,
100            below,
101            hysteresis: condition.hysteresis,
102        }
103    }
104}
105
106impl Condition {
107    /// A condition that holds while a topic's readings are above a line.
108    ///
109    /// # Arguments
110    ///
111    /// * `topic` - the topic whose readings are watched.
112    /// * `threshold` - the line.
113    ///
114    /// # Returns
115    ///
116    /// The condition, with no release band yet.
117    pub fn above(topic: impl Into<String>, threshold: f32) -> Self {
118        Self {
119            topic: topic.into(),
120            compare: Compare::Above,
121            threshold,
122            hysteresis: 0.0,
123        }
124    }
125
126    /// A condition that holds while a topic's readings are below a line.
127    ///
128    /// # Arguments
129    ///
130    /// * `topic` - the topic whose readings are watched.
131    /// * `threshold` - the line.
132    ///
133    /// # Returns
134    ///
135    /// The condition, with no release band yet.
136    pub fn below(topic: impl Into<String>, threshold: f32) -> Self {
137        Self {
138            topic: topic.into(),
139            compare: Compare::Below,
140            threshold,
141            hysteresis: 0.0,
142        }
143    }
144
145    /// Sets the release band on the far side of the line.
146    ///
147    /// # Arguments
148    ///
149    /// * `hysteresis` - how far back past the line a reading must come to clear the
150    ///   condition.
151    ///
152    /// # Returns
153    ///
154    /// The condition, for chaining.
155    pub fn with_hysteresis(mut self, hysteresis: f32) -> Self {
156        self.hysteresis = hysteresis;
157        self
158    }
159
160    /// Builds the trigger that decides this condition, reading by reading.
161    ///
162    /// # Returns
163    ///
164    /// A trigger that starts cleared.
165    pub fn trigger(&self) -> Trigger {
166        match self.compare {
167            Compare::Above => Trigger::above(self.threshold, self.hysteresis),
168            Compare::Below => Trigger::below(self.threshold, self.hysteresis),
169        }
170    }
171}
172
173/// What a rule does when its condition sets or clears.
174///
175/// In a rule file each action is an object named by what it does:
176///
177/// ```json
178/// { "drive": "bed-valve", "on": true }
179/// { "publish": "garden/bed-1/valve", "payload": "open" }
180/// ```
181#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
182#[serde(try_from = "ActionFile", into = "ActionFile")]
183pub enum Action {
184    /// Switches an actuator the engine was given under this name.
185    Drive {
186        /// The name the actuator was registered under.
187        actuator: String,
188        /// The setting to apply.
189        on: bool,
190    },
191    /// Publishes a message over the engine's link.
192    Publish {
193        /// The topic to publish to.
194        topic: String,
195        /// The text to publish.
196        payload: String,
197    },
198}
199
200/// An action as a rule file writes it: `drive` with `on`, or `publish` with `payload`.
201#[derive(Serialize, Deserialize)]
202#[serde(deny_unknown_fields)]
203struct ActionFile {
204    #[serde(default, skip_serializing_if = "Option::is_none")]
205    drive: Option<String>,
206    #[serde(default, skip_serializing_if = "Option::is_none")]
207    on: Option<bool>,
208    #[serde(default, skip_serializing_if = "Option::is_none")]
209    publish: Option<String>,
210    #[serde(default, skip_serializing_if = "Option::is_none")]
211    payload: Option<String>,
212}
213
214impl TryFrom<ActionFile> for Action {
215    type Error = String;
216
217    fn try_from(file: ActionFile) -> core::result::Result<Self, String> {
218        match file {
219            ActionFile {
220                drive: Some(actuator),
221                on: Some(on),
222                publish: None,
223                payload: None,
224            } => Ok(Action::Drive { actuator, on }),
225            ActionFile {
226                drive: None,
227                on: None,
228                publish: Some(topic),
229                payload: Some(payload),
230            } => Ok(Action::Publish { topic, payload }),
231            ActionFile {
232                drive: Some(_),
233                publish: Some(_),
234                ..
235            } => Err("an action does one thing: `drive` or `publish`, not both".to_owned()),
236            ActionFile {
237                drive: Some(actuator),
238                on: None,
239                ..
240            } => Err(format!(
241                "driving `{actuator}` needs `on`: true to switch it on, false to switch it off"
242            )),
243            ActionFile {
244                publish: Some(topic),
245                payload: None,
246                ..
247            } => Err(format!(
248                "publishing to `{topic}` needs a `payload`, the text to send"
249            )),
250            ActionFile { drive: Some(_), .. } => {
251                Err("`payload` belongs to `publish`, not to `drive`".to_owned())
252            }
253            ActionFile {
254                publish: Some(_), ..
255            } => Err("`on` belongs to `drive`, not to `publish`".to_owned()),
256            ActionFile { .. } => Err(
257                "an action names what it does: `drive` an actuator or `publish` to a topic"
258                    .to_owned(),
259            ),
260        }
261    }
262}
263
264impl From<Action> for ActionFile {
265    fn from(action: Action) -> Self {
266        match action {
267            Action::Drive { actuator, on } => ActionFile {
268                drive: Some(actuator),
269                on: Some(on),
270                publish: None,
271                payload: None,
272            },
273            Action::Publish { topic, payload } => ActionFile {
274                drive: None,
275                on: None,
276                publish: Some(topic),
277                payload: Some(payload),
278            },
279        }
280    }
281}
282
283/// One rule: a condition over a topic, and what to do as it sets and clears.
284///
285/// ```json
286/// {
287///   "name": "water-when-dry",
288///   "when": { "topic": "garden/bed-1/moisture", "below": 30.0, "hysteresis": 5.0 },
289///   "then": [ { "drive": "bed-valve", "on": true } ],
290///   "otherwise": [ { "drive": "bed-valve", "on": false } ]
291/// }
292/// ```
293///
294/// `then` runs once when the condition becomes true and `otherwise` once when it stops
295/// holding; either may be left out. The name is what the engine reports when the rule
296/// fires.
297#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
298#[serde(deny_unknown_fields)]
299pub struct Rule {
300    /// A stable name for the rule, such as `"water-when-dry"`.
301    pub name: String,
302    /// The condition the rule watches.
303    pub when: Condition,
304    /// What to do the moment the condition becomes true.
305    #[serde(default, skip_serializing_if = "Vec::is_empty")]
306    pub then: Vec<Action>,
307    /// What to do the moment the condition stops holding.
308    #[serde(default, skip_serializing_if = "Vec::is_empty")]
309    pub otherwise: Vec<Action>,
310}
311
312impl Rule {
313    /// Starts a rule with a name and a condition and nothing to do yet.
314    ///
315    /// # Arguments
316    ///
317    /// * `name` - a stable name for the rule.
318    /// * `when` - the condition it watches.
319    ///
320    /// # Returns
321    ///
322    /// The rule, for chaining.
323    pub fn new(name: impl Into<String>, when: Condition) -> Self {
324        Self {
325            name: name.into(),
326            when,
327            then: Vec::new(),
328            otherwise: Vec::new(),
329        }
330    }
331
332    /// Adds an action for the moment the condition becomes true.
333    ///
334    /// # Arguments
335    ///
336    /// * `action` - what to do.
337    ///
338    /// # Returns
339    ///
340    /// The rule, for chaining.
341    pub fn then(mut self, action: Action) -> Self {
342        self.then.push(action);
343        self
344    }
345
346    /// Adds an action for the moment the condition stops holding.
347    ///
348    /// # Arguments
349    ///
350    /// * `action` - what to do.
351    ///
352    /// # Returns
353    ///
354    /// The rule, for chaining.
355    pub fn otherwise(mut self, action: Action) -> Self {
356        self.otherwise.push(action);
357        self
358    }
359
360    /// Names every actuator the rule drives.
361    ///
362    /// # Returns
363    ///
364    /// Each actuator name, once, in name order.
365    pub fn actuators(&self) -> impl Iterator<Item = &str> {
366        let mut names: BTreeSet<&str> = BTreeSet::new();
367        for action in self.then.iter().chain(&self.otherwise) {
368            if let Action::Drive { actuator, .. } = action {
369                names.insert(actuator);
370            }
371        }
372        names.into_iter()
373    }
374}
375
376/// A set of rules, as a file a fleet shares.
377///
378/// ```json
379/// { "$schema": "https://pamoja.molex.cloud/schema/rules-1.json", "rules": [ ... ] }
380/// ```
381///
382/// The `$schema` is optional; it names the format the file is written in, and an editor
383/// reads it to check the file as it is typed. A field the format does not have is
384/// refused, with the nearest one it does.
385#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
386#[serde(try_from = "RulesFile", into = "RulesFile")]
387pub struct Rules {
388    /// The rules, in the order they are checked against each message.
389    pub rules: Vec<Rule>,
390}
391
392/// A rule file as written, with the `$schema` that names its format.
393#[derive(Serialize, Deserialize)]
394#[serde(deny_unknown_fields)]
395struct RulesFile {
396    #[serde(rename = "$schema", default, skip_serializing_if = "Option::is_none")]
397    schema: Option<String>,
398    rules: Vec<Rule>,
399}
400
401impl TryFrom<RulesFile> for Rules {
402    type Error = String;
403
404    fn try_from(file: RulesFile) -> core::result::Result<Self, String> {
405        crate::format::check_schema(file.schema.as_deref(), "rules", Rules::SCHEMA)?;
406        Ok(Rules { rules: file.rules })
407    }
408}
409
410impl From<Rules> for RulesFile {
411    fn from(rules: Rules) -> Self {
412        RulesFile {
413            schema: Some(Rules::SCHEMA.to_owned()),
414            rules: rules.rules,
415        }
416    }
417}
418
419impl Rules {
420    /// The address of the published JSON Schema for the rule file format this build
421    /// writes, which [`to_json`](Rules::to_json) names as the file's `$schema`.
422    pub const SCHEMA: &'static str = "https://pamoja.molex.cloud/schema/rules-1.json";
423
424    /// Starts with no rules.
425    ///
426    /// # Returns
427    ///
428    /// An empty set.
429    pub fn new() -> Self {
430        Self::default()
431    }
432
433    /// Adds a rule.
434    ///
435    /// # Arguments
436    ///
437    /// * `rule` - the rule to add.
438    ///
439    /// # Returns
440    ///
441    /// The set, for chaining.
442    pub fn with(mut self, rule: Rule) -> Self {
443        self.rules.push(rule);
444        self
445    }
446
447    /// Loads rules from their JSON file.
448    ///
449    /// # Arguments
450    ///
451    /// * `text` - the file's contents.
452    ///
453    /// # Returns
454    ///
455    /// The rules.
456    ///
457    /// # Errors
458    ///
459    /// Returns [`Error::Codec`] if the text is not a rule file, is written in a format
460    /// this build does not read, carries a field the format does not have, or holds a rule
461    /// [`check`](Rules::check) refuses. A misspelled field is reported with the name it was
462    /// probably meant to be.
463    ///
464    /// # Examples
465    ///
466    /// ```
467    /// use pamoja_profile::Rules;
468    ///
469    /// let file = r#"{
470    ///   "$schema": "https://pamoja.molex.cloud/schema/rules-1.json",
471    ///   "rules": [
472    ///     { "name": "vent-when-hot",
473    ///       "when": { "topic": "greenhouse/air", "above": 30.0, "hysteresis": 3.0 },
474    ///       "then": [ { "drive": "roof-vent", "on": true } ],
475    ///       "otherwise": [ { "drive": "roof-vent", "on": false } ] }
476    ///   ]
477    /// }"#;
478    /// let rules = Rules::from_json(file)?;
479    /// assert_eq!(rules.rules[0].when.threshold, 30.0);
480    ///
481    /// let typo = file.replace("\"hysteresis\"", "\"hysterisis\"");
482    /// let refused = Rules::from_json(&typo).unwrap_err().to_string();
483    /// assert!(refused.contains("did you mean `hysteresis`?"), "{refused}");
484    /// # Ok::<(), pamoja_core::Error>(())
485    /// ```
486    #[cfg(feature = "json")]
487    pub fn from_json(text: &str) -> Result<Self> {
488        let rules: Rules = serde_json::from_str(text)
489            .map_err(|error| Error::Codec(crate::format::explain(&error)))?;
490        rules.check()?;
491        Ok(rules)
492    }
493
494    /// Writes the rules as the JSON file a fleet shares.
495    ///
496    /// # Returns
497    ///
498    /// The pretty-printed JSON text.
499    ///
500    /// # Errors
501    ///
502    /// Returns [`Error::Codec`] if the rules cannot be serialized.
503    #[cfg(feature = "json")]
504    pub fn to_json(&self) -> Result<String> {
505        serde_json::to_string_pretty(self).map_err(|error| Error::Codec(error.to_string()))
506    }
507
508    /// Checks the rules for what a file can say that no engine could run.
509    ///
510    /// # Returns
511    ///
512    /// Nothing when every rule is usable.
513    ///
514    /// # Errors
515    ///
516    /// Returns [`Error::Codec`] naming the first problem and the rule it is in: a rule
517    /// without a name or with a name another rule has, a condition without a topic or
518    /// watching a filter with `+` or `#`, a threshold that is not a finite number, a
519    /// hysteresis that is not a finite number of zero or more, a rule with nothing to
520    /// do, or an action naming no actuator, no topic, or a filter to publish to.
521    ///
522    /// # Examples
523    ///
524    /// ```
525    /// use pamoja_profile::{Action, Condition, Rule, Rules};
526    ///
527    /// // A rule on a filter would subscribe to every bed and match none of them, since a
528    /// // rule compares each message's topic with its own exactly.
529    /// let every_bed = Rules::new().with(
530    ///     Rule::new("water-when-dry", Condition::below("garden/+/moisture", 30.0))
531    ///         .then(Action::Drive { actuator: "bed-valve".into(), on: true }),
532    /// );
533    /// let refused = every_bed.check().unwrap_err().to_string();
534    /// assert!(refused.contains("a rule watches one topic exactly"));
535    /// ```
536    pub fn check(&self) -> Result<()> {
537        let refuse = |reason: String| Err(Error::Codec(reason));
538        let filter = |topic: &str| topic.contains(['+', '#']);
539        let mut names = BTreeSet::new();
540        for rule in &self.rules {
541            let name = &rule.name;
542            if name.trim().is_empty() {
543                return refuse("a rule needs a name".to_owned());
544            }
545            if !names.insert(name.as_str()) {
546                return refuse(format!("two rules share the name `{name}`"));
547            }
548            let topic = &rule.when.topic;
549            if topic.trim().is_empty() {
550                return refuse(format!("the rule `{name}` needs a topic to watch"));
551            }
552            if filter(topic) {
553                return refuse(format!(
554                    "the rule `{name}` watches `{topic}`, a filter; a rule watches one topic exactly"
555                ));
556            }
557            if !rule.when.threshold.is_finite() {
558                return refuse(format!(
559                    "the rule `{name}` needs a threshold that is a finite number, not {}",
560                    rule.when.threshold
561                ));
562            }
563            if !(rule.when.hysteresis.is_finite() && rule.when.hysteresis >= 0.0) {
564                return refuse(format!(
565                    "the rule `{name}` needs a hysteresis that is a finite number of zero or more, not {}",
566                    rule.when.hysteresis
567                ));
568            }
569            if rule.then.is_empty() && rule.otherwise.is_empty() {
570                return refuse(format!("the rule `{name}` has nothing to do"));
571            }
572            for action in rule.then.iter().chain(&rule.otherwise) {
573                match action {
574                    Action::Drive { actuator, .. } if actuator.trim().is_empty() => {
575                        return refuse(format!(
576                            "the rule `{name}` drives an actuator with no name"
577                        ));
578                    }
579                    Action::Publish { topic, .. } if topic.trim().is_empty() => {
580                        return refuse(format!("the rule `{name}` publishes to no topic"));
581                    }
582                    Action::Publish { topic, .. } if filter(topic) => {
583                        return refuse(format!(
584                            "the rule `{name}` publishes to `{topic}`, a filter rather than a topic"
585                        ));
586                    }
587                    _ => {}
588                }
589            }
590        }
591        Ok(())
592    }
593}
594
595/// What a rule did on one message.
596#[derive(Clone, Debug, PartialEq)]
597pub struct Fired {
598    /// The rule's name.
599    pub rule: String,
600    /// Whether its condition set or cleared.
601    pub edge: Edge,
602    /// The reading that did it.
603    pub reading: f32,
604    /// What the rule calls for on this edge: its `then` actions when it set and its
605    /// `otherwise` actions when it cleared, in the order the file gives them.
606    pub actions: Vec<Action>,
607}
608
609/// The decisions a set of rules makes, reading by reading, with no link and no outputs
610/// of its own.
611///
612/// A [`RuleEngine`] receives each message, decodes it, and carries out what a rule calls
613/// for. This is its deciding half on its own, for a program that moves its own messages
614/// and drives its own outputs: hand it a topic and a reading, and it says which rules set
615/// or cleared and what each calls for. The engine runs on one, so the two decide alike.
616///
617/// # Examples
618///
619/// ```
620/// use pamoja_kit::Edge;
621/// use pamoja_profile::{Action, Condition, Rule, RuleEvaluator, Rules};
622///
623/// // Open a roof vent when a greenhouse passes 30 C, and close it once it is back
624/// // under 27 C.
625/// let rules = Rules::new().with(
626///     Rule::new("vent-when-hot", Condition::above("greenhouse/air", 30.0).with_hysteresis(3.0))
627///         .then(Action::Drive { actuator: "roof-vent".into(), on: true })
628///         .otherwise(Action::Drive { actuator: "roof-vent".into(), on: false }),
629/// );
630/// let mut evaluator = RuleEvaluator::new(rules)?;
631/// assert!(evaluator.evaluate("greenhouse/air", 29.0)?.is_empty());
632///
633/// let fired = evaluator.evaluate("greenhouse/air", 31.5)?;
634/// assert_eq!(fired[0].edge, Edge::Set);
635/// assert_eq!(fired[0].actions, [Action::Drive { actuator: "roof-vent".into(), on: true }]);
636/// # Ok::<(), pamoja_core::Error>(())
637/// ```
638pub struct RuleEvaluator {
639    armed: Vec<Armed>,
640}
641
642impl RuleEvaluator {
643    /// Arms a set of rules, every condition starting cleared.
644    ///
645    /// # Arguments
646    ///
647    /// * `rules` - the rules to judge readings by.
648    ///
649    /// # Returns
650    ///
651    /// The evaluator.
652    ///
653    /// # Errors
654    ///
655    /// Returns [`Error::Codec`] naming the first problem [`Rules::check`] finds.
656    pub fn new(rules: Rules) -> Result<Self> {
657        rules.check()?;
658        Ok(Self::unchecked(rules))
659    }
660
661    /// Arms a set of rules without checking them, for an engine that checks as it
662    /// connects.
663    fn unchecked(rules: Rules) -> Self {
664        Self {
665            armed: rules
666                .rules
667                .into_iter()
668                .map(|rule| Armed {
669                    trigger: rule.when.trigger(),
670                    rule,
671                })
672                .collect(),
673        }
674    }
675
676    /// Judges one reading from one topic against every rule that watches it.
677    ///
678    /// # Arguments
679    ///
680    /// * `topic` - the topic the reading arrived on.
681    /// * `reading` - the reading, already decoded from the message.
682    ///
683    /// # Returns
684    ///
685    /// What fired, in the order the rules are listed: each rule whose condition set or
686    /// cleared, with the actions it calls for. It is empty when no rule watches the
687    /// topic or the reading changed nothing.
688    ///
689    /// # Errors
690    ///
691    /// Returns [`Error::Codec`] when a rule watches the topic and the reading is not a
692    /// finite number, such as the NaN a failed probe reports, which would otherwise
693    /// leave every condition as it was with nothing to say why.
694    pub fn evaluate(&mut self, topic: &str, reading: f32) -> Result<Vec<Fired>> {
695        let watching = self.watching(topic);
696        if watching.is_empty() {
697            return Ok(Vec::new());
698        }
699        finite(topic, reading)?;
700        Ok(watching
701            .into_iter()
702            .filter_map(|index| self.judge(index, reading))
703            .collect())
704    }
705
706    /// Whether any rule watches a topic, so a program knows whether to decode a message
707    /// before handing it over.
708    ///
709    /// # Arguments
710    ///
711    /// * `topic` - the topic a message arrived on.
712    ///
713    /// # Returns
714    ///
715    /// `true` when some rule watches it.
716    pub fn watches(&self, topic: &str) -> bool {
717        self.armed
718            .iter()
719            .any(|armed| armed.rule.when.topic == topic)
720    }
721
722    /// Lists the topics the rules watch, which are the topics to subscribe to.
723    ///
724    /// # Returns
725    ///
726    /// Each topic once, in name order.
727    pub fn topics(&self) -> Vec<&str> {
728        let topics: BTreeSet<&str> = self
729            .armed
730            .iter()
731            .map(|armed| armed.rule.when.topic.as_str())
732            .collect();
733        topics.into_iter().collect()
734    }
735
736    /// Lists the actuators the rules drive, which are the outputs a program has to have.
737    ///
738    /// # Returns
739    ///
740    /// Each actuator once, in name order.
741    pub fn actuators(&self) -> Vec<&str> {
742        let actuators: BTreeSet<&str> = self
743            .armed
744            .iter()
745            .flat_map(|armed| armed.rule.actuators())
746            .collect();
747        actuators.into_iter().collect()
748    }
749
750    /// Walks the rules in the order they are checked.
751    ///
752    /// # Returns
753    ///
754    /// Each rule.
755    pub fn rules(&self) -> impl Iterator<Item = &Rule> {
756        self.armed.iter().map(|armed| &armed.rule)
757    }
758
759    /// Whether a rule's condition currently holds.
760    ///
761    /// # Arguments
762    ///
763    /// * `rule` - the rule's name.
764    ///
765    /// # Returns
766    ///
767    /// The condition's state, or `None` for a name no rule has.
768    pub fn is_set(&self, rule: &str) -> Option<bool> {
769        self.armed
770            .iter()
771            .find(|armed| armed.rule.name == rule)
772            .map(|armed| armed.trigger.is_set())
773    }
774
775    /// The positions of the rules that watch a topic, in the order they are listed.
776    fn watching(&self, topic: &str) -> Vec<usize> {
777        self.armed
778            .iter()
779            .enumerate()
780            .filter(|(_, armed)| armed.rule.when.topic == topic)
781            .map(|(index, _)| index)
782            .collect()
783    }
784
785    /// Hands one rule a reading, and says what it did if its condition moved.
786    fn judge(&mut self, index: usize, reading: f32) -> Option<Fired> {
787        let armed = &mut self.armed[index];
788        let edge = armed.trigger.update(reading)?;
789        let actions = match edge {
790            Edge::Set => armed.rule.then.clone(),
791            Edge::Cleared => armed.rule.otherwise.clone(),
792        };
793        Some(Fired {
794            rule: armed.rule.name.clone(),
795            edge,
796            reading,
797            actions,
798        })
799    }
800}
801
802/// Refuses a reading on a watched topic that is not a finite number.
803fn finite(topic: &str, reading: f32) -> Result<()> {
804    if reading.is_finite() {
805        Ok(())
806    } else {
807        Err(Error::Codec(format!(
808            "the reading on `{topic}` is {reading}, which is not a finite number"
809        )))
810    }
811}
812
813// An actuator behind a trait object, so the engine holds any number of them by name.
814// The core trait's futures are `Send`, which is what lets one be boxed like this.
815trait ErasedActuator: Send {
816    fn apply<'a>(&'a mut self, on: bool) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'a>>;
817}
818
819impl<A: Actuator<Command = bool> + Send> ErasedActuator for A {
820    fn apply<'a>(&'a mut self, on: bool) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'a>> {
821        Box::pin(Actuator::apply(self, on))
822    }
823}
824
825// A rule with the trigger that carries its condition's state between messages.
826struct Armed {
827    rule: Rule,
828    trigger: Trigger,
829}
830
831/// Runs a set of rules off a link: every reading that arrives on a watched topic is
832/// judged, and the actions a set or cleared condition calls for are carried out.
833///
834/// The engine subscribes to each rule's topic on [`connect`](RuleEngine::connect) and
835/// then handles one message per [`step`](RuleEngine::step), decoding the reading with
836/// the codec the nodes publish in, switching the actuators it was given by name, and
837/// publishing over the same link. The node that publishes the reading and the node whose
838/// actuator moves can be anywhere the link reaches; the engine is the third party
839/// between them, and its rules are a file.
840///
841/// # Examples
842///
843/// ```
844/// use pamoja_codec::JsonCodec;
845/// use pamoja_core::{Actuator, Result, Transport};
846/// use pamoja_kit::Edge;
847/// use pamoja_loopback::{LoopbackBroker, LoopbackTransport};
848/// use pamoja_profile::{Action, Condition, Rule, RuleEngine, Rules};
849///
850/// struct Valve(bool);
851/// impl Actuator for Valve {
852///     type Command = bool;
853///     async fn apply(&mut self, open: bool) -> Result<()> {
854///         self.0 = open;
855///         Ok(())
856///     }
857/// }
858///
859/// # async fn run() -> Result<()> {
860/// let broker = LoopbackBroker::new();
861/// let mut probe = LoopbackTransport::new(broker.clone());
862/// probe.connect().await?;
863///
864/// // Water the bed when it dries past 30, and stop once it is wetter than 35.
865/// let rules = Rules::new().with(
866///     Rule::new("water-when-dry", Condition::below("garden/bed-1/moisture", 30.0).with_hysteresis(5.0))
867///         .then(Action::Drive { actuator: "bed-valve".into(), on: true })
868///         .otherwise(Action::Drive { actuator: "bed-valve".into(), on: false }),
869/// );
870/// let mut engine = RuleEngine::new(rules, LoopbackTransport::new(broker), JsonCodec)
871///     .with_actuator("bed-valve", Valve(false));
872/// engine.connect().await?;
873///
874/// probe.send("garden/bed-1/moisture", b"28.0").await?;
875/// let fired = engine.step().await?.expect("the link is up");
876/// assert_eq!(fired[0].rule, "water-when-dry");
877/// assert_eq!(fired[0].edge, Edge::Set);
878/// # Ok(())
879/// # }
880/// ```
881pub struct RuleEngine<L, C> {
882    evaluator: RuleEvaluator,
883    link: L,
884    codec: C,
885    actuators: BTreeMap<String, Box<dyn ErasedActuator>>,
886}
887
888impl<L, C> RuleEngine<L, C> {
889    /// Assembles an engine around a link and the codec the readings arrive in.
890    ///
891    /// # Arguments
892    ///
893    /// * `rules` - the rules to run, checked when the engine connects.
894    /// * `link` - the link readings arrive on and actions publish over; connected by
895    ///   [`connect`](RuleEngine::connect).
896    /// * `codec` - the wire format the nodes publish their readings in.
897    ///
898    /// # Returns
899    ///
900    /// An engine with no actuators yet.
901    pub fn new(rules: Rules, link: L, codec: C) -> Self {
902        Self {
903            evaluator: RuleEvaluator::unchecked(rules),
904            link,
905            codec,
906            actuators: BTreeMap::new(),
907        }
908    }
909
910    /// Gives the engine an actuator a rule may drive by name.
911    ///
912    /// # Arguments
913    ///
914    /// * `name` - the name a `drive` action uses.
915    /// * `actuator` - the output, which the engine owns from here on.
916    ///
917    /// # Returns
918    ///
919    /// The engine, for chaining.
920    pub fn with_actuator(
921        mut self,
922        name: impl Into<String>,
923        actuator: impl Actuator<Command = bool> + Send + 'static,
924    ) -> Self {
925        self.actuators.insert(name.into(), Box::new(actuator));
926        self
927    }
928
929    /// Walks the rules in the order they are checked.
930    ///
931    /// # Returns
932    ///
933    /// Each rule.
934    pub fn rules(&self) -> impl Iterator<Item = &Rule> {
935        self.evaluator.rules()
936    }
937
938    /// Lists the topics the rules watch, which the engine subscribes to as it connects.
939    ///
940    /// # Returns
941    ///
942    /// Each topic once, in name order.
943    pub fn topics(&self) -> Vec<&str> {
944        self.evaluator.topics()
945    }
946
947    /// Lists the actuators the rules drive, each of which the engine has to be given.
948    ///
949    /// # Returns
950    ///
951    /// Each actuator once, in name order.
952    pub fn actuators(&self) -> Vec<&str> {
953        self.evaluator.actuators()
954    }
955
956    /// Whether a rule's condition currently holds.
957    ///
958    /// # Arguments
959    ///
960    /// * `rule` - the rule's name.
961    ///
962    /// # Returns
963    ///
964    /// The condition's state, or `None` for a name no rule has.
965    pub fn is_set(&self, rule: &str) -> Option<bool> {
966        self.evaluator.is_set(rule)
967    }
968}
969
970impl<L, C> RuleEngine<L, C>
971where
972    L: Transport + Receive,
973    C: Codec<f32>,
974{
975    /// Checks the rules, connects the link, and subscribes to every watched topic.
976    ///
977    /// # Returns
978    ///
979    /// Nothing once the engine is listening.
980    ///
981    /// # Errors
982    ///
983    /// Returns [`Error::Codec`] if a rule cannot be run (see [`Rules::check`]),
984    /// [`Error::Unsupported`] if a rule drives an actuator the engine was not given, and
985    /// the link's own error if it cannot connect or subscribe.
986    pub async fn connect(&mut self) -> Result<()> {
987        Rules {
988            rules: self.evaluator.rules().cloned().collect(),
989        }
990        .check()?;
991        if self
992            .evaluator
993            .actuators()
994            .iter()
995            .any(|actuator| !self.actuators.contains_key(*actuator))
996        {
997            return Err(Error::Unsupported(
998                "a rule drives an actuator the engine was not given",
999            ));
1000        }
1001        self.link.connect().await?;
1002        for topic in self.evaluator.topics() {
1003            self.link.subscribe(topic).await?;
1004        }
1005        Ok(())
1006    }
1007
1008    /// Waits for one message and runs every rule that watches its topic.
1009    ///
1010    /// # Returns
1011    ///
1012    /// What fired, which is empty when the message set or cleared nothing, or `None`
1013    /// once the link has ended.
1014    ///
1015    /// # Errors
1016    ///
1017    /// Returns the link's error if receiving or publishing fails,
1018    /// [`Error::Codec`] if a reading on a watched topic does not decode or is not a
1019    /// finite number, and the actuator's error if a drive fails.
1020    pub async fn step(&mut self) -> Result<Option<Vec<Fired>>> {
1021        let Some(message) = self.link.recv().await? else {
1022            return Ok(None);
1023        };
1024        let watching = self.evaluator.watching(&message.topic);
1025        let mut fired = Vec::new();
1026        if watching.is_empty() {
1027            return Ok(Some(fired));
1028        }
1029        let reading: f32 = self.codec.decode(&message.payload)?;
1030        finite(&message.topic, reading)?;
1031        for index in watching {
1032            let Some(one) = self.evaluator.judge(index, reading) else {
1033                continue;
1034            };
1035            for action in &one.actions {
1036                self.run(action).await?;
1037            }
1038            fired.push(one);
1039        }
1040        Ok(Some(fired))
1041    }
1042
1043    async fn run(&mut self, action: &Action) -> Result<()> {
1044        match action {
1045            Action::Drive { actuator, on } => {
1046                let actuator = self.actuators.get_mut(actuator).ok_or(Error::Unsupported(
1047                    "a rule drives an actuator the engine was not given",
1048                ))?;
1049                actuator.apply(*on).await
1050            }
1051            Action::Publish { topic, payload } => self.link.send(topic, payload.as_bytes()).await,
1052        }
1053    }
1054}
1055
1056#[cfg(test)]
1057mod tests {
1058    use std::sync::{Arc, Mutex};
1059
1060    use pamoja_codec::JsonCodec;
1061    use pamoja_loopback::{LoopbackBroker, LoopbackTransport};
1062
1063    use super::*;
1064
1065    const FILE: &str = r#"{
1066  "$schema": "https://pamoja.molex.cloud/schema/rules-1.json",
1067  "rules": [
1068    {
1069      "name": "water-when-dry",
1070      "when": { "topic": "garden/bed-1/moisture", "below": 30.0, "hysteresis": 5.0 },
1071      "then": [
1072        { "drive": "bed-valve", "on": true },
1073        { "publish": "garden/bed-1/valve", "payload": "open" }
1074      ],
1075      "otherwise": [
1076        { "drive": "bed-valve", "on": false },
1077        { "publish": "garden/bed-1/valve", "payload": "closed" }
1078      ]
1079    }
1080  ]
1081}"#;
1082
1083    #[derive(Clone, Default)]
1084    struct Valve(Arc<Mutex<Vec<bool>>>);
1085
1086    impl Actuator for Valve {
1087        type Command = bool;
1088
1089        async fn apply(&mut self, open: bool) -> Result<()> {
1090            self.0.lock().expect("valve lock").push(open);
1091            Ok(())
1092        }
1093    }
1094
1095    #[test]
1096    fn a_rule_file_round_trips_and_builds_its_trigger() {
1097        let rules = Rules::from_json(FILE).expect("parses");
1098        assert_eq!(rules.rules.len(), 1);
1099        let rule = &rules.rules[0];
1100        assert_eq!(rule.name, "water-when-dry");
1101        assert_eq!(rule.when.compare, Compare::Below);
1102        assert_eq!(rule.when.hysteresis, 5.0);
1103        assert_eq!(rule.actuators().collect::<Vec<_>>(), ["bed-valve"]);
1104        let mut trigger = rule.when.trigger();
1105        assert_eq!(trigger.update(28.0), Some(Edge::Set));
1106        let shared = rules.to_json().unwrap();
1107        assert!(
1108            shared.contains("\"publish\": \"garden/bed-1/valve\",\n"),
1109            "{shared}"
1110        );
1111        assert!(
1112            shared.starts_with(
1113                "{\n  \"$schema\": \"https://pamoja.molex.cloud/schema/rules-1.json\""
1114            ),
1115            "{shared}"
1116        );
1117        assert_eq!(Rules::from_json(&shared).unwrap(), rules);
1118
1119        let built = Rules::new().with(
1120            Rule::new(
1121                "water-when-dry",
1122                Condition::below("garden/bed-1/moisture", 30.0).with_hysteresis(5.0),
1123            )
1124            .then(Action::Drive {
1125                actuator: "bed-valve".into(),
1126                on: true,
1127            })
1128            .then(Action::Publish {
1129                topic: "garden/bed-1/valve".into(),
1130                payload: "open".into(),
1131            })
1132            .otherwise(Action::Drive {
1133                actuator: "bed-valve".into(),
1134                on: false,
1135            })
1136            .otherwise(Action::Publish {
1137                topic: "garden/bed-1/valve".into(),
1138                payload: "closed".into(),
1139            }),
1140        );
1141        assert_eq!(built, rules);
1142        assert_eq!(Condition::above("t", 1.0).compare, Compare::Above);
1143    }
1144
1145    #[test]
1146    fn a_rule_file_says_what_it_does_in_its_own_words_or_is_refused() {
1147        let rule = |when: &str, then: &str| {
1148            format!(r#"{{ "rules": [ {{ "name": "r", "when": {when}, "then": [ {then} ] }} ] }}"#)
1149        };
1150        let refused = |text: String| match Rules::from_json(&text) {
1151            Err(Error::Codec(reason)) => reason,
1152            other => panic!("expected {text} to be refused, got {other:?}"),
1153        };
1154        let drive = r#"{ "drive": "fan", "on": true }"#;
1155        let when = r#"{ "topic": "t", "above": 1.0 }"#;
1156        Rules::from_json(&rule(when, drive)).expect("a usable rule");
1157
1158        assert!(refused(rule(r#"{ "topic": "t" }"#, drive)).contains("needs a line to cross"));
1159        assert!(refused(rule(
1160            r#"{ "topic": "t", "above": 1.0, "below": 0.0 }"#,
1161            drive
1162        ))
1163        .contains("names both `above` and `below`"));
1164        assert!(refused(rule(
1165            r#"{ "topic": "t", "compare": "above", "threshold": 1.0 }"#,
1166            drive
1167        ))
1168        .starts_with("unknown field `compare`, expected one of `topic`, `above`, `below`"));
1169        assert!(refused(rule(r#"{ "topic": "t", "abve": 1.0 }"#, drive))
1170            .starts_with("unknown field `abve`, did you mean `above`?"));
1171
1172        assert!(refused(rule(when, r#"{ "drive": "fan" }"#)).contains("needs `on`"));
1173        assert!(refused(rule(when, r#"{ "publish": "t/out" }"#)).contains("needs a `payload`"));
1174        assert!(refused(rule(
1175            when,
1176            r#"{ "drive": "fan", "on": true, "publish": "t/out", "payload": "x" }"#
1177        ))
1178        .contains("not both"));
1179        assert!(refused(rule(
1180            when,
1181            r#"{ "drive": "fan", "on": true, "payload": "x" }"#
1182        ))
1183        .contains("`payload` belongs to `publish`"));
1184        assert!(refused(rule(when, r#"{ "on": true }"#)).contains("an action names what it does"));
1185        assert!(refused(rule(
1186            when,
1187            r#"{ "do": "drive", "actuator": "fan", "on": true }"#
1188        ))
1189        .starts_with("unknown field `do`"));
1190
1191        let newer = FILE.replace("rules-1.json", "rules-2.json");
1192        assert!(refused(newer).contains("rules format 2"));
1193        let idle = rule(when, "").replace(r#", "then": [  ]"#, "");
1194        assert_eq!(refused(idle), "the rule `r` has nothing to do");
1195    }
1196
1197    #[test]
1198    fn a_rule_file_is_checked_before_it_runs() {
1199        let ok = Rules::from_json(FILE).unwrap();
1200        ok.check().expect("a usable file");
1201        let publish = || Action::Publish {
1202            topic: "x".into(),
1203            payload: "y".into(),
1204        };
1205        let refused = |rules: Rules| match rules.check() {
1206            Err(Error::Codec(reason)) => reason,
1207            other => panic!("expected a codec error, got {other:?}"),
1208        };
1209
1210        let unnamed = Rules::new().with(Rule::new("", Condition::above("t", 1.0)).then(publish()));
1211        assert_eq!(refused(unnamed), "a rule needs a name");
1212        let idle = Rules::new().with(Rule::new("idle", Condition::above("t", 1.0)));
1213        assert_eq!(refused(idle), "the rule `idle` has nothing to do");
1214        let twice = Rules::new()
1215            .with(Rule::new("a", Condition::above("t", 1.0)).then(publish()))
1216            .with(Rule::new("a", Condition::above("t", 1.0)).then(publish()));
1217        assert_eq!(refused(twice), "two rules share the name `a`");
1218        let negative = Rules::new()
1219            .with(Rule::new("a", Condition::above("t", 1.0).with_hysteresis(-1.0)).then(publish()));
1220        assert!(refused(negative).contains("a finite number of zero or more, not -1"));
1221        let nameless_actuator = Rules::new().with(Rule::new("a", Condition::above("t", 1.0)).then(
1222            Action::Drive {
1223                actuator: " ".into(),
1224                on: true,
1225            },
1226        ));
1227        assert_eq!(
1228            refused(nameless_actuator),
1229            "the rule `a` drives an actuator with no name"
1230        );
1231        let filter =
1232            Rules::new().with(Rule::new("a", Condition::above("t/+", 1.0)).then(publish()));
1233        assert!(refused(filter).contains("a rule watches one topic exactly"));
1234        let broadcast = Rules::new().with(Rule::new("a", Condition::above("t", 1.0)).then(
1235            Action::Publish {
1236                topic: "alerts/#".into(),
1237                payload: "hot".into(),
1238            },
1239        ));
1240        assert!(refused(broadcast).contains("a filter rather than a topic"));
1241    }
1242
1243    #[test]
1244    fn an_evaluator_decides_what_the_engine_would_and_says_what_to_do() {
1245        let file = Rules::from_json(FILE).unwrap().with(
1246            Rule::new(
1247                "flood-alarm",
1248                Condition::above("garden/bed-1/moisture", 60.0).with_hysteresis(5.0),
1249            )
1250            .then(Action::Publish {
1251                topic: "garden/alarm".into(),
1252                payload: "waterlogged".into(),
1253            }),
1254        );
1255        let mut evaluator = RuleEvaluator::new(file).expect("a usable file");
1256        assert_eq!(evaluator.topics(), ["garden/bed-1/moisture"]);
1257        assert_eq!(evaluator.actuators(), ["bed-valve"]);
1258        assert!(evaluator.watches("garden/bed-1/moisture"));
1259        assert!(!evaluator.watches("garden/bed-2/moisture"));
1260        assert!(evaluator
1261            .evaluate("garden/bed-2/moisture", 5.0)
1262            .unwrap()
1263            .is_empty());
1264
1265        let dry = evaluator.evaluate("garden/bed-1/moisture", 28.0).unwrap();
1266        assert_eq!(dry.len(), 1);
1267        assert_eq!(
1268            (dry[0].rule.as_str(), dry[0].edge),
1269            ("water-when-dry", Edge::Set)
1270        );
1271        assert_eq!(dry[0].actions.len(), 2);
1272        assert_eq!(evaluator.is_set("water-when-dry"), Some(true));
1273
1274        let soaked = evaluator.evaluate("garden/bed-1/moisture", 65.0).unwrap();
1275        let names: Vec<&str> = soaked.iter().map(|fired| fired.rule.as_str()).collect();
1276        assert_eq!(
1277            names,
1278            ["water-when-dry", "flood-alarm"],
1279            "in the order the file lists them"
1280        );
1281        assert_eq!(soaked[0].edge, Edge::Cleared);
1282        assert_eq!(
1283            soaked[1].actions,
1284            [Action::Publish {
1285                topic: "garden/alarm".into(),
1286                payload: "waterlogged".into(),
1287            }]
1288        );
1289
1290        let failed = evaluator.evaluate("garden/bed-1/moisture", f32::NAN);
1291        assert!(
1292            matches!(&failed, Err(Error::Codec(reason)) if reason.contains("not a finite number")),
1293            "{failed:?}"
1294        );
1295        assert!(
1296            evaluator
1297                .evaluate("garden/bed-2/moisture", f32::NAN)
1298                .unwrap()
1299                .is_empty(),
1300            "a topic no rule watches is not judged"
1301        );
1302        assert_eq!(evaluator.is_set("flood-alarm"), Some(true));
1303
1304        let unusable = Rules::new().with(Rule::new("idle", Condition::above("t", 1.0)));
1305        assert!(matches!(RuleEvaluator::new(unusable), Err(Error::Codec(_))));
1306    }
1307
1308    #[tokio::test]
1309    async fn one_node_reading_drives_another_node_actuator_over_the_broker() {
1310        let broker = LoopbackBroker::new();
1311        let mut probe = LoopbackTransport::new(broker.clone());
1312        let mut watcher = LoopbackTransport::new(broker.clone());
1313        probe.connect().await.unwrap();
1314        watcher.connect().await.unwrap();
1315        watcher.subscribe("garden/bed-1/valve").await.unwrap();
1316
1317        // The engine's link carries one topic no rule watches, subscribed before the
1318        // engine takes it over.
1319        let mut link = LoopbackTransport::new(broker);
1320        link.connect().await.unwrap();
1321        link.subscribe("garden/bed-1/note").await.unwrap();
1322
1323        let valve = Valve::default();
1324        let mut engine = RuleEngine::new(Rules::from_json(FILE).unwrap(), link, JsonCodec)
1325            .with_actuator("bed-valve", valve.clone());
1326        engine.connect().await.unwrap();
1327        assert_eq!(engine.rules().count(), 1);
1328        assert_eq!(engine.is_set("water-when-dry"), Some(false));
1329        assert_eq!(engine.is_set("nowhere"), None);
1330
1331        let mut edges = Vec::new();
1332        for reading in [42.0f32, 31.0, 28.0, 33.0, 36.0] {
1333            let payload = JsonCodec.encode(&reading).unwrap();
1334            probe.send("garden/bed-1/moisture", &payload).await.unwrap();
1335            let fired = engine.step().await.unwrap().expect("the link is up");
1336            edges.push(fired.first().map(|fired| (fired.edge, fired.reading)));
1337        }
1338        assert_eq!(
1339            edges,
1340            [
1341                None,
1342                None,
1343                Some((Edge::Set, 28.0)),
1344                None,
1345                Some((Edge::Cleared, 36.0))
1346            ]
1347        );
1348        assert_eq!(*valve.0.lock().unwrap(), vec![true, false]);
1349        assert!(!engine.is_set("water-when-dry").unwrap());
1350
1351        let open = watcher.recv().await.unwrap().expect("open");
1352        let closed = watcher.recv().await.unwrap().expect("closed");
1353        assert_eq!(
1354            (open.topic.as_str(), open.payload.as_slice()),
1355            ("garden/bed-1/valve", b"open".as_slice())
1356        );
1357        assert_eq!(closed.payload, b"closed");
1358
1359        // A message on a topic no rule watches is not decoded at all.
1360        probe
1361            .send("garden/bed-1/note", b"not a number")
1362            .await
1363            .unwrap();
1364        assert_eq!(engine.step().await.unwrap(), Some(Vec::new()));
1365
1366        // The engine reports each edge with the actions it carried out.
1367        let payload = JsonCodec.encode(&27.0f32).unwrap();
1368        probe.send("garden/bed-1/moisture", &payload).await.unwrap();
1369        let fired = engine.step().await.unwrap().expect("the link is up");
1370        assert_eq!(
1371            fired[0].actions,
1372            Rules::from_json(FILE).unwrap().rules[0].then
1373        );
1374    }
1375
1376    #[tokio::test]
1377    async fn an_actuator_a_rule_names_must_be_given() {
1378        let broker = LoopbackBroker::new();
1379        let mut engine = RuleEngine::new(
1380            Rules::from_json(FILE).unwrap(),
1381            LoopbackTransport::new(broker),
1382            JsonCodec,
1383        );
1384        assert!(matches!(engine.connect().await, Err(Error::Unsupported(_))));
1385    }
1386}
pamoja_profile/rules.rs

pamoja_profile/
rules.rs
