harmos_operations/lifecycle/
watch.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#[cfg(feature = "runtime")]
17use harmos_sdk::operations::instances::InstanceAccess;
18
19#[cfg(feature = "runtime")]
20use crate::{Error, Operation, Response};
21use harmos_domain::operations::instance::Id;
22
23/* === Constants === */
24
25/* === Macros === */
26
27/* === Type Aliases === */
28
29/* === Structs === */
30
31#[harmos::operation(
32    harmos_path = crate,
33    resources = [InstanceAccess],
34    id = "host.operations.watch",
35    response = super::InstanceState,
36    output = super::InstanceState,
37    policy = super::WATCH_POLICY,
38    visibility = Shared,
39    grpc(package = "harmos.host.v1", service = "Operations", method = "Watch"),
40)]
41#[derive(Clone, Default, PartialEq)]
42pub struct Watch {
43    pub instance: u64,
44}
45
46/* === Enums === */
47
48/* === Traits === */
49
50/* === Implementations === */
51
52impl Watch {
53    pub fn of(instance: Id) -> Self {
54        Self { instance: instance.value() }
55    }
56}
57
58#[cfg(feature = "runtime")]
59impl Operation for Watch {
60    type Error = Error;
61
62    async fn run(self, context: harmos_sdk::Context<'_, Self>) -> Result<Response<super::InstanceState>, Error> {
63        let access = context.resource::<harmos_sdk::operations::instances::InstanceAccess>();
64        let output = context.streams();
65
66        let instance = access.0.owned(Id::new(self.instance)).ok_or(Error::UnknownInstance(self.instance))?;
67        use harmos_domain::MemorySize;
68        let mut changes = instance.subscribe();
69        drop(instance);
70        loop {
71            let status = changes.current();
72            let state = super::InstanceState::from(status);
73            output.send_with(state.memory_size(), || Ok(state.clone())).await?;
74            if status.is_settled() || !changes.changed().await {
75                return Ok(state.into());
76            }
77        }
78    }
79}
80
81/* === Public Functions === */
82
83/* === Private Functions === */
84
85/* === Tests === */
harmos_operations/lifecycle/watch.rs

harmos_operations/lifecycle/
watch.rs
