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

harmos_operations/lifecycle/
watch.rs
