harmos-runtime 0.1.6

Harmos runtime: managed state, undo, recovery, background work and typed streams.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140

/* ***
 *
 * Harmos
 *
 * SPDX-License-Identifier: Apache-2.0
 *
 * ***
 */

/* === Modules === */

mod abandon_file;
mod append_file;
mod capture_file;
mod create_file;
mod delete_file;
mod finish_file;
mod follow_file;
mod follow_objects;
mod follow_production;
mod follow_records;
mod follow_stream;
mod hash_file;
mod list_directory;
mod query;
mod read_file;
mod resume_file;
mod storage_status;
mod watch_directory;
mod write_file;
mod write_objects;
mod write_records;
mod write_schema;
mod write_streams;

/* === Re-Exports === */

pub use harmos_domain::operations::storage::*;

/* === Imports === */

// Standard Library Imports

use std::sync::Arc;

// Third Party Imports

// Local Imports
use crate::catalog::{Catalog, Member, Visitor};

use crate::operations::OperationManager;

use crate::storage::StorageManager;

use crate::{Result, operations::Lane};

/* === Constants === */

const IO_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);

/* === Macros === */

// Bind Each Declaration To Its Original Storage Handler.
macro_rules! provider {
    ($($module:ident),* $(,)?) => {
        impl<C: 'static, V> Catalog<V> for Provider<C>
        where V: Visitor $(+ Member<$module::Handler<C>>)* {
            type Entries = [V::Entry; [$(stringify!($module)),*].len()];
            const ENTRIES: Self::Entries = [$(<V as Member<$module::Handler<C>>>::ENTRY),*];
        }
    };
}

provider! {
    list_directory,
    watch_directory,
    query,
    read_file,
    hash_file,
    capture_file,
    follow_production,
    follow_records,
    follow_stream,
    follow_file,
    storage_status,
    resume_file,
    create_file,
    append_file,
    finish_file,
    abandon_file,
    write_records,
    write_streams,
    write_file,
    delete_file,
    write_schema,
    write_objects,
    follow_objects,
}

/* === Type Aliases === */

/* === Structs === */

struct Provider<C>(std::marker::PhantomData<fn() -> C>);

/* === Enums === */

/* === Traits === */

/* === Implementations === */

/* === Public Functions === */

// Serve The Storage Operations Over One Storage Manager.
pub fn register<C: 'static>(operations: &OperationManager, owner: &str, lane: impl Lane, storage: Arc<StorageManager<C>>) -> Result<()> {
    operations.register_wire_operations::<Provider<C>, _>(owner, lane, storage)
}

/* === Private Functions === */

fn output_timeout() -> crate::Error {
    crate::Error::backend(
        "storage output",
        crate::BackendFailure::Transient,
        std::io::Error::new(std::io::ErrorKind::TimedOut, "consumer delivery deadline exceeded"),
    )
}

fn classify(error: &crate::Error) -> crate::operations::FailureKind {
    let mut cause = error;
    while let crate::Error::Persistence { source } = cause {
        cause = source;
    }
    match cause {
        crate::Error::Backend { kind: crate::BackendFailure::Transient, .. } => crate::operations::FailureKind::Transient,
        _ => crate::operations::FailureKind::Permanent,
    }
}

/* === Tests === */