grpc_webnext_client/
state.rs

1//! Connectivity state, and watching it change.
2//!
3//! gRPC channels expose this (`GetState` / `WaitForStateChange` in grpc-go) because a
4//! reconnecting channel is otherwise a black box: an app cannot tell "the server is
5//! down" from "nothing has been asked of it yet". tonic happens not to surface it;
6//! a frontend needs it more than a backend does, because the answer is usually a
7//! banner in the UI.
8
9use std::cell::RefCell;
10use std::rc::Rc;
11
12use futures::channel::mpsc;
13use futures::stream::Stream;
14
15/// Where the tunnel is, using gRPC's connectivity-state vocabulary.
16#[derive(Debug, Clone, Copy, PartialEq, Eq)]
17pub enum ConnectivityState {
18    /// No tunnel, and none being opened — the state before the first call, and
19    /// after one drops.
20    Idle,
21    /// A tunnel is being opened.
22    Connecting,
23    /// A tunnel is open and calls can be made.
24    Ready,
25    /// The last attempt to open a tunnel failed. The next call tries again.
26    TransientFailure,
27}
28
29#[derive(Default)]
30struct Inner {
31    state: Option<ConnectivityState>,
32    watchers: Vec<mpsc::UnboundedSender<ConnectivityState>>,
33}
34
35/// The current state plus a fan-out to anyone watching.
36#[derive(Clone, Default)]
37pub(crate) struct StateWatch {
38    inner: Rc<RefCell<Inner>>,
39}
40
41impl StateWatch {
42    pub(crate) fn get(&self) -> ConnectivityState {
43        self.inner.borrow().state.unwrap_or(ConnectivityState::Idle)
44    }
45
46    /// Record a state. Repeats are dropped, so a watcher sees transitions rather
47    /// than a stream of the same value.
48    pub(crate) fn set(&self, state: ConnectivityState) {
49        let mut inner = self.inner.borrow_mut();
50        if inner.state == Some(state) {
51            return;
52        }
53        inner.state = Some(state);
54        // Drop watchers whose receiver is gone — otherwise a long-lived client
55        // accumulates one dead sender per abandoned watcher.
56        inner.watchers.retain(|w| w.unbounded_send(state).is_ok());
57    }
58
59    pub(crate) fn watch(&self) -> impl Stream<Item = ConnectivityState> + 'static {
60        let (tx, rx) = mpsc::unbounded();
61        self.inner.borrow_mut().watchers.push(tx);
62        rx
63    }
64}
65
66#[cfg(test)]
67mod tests {
68    use super::*;
69    use futures::StreamExt;
70
71    #[test]
72    fn starts_idle() {
73        assert_eq!(StateWatch::default().get(), ConnectivityState::Idle);
74    }
75
76    #[test]
77    fn watchers_see_transitions_but_not_repeats() {
78        futures::executor::block_on(async {
79            let watch = StateWatch::default();
80            let mut seen = watch.watch();
81
82            watch.set(ConnectivityState::Connecting);
83            watch.set(ConnectivityState::Connecting); // same value: not an event
84            watch.set(ConnectivityState::Ready);
85            watch.set(ConnectivityState::Idle);
86            drop(watch);
87
88            let all: Vec<_> = seen.by_ref().collect().await;
89            assert_eq!(
90                all,
91                vec![
92                    ConnectivityState::Connecting,
93                    ConnectivityState::Ready,
94                    ConnectivityState::Idle
95                ]
96            );
97        });
98    }
99
100    #[test]
101    fn a_watcher_added_later_sees_only_what_follows() {
102        futures::executor::block_on(async {
103            let watch = StateWatch::default();
104            watch.set(ConnectivityState::Ready);
105            let mut late = watch.watch();
106            watch.set(ConnectivityState::Idle);
107            drop(watch);
108
109            let all: Vec<_> = late.by_ref().collect().await;
110            assert_eq!(all, vec![ConnectivityState::Idle]);
111        });
112    }
113
114    #[test]
115    fn a_dropped_watcher_does_not_accumulate() {
116        let watch = StateWatch::default();
117        drop(watch.watch());
118        watch.set(ConnectivityState::Connecting);
119        watch.set(ConnectivityState::Ready);
120        assert!(watch.inner.borrow().watchers.is_empty());
121    }
122}
grpc_webnext_client/state.rs

grpc_webnext_client/
state.rs
