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

state_m/
reader.rs
