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

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

/* === Modules === */

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

/* === Imports === */

// Standard Library Imports
use std::future::Future;
use std::sync::Arc;
// Third Party Imports
use futures_util::stream::{FuturesUnordered, StreamExt};
use tokio::sync::{Mutex, watch};
use tokio::task::{JoinError, JoinHandle};
// Local Imports
use super::executor;

/* === Constants === */

/* === Macros === */

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

/* === Structs === */

pub(crate) struct Task<T> {
    stopping: watch::Sender<bool>,
    join: Mutex<Join<T>>,
}

/* === Enums === */

enum Join<T> {
    Running(JoinHandle<T>),
    Settled(Result<Arc<T>, Arc<JoinError>>),
}

/* === Traits === */

/* === Implementations === */

impl<T: Send + Sync + 'static> Task<T> {
    pub(crate) fn spawn<F, W>(work: W) -> Self
    where
        W: FnOnce(watch::Sender<bool>) -> F,
        F: Future<Output = T> + Send + 'static,
    {
        Self::with_cancellation(watch::channel(false).0, work)
    }

    pub(crate) fn with_cancellation<F, W>(stopping: watch::Sender<bool>, work: W) -> Self
    where
        W: FnOnce(watch::Sender<bool>) -> F,
        F: Future<Output = T> + Send + 'static,
    {
        let task = executor::task(work(stopping.clone()));
        Self { stopping, join: Mutex::new(Join::Running(task)) }
    }

    pub(crate) fn stop(&self) {
        self.stopping.send_replace(true);
    }

    pub(crate) async fn settle(&self) -> Result<Arc<T>, Arc<JoinError>> {
        self.settle_retry(|| None::<std::future::Ready<T>>).await
    }

    pub(crate) async fn settle_retry<F>(&self, retry: impl FnOnce() -> Option<F>) -> Result<Arc<T>, Arc<JoinError>>
    where
        F: Future<Output = T> + Send + 'static,
    {
        let mut join = self.join.lock().await;
        if matches!(&*join, Join::Settled(_))
            && let Some(future) = retry()
        {
            *join = Join::Running(executor::task(future));
        }
        if let Join::Running(task) = &mut *join {
            let result = task.await.map(Arc::new).map_err(Arc::new);
            *join = Join::Settled(result);
        }
        let Join::Settled(result) = &*join else { unreachable!("task was joined") };
        result.clone()
    }
}

impl<T> Drop for Task<T> {
    fn drop(&mut self) {
        self.stopping.send_replace(true);
    }
}

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

pub(crate) async fn settle_all<E, F>(tasks: impl IntoIterator<Item = F>) -> Result<(), E>
where
    F: Future<Output = Result<(), E>>,
{
    let mut pending: FuturesUnordered<_> = tasks.into_iter().enumerate().map(|(index, task)| async move { (index, task.await) }).collect();
    let mut failure = None;
    while let Some((index, result)) = pending.next().await {
        if let Err(error) = result
            && failure.as_ref().is_none_or(|(first, _)| index < *first)
        {
            failure = Some((index, error));
        }
    }
    failure.map_or(Ok(()), |(_, error)| Err(error))
}

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

/* === Tests === */