harmos_runtime/observers/
mod.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10// The observer kind of resident operation. An observer's declaration (its
11// interval, streams, demand keys and activation) is attribute metadata; the
12// operation manager activates it through `submit_resident`, shares one
13// instance among runtime callers and reports its
14// phases through the shared operation status. A backend drives one
15// activation: in process over an authored observer, or remotely through a
16// sidecar's observe control.
17
18/* === Modules === */
19
20mod driver;
21
22#[cfg(feature = "plugin-sidecar")]
23mod native;
24
25/* === Re-Exports === */
26
27pub use crate::operations::resident::Staged as ObserverRegistration;
28pub use driver::ObserverActivation;
29
30pub(crate) use driver::Backend;
31#[cfg(feature = "plugin-sidecar")]
32pub(crate) use driver::ChildSettlement;
33#[cfg(feature = "plugin-sidecar")]
34pub(crate) use native::Native;
35
36/* === Imports === */
37
38// Standard Library Imports
39use std::collections::BTreeSet;
40use std::future::Future;
41use std::sync::Arc;
42use std::time::Duration;
43// Third Party Imports
44
45// Local Imports
46use crate::{Error, Result};
47use harmos_domain::inventory::StreamDeclaration;
48
49/* === Constants === */
50
51/* === Macros === */
52
53/* === Type Aliases === */
54
55/* === Structs === */
56
57#[derive(Clone, Debug)]
58pub struct ObserverDefinition {
59    pub id: String,
60    pub interval: Duration,
61    pub streams: Vec<StreamDeclaration>,
62    pub demand_keys: Vec<String>,
63}
64
65/* === Enums === */
66
67/* === Traits === */
68
69pub trait ObserverBackend: Send + Sync + 'static {
70    fn definition(&self) -> crate::Result<ObserverDefinition>;
71    fn run(
72        &self, resources: Arc<harmos_sdk::ExecutionContext>, activation: ObserverActivation,
73    ) -> impl Future<Output = crate::Result<()>> + Send;
74}
75
76/* === Implementations === */
77
78/* === Public Functions === */
79
80/* === Private Functions === */
81
82pub(crate) fn validate(definition: &ObserverDefinition) -> Result<()> {
83    if definition.id.trim().is_empty() || definition.id.chars().any(char::is_control) {
84        return Err(Error::InvalidInput {
85            field: "observer identity",
86            reason: "must be nonempty and contain no control characters",
87        });
88    }
89    if definition.interval.is_zero() {
90        return Err(Error::InvalidInput {
91            field: "observer interval",
92            reason: "must be positive",
93        });
94    }
95    let mut names = BTreeSet::new();
96    for stream in &definition.streams {
97        if stream.id.is_empty() || stream.schema == 0 || stream.window == 0 || !names.insert(stream.id.as_str()) {
98            return Err(Error::InvalidInput {
99                field: "observer streams",
100                reason: "stream identities must be unique and schema/window must be positive",
101            });
102        }
103    }
104    for key in &definition.demand_keys {
105        if key.is_empty() || !names.insert(key.as_str()) {
106            return Err(Error::InvalidInput {
107                field: "observer demand",
108                reason: "demand keys must be nonempty, unique, and distinct from streams",
109            });
110        }
111    }
112    Ok(())
113}
114
115/* === Tests === */
harmos_runtime/observers/mod.rs

harmos_runtime/observers/
mod.rs
