state_m/
state_machine.rs

1use crate::{
2    AsState, AsTag, KvAssoc, State, StateEvent,
3    handle::{ArcHandle, AsHandle, Handle, StateChangeError as HandleStateChangeError},
4    reader::Reader,
5    source::Source,
6};
7use async_trait::async_trait;
8use dashmap::DashMap;
9use itertools::Itertools;
10use state_m_macro::*;
11use std::{
12    fmt::{self, Debug, Display},
13    ops::Deref,
14    pin::Pin,
15    sync::Arc,
16};
17use thiserror::Error;
18use tracing::instrument;
19
20/// StateMachine: data structure to store handles.
21/// * `K` - the `Tag` type to distinguish different handles.
22#[derive(Clone)]
23pub struct StateMachine<K>(Arc<DashMap<K, (String, Box<dyn AsHandle>)>>)
24where
25    K: AsTag;
26
27impl<K> Debug for StateMachine<K>
28where
29    K: AsTag,
30{
31    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
32        for item in self.iter() {
33            let (_, (l, h)) = item.pair();
34            write!(f, "{:<20} | {:?}\n", l, h.debug_state())?
35        }
36        Ok(())
37    }
38}
39
40impl<K> Default for StateMachine<K>
41where
42    K: AsTag,
43{
44    fn default() -> Self {
45        Self(Default::default())
46    }
47}
48
49impl<K> Deref for StateMachine<K>
50where
51    K: AsTag,
52{
53    type Target = Arc<DashMap<K, (String, Box<dyn AsHandle>)>>;
54
55    fn deref(&self) -> &Self::Target {
56        &self.0
57    }
58}
59
60impl<K> StateMachine<K>
61where
62    K: 'static + AsTag,
63{
64    fn get_handle<T>(&self, tag: T) -> Result<ArcHandle<T::Value>, GetHandleError<K>>
65    where
66        T: Clone + Debug + Into<K> + KvAssoc,
67        T::Value: AsState + Send + Sync,
68    {
69        let k = tag.clone().into();
70        match self.get(&k) {
71            Some(v) => match v.value().1.downcast_ref::<ArcHandle<T::Value>>() {
72                Some(h) => Ok(h.clone()),
73                None => Err(GetHandleError::TypeNotMatch),
74            },
75            None => Err(GetHandleError::HandleNotExist(tag.into())),
76        }
77    }
78}
79
80impl<K> StateMachine<K>
81where
82    K: 'static + AsTag,
83{
84    /// Add state source into state machine.
85    pub async fn add_source<T>(&self, tag: T, capacity: usize) -> Result<(), AddHandleError<K>>
86    where
87        T: 'static + Clone + Debug + Display + Into<K> + KvAssoc + Send + Sync,
88        T::Value: 'static + AsState + Send + Sync,
89    {
90        let k = tag.clone().into();
91        if self.contains_key(&k) {
92            return Err(AddHandleError::AlreadyExist(tag.into()));
93        }
94        let h = ArcHandle(Arc::new(Handle::from_source(Source::<T::Value>::new(
95            capacity,
96        ))));
97        h.init(tag.clone()).await;
98        self.insert(k, (tag.to_string(), Box::new(h)));
99        Ok(())
100    }
101
102    /// Add state reader into state machine.
103    pub async fn add_reader<T>(
104        &self,
105        tag: T,
106        reader: Reader<T::Value>,
107    ) -> Result<(), AddHandleError<K>>
108    where
109        T: 'static + Clone + Debug + Display + Into<K> + KvAssoc + Send + Sync,
110        T::Value: 'static + AsState + Send + Sync,
111    {
112        let k = tag.clone().into();
113        if self.contains_key(&k) {
114            return Err(AddHandleError::AlreadyExist(tag.into()));
115        }
116        if reader.is_closed() {
117            return Err(AddHandleError::ChannelClosed);
118        }
119        let h = ArcHandle(Arc::new(Handle::from_reader(reader)));
120        h.init(tag.clone()).await;
121        self.insert(k, (tag.to_string(), Box::new(h)));
122        Ok(())
123    }
124
125    /// Remove state source from state machine.
126    pub fn del_handle<T>(&self, tag: &T) -> Result<bool, GetHandleError<K>>
127    where
128        T: 'static + Clone + Debug + Into<K> + KvAssoc,
129        T::Value: AsState + Send + Sync,
130    {
131        match self.remove(&tag.clone().into()) {
132            Some((_, v)) => match v.1.downcast_ref::<ArcHandle<T::Value>>() {
133                Some(h) => {
134                    h.close();
135                    Ok(true)
136                }
137                None => Err(GetHandleError::TypeNotMatch),
138            },
139            None => Ok(false),
140        }
141    }
142
143    /// If state source of tag exists in state machine.
144    pub fn has_handle<T>(&self, tag: &T) -> bool
145    where
146        T: Clone + Into<K>,
147    {
148        self.contains_key(&tag.clone().into())
149    }
150
151    pub fn reader<T>(&self, tag: T) -> Result<Reader<T::Value>, GetHandleError<K>>
152    where
153        T: Clone + Debug + Into<K> + KvAssoc,
154        T::Value: AsState + Send + Sync,
155    {
156        Ok(self.get_handle(tag)?.reader())
157    }
158
159    pub fn value<T>(&self, tag: T) -> Result<T::Value, GetHandleError<K>>
160    where
161        T: Clone + Debug + Into<K> + KvAssoc,
162        T::Value: 'static + AsState + Send + Sync,
163    {
164        Ok(self.get_handle(tag)?.value())
165    }
166
167    pub fn state<T>(&self, tag: T) -> Result<State<T::Value>, GetHandleError<K>>
168    where
169        T: Clone + Debug + Into<K> + KvAssoc,
170        T::Value: 'static + AsState + Send + Sync,
171    {
172        Ok(self.get_handle(tag)?.state())
173    }
174
175    pub async fn touch<T>(&self, tag: T) -> Result<(), StateChangeError<T, K>>
176    where
177        T: Clone + Debug + Into<K> + KvAssoc,
178        T::Value: 'static + AsState + Send + Sync,
179    {
180        Ok(self.get_handle(tag)?.touch().await?)
181    }
182
183    pub async fn wait_touch<T>(&self, tag: T) -> Result<(), StateChangeError<T, K>>
184    where
185        T: Clone + Debug + Into<K> + KvAssoc,
186        T::Value: 'static + AsState + Send + Sync,
187    {
188        Ok(self.get_handle(tag)?.wait_touch().await?)
189    }
190
191    pub async fn alter<T>(&self, tag: T, s: T::Value) -> Result<(), StateChangeError<T, K>>
192    where
193        T: Clone + Debug + Into<K> + KvAssoc,
194        T::Value: 'static + AsState + Send + Sync,
195    {
196        Ok(self.get_handle(tag)?.alter(s).await?)
197    }
198
199    pub async fn wait_alter<T>(&self, tag: T, s: T::Value) -> Result<(), StateChangeError<T, K>>
200    where
201        T: Clone + Debug + Into<K> + KvAssoc,
202        T::Value: 'static + AsState + Send + Sync,
203    {
204        Ok(self.get_handle(tag)?.wait_alter(s).await?)
205    }
206
207    pub async fn amend<T>(
208        &self,
209        tag: T,
210        f: impl FnOnce(T::Value) -> T::Value,
211    ) -> Result<(), StateChangeError<T, K>>
212    where
213        T: Clone + Debug + Into<K> + KvAssoc,
214        T::Value: 'static + AsState + Send + Sync,
215    {
216        Ok(self.get_handle(tag)?.amend(f).await?)
217    }
218
219    pub async fn wait_amend<T>(
220        &self,
221        tag: T,
222        f: impl FnOnce(T::Value) -> T::Value,
223    ) -> Result<(), StateChangeError<T, K>>
224    where
225        T: Clone + Debug + Into<K> + KvAssoc,
226        T::Value: 'static + AsState + Send + Sync,
227    {
228        Ok(self.get_handle(tag)?.wait_amend(f).await?)
229    }
230}
231
232/// State change result.
233#[derive(Clone, Debug)]
234pub enum StateChange<T>
235where
236    T: KvAssoc,
237    T::Value: AsState + Send + Sync,
238{
239    /// State changed.
240    /// * `0` - cur state.
241    /// * `1` - old state.
242    Change(State<T::Value>, State<T::Value>),
243    /// State unchange.
244    /// * `0` - cur state.
245    UnChange(State<T::Value>),
246}
247
248impl<T> StateChange<T>
249where
250    T: KvAssoc,
251    T::Value: AsState + Send + Sync,
252{
253    pub fn cur(&self) -> State<T::Value> {
254        match self {
255            StateChange::Change(v, _) => v.clone(),
256            StateChange::UnChange(v) => v.clone(),
257        }
258    }
259
260    pub fn old(&self) -> State<T::Value> {
261        match self {
262            StateChange::Change(_, v) => v.clone(),
263            StateChange::UnChange(v) => v.clone(),
264        }
265    }
266}
267
268impl<K> StateMachine<K>
269where
270    K: 'static + AsTag,
271{
272    sm_watch!(1);
273    sm_watch!(2);
274    sm_watch!(3);
275    sm_watch!(4);
276    sm_watch!(5);
277    sm_watch!(6);
278    sm_watch!(7);
279    sm_watch!(8);
280    sm_watch!(9);
281    sm_watch!(10);
282
283    sm_merge_reader!(2);
284    sm_merge_reader!(3);
285    sm_merge_reader!(4);
286    sm_merge_reader!(5);
287    sm_merge_reader!(6);
288    sm_merge_reader!(7);
289    sm_merge_reader!(8);
290    sm_merge_reader!(9);
291    sm_merge_reader!(10);
292
293    sm_split_reader!(2);
294    sm_split_reader!(3);
295    sm_split_reader!(4);
296    sm_split_reader!(5);
297    sm_split_reader!(6);
298    sm_split_reader!(7);
299    sm_split_reader!(8);
300    sm_split_reader!(9);
301    sm_split_reader!(10);
302}
303
304pub trait HasStateMachine {
305    type K: AsTag;
306
307    fn state_machine(&self) -> &StateMachine<Self::K>;
308}
309
310#[async_trait]
311pub trait UseStateMachine: HasStateMachine {
312    /// Add state source into state machine.
313    /// * `tag` - the `Tag` of the source.
314    /// * `capacity` - the capacity of broadcast channel.
315    async fn add_source<T>(&self, tag: T, capacity: usize) -> Result<(), AddHandleError<Self::K>>
316    where
317        T: 'static + Clone + Debug + Display + Into<Self::K> + KvAssoc + Send + Sync,
318        T::Value: 'static + AsState + Send + Sync;
319
320    /// Add state reader into state machine.
321    /// * `tag` - the `Tag` of the reader.
322    /// * `reader` - the reader to be added into state machine.
323    async fn add_reader<T>(
324        &self,
325        tag: T,
326        reader: Reader<T::Value>,
327    ) -> Result<(), AddHandleError<Self::K>>
328    where
329        T: 'static + Clone + Debug + Display + Into<Self::K> + KvAssoc + Send + Sync,
330        T::Value: 'static + AsState + Send + Sync;
331
332    /// Delete state handle (source or reader) from state machine.
333    /// * `tag` - the `Tag` of the handle to be deleted.
334    fn del_handle<T>(&self, tag: &T) -> Result<bool, GetHandleError<Self::K>>
335    where
336        T: 'static + Clone + Debug + Into<Self::K> + KvAssoc,
337        T::Value: AsState + Send + Sync;
338
339    /// If there is a handle (source or reader) in state machine for a tag.
340    /// * `tag` - the `Tag` to find handle associated with it.
341    fn has_handle<T>(&self, tag: &T) -> bool
342    where
343        T: Clone + Into<Self::K>;
344
345    /// Get a new reader from state machine, which will receive state change events individually.
346    /// * `tag` - the `Tag` of the handle which you want to get a reader from it.
347    fn reader<T>(&self, tag: T) -> Result<Reader<T::Value>, GetHandleError<Self::K>>
348    where
349        T: Clone + Debug + Into<Self::K> + KvAssoc,
350        T::Value: AsState + Send + Sync;
351
352    /// Get current state value of a tag in state machine.
353    /// * `tag` - the `Tag` of the handle which you want to get state value from it.
354    fn value<T>(&self, tag: T) -> Result<T::Value, GetHandleError<Self::K>>
355    where
356        T: 'static + Clone + Debug + Into<Self::K> + KvAssoc + Send + Sync,
357        T::Value: 'static + AsState + Send + Sync;
358
359    /// Get current state of a tag in state machine.
360    /// * `tag` - the `Tag` of the handle which you want to get state from it.
361    fn state<T>(&self, tag: T) -> Result<State<T::Value>, GetHandleError<Self::K>>
362    where
363        T: 'static + Clone + Debug + Into<Self::K> + KvAssoc + Send + Sync,
364        T::Value: 'static + AsState + Send + Sync;
365
366    /// Trigger a state change event, but doesn't change the state value.
367    /// * `tag` - the `Tag` of the handle which you want to touch.
368    async fn touch<T>(&self, tag: T) -> Result<(), StateChangeError<T, Self::K>>
369    where
370        T: 'static + Clone + Debug + Into<Self::K> + KvAssoc + Send + Sync,
371        T::Value: 'static + AsState + Send + Sync;
372
373    /// Trigger a state change event, but doesn't change the state value, and wait for all readers finishing responding actions.
374    /// * `tag` - the `Tag` of the handle which you want to touch.
375    async fn wait_touch<T>(&self, tag: T) -> Result<(), StateChangeError<T, Self::K>>
376    where
377        T: 'static + Clone + Debug + Into<Self::K> + KvAssoc + Send + Sync,
378        T::Value: 'static + AsState + Send + Sync;
379
380    /// Alter a state in state machine.
381    /// * `tag` - the `Tag` of the handle which you want to alter.
382    async fn alter<T>(&self, tag: T, s: T::Value) -> Result<(), StateChangeError<T, Self::K>>
383    where
384        T: 'static + Clone + Debug + Into<Self::K> + KvAssoc + Send + Sync,
385        T::Value: 'static + AsState + Send + Sync;
386
387    /// Alter a state in state machine, and wait for all readers finishing responding actions.
388    /// * `tag` - the `Tag` of the handle which you want to alter.
389    async fn wait_alter<T>(&self, tag: T, s: T::Value) -> Result<(), StateChangeError<T, Self::K>>
390    where
391        T: 'static + Clone + Debug + Into<Self::K> + KvAssoc + Send + Sync,
392        T::Value: 'static + AsState + Send + Sync;
393
394    /// Amend a state in state machine, with a closure which take the current state value as parameter and return new state value.
395    /// * `tag` - the `Tag` of the handle which you want to amend.
396    async fn amend<T>(
397        &self,
398        tag: T,
399        f: impl FnOnce(T::Value) -> T::Value + Send,
400    ) -> Result<(), StateChangeError<T, Self::K>>
401    where
402        T: 'static + Clone + Debug + Into<Self::K> + KvAssoc + Send + Sync,
403        T::Value: 'static + AsState + Send + Sync;
404
405    /// Amend a state in state machine, with a closure which take the current state value as parameter and return new state value, and wait for all readers finishing responding actions.
406    /// * `tag` - the `Tag` of the handle which you want to amend.
407    async fn wait_amend<T>(
408        &self,
409        tag: T,
410        f: impl FnOnce(T::Value) -> T::Value + Send,
411    ) -> Result<(), StateChangeError<T, Self::K>>
412    where
413        T: 'static + Clone + Debug + Into<Self::K> + KvAssoc + Send + Sync,
414        T::Value: 'static + AsState + Send + Sync;
415
416    watch_decl!(1);
417    watch_decl!(2);
418    watch_decl!(3);
419    watch_decl!(4);
420    watch_decl!(5);
421    watch_decl!(6);
422    watch_decl!(7);
423    watch_decl!(8);
424    watch_decl!(9);
425    watch_decl!(10);
426
427    merge_reader_decl!(2);
428    merge_reader_decl!(3);
429    merge_reader_decl!(4);
430    merge_reader_decl!(5);
431    merge_reader_decl!(6);
432    merge_reader_decl!(7);
433    merge_reader_decl!(8);
434    merge_reader_decl!(9);
435    merge_reader_decl!(10);
436
437    split_reader_decl!(2);
438    split_reader_decl!(3);
439    split_reader_decl!(4);
440    split_reader_decl!(5);
441    split_reader_decl!(6);
442    split_reader_decl!(7);
443    split_reader_decl!(8);
444    split_reader_decl!(9);
445    split_reader_decl!(10);
446}
447
448#[async_trait]
449impl<M> UseStateMachine for M
450where
451    M: 'static + HasStateMachine + Send + Sync,
452{
453    async fn add_source<T>(&self, tag: T, capacity: usize) -> Result<(), AddHandleError<Self::K>>
454    where
455        T: 'static + Clone + Debug + Display + Into<Self::K> + KvAssoc + Send + Sync,
456        T::Value: 'static + AsState + Send + Sync,
457    {
458        self.state_machine()
459            .add_source(tag.clone(), capacity)
460            .await?;
461        Ok(())
462    }
463
464    fn del_handle<T>(&self, tag: &T) -> Result<bool, GetHandleError<Self::K>>
465    where
466        T: 'static + Clone + Debug + Into<Self::K> + KvAssoc,
467        T::Value: AsState + Send + Sync,
468    {
469        self.state_machine().del_handle(tag)
470    }
471
472    fn has_handle<T>(&self, tag: &T) -> bool
473    where
474        T: Clone + Into<Self::K>,
475    {
476        self.state_machine().has_handle(tag)
477    }
478
479    fn reader<T>(&self, tag: T) -> Result<Reader<T::Value>, GetHandleError<Self::K>>
480    where
481        T: Clone + Debug + Into<Self::K> + KvAssoc,
482        T::Value: AsState + Send + Sync,
483    {
484        self.state_machine().reader(tag)
485    }
486
487    async fn add_reader<T>(
488        &self,
489        tag: T,
490        reader: Reader<T::Value>,
491    ) -> Result<(), AddHandleError<Self::K>>
492    where
493        T: 'static + Clone + Debug + Display + Into<Self::K> + KvAssoc + Send + Sync,
494        T::Value: 'static + AsState + Send + Sync,
495    {
496        self.state_machine().add_reader(tag, reader).await?;
497        Ok(())
498    }
499
500    fn value<T>(&self, tag: T) -> Result<T::Value, GetHandleError<Self::K>>
501    where
502        T: 'static + Clone + Debug + Into<Self::K> + KvAssoc + Send + Sync,
503        T::Value: 'static + AsState + Send + Sync,
504    {
505        self.state_machine().value(tag)
506    }
507
508    fn state<T>(&self, tag: T) -> Result<State<T::Value>, GetHandleError<Self::K>>
509    where
510        T: 'static + Clone + Debug + Into<Self::K> + KvAssoc + Send + Sync,
511        T::Value: 'static + AsState + Send + Sync,
512    {
513        self.state_machine().state(tag)
514    }
515
516    #[instrument(level = "trace", skip(self))]
517    async fn touch<T>(&self, tag: T) -> Result<(), StateChangeError<T, Self::K>>
518    where
519        T: 'static + Clone + Debug + Into<Self::K> + KvAssoc + Send + Sync,
520        T::Value: 'static + AsState + Send + Sync,
521    {
522        self.state_machine().touch(tag).await
523    }
524
525    #[instrument(level = "trace", skip(self))]
526    async fn wait_touch<T>(&self, tag: T) -> Result<(), StateChangeError<T, Self::K>>
527    where
528        T: 'static + Clone + Debug + Into<Self::K> + KvAssoc + Send + Sync,
529        T::Value: 'static + AsState + Send + Sync,
530    {
531        self.state_machine().wait_touch(tag).await
532    }
533
534    #[instrument(level = "trace", skip(self))]
535    async fn alter<T>(&self, tag: T, s: T::Value) -> Result<(), StateChangeError<T, Self::K>>
536    where
537        T: 'static + Clone + Debug + Into<Self::K> + KvAssoc + Send + Sync,
538        T::Value: 'static + AsState + Send + Sync,
539    {
540        self.state_machine().alter(tag, s).await
541    }
542
543    #[instrument(level = "trace", skip(self))]
544    async fn wait_alter<T>(&self, tag: T, s: T::Value) -> Result<(), StateChangeError<T, Self::K>>
545    where
546        T: 'static + Clone + Debug + Into<Self::K> + KvAssoc + Send + Sync,
547        T::Value: 'static + AsState + Send + Sync,
548    {
549        self.state_machine().wait_alter(tag, s).await
550    }
551
552    #[instrument(level = "trace", skip(self, f))]
553    async fn amend<T>(
554        &self,
555        tag: T,
556        f: impl FnOnce(T::Value) -> T::Value + Send,
557    ) -> Result<(), StateChangeError<T, Self::K>>
558    where
559        T: 'static + Clone + Debug + Into<Self::K> + KvAssoc + Send + Sync,
560        T::Value: 'static + AsState + Send + Sync,
561    {
562        self.state_machine().amend(tag, f).await
563    }
564
565    #[instrument(level = "trace", skip(self, f))]
566    async fn wait_amend<T>(
567        &self,
568        tag: T,
569        f: impl FnOnce(T::Value) -> T::Value + Send,
570    ) -> Result<(), StateChangeError<T, Self::K>>
571    where
572        T: 'static + Clone + Debug + Into<Self::K> + KvAssoc + Send + Sync,
573        T::Value: 'static + AsState + Send + Sync,
574    {
575        self.state_machine().wait_amend(tag, f).await
576    }
577
578    watch_impl!(1);
579    watch_impl!(2);
580    watch_impl!(3);
581    watch_impl!(4);
582    watch_impl!(5);
583    watch_impl!(6);
584    watch_impl!(7);
585    watch_impl!(8);
586    watch_impl!(9);
587    watch_impl!(10);
588
589    merge_reader_impl!(2);
590    merge_reader_impl!(3);
591    merge_reader_impl!(4);
592    merge_reader_impl!(5);
593    merge_reader_impl!(6);
594    merge_reader_impl!(7);
595    merge_reader_impl!(8);
596    merge_reader_impl!(9);
597    merge_reader_impl!(10);
598
599    split_reader_impl!(2);
600    split_reader_impl!(3);
601    split_reader_impl!(4);
602    split_reader_impl!(5);
603    split_reader_impl!(6);
604    split_reader_impl!(7);
605    split_reader_impl!(8);
606    split_reader_impl!(9);
607    split_reader_impl!(10);
608}
609
610/// StateChangeError
611#[derive(Debug, Error)]
612pub enum StateChangeError<T, K>
613where
614    T: Debug + Into<K> + KvAssoc,
615    T::Value: Default,
616    K: AsTag,
617{
618    #[error(transparent)]
619    GetHandleError(#[from] GetHandleError<K>),
620    #[error(transparent)]
621    StateChangeError(#[from] HandleStateChangeError<T::Value>),
622}
623
624/// GetHandleError
625#[derive(Debug, Error)]
626pub enum GetHandleError<K>
627where
628    K: AsTag,
629{
630    #[error("State handle for tag [{0:?}] not exist.")]
631    HandleNotExist(K),
632    #[error("Type of state value does not match.")]
633    TypeNotMatch,
634}
635
636/// AddHandleError
637#[derive(Debug, Error)]
638pub enum AddHandleError<K>
639where
640    K: AsTag,
641{
642    #[error("State handle for tag [{0:?}] already exist.")]
643    AlreadyExist(K),
644    #[error("The state channel has been closed.")]
645    ChannelClosed,
646}
state_m/state_machine.rs

state_m/
state_machine.rs
