organix 0.1.3

Building organic application in rust, leveraging tokio and defining inter communication layer mechanism
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
125
126
127
128
129
130
131
132
133

use std::{
    fmt,
    pin::Pin,
    sync::Arc,
    task::{Context, Poll},
    time::SystemTime,
};
use tokio::{
    stream::Stream,
    sync::watch::{self, Receiver, Sender},
};

#[derive(Debug, Clone)]
pub struct StatusReader {
    status: Receiver<Status>,
    updater: Arc<Sender<Status>>,
}

#[derive(Debug)]
pub struct StatusUpdater {
    updater: Arc<Sender<Status>>,
}

/// these are the different status of the service
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub enum Status {
    Starting { since: SystemTime },
    Started { since: SystemTime },
    ShuttingDown { since: SystemTime },
    Shutdown { since: SystemTime },
}

impl Status {
    pub fn starting() -> Self {
        Status::Starting {
            since: SystemTime::now(),
        }
    }

    pub fn started() -> Self {
        Status::Started {
            since: SystemTime::now(),
        }
    }

    pub fn shutting_down() -> Self {
        Status::ShuttingDown {
            since: SystemTime::now(),
        }
    }

    pub fn shutdown() -> Self {
        Status::Shutdown {
            since: SystemTime::now(),
        }
    }

    pub fn is_shutdown(&self) -> bool {
        if let Status::Shutdown { .. } = self {
            true
        } else {
            false
        }
    }
}

impl StatusReader {
    /// create a new StatusReader
    #[allow(clippy::new_without_default)]
    pub fn new(initial: Status) -> Self {
        let (updater, status) = watch::channel(initial);
        let updater = Arc::new(updater);

        StatusReader { status, updater }
    }

    /// create a `StatusUpdater` from the given reader
    pub fn updater(&self) -> StatusUpdater {
        StatusUpdater {
            updater: Arc::clone(&self.updater),
        }
    }

    /// get the current `Status`
    pub fn status(&self) -> Status {
        *self.status.borrow()
    }

    /// be notified on status update
    pub async fn updated(&mut self) -> Option<Status> {
        self.status.recv().await
    }
}

impl StatusUpdater {
    pub fn update(&self, status: Status) {
        if self.updater.broadcast(status).is_err() {
            // if the receiver is gone, it means the watchdog dropped the
            // associated StatusReader and that it is not important to monitor
            // the status: we can ignore the broadcast error then
        }
    }
}

impl Stream for StatusReader {
    type Item = Status;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Status>> {
        Pin::new(&mut self.get_mut().status).poll_next(cx)
    }
}

impl Drop for StatusUpdater {
    fn drop(&mut self) {
        self.update(Status::ShuttingDown {
            since: SystemTime::now(),
        })
    }
}

impl fmt::Display for Status {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        let (v, since) = match self {
            Status::Starting { since } => ("starting", since),
            Status::Started { since } => ("started", since),
            Status::ShuttingDown { since } => ("shutting down", since),
            Status::Shutdown { since } => ("shutdown", since),
        };

        // TODO: display time in better format
        write!(f, "{} (since {:?})", v, since)
    }
}