harmos_operations/storage/
follow_records.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
17use std::borrow::Cow;
18// Third Party Imports
19// Local Imports
20use harmos_domain::model::storage::codec::{Nested, Optional, product};
21use harmos_domain::model::storage::{RecordSelection, StorageCursor, StoredRecord};
22use harmos_domain::schema::declaration::{Field, Kind};
23use harmos_domain::schema::{Cardinality, Described};
24use harmos_domain::{MemorySize, MessageError};
25#[cfg(feature = "runtime")]
26use harmos_storage::{ErasedCatalog, StorageManager};
27
28/* === Constants === */
29
30/* === Macros === */
31
32product!(FollowBoundaries {
33    saved: Nested<StorageCursor> => 1,
34    accepted: Nested<StorageCursor> => 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_records",
43    response = StorageCursor,
44    policy = super::STREAM_POLICY,
45    output = Followed,
46    visibility = Shared,
47    grpc(package = "harmos.storage.v1", service =  "Storage", method =  "FollowRecords"),
48    grant = Self::grant_for)]
49#[derive(Clone, Debug, PartialEq, Eq)]
50pub struct FollowRecords {
51    pub after: Option<StorageCursor>,
52    pub selection: RecordSelection,
53}
54
55#[derive(Clone, Debug, PartialEq, Eq)]
56pub struct FollowBoundaries {
57    pub saved: StorageCursor,
58    pub accepted: StorageCursor,
59}
60
61/* === Enums === */
62
63#[derive(Clone, Debug, PartialEq)]
64pub enum Followed {
65    Boundaries(FollowBoundaries),
66    Record(StoredRecord),
67}
68
69/* === Traits === */
70
71/* === Implementations === */
72
73product!(FollowRecords {
74    after: Optional<Nested<StorageCursor>> => 1,
75    selection: Nested<RecordSelection> => 2,
76});
77
78impl FollowRecords {
79    fn grant_for(request: &FollowRecords) -> Option<harmos_domain::plugin::Grant> {
80        Some(harmos_domain::plugin::Grant::of(
81            <Self as harmos_domain::OperationDeclaration>::ID,
82            request.selection.schema.id.as_str().to_owned(),
83        ))
84    }
85}
86
87impl MemorySize for FollowRecords {
88    fn memory_size(&self) -> usize {
89        std::mem::size_of::<Self>()
90            .saturating_add(self.after.memory_size().saturating_sub(std::mem::size_of_val(&self.after)))
91            .saturating_add(self.selection.memory_size().saturating_sub(std::mem::size_of_val(&self.selection)))
92    }
93}
94
95impl MemorySize for FollowBoundaries {
96    fn memory_size(&self) -> usize {
97        std::mem::size_of::<Self>()
98            .saturating_add(self.saved.memory_size().saturating_sub(std::mem::size_of::<StorageCursor>()))
99            .saturating_add(self.accepted.memory_size().saturating_sub(std::mem::size_of::<StorageCursor>()))
100    }
101}
102
103impl MemorySize for Followed {
104    fn memory_size(&self) -> usize {
105        let heap = match self {
106            Self::Boundaries(value) => value.memory_size().saturating_sub(std::mem::size_of::<FollowBoundaries>()),
107            Self::Record(value) => value.memory_size().saturating_sub(std::mem::size_of::<StoredRecord>()),
108        };
109        std::mem::size_of::<Self>().saturating_add(heap)
110    }
111}
112
113impl Described for Followed {
114    const NAME: &'static str = concat!(module_path!(), "::", stringify!(Followed));
115    const FIELDS: &'static [Field] = &[
116        Field {
117            name: Cow::Borrowed("boundaries"),
118            tag: 1,
119            cardinality: Cardinality::Variant,
120            kind: Kind::Message { definition: FollowBoundaries::DEFINITION },
121        },
122        Field {
123            name: Cow::Borrowed("record"),
124            tag: 2,
125            cardinality: Cardinality::Variant,
126            kind: Kind::Message { definition: StoredRecord::DEFINITION },
127        },
128    ];
129
130    const MARKER: &'static [u8] = &harmos_domain::schema::declaration::encode::<
131        { harmos_domain::schema::declaration::encoded_len(Self::NAME, Self::FIELDS) },
132    >(Self::NAME, Self::FIELDS);
133}
134
135impl harmos_domain::__private::PostcardValue for Followed {
136    fn write(&self, bytes: &mut Vec<u8>) {
137        match self {
138            Self::Boundaries(value) => {
139                0u32.write(bytes);
140                value.write(bytes);
141            }
142            Self::Record(value) => {
143                1u32.write(bytes);
144                value.write(bytes);
145            }
146        }
147    }
148
149    fn size(&self) -> usize {
150        1usize.saturating_add(match self {
151            Self::Boundaries(value) => value.size(),
152            Self::Record(value) => value.size(),
153        })
154    }
155
156    fn memory(input: &mut harmos_domain::__private::PostcardReader<'_>) -> std::result::Result<usize, MessageError> {
157        input.nested(|input| {
158            let heap = match input.value::<u32>()? {
159                0 => <FollowBoundaries as harmos_domain::__private::PostcardValue>::memory(input)?
160                    .saturating_sub(std::mem::size_of::<FollowBoundaries>()),
161                1 => <StoredRecord as harmos_domain::__private::PostcardValue>::memory(input)?
162                    .saturating_sub(std::mem::size_of::<StoredRecord>()),
163                _ => return Err(MessageError::borrowed("unknown follow output variant")),
164            };
165            harmos_domain::__private::decoding_add(std::mem::size_of::<Self>(), heap)
166        })
167    }
168
169    fn read(input: &mut harmos_domain::__private::PostcardReader<'_>) -> std::result::Result<Self, MessageError> {
170        input.nested(|input| match input.value::<u32>()? {
171            0 => <FollowBoundaries as harmos_domain::__private::PostcardValue>::read(input).map(Self::Boundaries),
172            1 => <StoredRecord as harmos_domain::__private::PostcardValue>::read(input).map(Self::Record),
173            _ => Err(MessageError::borrowed("unknown follow output variant")),
174        })
175    }
176}
177
178impl Followed {
179    pub fn wrapping_memory(record: usize) -> usize {
180        std::mem::size_of::<Self>().saturating_add(record.saturating_sub(std::mem::size_of::<StoredRecord>()))
181    }
182}
183
184#[cfg(feature = "runtime")]
185mod execution {
186    use super::*;
187
188    // Standard Library Imports
189
190    // Third Party Imports
191
192    // Local Imports
193
194    use harmos_sdk::operations::Operation;
195
196    use harmos_storage::{StorageCursor, StoredRecord};
197
198    use crate::MemorySize;
199    use crate::{Error, Result};
200
201    impl Operation for super::FollowRecords {
202        type Error = Error;
203
204        async fn run(self, context: harmos_sdk::Context<'_, Self>) -> Result<harmos_sdk::operations::Response<StorageCursor>> {
205            let storage = context.resource::<harmos_storage::StorageManager<harmos_storage::ErasedCatalog>>();
206            let output = context.streams();
207
208            harmos_storage::query::Source::Records(self.selection.clone()).validate()?;
209            let mut follower = storage.follow::<StoredRecord>(self.after.as_ref(), self.selection.clone()).await?;
210            let first = Followed::Boundaries(FollowBoundaries {
211                saved: follower.saved_boundary().clone(),
212                accepted: follower.accepted_boundary().clone(),
213            });
214            output.send_with(first.memory_size(), || Ok(first.clone())).await?;
215            loop {
216                let page = follower.next().await?;
217                for entry in page.entries() {
218                    let wrapped = Followed::wrapping_memory(entry.memory_size());
219                    output.send_with(wrapped, || Ok(Followed::Record(entry.clone()))).await?;
220                }
221            }
222        }
223    }
224}
225
226/* === Public Functions === */
227
228/* === Private Functions === */
229
230/* === Tests === */
harmos_operations/storage/follow_records.rs

harmos_operations/storage/
follow_records.rs
