qb 0.3.8

EMACS for Kubernetes — a powerful, extensible terminal UI for managing Kubernetes clusters
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
134
135

use {
    std::{
        future::Future,
        time::{
            Duration,
            Instant,
            SystemTime,
        },
    },
    tokio::{
        sync::oneshot,
        task::JoinHandle,
    },
};

/// One background job. Dropping the slot aborts the task and discards any
/// result that has not been polled yet, so a superseded read cannot land
/// on the UI or keep allocating after the user has moved on.
pub struct Slot<T> {
    handle: JoinHandle<()>,
    rx: oneshot::Receiver<T>,
}

pub enum SlotPoll<T> {
    Pending,
    Ready(T),
    Closed,
}

impl<T: Send+'static> Slot<T> {
    pub fn spawn(fut: impl Future<Output=T>+Send+'static) -> Self {
        let (tx, rx) = oneshot::channel();
        let handle = tokio::spawn(async move {
            let value = fut.await;
            let _ = tx.send(value);
        });
        Self { handle, rx }
    }

    pub fn poll(&mut self) -> SlotPoll<T> {
        match self.rx.try_recv() {
            | Ok(value) => SlotPoll::Ready(value),
            | Err(oneshot::error::TryRecvError::Empty) => SlotPoll::Pending,
            | Err(oneshot::error::TryRecvError::Closed) => SlotPoll::Closed,
        }
    }
}

impl<T> Drop for Slot<T> {
    fn drop(&mut self) {
        self.handle.abort();
    }
}

/// Detects a sleep/wake (or a wall-clock step) by comparing wall time to the
/// monotonic clock. `Instant` does not advance while the machine is suspended.
pub struct WakeClock {
    last_wall: SystemTime,
    last_mono: Instant,
}

impl WakeClock {
    pub fn new() -> Self {
        Self {
            last_wall: SystemTime::now(),
            last_mono: Instant::now(),
        }
    }

    pub fn jumped(&mut self, threshold: Duration) -> bool {
        let wall = SystemTime::now();
        let mono = Instant::now();
        let jumped = wall_ahead(self.last_wall, wall, self.last_mono, mono, threshold);
        self.last_wall = wall;
        self.last_mono = mono;
        jumped
    }
}

pub fn wall_ahead(
    prev_wall: SystemTime,
    wall: SystemTime,
    prev_mono: Instant,
    mono: Instant,
    threshold: Duration,
) -> bool {
    let wall_delta = wall.duration_since(prev_wall).unwrap_or_default();
    let mono_delta = mono.saturating_duration_since(prev_mono);
    wall_delta.saturating_sub(mono_delta) >= threshold
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn sleep_is_wall_time_without_monotonic_progress() {
        let wall = SystemTime::UNIX_EPOCH + Duration::from_secs(1_000);
        let mono = Instant::now();
        assert!(wall_ahead(
            wall,
            wall + Duration::from_secs(30),
            mono,
            mono,
            Duration::from_secs(2)
        ));
        assert!(!wall_ahead(
            wall,
            wall + Duration::from_millis(40),
            mono,
            mono + Duration::from_millis(40),
            Duration::from_secs(2)
        ));
        assert!(!wall_ahead(
            wall,
            wall - Duration::from_secs(5),
            mono,
            mono,
            Duration::from_secs(2)
        ));
    }

    #[tokio::test]
    async fn slot_delivers_without_the_caller_awaiting() {
        let mut slot = Slot::spawn(async { 7u8 });
        let value = loop {
            match slot.poll() {
                | SlotPoll::Ready(value) => break value,
                | SlotPoll::Pending => tokio::task::yield_now().await,
                | SlotPoll::Closed => panic!("slot closed before a value"),
            }
        };
        assert_eq!(value, 7);
    }
}