harmos_runtime/
builtin.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10// Operations Harmos implements over its own state. Nothing here is served by
11// default: a host registers the catalog explicitly, after which plugins reach
12// it by id through the ordinary call path and gRPC through its declared routes.
13
14/* === Modules === */
15
16/* === Re-Exports === */
17
18pub use harmos_domain::operations::lifecycle::{Cancel, InstanceState, Status, Watch};
19
20/* === Imports === */
21
22// Standard Library Imports
23use std::sync::Arc;
24
25// Third Party Imports
26
27// Local Imports
28use crate::MemorySize;
29use crate::operations::{Context, Lane, Operation, OperationInstanceId, OperationManager, Response, StreamOutput};
30use crate::{Error, Result};
31
32/* === Constants === */
33
34/* === Macros === */
35
36/* === Type Aliases === */
37
38/* === Structs === */
39
40// The built-in operations served so far: `host.operations.status`,
41// `host.operations.cancel` and `host.operations.watch`.
42pub struct Catalog;
43
44// Reports an instance the caller submitted. Settled instances leave the index,
45// so their status is unknown here; a live handle keeps reporting them.
46pub struct Reporting;
47
48// Requests cancellation of an instance the caller submitted.
49pub struct Cancelling;
50
51// Streams each status change of an instance the caller submitted, from its
52// status when the watch begins until it stops or settles.
53pub struct Watching;
54
55/* === Enums === */
56
57/* === Traits === */
58
59/* === Implementations === */
60
61impl Catalog {
62    pub fn register(operations: &OperationManager, owner: impl Into<String>, lane: impl Lane + Copy) -> Result<()> {
63        let owner = owner.into();
64        operations.register_wire_operation::<Reporting>(owner.clone(), lane, Arc::new(()))?;
65        operations.register_wire_operation::<Cancelling>(owner.clone(), lane, Arc::new(()))?;
66        operations.register_wire_operation::<Watching>(owner, lane, Arc::new(()))
67    }
68}
69
70impl Operation for Reporting {
71    type Declaration = Status;
72    type Resources = ();
73    type Error = Error;
74    type Context<'a> = Context<'a>;
75    type Endpoints = ();
76
77    async fn initialize<'ctx>(_: Arc<()>, _: Self::Context<'ctx>) -> Result<Self> {
78        Ok(Self)
79    }
80
81    async fn run<'ctx>(
82        &mut self, context: Self::Context<'ctx>, request: &mut Status, _: &mut Self::Endpoints,
83    ) -> Result<Response<InstanceState>> {
84        let id = OperationInstanceId::new(request.instance);
85        let instance = context.owned_instance(id).ok_or_else(|| unknown(id))?;
86        let state = InstanceState::from(instance.status().portable());
87        Ok(state.into())
88    }
89}
90
91impl Operation for Cancelling {
92    type Declaration = Cancel;
93    type Resources = ();
94    type Error = Error;
95    type Context<'a> = Context<'a>;
96    type Endpoints = ();
97
98    async fn initialize<'ctx>(_: Arc<()>, _: Self::Context<'ctx>) -> Result<Self> {
99        Ok(Self)
100    }
101
102    async fn run<'ctx>(
103        &mut self, context: Self::Context<'ctx>, request: &mut Cancel, _: &mut Self::Endpoints,
104    ) -> Result<Response<InstanceState>> {
105        let id = OperationInstanceId::new(request.instance);
106        let instance = context.owned_instance(id).ok_or_else(|| unknown(id))?;
107        instance.cancel();
108        let state = InstanceState::from(instance.status().portable());
109        Ok(state.into())
110    }
111}
112
113impl Operation for Watching {
114    type Declaration = Watch;
115    type Resources = ();
116    type Error = Error;
117    type Context<'a> = Context<'a>;
118    type Endpoints = StreamOutput<InstanceState>;
119
120    async fn initialize<'ctx>(_: Arc<()>, _: Self::Context<'ctx>) -> Result<Self> {
121        Ok(Self)
122    }
123
124    async fn run<'ctx>(
125        &mut self, context: Self::Context<'ctx>, request: &mut Watch, output: &mut Self::Endpoints,
126    ) -> Result<Response<InstanceState>> {
127        let id = OperationInstanceId::new(request.instance);
128        let instance = context.owned_instance(id).ok_or_else(|| unknown(id))?;
129        // Subscribe before reading so no change between them is lost.
130        let mut changes = instance.settlement();
131        drop(instance);
132        loop {
133            let status = changes.borrow_and_update().portable();
134            let state = InstanceState::from(status);
135            output.send_with(state.memory_size(), || Ok(state.clone())).await?;
136            if status.is_settled() {
137                return Ok(state.into());
138            }
139            if changes.changed().await.is_err() {
140                return Ok(state.into());
141            }
142        }
143    }
144}
145
146/* === Public Functions === */
147
148/* === Private Functions === */
149
150fn unknown(id: OperationInstanceId) -> Error {
151    Error::NotRegistered { domain: "operation instance", name: id.to_string() }
152}
153
154/* === Tests === */
harmos_runtime/builtin.rs

harmos_runtime/
builtin.rs
