harmos_operations/storage/
follow_stream.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::{StorageCursor, StoredRow, StreamSelection};
25#[cfg(feature = "runtime")]
26use harmos_storage::{ErasedCatalog, StorageManager};
27
28/* === Constants === */
29
30/* === Macros === */
31
32product!(FollowStream {
33    after: Optional<Nested<StorageCursor>> => 1,
34    selection: Nested<StreamSelection> => 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_stream",
43    response = StorageCursor,
44    policy = STREAM_POLICY,
45    output = StoredRow,
46    visibility = Host)]
47#[derive(Clone, Debug, PartialEq, Eq)]
48pub struct FollowStream {
49    pub after: Option<StorageCursor>,
50    pub selection: StreamSelection,
51}
52
53/* === Enums === */
54
55/* === Traits === */
56
57/* === Implementations === */
58
59impl MemorySize for FollowStream {
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    use harmos_storage::{StorageCursor, StoredRow};
78
79    use crate::{Error, MemorySize, Result};
80
81    impl Operation for super::FollowStream {
82        type Error = Error;
83
84        async fn run(self, context: harmos_sdk::Context<'_, Self>) -> Result<harmos_sdk::operations::Response<StorageCursor>> {
85            let storage = context.resource::<harmos_storage::StorageManager<harmos_storage::ErasedCatalog>>();
86            let output = context.streams();
87
88            harmos_storage::query::Source::Stream(self.selection.clone()).validate()?;
89            let mut follower = storage.follow::<StoredRow>(self.after.as_ref(), self.selection.clone()).await?;
90            loop {
91                let page = follower.next().await?;
92                for entry in page.entries() {
93                    output.send_with(entry.memory_size(), || Ok(entry.clone())).await?;
94                }
95            }
96        }
97    }
98}
99
100/* === Public Functions === */
101
102/* === Private Functions === */
103
104/* === Tests === */
harmos_operations/storage/follow_stream.rs

harmos_operations/storage/
follow_stream.rs
