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

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

/* === Modules === */

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

/* === Imports === */

// Standard Library Imports
use std::collections::BTreeMap;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
// Third Party Imports
use tokio::sync::watch;
// Local Imports

/* === Constants === */

/* === Macros === */

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

/* === Structs === */

// The cursor each live feed has taken in. A plugin publication never evicts
// a row past the slowest feed, and a sidecar publication waits for it.
#[derive(Debug)]
pub(crate) struct Ingest {
    taken: watch::Sender<BTreeMap<u64, u64>>,
    next: AtomicU64,
}

#[derive(Debug)]
pub(crate) struct Intake {
    ingest: Arc<Ingest>,
    id: u64,
}

/* === Enums === */

/* === Traits === */

/* === Implementations === */

impl Ingest {
    pub(crate) fn new() -> Self {
        Self { taken: watch::channel(BTreeMap::new()).0, next: AtomicU64::new(0) }
    }

    pub(crate) fn engage(self: &Arc<Self>, resume: u64) -> Intake {
        let id = self.next.fetch_add(1, Ordering::Relaxed);
        self.taken.send_modify(|taken| {
            taken.insert(id, resume);
        });
        Intake { ingest: Arc::clone(self), id }
    }

    // Every row at or below this cursor is in every live feed's hands.
    pub(crate) fn taken(&self) -> u64 {
        self.taken.borrow().values().min().copied().unwrap_or(u64::MAX)
    }

    pub(crate) fn watch(&self) -> watch::Receiver<BTreeMap<u64, u64>> {
        self.taken.subscribe()
    }

    // Revocation and close re-evaluate every waiting publication.
    pub(crate) fn wake(&self) {
        self.taken.send_modify(|_| {});
    }
}

impl Intake {
    pub(crate) fn took(&self, cursor: u64) {
        self.ingest.taken.send_if_modified(|taken| {
            let entry = taken.entry(self.id).or_default();
            let advanced = *entry < cursor;
            *entry = (*entry).max(cursor);
            advanced
        });
    }
}

impl Drop for Intake {
    fn drop(&mut self) {
        self.ingest.taken.send_modify(|taken| {
            taken.remove(&self.id);
        });
    }
}

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

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

/* === Tests === */