harmos-runtime 0.1.5

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

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

/* === Modules === */

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

/* === Imports === */

// Standard Library Imports
use std::sync::Arc;
// Third Party Imports
use tokio::sync::Semaphore;
// Local Imports
use super::{Definition, Registry};
use crate::state::dispatch::Dispatch;
use crate::storage::SchemaKey;
use crate::{
    Application, Error, Result, state::Position, state::Prepared, state::PrincipalId, state::Receipt, state::Recorded, state::StateManager,
    state::SystemFact,
};

/* === Constants === */

/* === Macros === */

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

/* === Structs === */

pub struct RecordManager<A: Application> {
    pub(super) state: StateManager<A>,
    pub(super) catalog: Arc<Dispatch<A, crate::storage::Payload<A>>>,
    preparing: Arc<Semaphore>,
    pub(super) schemas: Arc<Registry>,
}

/* === Enums === */

/* === Traits === */

/* === Implementations === */

impl<A: Application> RecordManager<A> {
    pub fn new(state: StateManager<A>) -> Self {
        let (catalog, preparing) = state.record_resources();
        let schemas = state.record_schemas();
        Self { state, catalog, preparing, schemas }
    }

    pub async fn register(&self, owner: &str, schema: &SchemaKey, graph: &crate::schema::graph::Schema) -> Result<Arc<Definition>> {
        let _preparation =
            Arc::clone(&self.preparing).acquire_owned().await.expect("record preparation admission remains owned by the original state");
        let pending = self.schemas.reserve(owner, schema, graph)?;
        let id = crate::DefinitionId::new(schema.id.as_str());
        if !matches!(self.catalog.has_record(&id, schema.version.get()), Ok(false)) {
            return Err(Error::AlreadyRegistered { domain: "native record schema", name: schema.id.as_str().to_owned() });
        }
        if !pending.is_committed()
            && let Some(storage) = &self.schemas.storage
        {
            storage.write_schema(pending.definition.schema(), pending.definition.graph()).await?;
        }
        Ok(pending.commit())
    }

    pub fn definition(&self, schema: &SchemaKey) -> Option<Arc<Definition>> {
        self.schemas.definition(schema)
    }

    pub fn definitions(&self) -> Vec<Arc<Definition>> {
        self.schemas.definitions()
    }

    pub async fn publish<R: Into<A::Record>>(&self, record: R) -> Result<Receipt> {
        let record = record.into();
        let preparation =
            Arc::clone(&self.preparing).acquire_owned().await.expect("record preparation admission remains owned by the original state");
        let prepared = Prepared::typed_record(&record).map_err(Error::Preparation)?;
        let declared = self.catalog.has_record(&prepared.id, prepared.version).map_err(|source| Error::RecordDeclaration { source })?;
        if !declared {
            return Err(Error::RecordDeclaration { source: crate::state::DecodeError::UnknownDefinition { id: prepared.id } });
        }
        self.state.append_prepared(Recorded::Record(record), prepared, preparation).await
    }

    pub async fn publish_system(&self, fact: SystemFact) -> Result<Receipt> {
        let preparation =
            Arc::clone(&self.preparing).acquire_owned().await.expect("record preparation admission remains owned by the original state");
        let prepared = Prepared::system(&fact).map_err(Error::Preparation)?;
        self.state.append_prepared(Recorded::System(fact), prepared, preparation).await
    }

    pub fn as_principal(&self, principal: PrincipalId) -> Self {
        Self::new(self.state.as_principal(principal))
    }

    pub fn caused_by(&self, cause: Position) -> Self {
        Self::new(self.state.caused_by(cause))
    }
}

impl<A: Application> Clone for RecordManager<A> {
    fn clone(&self) -> Self {
        Self::new(self.state.clone())
    }
}

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

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

/* === Tests === */