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    /// # Returns
58    /// Reader of new data type.
59    pub fn derive<T>(&self) -> Reader<T>
60    where
61        T: AsState + From<S> + Send,
62    {
63        self.derive_by(T::from)
64    }
65
66    /// Convert data type of state reader, with an closure.
67    /// # Arguments
68    /// * `f` - an closure, which takes the old state value as parameter, and return the new state value.
69    /// # Returns
70    /// Reader of new data type.
71    pub fn derive_by<T, F>(&self, f: F) -> Reader<T>
72    where
73        T: AsState + Send,
74        F: Fn(S) -> T + Send + 'static,
75    {
76        let (tx, _) = channel(self.capacity);
77        let tx_c = tx.clone();
78        let mut rx_o = self.sender.subscribe();
79        tokio::spawn(async move {
80            loop {
81                select! {
82                    res = rx_o.recv() => {
83                        match res {
84                            Ok(s) => {
85                                tracing::trace!("recv -- {:?}", s.state);
86                                let s_new = StateEvent {
87                                    state: State {
88                                        value: f(s.state.value),
89                                        timestamp: s.state.timestamp,
90                                    },
91                                    is_touch: s.is_touch,
92                                    close_handle: s.close_handle,
93                                };
94                                if tx_c.send(s_new).is_err() {
95                                    break;
96                                }
97                            },
98                            Err(e) => {
99                                match e {
100                                    RecvError::Closed => break,
101                                    RecvError::Lagged(n) => {
102                                        tracing::warn!("lagged | skipped {n} messages.")
103                                    },
104                                }
105                            },
106                        }
107                    }
108                }
109            }
110        });
111        Reader(Inner {
112            capacity: self.capacity,
113            sender: tx,
114            pass_checks: self.pass_checks.clone(),
115        })
116    }
117
118    /// Convert data type of state reader, with an async closure.
119    /// # Arguments
120    /// * `f` - an async closure, which takes the old state value as parameter, and return the new state value.
121    /// # Returns
122    /// Reader of new data type.
123    pub fn derive_by_async<T, F, Fut>(&self, f: F) -> Reader<T>
124    where
125        T: 'static + AsState + Send,
126        F: Fn(S) -> Fut + Send + Sync + 'static,
127        Fut: Future<Output = T> + Send,
128    {
129        let (tx, _) = channel(self.capacity);
130        let tx_c = tx.clone();
131        let mut rx_o = self.sender.subscribe();
132        tokio::spawn(async move {
133            loop {
134                select! {
135                    res = rx_o.recv() => {
136                        match res {
137                            Ok(s) => {
138                                tracing::trace!("recv -- {:?}", s.state);
139                                let s_new = StateEvent {
140                                    state: State {
141                                        value: f(s.state.value).await,
142                                        timestamp: s.state.timestamp,
143                                    },
144                                    is_touch: s.is_touch,
145                                    close_handle: s.close_handle,
146                                };
147                                if tx_c.send(s_new).is_err() {
148                                    break;
149                                }
150                            },
151                            Err(e) => {
152                                match e {
153                                    RecvError::Closed => break,
154                                    RecvError::Lagged(n) => {
155                                        tracing::warn!("lagged | skipped {n} messages.")
156                                    },
157                                }
158                            },
159                        }
160                    }
161                }
162            }
163        });
164        Reader(Inner {
165            capacity: self.capacity,
166            sender: tx,
167            pass_checks: self.pass_checks.clone(),
168        })
169    }
170}
state_m/reader.rs

state_m/
reader.rs
