harmos_runtime/state/
managed.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
17use std::sync::Arc;
18
19// Third Party Imports
20
21// Local Imports
22use super::{At, Error, Origin, Position, TrackedState};
23use crate::Application;
24
25/* === Constants === */
26
27/* === Macros === */
28
29/* === Type Aliases === */
30
31/* === Structs === */
32
33pub struct ManagedState<A: Application> {
34    tracked: Arc<TrackedState<A>>,
35}
36
37/* === Enums === */
38
39/* === Traits === */
40
41/* === Implementations === */
42
43impl<A: Application> ManagedState<A> {
44    pub(super) fn new(tracked: Arc<TrackedState<A>>) -> Self {
45        Self { tracked }
46    }
47
48    #[tracing::instrument(target = "harmos::state", name = "state.read", level = "debug", skip_all)]
49    pub async fn read<R>(&self, read: impl FnOnce(&A::State) -> R) -> Result<At<R>, Error> {
50        if self.tracked.poisoned() {
51            return Err(self.tracked.unavailable());
52        }
53        let state = self.tracked.state.read().await;
54        // The Writer May Have Panicked While This Reader Awaited The Lock.
55        if self.tracked.poisoned() {
56            return Err(self.tracked.unavailable());
57        }
58
59        Ok(At {
60            value: read(&state.value),
61            position: state.position,
62        })
63    }
64
65    #[tracing::instrument(target = "harmos::state", name = "state.at", level = "debug", skip_all)]
66    pub async fn at(&self, origin: Origin<A::State>, at: Position) -> Result<At<A::State>, Error> {
67        if self.tracked.poisoned() {
68            return Err(self.tracked.unavailable());
69        }
70
71        {
72            let tail = self.tracked.tail.read().await;
73            if tail.splits_commit(at) || tail.splits_commit(origin.position) {
74                return Err(Error::IncompleteCommit);
75            }
76        }
77        self.tracked.reads.state(origin, at).await
78    }
79}
80
81impl<A: Application> Clone for ManagedState<A> {
82    fn clone(&self) -> Self {
83        Self::new(Arc::clone(&self.tracked))
84    }
85}
86
87/* === Public Functions === */
88
89/* === Private Functions === */
90
91/* === Tests === */
harmos_runtime/state/managed.rs

harmos_runtime/state/
managed.rs
