state_m/
state_machine.rs

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

state_m/
state_machine.rs
