harmos_operations/storage/
follow_objects.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::STREAM_POLICY;
22use harmos_domain::MemorySize;
23use harmos_domain::model::storage::codec::{Nested, Optional, product};
24use harmos_domain::model::storage::{ObjectEntry, ObjectSelection, StorageCursor};
25#[cfg(feature = "runtime")]
26use harmos_storage::{ErasedCatalog, StorageManager};
27
28/* === Constants === */
29
30/* === Macros === */
31
32product!(FollowObjects {
33    after: Optional<Nested<StorageCursor>> => 1,
34    selection: Nested<ObjectSelection> => 2,
35});
36
37/* === Type Aliases === */
38
39/* === Structs === */
40
41#[harmos::operation(harmos_path = crate, resources = [StorageManager<ErasedCatalog>], codec = manual,
42    id = "storage.follow_objects",
43    response = StorageCursor,
44    policy = STREAM_POLICY,
45    output = ObjectEntry,
46    visibility = Host)]
47#[derive(Clone, Debug, PartialEq, Eq)]
48pub struct FollowObjects {
49    pub after: Option<StorageCursor>,
50    pub selection: ObjectSelection,
51}
52
53/* === Enums === */
54
55/* === Traits === */
56
57/* === Implementations === */
58
59impl MemorySize for FollowObjects {
60    fn memory_size(&self) -> usize {
61        std::mem::size_of::<Self>()
62            .saturating_add(self.after.memory_size().saturating_sub(std::mem::size_of_val(&self.after)))
63            .saturating_add(self.selection.memory_size().saturating_sub(std::mem::size_of_val(&self.selection)))
64    }
65}
66
67#[cfg(feature = "runtime")]
68mod execution {
69
70    // Standard Library Imports
71
72    // Third Party Imports
73
74    // Local Imports
75
76    use harmos_sdk::operations::Operation;
77
78    use harmos_storage::{ObjectEntry, StorageCursor};
79
80    use crate::{Error, MemorySize, Result};
81
82    impl Operation for super::FollowObjects {
83        type Error = Error;
84
85        async fn run(self, context: harmos_sdk::Context<'_, Self>) -> Result<harmos_sdk::operations::Response<StorageCursor>> {
86            let storage = context.resource::<harmos_storage::StorageManager<harmos_storage::ErasedCatalog>>();
87            let output = context.streams();
88
89            if !self.selection.follows_revisions() {
90                return Err(Error::InvalidInput {
91                    field: "storage follow",
92                    reason: "following delivers every object revision",
93                });
94            }
95            let mut follower = storage.follow::<ObjectEntry>(self.after.as_ref(), self.selection.clone()).await?;
96            loop {
97                let page = follower.next().await?;
98                for entry in page.entries() {
99                    output.send_with(entry.memory_size(), || Ok(entry.clone())).await?;
100                }
101            }
102        }
103    }
104}
105
106/* === Public Functions === */
107
108/* === Private Functions === */
109
110/* === Tests === */
harmos_operations/storage/follow_objects.rs

harmos_operations/storage/
follow_objects.rs
