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_with_resources`, shares one
13// instance among every activation over the same resources 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    type Resources: Send + Sync + 'static;
71    fn definition(&self) -> crate::Result<ObserverDefinition>;
72    fn run(&self, resources: Arc<Self::Resources>, activation: ObserverActivation) -> impl Future<Output = crate::Result<()>> + Send;
73}
74
75/* === Implementations === */
76
77/* === Public Functions === */
78
79/* === Private Functions === */
80
81pub(crate) fn validate(definition: &ObserverDefinition) -> Result<()> {
82    if definition.id.trim().is_empty() || definition.id.chars().any(char::is_control) {
83        return Err(Error::InvalidInput {
84            field: "observer identity",
85            reason: "must be nonempty and contain no control characters",
86        });
87    }
88    if definition.interval.is_zero() {
89        return Err(Error::InvalidInput {
90            field: "observer interval",
91            reason: "must be positive",
92        });
93    }
94    let mut names = BTreeSet::new();
95    for stream in &definition.streams {
96        if stream.id.is_empty() || stream.schema == 0 || stream.window == 0 || !names.insert(stream.id.as_str()) {
97            return Err(Error::InvalidInput {
98                field: "observer streams",
99                reason: "stream identities must be unique and schema/window must be positive",
100            });
101        }
102    }
103    for key in &definition.demand_keys {
104        if key.is_empty() || !names.insert(key.as_str()) {
105            return Err(Error::InvalidInput {
106                field: "observer demand",
107                reason: "demand keys must be nonempty, unique, and distinct from streams",
108            });
109        }
110    }
111    Ok(())
112}
113
114/* === Tests === */
harmos_runtime/observers/mod.rs

harmos_runtime/observers/
mod.rs
