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

harmos_runtime/storage/operations/
mod.rs
