state_m/
reader.rs

1use crate::{
2    AsState,
3    source::Inner,
4    state::{State, StateEvent},
5};
6use std::{fmt::Debug, ops::Deref};
7use tokio::{
8    select,
9    sync::broadcast::{Sender, channel, error::RecvError},
10};
11
12/// Reader of state, to receive state change events.
13#[derive(Clone, Debug)]
14pub struct Reader<S>(Inner<S>)
15where
16    S: 'static + AsState;
17
18impl<S> Deref for Reader<S>
19where
20    S: 'static + AsState,
21{
22    type Target = Inner<S>;
23
24    fn deref(&self) -> &Self::Target {
25        &self.0
26    }
27}
28
29impl<S> Reader<S>
30where
31    S: 'static + AsState,
32{
33    pub(crate) fn from(inner: Inner<S>) -> Self {
34        Self(inner)
35    }
36
37    pub(crate) fn new(capacity: usize, sender: Sender<StateEvent<S>>) -> Self {
38        Self(Inner {
39            capacity,
40            sender,
41            pass_checks: Default::default(),
42        })
43    }
44}
45
46impl<S> Reader<S>
47where
48    S: 'static + AsState + Send,
49{
50    /// Check is the channel has been closed.
51    pub fn is_closed(&self) -> bool {
52        self.sender.subscribe().is_closed()
53    }
54
55    /// Convert data type of state reader.
56    /// # Arguments
57    /// * `capacity` - capacity of the new broadcast channel will be created.
58    /// # Returns
59    /// Reader of new data type.
60    pub fn extend<T>(&self, capacity: usize) -> Reader<T>
61    where
62        T: AsState + From<S> + Send,
63    {
64        self.extend_with(capacity, T::from)
65    }
66
67    /// Convert data type of state reader, with an closure.
68    /// # Arguments
69    /// * `capacity` - capacity of the new broadcast channel will be created.
70    /// * `f` - an closure, which takes the old state value as parameter, and return the new state value.
71    /// # Returns
72    /// Reader of new data type.
73    pub fn extend_with<T, F>(&self, capacity: usize, f: F) -> Reader<T>
74    where
75        T: AsState + Send,
76        F: Fn(S) -> T + Send + 'static,
77    {
78        let (tx, _) = channel(capacity);
79        let tx_c = tx.clone();
80        let mut rx_o = self.sender.subscribe();
81        tokio::spawn(async move {
82            loop {
83                select! {
84                    res = rx_o.recv() => {
85                        match res {
86                            Ok(s) => {
87                                tracing::trace!("recv -- {:?}", s.state);
88                                let s_new = StateEvent {
89                                    state: State {
90                                        value: f(s.state.value),
91                                        timestamp: s.state.timestamp,
92                                    },
93                                    is_touch: s.is_touch,
94                                    close_handle: s.close_handle,
95                                };
96                                if tx_c.send(s_new).is_err() {
97                                    break;
98                                }
99                            },
100                            Err(e) => {
101                                match e {
102                                    RecvError::Closed => break,
103                                    RecvError::Lagged(n) => {
104                                        tracing::warn!("lagged | skipped {n} messages.")
105                                    },
106                                }
107                            },
108                        }
109                    }
110                }
111            }
112        });
113        Reader(Inner {
114            capacity: self.capacity,
115            sender: tx,
116            pass_checks: self.pass_checks.clone(),
117        })
118    }
119
120    /// Convert data type of state reader, with an async closure.
121    /// # Arguments
122    /// * `capacity` - capacity of the new broadcast channel will be created.
123    /// * `f` - an async closure, which takes the old state value as parameter, and return the new state value.
124    /// # Returns
125    /// Reader of new data type.
126    pub fn async_entend_with<T, F, Fut>(&self, capacity: usize, f: F) -> Reader<T>
127    where
128        T: 'static + AsState + Send,
129        F: Fn(S) -> Fut + Send + Sync + 'static,
130        Fut: Future<Output = T> + Send,
131    {
132        let (tx, _) = channel(capacity);
133        let tx_c = tx.clone();
134        let mut rx_o = self.sender.subscribe();
135        tokio::spawn(async move {
136            loop {
137                select! {
138                    res = rx_o.recv() => {
139                        match res {
140                            Ok(s) => {
141                                tracing::trace!("recv -- {:?}", s.state);
142                                let s_new = StateEvent {
143                                    state: State {
144                                        value: f(s.state.value).await,
145                                        timestamp: s.state.timestamp,
146                                    },
147                                    is_touch: s.is_touch,
148                                    close_handle: s.close_handle,
149                                };
150                                if tx_c.send(s_new).is_err() {
151                                    break;
152                                }
153                            },
154                            Err(e) => {
155                                match e {
156                                    RecvError::Closed => break,
157                                    RecvError::Lagged(n) => {
158                                        tracing::warn!("lagged | skipped {n} messages.")
159                                    },
160                                }
161                            },
162                        }
163                    }
164                }
165            }
166        });
167        Reader(Inner {
168            capacity,
169            sender: tx,
170            pass_checks: self.pass_checks.clone(),
171        })
172    }
173}
state_m/reader.rs

state_m/
reader.rs
