harmos_operations/storage/
watch_directory.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// Third Party Imports
18// Local Imports
19use super::IO_POLICY;
20use harmos_domain::MemorySize;
21use harmos_domain::model::storage::codec::{Flag, Nested, product};
22use harmos_domain::model::storage::{DirectoryCompletion, DirectoryEvent, DirectorySelection};
23use harmos_domain::operations::ReplaySafety;
24#[cfg(feature = "runtime")]
25use harmos_storage::{ErasedCatalog, StorageManager};
26
27/* === Constants === */
28
29/* === Macros === */
30
31product!(WatchDirectory { selection: Nested<DirectorySelection> => 1, include_existing: Flag => 2 });
32
33/* === Type Aliases === */
34
35/* === Structs === */
36
37#[harmos::operation(harmos_path = crate, resources = [StorageManager<ErasedCatalog>], codec = manual,
38    id = "storage.watch_directory",
39    response = DirectoryCompletion,
40    policy = IO_POLICY,
41    output = DirectoryEvent,
42    visibility = Shared,
43    replay = ReplaySafety::Idempotent,
44    grpc(package = "harmos.storage.v1", service =  "Storage", method =  "WatchDirectory"))]
45#[derive(Clone, Debug, PartialEq, Eq)]
46pub struct WatchDirectory {
47    pub selection: DirectorySelection,
48    pub include_existing: bool,
49}
50
51/* === Enums === */
52
53/* === Traits === */
54
55/* === Implementations === */
56
57impl MemorySize for WatchDirectory {
58    fn memory_size(&self) -> usize {
59        std::mem::size_of::<Self>().saturating_add(self.selection.memory_size().saturating_sub(std::mem::size_of_val(&self.selection)))
60    }
61}
62
63#[cfg(feature = "runtime")]
64mod execution {
65
66    // Finite Storage Query Operation
67
68    // Standard Library Imports
69    // Third Party Imports
70    // Local Imports
71    use crate::{Error, Result};
72    use harmos_domain::model::storage::{DirectoryCompletion, DirectoryEvent};
73    use harmos_sdk::operations::{Operation, Response};
74
75    impl Operation for super::WatchDirectory {
76        type Error = Error;
77
78        #[tracing::instrument(target = "harmos::storage", name = "storage.operation.watch_directory", level = "trace", skip_all)]
79        async fn run(self, context: harmos_sdk::Context<'_, Self>) -> Result<Response<DirectoryCompletion>> {
80            let storage = context.resource::<harmos_storage::StorageManager<harmos_storage::ErasedCatalog>>();
81            let output = context.streams();
82
83            let mut reader = storage.watch_directory(self.selection.clone(), self.include_existing).await?;
84            let mut entries = 0_u64;
85            while let Some(item) = reader.next().await? {
86                if matches!(&*item, DirectoryEvent::Existing(_) | DirectoryEvent::Created(_) | DirectoryEvent::Modified(_)) {
87                    entries = entries.checked_add(1).ok_or(Error::ResourceExhausted { resource: "directory observation count" })?;
88                }
89                let response = Response::from_owner(item.owner());
90                tokio::time::timeout(crate::storage::IO_TIMEOUT, output.send_owned(response))
91                    .await
92                    .map_err(|_| crate::storage::output_timeout())??;
93            }
94            Ok(DirectoryCompletion { entries }.into())
95        }
96
97        fn classify(error: &Error) -> harmos_sdk::operations::FailureKind {
98            crate::storage::classify(error)
99        }
100    }
101}
102
103/* === Public Functions === */
104
105/* === Private Functions === */
106
107/* === Tests === */
harmos_operations/storage/watch_directory.rs

harmos_operations/storage/
watch_directory.rs
