mj_controller/pollers/
feed.rs

1use super::*;
2
3/// Something a control loop waits on and then drains: one awaited receive for
4/// the `select!` arm, and a non-blocking receive for the batch that follows.
5pub trait FeedSource {
6    type Item;
7
8    /// Cancel-safe: a wait that loses the race must not drop a message.
9    fn wait(&mut self) -> impl Future<Output = Option<Self::Item>>;
10
11    fn poll_now(&mut self) -> Option<Self::Item>;
12}
13
14impl<T> FeedSource for tokio::sync::mpsc::Receiver<T> {
15    type Item = T;
16
17    fn wait(&mut self) -> impl Future<Output = Option<T>> {
18        self.recv()
19    }
20
21    fn poll_now(&mut self) -> Option<T> {
22        self.try_recv().ok()
23    }
24}
25
26impl<T> FeedSource for tokio::sync::mpsc::UnboundedReceiver<T> {
27    type Item = T;
28
29    fn wait(&mut self) -> impl Future<Output = Option<T>> {
30        self.recv()
31    }
32
33    fn poll_now(&mut self) -> Option<T> {
34        self.try_recv().ok()
35    }
36}
37
38impl<T: Clone> FeedSource for tokio::sync::watch::Receiver<T> {
39    type Item = T;
40
41    async fn wait(&mut self) -> Option<T> {
42        self.changed().await.ok()?;
43        Some(self.borrow_and_update().clone())
44    }
45
46    fn poll_now(&mut self) -> Option<T> {
47        self.has_changed()
48            .ok()
49            .filter(|changed| *changed)
50            .map(|_| self.borrow_and_update().clone())
51    }
52}
53
54impl FeedSource for SessionManagerUpdates {
55    type Item = SessionManagerUpdate;
56
57    fn wait(&mut self) -> impl Future<Output = Option<SessionManagerUpdate>> {
58        self.recv()
59    }
60
61    fn poll_now(&mut self) -> Option<SessionManagerUpdate> {
62        self.try_recv().ok()
63    }
64}
65
66impl FeedSource for RecoveryCoordinator {
67    type Item = RecoveryResult;
68
69    fn wait(&mut self) -> impl Future<Output = Option<RecoveryResult>> {
70        self.result()
71    }
72
73    fn poll_now(&mut self) -> Option<RecoveryResult> {
74        self.try_result()
75    }
76}
77
78impl FeedSource for CredentialSyncCoordinator {
79    type Item = mj_core::credentials::CredentialSyncResult;
80
81    fn wait(&mut self) -> impl Future<Output = Option<Self::Item>> {
82        self.result()
83    }
84
85    fn poll_now(&mut self) -> Option<Self::Item> {
86        self.try_result()
87    }
88}
89
90/// One background feed as a control loop uses it.
91///
92/// The `select!` arm hands the message that woke the loop to [`Feed::accept`],
93/// and the drain that follows walks [`Feed::next_ready`] until the feed is
94/// empty, so a burst of updates costs one draw. A closed channel reports `None`
95/// for ever, which would leave its arm permanently ready; `accept` retires the
96/// feed instead, and [`Feed::is_open`] gates the arm.
97pub struct Feed<S: FeedSource> {
98    pub(super) source: S,
99    pub(super) pending: Option<S::Item>,
100    pub(super) open: bool,
101    /// Whether the drain has taken a message since it was last asked. A loop
102    /// that gates a frame on a timer still has to draw when a message rode
103    /// along with that timer.
104    pub(super) delivered: bool,
105}
106
107impl<S: FeedSource> Feed<S> {
108    pub fn new(source: S) -> Self {
109        Self {
110            source,
111            pending: None,
112            open: true,
113            delivered: false,
114        }
115    }
116
117    pub fn is_open(&self) -> bool {
118        self.open
119    }
120
121    pub fn wait(&mut self) -> impl Future<Output = Option<S::Item>> {
122        self.source.wait()
123    }
124
125    /// Latches the message that won the select and reports whether one arrived.
126    /// Applying it determines whether the visible state needs a redraw.
127    pub fn accept(&mut self, message: Option<S::Item>) -> bool {
128        match message {
129            Some(message) => {
130                self.pending = Some(message);
131                true
132            }
133            None => {
134                self.open = false;
135                false
136            }
137        }
138    }
139
140    /// The next message for the batch drain: the one that won the select
141    /// first, then whatever queued behind it.
142    pub fn next_ready(&mut self) -> Option<S::Item> {
143        let message = self.pending.take().or_else(|| self.source.poll_now());
144        self.delivered |= message.is_some();
145        message
146    }
147
148    /// Whether [`Self::next_ready`] produced a message since the last call.
149    pub fn take_delivered(&mut self) -> bool {
150        std::mem::take(&mut self.delivered)
151    }
152}
mj_controller/pollers/feed.rs

mj_controller/pollers/
feed.rs
