harmos_operations/storage/
read_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::FileSource;
25use harmos_domain::model::storage::{ByteRange, FileBatch, FileCompletion};
26use harmos_domain::operations::ReplaySafety;
27#[cfg(feature = "runtime")]
28use harmos_sdk::operations::access::GrantAccess;
29#[cfg(feature = "runtime")]
30use harmos_storage::{ErasedCatalog, StorageManager};
31
32/* === Constants === */
33
34/* === Macros === */
35
36product!(ReadFile {
37    source: Nested<FileSource> => 1,
38    range: Nested<ByteRange> => 2,
39});
40
41/* === Type Aliases === */
42
43/* === Structs === */
44
45// Read A Bounded Byte Range Of A Stored Or Local File.
46#[harmos::operation(
47    harmos_path = crate,
48    resources = [StorageManager<ErasedCatalog>, GrantAccess],
49    codec = manual,
50    id = "storage.read_file",
51    response = FileCompletion,
52    policy = IO_POLICY,
53    output = FileBatch,
54    visibility = Shared,
55    replay = ReplaySafety::Idempotent,
56    grpc(package = "harmos.storage.v1", service =  "Storage", method =  "ReadFile"),
57    grant = Self::grant_for,
58)]
59#[derive(Clone, Debug, PartialEq, Eq)]
60pub struct ReadFile {
61    pub source: FileSource,
62    pub range: ByteRange,
63}
64
65/* === Enums === */
66
67/* === Traits === */
68
69/* === Implementations === */
70
71impl ReadFile {
72    fn grant_for(request: &ReadFile) -> Option<harmos_domain::plugin::Grant> {
73        request.source.grant_scope().map(|scope| harmos_domain::plugin::Grant::of(<Self as harmos_domain::OperationDeclaration>::ID, scope))
74    }
75}
76
77impl MemorySize for ReadFile {
78    fn memory_size(&self) -> usize {
79        std::mem::size_of::<Self>()
80            .saturating_add(self.source.memory_size().saturating_sub(std::mem::size_of_val(&self.source)))
81            .saturating_add(self.range.memory_size().saturating_sub(std::mem::size_of_val(&self.range)))
82    }
83}
84
85#[cfg(feature = "runtime")]
86mod execution {
87    use super::*;
88
89    // Standard Library Imports
90    // Third Party Imports
91    // Local Imports
92    use crate::storage::IO_TIMEOUT;
93    use crate::{Error, Result};
94    use harmos_domain::model::storage::files::FileCompletion;
95    use harmos_sdk::operations::{Operation, OperationDeclaration};
96
97    impl Operation for super::ReadFile {
98        type Error = Error;
99
100        #[tracing::instrument(target = "harmos::storage", name = "storage.operation.read_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(ReadFile::ID, scope));
109            if let Some(grant) = &grant {
110                ctx.admit(grant).map_err(Error::Access)?;
111            }
112            let mut reader = storage
113                .read_file(harmos_storage::FileRead {
114                    source: self.source.clone(),
115                    range: self.range,
116                })
117                .await?;
118            while let Some(batch) = reader.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(batch.owner());
123                tokio::time::timeout(IO_TIMEOUT, output.send_owned(response)).await.map_err(|_| crate::storage::output_timeout())??;
124            }
125            Ok(reader.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/read_file.rs

harmos_operations/storage/
read_file.rs
