harmos-runtime 0.1.2

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

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

/* === Modules === */

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

/* === Imports === */

// Standard Library Imports
use std::future::Future;
use std::io;
use std::pin::Pin;
// Third Party Imports
use tokio::task::JoinHandle;
// Local Imports
use super::Arrival;
use crate::supervision::executor;
use harmos_sidecar::attach::{End, Ending};
use harmos_sidecar::{Frame, transport};

/* === Constants === */

/* === Macros === */

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

/* === Structs === */

struct DrivenEnd {
    sender: transport::Sender,
    task: Option<JoinHandle<io::Result<()>>>,
    original: Option<Ending>,
    failure: Option<String>,
}

/* === Enums === */

pub(super) enum Connecting {
    Unavailable,
    Binary(transport::Sender, transport::Receiver),
}

pub(super) struct Sender(transport::Sender);
pub(super) struct Receiver(transport::Receiver);

/* === Traits === */

/* === Implementations === */

impl Connecting {
    pub fn new(data: transport::Io, control: Option<transport::Io>, ending: Option<Ending>) -> (Self, Option<Ending>) {
        let Some(control) = control else {
            return (Self::Unavailable, ending);
        };
        let transport::Connection { sender, receiver, mut driver } = transport::Connection::new(data, control);
        let task = executor::task(async move { driver.run().await });
        let ending = Ending::new(DrivenEnd { sender: sender.clone(), task: Some(task), original: ending, failure: None });
        (Self::Binary(sender, receiver), Some(ending))
    }

    pub async fn connect(self) -> io::Result<(Sender, Receiver)> {
        match self {
            Self::Unavailable => {
                Err(io::Error::new(io::ErrorKind::Unsupported, "attachment requires an independent binary control channel"))
            }
            Self::Binary(sender, receiver) => Ok((Sender(sender), Receiver(receiver))),
        }
    }
}

impl Sender {
    pub async fn send(&self, frame: Frame) -> io::Result<()> {
        self.0.send(frame).await
    }
}

impl Receiver {
    pub async fn receive(&mut self) -> io::Result<Option<Arrival>> {
        Ok(self.0.receive().await?.map(Arrival::Binary))
    }
}

impl End for DrivenEnd {
    fn settle(&mut self) -> Pin<Box<dyn Future<Output = io::Result<()>> + Send + '_>> {
        Box::pin(async move {
            self.sender.close();
            if let Some(task) = self.task.as_mut() {
                let joined = task.await;
                self.task.take();
                if let Err(error) = joined {
                    self.failure = Some(error.to_string());
                }
                // Protocol EOF/Failure Is Reported By The Receiver; Joining Proves Physical I/O Ended.
            }
            if let Some(original) = self.original.as_mut() {
                original.settle().await?;
            }
            if let Some(failure) = &self.failure {
                return Err(io::Error::other(failure.clone()));
            }
            self.original.take();
            Ok(())
        })
    }
}

impl Drop for DrivenEnd {
    fn drop(&mut self) {
        self.sender.close();
    }
}

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

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

/* === Tests === */