rmux-server 0.10.0

Tokio daemon and request dispatcher for the RMUX terminal multiplexer.
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

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) struct ShutdownRetryToken(u64);

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum DetachedConnectionScope {
    CountAll,
    Exclude(u64),
}

impl DetachedConnectionScope {
    fn from_exclusion(excluded_connection_id: Option<u64>) -> Self {
        match excluded_connection_id {
            Some(connection_id) => Self::Exclude(connection_id),
            None => Self::CountAll,
        }
    }

    fn tighten(self, requested: Self) -> Self {
        match (self, requested) {
            (Self::CountAll, _) | (_, Self::CountAll) => Self::CountAll,
            (Self::Exclude(current), Self::Exclude(requested)) if current == requested => self,
            (Self::Exclude(_), Self::Exclude(_)) => Self::CountAll,
        }
    }

    fn exclusion(self) -> Option<u64> {
        match self {
            Self::CountAll => None,
            Self::Exclude(connection_id) => Some(connection_id),
        }
    }
}

pub(super) fn tighten_exclusions(current: Option<u64>, requested: Option<u64>) -> Option<u64> {
    DetachedConnectionScope::from_exclusion(current)
        .tighten(DetachedConnectionScope::from_exclusion(requested))
        .exclusion()
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
struct ScheduledRetry {
    token: ShutdownRetryToken,
    scope: DetachedConnectionScope,
}

#[derive(Debug)]
pub(in crate::handler) struct ShutdownRetryState {
    scheduled: Option<ScheduledRetry>,
    next_token: u64,
}

impl ShutdownRetryState {
    pub(in crate::handler) fn new() -> Self {
        Self {
            scheduled: None,
            next_token: 1,
        }
    }

    pub(super) fn tighten_if_scheduled(&mut self, excluded_connection_id: Option<u64>) -> bool {
        if self.scheduled.is_none() {
            return false;
        }
        let _ = self.effective_exclusion(excluded_connection_id);
        true
    }

    pub(super) fn effective_exclusion(
        &mut self,
        excluded_connection_id: Option<u64>,
    ) -> Option<u64> {
        let Some(scheduled) = self.scheduled.as_mut() else {
            return excluded_connection_id;
        };
        scheduled.scope = scheduled
            .scope
            .tighten(DetachedConnectionScope::from_exclusion(
                excluded_connection_id,
            ));
        scheduled.scope.exclusion()
    }

    pub(super) fn schedule_or_tighten(
        &mut self,
        excluded_connection_id: Option<u64>,
    ) -> Option<ShutdownRetryToken> {
        if self.tighten_if_scheduled(excluded_connection_id) {
            return None;
        }

        let token = ShutdownRetryToken(self.next_token);
        self.next_token = self.next_token.wrapping_add(1);
        if self.next_token == 0 {
            self.next_token = 1;
        }
        self.scheduled = Some(ScheduledRetry {
            token,
            scope: DetachedConnectionScope::from_exclusion(excluded_connection_id),
        });
        Some(token)
    }

    pub(super) fn take(&mut self, token: ShutdownRetryToken) -> Option<Option<u64>> {
        let scheduled = self
            .scheduled
            .filter(|scheduled| scheduled.token == token)?;
        self.scheduled = None;
        Some(scheduled.scope.exclusion())
    }

    pub(super) fn cancel(&mut self, token: ShutdownRetryToken) {
        if self
            .scheduled
            .is_some_and(|scheduled| scheduled.token == token)
        {
            self.scheduled = None;
        }
    }
}