ebus_homie/controller/
types.rs

1//! Public value types of the controller: ids, configuration, events, views.
2
3use alloc::{string::String, sync::Arc, vec::Vec};
4use core::fmt;
5
6use ebus_mqtt::{Link, QoS, filter};
7use rpds::RedBlackTreeMapSync;
8
9pub use crate::{Id, PropertyAddr};
10pub use ebus_mqtt::Handled;
11
12use crate::EBUS_DOMAIN;
13use crate::datatype::Problem;
14use crate::description::{DescriptionError, DeviceDescription, Ignored};
15use crate::id::is_valid_id;
16use crate::state::DeviceState;
17
18/// The topic domain (first topic level) the controller watches.
19#[derive(Clone, Debug, PartialEq, Eq)]
20pub enum Domain {
21    /// One domain, `ebus` by default (framework.md:85).
22    Named(String),
23    /// Every domain (`+/5/+/$state`, convention.md:161-162); wildcard discovery
24    /// only.
25    Any,
26}
27
28/// Which devices the controller tracks.
29#[derive(Clone, Debug, PartialEq, Eq)]
30pub enum Discovery {
31    /// Every device that publishes a `$state` in the domain.
32    Wildcard,
33    /// One device, plus the `$state` of the root its description names.
34    Device(Id),
35    /// A root device and every device its `children` declare, transitively.
36    Tree {
37        /// The root device.
38        root: Id,
39    },
40}
41
42/// Limits on how many devices replay their retained property messages at once
43/// (python-sdk#97). Start from [`Pacing::default`] and set the fields.
44#[derive(Clone, Copy, Debug, PartialEq, Eq)]
45#[non_exhaustive]
46pub struct Pacing {
47    /// Devices whose property burst may be in flight at once.
48    pub batch_devices: usize,
49    /// Retained messages the devices in flight may owe in total, two per
50    /// retained property (a value and a `$target`).
51    pub batch_values: usize,
52}
53
54impl Default for Pacing {
55    /// python-sdk's constants: 8 devices, 500 messages.
56    fn default() -> Self {
57        Pacing {
58            batch_devices: 8,
59            batch_values: 500,
60        }
61    }
62}
63
64/// Controller configuration. Start from [`Config::wildcard`],
65/// [`Config::tree`] or [`Config::device`] and adjust the fields.
66#[derive(Clone, Debug, PartialEq, Eq)]
67#[non_exhaustive]
68pub struct Config {
69    /// Topic domain; [`Domain::Named`] `"ebus"` by default.
70    pub domain: Domain,
71    /// Discovery mode.
72    pub discovery: Discovery,
73    /// QoS of every subscription; `ExactlyOnce` by default (convention.md:45).
74    pub subscribe_qos: QoS,
75    /// Subscription pacing; `None` subscribes every device's property filters
76    /// with its `$description`.
77    pub pacing: Option<Pacing>,
78    /// Milliseconds a device may wait for its retained messages before it is
79    /// resubscribed; `None` disables stuck healing. 10 000 by default.
80    pub stuck_timeout_ms: Option<u64>,
81    /// Resubscribe attempts before a stuck device is given up; 3 by default.
82    pub max_resubscribe_attempts: u8,
83    /// Milliseconds after a link-up within which a known device must be heard
84    /// again, or [`Event::DeviceAbsent`] fires; 30 000 by default.
85    pub absent_after_ms: u64,
86    /// Milliseconds after a description is applied, or after a link-up, before
87    /// [`Event::DeviceSynced`] reports missing values; 10 000 by default.
88    pub sync_timeout_ms: u64,
89    /// Trust a CONNACK with `session_present: true` and keep every
90    /// subscription; `false` (the default) re-paces as after a clean session.
91    pub trust_session: bool,
92    /// The host never sends, after a `Link::Up { session_present: false }`,
93    /// a SUBSCRIBE the controller queued before handling it. Replay matching
94    /// then restarts at each clean link-up, so a replay lost with a
95    /// connection delays no later re-sync; `false` (the default) keeps the
96    /// unmatched SUBSCRIBEs across it.
97    pub link_up_discards_queued: bool,
98    /// Values buffered per device before its `$description` arrives; 1024 by
99    /// default.
100    pub max_early_values: usize,
101    /// Alerts held per device; a new alert id past it is dropped. 256 by
102    /// default.
103    pub max_alerts: usize,
104    /// Most devices tracked at once; `None` (the default) is unbounded.
105    pub max_devices: Option<usize>,
106    /// Filters the host subscribes on the same client. The controller never
107    /// unsubscribes them.
108    pub host_filters: Vec<String>,
109}
110
111impl Config {
112    /// Wildcard discovery in the `ebus` domain, with every default.
113    pub fn wildcard() -> Self {
114        Config {
115            domain: Domain::Named(String::from(EBUS_DOMAIN)),
116            discovery: Discovery::Wildcard,
117            subscribe_qos: QoS::ExactlyOnce,
118            pacing: Some(Pacing::default()),
119            stuck_timeout_ms: Some(10_000),
120            max_resubscribe_attempts: 3,
121            absent_after_ms: 30_000,
122            sync_timeout_ms: 10_000,
123            trust_session: false,
124            link_up_discards_queued: false,
125            max_early_values: 1024,
126            max_alerts: 256,
127            max_devices: None,
128            host_filters: Vec::new(),
129        }
130    }
131
132    /// Tree discovery from `root`, otherwise as [`Config::wildcard`].
133    pub fn tree(root: &str) -> Result<Self, ConfigError> {
134        let root = Id::new(root).ok_or(ConfigError::InvalidDeviceId)?;
135        Ok(Config {
136            discovery: Discovery::Tree { root },
137            ..Config::wildcard()
138        })
139    }
140
141    /// Single-device discovery of `id`, otherwise as [`Config::wildcard`].
142    pub fn device(id: &str) -> Result<Self, ConfigError> {
143        let id = Id::new(id).ok_or(ConfigError::InvalidDeviceId)?;
144        Ok(Config {
145            discovery: Discovery::Device(id),
146            ..Config::wildcard()
147        })
148    }
149
150    pub(super) fn validate(&self) -> Result<(), ConfigError> {
151        match (&self.domain, &self.discovery) {
152            (Domain::Named(d), _) if !is_valid_id(d) => return Err(ConfigError::InvalidDomain),
153            (Domain::Any, Discovery::Device(_) | Discovery::Tree { .. }) => {
154                return Err(ConfigError::InvalidDomain);
155            }
156            _ => {}
157        }
158        if self
159            .pacing
160            .is_some_and(|p| p.batch_devices == 0 || p.batch_values == 0)
161        {
162            return Err(ConfigError::ZeroBatch);
163        }
164        if !self.host_filters.iter().all(|f| filter::is_valid_filter(f)) {
165            return Err(ConfigError::InvalidHostFilter);
166        }
167        Ok(())
168    }
169}
170
171impl Default for Config {
172    /// [`Config::wildcard`].
173    fn default() -> Self {
174        Config::wildcard()
175    }
176}
177
178/// Why a [`Config`] was refused.
179#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
180#[non_exhaustive]
181pub enum ConfigError {
182    /// [`Domain::Named`] holds an invalid id, or [`Domain::Any`] is combined
183    /// with device or tree discovery.
184    InvalidDomain,
185    /// The device or root id is not a valid id.
186    InvalidDeviceId,
187    /// A [`Pacing`] limit is zero.
188    ZeroBatch,
189    /// A host filter is not a valid MQTT topic filter.
190    InvalidHostFilter,
191}
192
193impl ConfigError {
194    /// camelCase reason code.
195    pub const fn as_str(self) -> &'static str {
196        match self {
197            ConfigError::InvalidDomain => "invalidDomain",
198            ConfigError::InvalidDeviceId => "invalidDeviceId",
199            ConfigError::ZeroBatch => "zeroBatch",
200            ConfigError::InvalidHostFilter => "invalidHostFilter",
201        }
202    }
203}
204
205impl fmt::Display for ConfigError {
206    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
207        f.write_str(self.as_str())
208    }
209}
210
211impl core::error::Error for ConfigError {}
212
213/// Where a property message came from, read from [`ebus_mqtt::Message::retain`].
214#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
215pub enum Origin {
216    /// The broker's retained copy (retain bit set).
217    Retained,
218    /// A live publish (retain bit clear).
219    Live,
220    /// The client does not report the retain bit.
221    Unknown,
222}
223
224impl Origin {
225    pub(super) fn of(retain: Option<bool>) -> Self {
226        match retain {
227            Some(true) => Origin::Retained,
228            Some(false) => Origin::Live,
229            None => Origin::Unknown,
230        }
231    }
232}
233
234/// A received value or `$target`, delivered unmodified (convention.md:78).
235#[derive(Clone, Debug, PartialEq, Eq, Hash)]
236#[non_exhaustive]
237pub struct Value {
238    /// The payload text, with a single `0x00` byte decoded to `""`.
239    pub raw: Arc<str>,
240    /// [`check_value`](crate::datatype::check_value) against the property's
241    /// declaration.
242    pub check: Result<(), Problem>,
243}
244
245/// What a stuck device still lacks.
246#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
247pub enum Missing {
248    /// Its `$state`.
249    State,
250    /// Its `$description`.
251    Description,
252    /// Retained property values.
253    Properties,
254}
255
256/// Something the controller observed. Events are hints; the views returned by
257/// [`Controller::device`](super::Controller::device) are the truth.
258///
259/// Each struct variant is `#[non_exhaustive]`, so a pattern names the fields
260/// it needs and ends with `..`.
261#[derive(Clone, Debug, PartialEq)]
262#[non_exhaustive]
263pub enum Event {
264    /// The broker link changed, emitted after the commands it caused are queued.
265    Link(Link),
266    /// A device's first valid `$state`.
267    #[non_exhaustive]
268    DeviceDiscovered {
269        /// Device id.
270        device: Id,
271        /// Its state.
272        state: DeviceState,
273    },
274    /// A device's `$state` changed; also emitted, with `old: None`, after
275    /// [`Event::DeviceDiscovered`].
276    #[non_exhaustive]
277    StateChanged {
278        /// Device id.
279        device: Id,
280        /// The previous state.
281        old: Option<DeviceState>,
282        /// The new state.
283        new: DeviceState,
284        /// The device's effective state after the change.
285        effective: DeviceState,
286    },
287    /// A device's effective state changed while its own `$state` did not:
288    /// its root changed state or went away.
289    #[non_exhaustive]
290    EffectiveStateChanged {
291        /// Device id.
292        device: Id,
293        /// The previous effective state.
294        old: DeviceState,
295        /// The new effective state.
296        new: DeviceState,
297    },
298    /// A device announced a description: on its ready edge, or at once when
299    /// it changed it while `ready`. A change while `sleeping` is announced on
300    /// the next ready edge.
301    #[non_exhaustive]
302    DescriptionChanged {
303        /// Device id.
304        device: Id,
305        /// The parsed description.
306        description: Arc<DeviceDescription>,
307        /// Nodes and properties left out as illegal.
308        ignored: Vec<Ignored>,
309        /// The description changed while the device was `ready` or
310        /// `sleeping`, which convention.md:207 forbids.
311        while_ready: bool,
312    },
313    /// A `$description` with an illegal device-level field. The device keeps
314    /// its `$state` and has no model.
315    #[non_exhaustive]
316    DescriptionRejected {
317        /// Device id.
318        device: Id,
319        /// Why it was rejected.
320        error: DescriptionError,
321    },
322    /// A retained property's value changed.
323    #[non_exhaustive]
324    PropertyChanged {
325        /// The property.
326        addr: PropertyAddr,
327        /// The value.
328        value: Value,
329        /// Where it came from.
330        origin: Origin,
331    },
332    /// A message for a property declared `retained: false`; never cached
333    /// (convention.md:370-371). An event with [`Origin::Retained`] is a stale
334    /// copy the broker replayed on a subscribe, which convention.md:50 forbids
335    /// devices to publish.
336    #[non_exhaustive]
337    PropertyEvent {
338        /// The property.
339        addr: PropertyAddr,
340        /// The value.
341        value: Value,
342        /// Where it came from.
343        origin: Origin,
344    },
345    /// A property's value was cleared: by a zero-length payload, or because
346    /// a re-sync found it no longer retained.
347    #[non_exhaustive]
348    PropertyCleared {
349        /// The property.
350        addr: PropertyAddr,
351    },
352    /// A property's `$target` changed; `None` when it was cleared.
353    #[non_exhaustive]
354    TargetChanged {
355        /// The property.
356        addr: PropertyAddr,
357        /// The target.
358        target: Option<Value>,
359    },
360    /// A device's `$alert` changed.
361    #[non_exhaustive]
362    Alert {
363        /// Device id.
364        device: Id,
365        /// Alert id.
366        alert: Id,
367        /// Alert text; `None` when the alert was cleared.
368        message: Option<String>,
369    },
370    /// A device's description is held, its property burst is over, and every
371    /// retained property it declares has a value (`missing: 0`), or
372    /// [`Config::sync_timeout_ms`] passed with `missing` still absent.
373    /// Re-armed by a description change and by a link-up.
374    #[non_exhaustive]
375    DeviceSynced {
376        /// Device id.
377        device: Id,
378        /// Declared retained properties without a value.
379        missing: u32,
380    },
381    /// A device's retained messages did not arrive in time.
382    #[non_exhaustive]
383    DeviceStuck {
384        /// Device id.
385        device: Id,
386        /// What it lacks.
387        missing: Missing,
388        /// The resubscribe attempt now made, or with `gave_up` the attempts made.
389        attempt: u8,
390        /// No more attempts follow; the subscriptions stay in place.
391        gave_up: bool,
392    },
393    /// A known device was not heard within [`Config::absent_after_ms`] of a
394    /// link-up. It stays known and heals on its next message. This is also
395    /// the only signal for a device whose `$state` was deleted while the
396    /// controller did not hear it, since nothing retained is replayed for it.
397    #[non_exhaustive]
398    DeviceAbsent {
399        /// Device id.
400        device: Id,
401    },
402    /// A device left the registry: a zero-length `$state`, or, in tree mode,
403    /// its parent no longer declares it.
404    #[non_exhaustive]
405    DeviceRemoved {
406        /// Device id.
407        device: Id,
408    },
409    /// Every device declared under a root has published its description.
410    /// Fires on each incomplete-to-complete edge.
411    #[non_exhaustive]
412    TreeComplete {
413        /// The root.
414        root: Id,
415    },
416}
417
418/// Options for [`Controller::set`](super::Controller::set). Start from
419/// [`SetOptions::default`] and set the fields or chain the setters.
420#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash)]
421#[non_exhaustive]
422pub struct SetOptions {
423    /// Send even when the device's effective state is not `ready`.
424    pub allow_not_ready: bool,
425    /// Skip the payload check.
426    pub unchecked: bool,
427}
428
429impl SetOptions {
430    /// With `allow_not_ready` set.
431    #[must_use]
432    pub const fn allow_not_ready(mut self, allow: bool) -> Self {
433        self.allow_not_ready = allow;
434        self
435    }
436
437    /// With `unchecked` set.
438    #[must_use]
439    pub const fn unchecked(mut self, unchecked: bool) -> Self {
440        self.unchecked = unchecked;
441        self
442    }
443}
444
445/// Why a `/set` or `$broadcast` was not sent.
446#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
447#[non_exhaustive]
448pub enum SetError {
449    /// The broker link is down; QoS 0 is never held (convention.md:695).
450    NotConnected,
451    /// No device with that id is known.
452    UnknownDevice,
453    /// The device has no model, or declares no such property.
454    UnknownProperty,
455    /// The property is not settable.
456    NotSettable,
457    /// The device's effective state is not `ready` (convention.md:279).
458    DeviceNotReady(DeviceState),
459    /// The payload fails the check the device runs.
460    Invalid(Problem),
461    /// A `$broadcast` subtopic level is not a valid id, or the domain is
462    /// [`Domain::Any`].
463    InvalidTopic,
464}
465
466impl SetError {
467    /// camelCase reason code.
468    pub const fn as_str(self) -> &'static str {
469        match self {
470            SetError::NotConnected => "notConnected",
471            SetError::UnknownDevice => "unknownDevice",
472            SetError::UnknownProperty => "unknownProperty",
473            SetError::NotSettable => "notSettable",
474            SetError::DeviceNotReady(_) => "deviceNotReady",
475            SetError::Invalid(_) => "invalid",
476            SetError::InvalidTopic => "invalidTopic",
477        }
478    }
479}
480
481impl fmt::Display for SetError {
482    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
483        match self {
484            SetError::DeviceNotReady(s) => write!(f, "deviceNotReady({s})"),
485            SetError::Invalid(p) => write!(f, "invalid({p})"),
486            other => f.write_str(other.as_str()),
487        }
488    }
489}
490
491impl core::error::Error for SetError {}
492
493/// What the controller knows about one property.
494#[derive(Clone, Debug, Default, PartialEq, Eq)]
495#[non_exhaustive]
496pub struct PropertyView {
497    /// The last value, checked; always `None` for a `retained: false` property.
498    pub value: Option<Value>,
499    /// The last value text that passed the check.
500    pub last_valid: Option<Arc<str>>,
501    /// The last `$target`.
502    pub target: Option<Value>,
503}
504
505#[derive(Clone, Debug, PartialEq, Eq)]
506pub(super) struct Slot {
507    pub(super) view: PropertyView,
508    pub(super) node: Id,
509    pub(super) prop: Id,
510    /// Position in `DeviceDescription::nodes` and the node's `properties`.
511    pub(super) node_idx: usize,
512    pub(super) prop_idx: usize,
513    pub(super) retained: bool,
514}
515
516/// A device's slots by node, then property. Persistent: a clone shares every
517/// entry, and a change copies only the path to it.
518pub(super) type Props = RedBlackTreeMapSync<Id, RedBlackTreeMapSync<Id, Slot>>;
519
520/// What the controller knows about one device.
521#[derive(Clone, Debug)]
522pub struct DeviceView {
523    /// Device id.
524    pub id: Id,
525    /// Topic domain it publishes in.
526    pub domain: Arc<str>,
527    /// Last valid `$state`; always `Some` for a device the queries return.
528    pub state: Option<DeviceState>,
529    /// Effective state (convention.md:262-267): `lost` when its root is lost,
530    /// else its own state.
531    pub effective: Option<DeviceState>,
532    /// The applied description, `None` before the first valid one.
533    pub description: Option<Arc<DeviceDescription>>,
534    /// Heard since the last link-up.
535    pub confirmed: bool,
536    /// [`Event::DeviceSynced`] fired since the last description change or
537    /// link-up.
538    pub synced: bool,
539    pub(super) props: Props,
540    pub(super) alerts: RedBlackTreeMapSync<Id, String>,
541}
542
543impl DeviceView {
544    pub(super) fn new(id: Id, domain: Arc<str>) -> Self {
545        DeviceView {
546            id,
547            domain,
548            state: None,
549            effective: None,
550            description: None,
551            confirmed: true,
552            synced: false,
553            props: Props::new_sync(),
554            alerts: RedBlackTreeMapSync::new_sync(),
555        }
556    }
557
558    /// True when `other` holds the very maps of properties and alerts this
559    /// view holds, as a clone does until either is changed. Constant time;
560    /// `false` does not prove the contents differ.
561    pub fn shares_values(&self, other: &DeviceView) -> bool {
562        self.props.ptr_eq(&other.props) && self.alerts.ptr_eq(&other.alerts)
563    }
564
565    /// The property `node/property`, when the description declares it.
566    pub fn property(&self, node: &str, property: &str) -> Option<&PropertyView> {
567        self.props.get(node)?.get(property).map(|s| &s.view)
568    }
569
570    /// Every declared property as `(node, property, view)`, in id order.
571    pub fn properties(&self) -> impl Iterator<Item = (&Id, &Id, &PropertyView)> + '_ {
572        self.props
573            .values()
574            .flat_map(|m| m.values())
575            .map(|s| (&s.node, &s.prop, &s.view))
576    }
577
578    /// The text of alert `id`, when it is set.
579    pub fn alert(&self, id: &str) -> Option<&str> {
580        self.alerts.get(id).map(String::as_str)
581    }
582
583    /// Every set alert as `(id, text)`, in id order.
584    pub fn alerts(&self) -> impl Iterator<Item = (&Id, &str)> + '_ {
585        self.alerts.iter().map(|(k, v)| (k, v.as_str()))
586    }
587}
ebus_homie/controller/types.rs

ebus_homie/controller/
types.rs
