harmos_operations/storage/
follow_file.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::IO_POLICY;
22use harmos_domain::MemorySize;
23use harmos_domain::model::storage::codec::{Nested, product};
24use harmos_domain::model::storage::files::{FileCompletion, FileEvent, FileSource, FollowStart};
25#[cfg(feature = "runtime")]
26use harmos_sdk::operations::access::GrantAccess;
27#[cfg(feature = "runtime")]
28use harmos_storage::{ErasedCatalog, StorageManager};
29
30/* === Constants === */
31
32/* === Macros === */
33
34product!(FollowFile {
35    source: Nested<FileSource> => 1,
36    start: Nested<FollowStart> => 2,
37});
38
39/* === Type Aliases === */
40
41/* === Structs === */
42
43#[harmos::operation(
44    harmos_path = crate,
45    resources = [StorageManager<ErasedCatalog>, GrantAccess],
46    codec = manual,
47    id = "storage.follow_file",
48    response = FileCompletion,
49    policy = IO_POLICY,
50    output = FileEvent,
51    visibility = Shared,
52    grpc(package = "harmos.storage.v1", service =  "Storage", method =  "FollowFile"),
53    grant = Self::grant_for,
54)]
55#[derive(Clone, Debug, PartialEq, Eq)]
56pub struct FollowFile {
57    pub source: FileSource,
58    pub start: FollowStart,
59}
60
61/* === Enums === */
62
63/* === Traits === */
64
65/* === Implementations === */
66
67impl FollowFile {
68    fn grant_for(request: &FollowFile) -> Option<harmos_domain::plugin::Grant> {
69        request.source.grant_scope().map(|scope| harmos_domain::plugin::Grant::of(<Self as harmos_domain::OperationDeclaration>::ID, scope))
70    }
71}
72
73impl MemorySize for FollowFile {
74    fn memory_size(&self) -> usize {
75        std::mem::size_of::<Self>()
76            .saturating_add(self.source.memory_size().saturating_sub(std::mem::size_of_val(&self.source)))
77            .saturating_add(self.start.memory_size().saturating_sub(std::mem::size_of_val(&self.start)))
78    }
79}
80
81#[cfg(feature = "runtime")]
82mod execution {
83    use super::*;
84
85    // Standard Library Imports
86
87    // Third Party Imports
88
89    // Local Imports
90
91    use harmos_domain::model::storage::files::FileCompletion;
92    use harmos_sdk::operations::{Operation, OperationDeclaration};
93
94    use crate::storage::IO_TIMEOUT;
95    use crate::{Error, Result};
96
97    impl Operation for super::FollowFile {
98        type Error = Error;
99
100        #[tracing::instrument(target = "harmos::storage", name = "storage.operation.follow_file", level = "trace", skip_all)]
101        async fn run(self, context: harmos_sdk::Context<'_, Self>) -> Result<harmos_sdk::operations::Response<FileCompletion>> {
102            let storage = context.resource::<harmos_storage::StorageManager<harmos_storage::ErasedCatalog>>();
103            let grants = context.resource::<harmos_sdk::operations::access::GrantAccess>();
104            let output = context.streams();
105
106            let host = grants;
107            let ctx = &host.0;
108            let grant = self.source.grant_scope().map(|scope| harmos_domain::plugin::Grant::of(FollowFile::ID, scope));
109            if let Some(grant) = &grant {
110                ctx.admit(grant).map_err(Error::Access)?;
111            }
112            let mut follower = storage
113                .follow_file(harmos_storage::FileFollow {
114                    source: self.source.clone(),
115                    start: self.start,
116                })
117                .await?;
118            while let Some(event) = follower.next().await? {
119                if let Some(grant) = &grant {
120                    ctx.admit(grant).map_err(Error::Access)?;
121                }
122                let response = harmos_sdk::operations::Response::from_owner(event.owner());
123                tokio::time::timeout(IO_TIMEOUT, output.send_owned(response)).await.map_err(|_| crate::storage::output_timeout())??;
124            }
125            Ok(follower.completion()?.into())
126        }
127
128        fn classify(error: &Error) -> harmos_sdk::operations::FailureKind {
129            crate::storage::classify(error)
130        }
131    }
132}
133
134/* === Public Functions === */
135
136/* === Private Functions === */
137
138/* === Tests === */
harmos_operations/storage/follow_file.rs

harmos_operations/storage/
follow_file.rs
