harmos_runtime/storage/operations/
mod.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12mod abandon_file;
13mod append_file;
14mod capture_file;
15mod create_file;
16mod delete_file;
17mod finish_file;
18mod follow_file;
19mod follow_objects;
20mod follow_production;
21mod follow_records;
22mod follow_stream;
23mod hash_file;
24mod list_directory;
25mod query;
26mod read_file;
27mod resume_file;
28mod storage_status;
29mod watch_directory;
30mod write_file;
31mod write_objects;
32mod write_records;
33mod write_schema;
34mod write_streams;
35
36/* === Re-Exports === */
37
38pub use harmos_domain::operations::storage::*;
39
40/* === Imports === */
41
42// Standard Library Imports
43
44use std::sync::Arc;
45
46// Third Party Imports
47
48// Local Imports
49use crate::catalog::{Catalog, Member, Visitor};
50
51use crate::operations::OperationManager;
52
53use crate::storage::StorageManager;
54
55use crate::{Result, operations::Lane};
56
57/* === Constants === */
58
59const IO_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
60
61/* === Macros === */
62
63// Bind Each Declaration To Its Original Storage Handler.
64macro_rules! provider {
65    ($($module:ident),* $(,)?) => {
66        impl<C: 'static, V> Catalog<V> for Provider<C>
67        where V: Visitor $(+ Member<$module::Handler<C>>)* {
68            type Entries = [V::Entry; [$(stringify!($module)),*].len()];
69            const ENTRIES: Self::Entries = [$(<V as Member<$module::Handler<C>>>::ENTRY),*];
70        }
71    };
72}
73
74provider! {
75    list_directory,
76    watch_directory,
77    query,
78    read_file,
79    hash_file,
80    capture_file,
81    follow_production,
82    follow_records,
83    follow_stream,
84    follow_file,
85    storage_status,
86    resume_file,
87    create_file,
88    append_file,
89    finish_file,
90    abandon_file,
91    write_records,
92    write_streams,
93    write_file,
94    delete_file,
95    write_schema,
96    write_objects,
97    follow_objects,
98}
99
100/* === Type Aliases === */
101
102/* === Structs === */
103
104struct Provider<C>(std::marker::PhantomData<fn() -> C>);
105
106/* === Enums === */
107
108/* === Traits === */
109
110/* === Implementations === */
111
112/* === Public Functions === */
113
114// Serve The Storage Operations Over One Storage Manager.
115pub fn register<C: 'static>(operations: &OperationManager, owner: &str, lane: impl Lane, storage: Arc<StorageManager<C>>) -> Result<()> {
116    operations.register_wire_operations::<Provider<C>, _>(owner, lane, storage)
117}
118
119/* === Private Functions === */
120
121fn output_timeout() -> crate::Error {
122    crate::Error::backend(
123        "storage output",
124        crate::BackendFailure::Transient,
125        std::io::Error::new(std::io::ErrorKind::TimedOut, "consumer delivery deadline exceeded"),
126    )
127}
128
129fn classify(error: &crate::Error) -> crate::operations::FailureKind {
130    let mut cause = error;
131    while let crate::Error::Persistence { source } = cause {
132        cause = source;
133    }
134    match cause {
135        crate::Error::Backend { kind: crate::BackendFailure::Transient, .. } => crate::operations::FailureKind::Transient,
136        _ => crate::operations::FailureKind::Permanent,
137    }
138}
139
140/* === Tests === */
harmos_runtime/storage/operations/mod.rs

harmos_runtime/storage/operations/
mod.rs
