harmos_runtime/operations/manager/
registration.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12/* === Re-Exports === */
13
14/* === Imports === */
15
16// Standard Library Imports
17
18// Third Party Imports
19
20// Local Imports
21use super::*;
22
23/* === Constants === */
24
25/* === Macros === */
26
27/* === Type Aliases === */
28
29/* === Structs === */
30
31/* === Enums === */
32
33/* === Traits === */
34
35/* === Implementations === */
36
37impl OperationManager {
38    pub fn register_operation<T>(&self, owner: impl Into<String>, lane: impl Lane, resources: Arc<T::Resources>) -> Result<()>
39    where
40        T: for<'ctx> crate::operations::Operation<Context<'ctx> = Context<'ctx>>,
41    {
42        self.register_operation_at::<T>(owner, lane.name(), resources)
43    }
44
45    pub fn register_operation_id<T>(
46        &self, owner: impl Into<String>, id: impl Into<String>, lane: impl Lane, resources: Arc<T::Resources>,
47    ) -> Result<()>
48    where
49        T: for<'ctx> crate::operations::Operation<Context<'ctx> = Context<'ctx>>,
50    {
51        self.register_with(owner, |bindings| bindings.register_operation_id::<T>(id, lane, resources))
52    }
53
54    pub(in crate::operations) fn register_operation_at<T>(
55        &self, owner: impl Into<String>, lane: &'static str, resources: Arc<T::Resources>,
56    ) -> Result<()>
57    where
58        T: for<'ctx> crate::operations::Operation<Context<'ctx> = Context<'ctx>>,
59    {
60        self.register_with(owner, |bindings| bindings.register_owned_at::<T>(lane, resources))
61    }
62
63    pub fn register_wire_operation<T>(&self, owner: impl Into<String>, lane: impl Lane, resources: Arc<T::Resources>) -> Result<()>
64    where
65        T: for<'ctx> crate::operations::Operation<Context<'ctx> = Context<'ctx>>,
66    {
67        self.register_with(owner, |bindings| bindings.register_wire_operation::<T>(lane, resources))
68    }
69
70    pub fn register_wire_contract<T>(
71        &self, owner: impl Into<String>, contract: &crate::operations::Contract, lane: impl Lane, resources: Arc<T::Resources>,
72    ) -> Result<()>
73    where
74        T: for<'ctx> crate::operations::Operation<Context<'ctx> = Context<'ctx>>,
75    {
76        self.register_with(owner, |bindings| bindings.register_wire_contract::<T>(contract, lane, resources))
77    }
78
79    pub(in crate::operations) fn binding_owned_definition<T>(
80        &self, owner: impl Into<String>, lane: &'static str, resources: Arc<T::Resources>, definition: Definition,
81        wire: Option<crate::operations::wire::Dispatch>,
82        #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))] plugin: Option<PluginScope>,
83    ) -> Result<Arc<Entry>>
84    where
85        T: for<'ctx> crate::operations::Operation<Context<'ctx> = Context<'ctx>>,
86    {
87        // Only the scope's built-in observe control may carry a reserved id.
88        #[cfg(feature = "plugin-sidecar")]
89        let built_in = plugin.as_ref().and_then(PluginScope::observe_contract).is_some_and(|contract| *contract.operation() == definition);
90        #[cfg(not(feature = "plugin-sidecar"))]
91        let built_in = false;
92        if built_in { definition.policy.validate().map_err(harmos_domain::Error::from) } else { definition.validate() }
93            .map_err(|source| Error::OperationDeclaration { source })?;
94        if <T::Endpoints as crate::operations::EndpointSet>::item_types() != crate::operations::item_types::<T::Declaration>() {
95            return Err(Error::InvalidInput {
96                field: "operation endpoint types",
97                reason: "authored endpoints do not match declared stream item types",
98            });
99        }
100        let binding = crate::operations::authored::factory::<T>(self, resources);
101        self.binding_typed(
102            owner,
103            lane,
104            binding,
105            definition,
106            wire,
107            #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
108            plugin,
109        )
110    }
111
112    pub(super) fn binding_typed<O: OperationDeclaration, P: Send + 'static>(
113        &self, owner: impl Into<String>, lane: &'static str, binding: Typed<O, P>, definition: Definition,
114        wire: Option<crate::operations::wire::Dispatch>,
115        #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))] plugin: Option<PluginScope>,
116    ) -> Result<Arc<Entry>> {
117        let owner = Identifier::new(owner.into()).map_err(|source| Error::OperationDeclaration { source: source.into() })?;
118        let bounded =
119            self.instances.lane_bounded(lane).ok_or_else(|| Error::NotRegistered { domain: "operation lane", name: lane.to_owned() })?;
120        let generation = OperationGeneration::fresh()?;
121        Ok(Arc::new(Entry {
122            owner,
123            definition,
124            contract: None,
125            #[cfg(feature = "grpc")]
126            grpc: None,
127            generation,
128            lane,
129            bounded,
130            #[cfg(feature = "plugin-sidecar")]
131            resident: false,
132            active: AtomicBool::new(true),
133            retirement: AtomicU8::new(Retirement::Retained as u8),
134            pending: Mutex::new(BTreeMap::new()),
135            binding: Arc::new(binding),
136            #[cfg(feature = "plugin-sidecar")]
137            native_binding: None,
138            wire,
139            #[cfg(feature = "plugin-sidecar")]
140            native: None,
141            #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
142            plugin,
143            #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
144            sdk: None,
145        }))
146    }
147}
148
149/* === Public Functions === */
150
151/* === Private Functions === */
152
153/* === Tests === */
harmos_runtime/operations/manager/registration.rs

harmos_runtime/operations/manager/
registration.rs
