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() {
34            let (_, (l, h)) = item.pair();
35            write!(f, "{:<20} | {:?}\n", l, h.debug_state())?
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
239/// State change result.
240#[derive(Clone, Debug)]
241pub enum StateChange<T>
242where
243    T: KvAssoc,
244    T::Value: AsState + Send + Sync,
245{
246    /// State changed.
247    /// * `0` - cur state.
248    /// * `1` - old state.
249    Change(State<T::Value>, State<T::Value>),
250    /// State unchange.
251    /// * `0` - cur state.
252    UnChange(State<T::Value>),
253}
254
255impl<T> StateChange<T>
256where
257    T: KvAssoc,
258    T::Value: AsState + Send + Sync,
259{
260    pub fn cur(&self) -> State<T::Value> {
261        match self {
262            StateChange::Change(v, _) => v.clone(),
263            StateChange::UnChange(v) => v.clone(),
264        }
265    }
266
267    pub fn old(&self) -> State<T::Value> {
268        match self {
269            StateChange::Change(_, v) => v.clone(),
270            StateChange::UnChange(v) => v.clone(),
271        }
272    }
273}
274
275impl<K> StateMachine<K>
276where
277    K: 'static + AsTag,
278{
279    sm_watch!(1);
280    sm_watch!(2);
281    sm_watch!(3);
282    sm_watch!(4);
283    sm_watch!(5);
284    sm_watch!(6);
285    sm_watch!(7);
286    sm_watch!(8);
287    sm_watch!(9);
288    sm_watch!(10);
289
290    sm_merge_reader!(2);
291    sm_merge_reader!(3);
292    sm_merge_reader!(4);
293    sm_merge_reader!(5);
294    sm_merge_reader!(6);
295    sm_merge_reader!(7);
296    sm_merge_reader!(8);
297    sm_merge_reader!(9);
298    sm_merge_reader!(10);
299
300    sm_split_reader!(2);
301    sm_split_reader!(3);
302    sm_split_reader!(4);
303    sm_split_reader!(5);
304    sm_split_reader!(6);
305    sm_split_reader!(7);
306    sm_split_reader!(8);
307    sm_split_reader!(9);
308    sm_split_reader!(10);
309}
310
311pub trait HasStateMachine {
312    type K: AsTag;
313
314    fn state_machine(&self) -> &StateMachine<Self::K>;
315}
316
317#[async_trait]
318pub trait UseStateMachine: HasStateMachine {
319    /// Add state source into state machine.
320    /// * `tag` - the `Tag` of the source.
321    /// * `capacity` - the capacity of broadcast channel.
322    /// * `pass_checks` - check conditions to continue to recieve state change events.
323    async fn add_source<T>(
324        &self,
325        tag: T,
326        capacity: usize,
327        pass_checks: Option<Vec<Box<dyn AsPassCheck + Send + Sync>>>,
328    ) -> Result<(), AddHandleError<Self::K>>
329    where
330        T: 'static + Clone + Debug + Display + Into<Self::K> + KvAssoc + Send + Sync,
331        T::Value: 'static + AsState + Send + Sync;
332
333    /// Add state reader into state machine.
334    /// * `tag` - the `Tag` of the reader.
335    /// * `reader` - the reader to be added into state machine.
336    async fn add_reader<T>(
337        &self,
338        tag: T,
339        reader: Reader<T::Value>,
340    ) -> Result<(), AddHandleError<Self::K>>
341    where
342        T: 'static + Clone + Debug + Display + Into<Self::K> + KvAssoc + Send + Sync,
343        T::Value: 'static + AsState + Send + Sync;
344
345    /// Delete state handle (source or reader) from state machine.
346    /// * `tag` - the `Tag` of the handle to be deleted.
347    fn del_handle<T>(&self, tag: &T) -> Result<bool, GetHandleError<Self::K>>
348    where
349        T: 'static + Clone + Debug + Into<Self::K> + KvAssoc,
350        T::Value: AsState + Send + Sync;
351
352    /// If there is a handle (source or reader) in state machine for a tag.
353    /// * `tag` - the `Tag` to find handle associated with it.
354    fn has_handle<T>(&self, tag: &T) -> bool
355    where
356        T: Clone + Into<Self::K>;
357
358    /// Get a new reader from state machine, which will receive state change events individually.
359    /// * `tag` - the `Tag` of the handle which you want to get a reader from it.
360    fn reader<T>(&self, tag: T) -> Result<Reader<T::Value>, GetHandleError<Self::K>>
361    where
362        T: Clone + Debug + Into<Self::K> + KvAssoc,
363        T::Value: AsState + Send + Sync;
364
365    /// Get current state value of a tag in state machine.
366    /// * `tag` - the `Tag` of the handle which you want to get state value from it.
367    fn value<T>(&self, tag: T) -> Result<T::Value, GetHandleError<Self::K>>
368    where
369        T: 'static + Clone + Debug + Into<Self::K> + KvAssoc + Send + Sync,
370        T::Value: 'static + AsState + Send + Sync;
371
372    /// Get current state of a tag in state machine.
373    /// * `tag` - the `Tag` of the handle which you want to get state from it.
374    fn state<T>(&self, tag: T) -> Result<State<T::Value>, GetHandleError<Self::K>>
375    where
376        T: 'static + Clone + Debug + Into<Self::K> + KvAssoc + Send + Sync,
377        T::Value: 'static + AsState + Send + Sync;
378
379    /// Trigger a state change event, but doesn't change the state value.
380    /// * `tag` - the `Tag` of the handle which you want to touch.
381    async fn touch<T>(&self, tag: T) -> Result<(), StateChangeError<T, Self::K>>
382    where
383        T: 'static + Clone + Debug + Into<Self::K> + KvAssoc + Send + Sync,
384        T::Value: 'static + AsState + Send + Sync;
385
386    /// Trigger a state change event, but doesn't change the state value, and wait for all readers finishing responding actions.
387    /// * `tag` - the `Tag` of the handle which you want to touch.
388    async fn wait_touch<T>(&self, tag: T) -> Result<(), StateChangeError<T, Self::K>>
389    where
390        T: 'static + Clone + Debug + Into<Self::K> + KvAssoc + Send + Sync,
391        T::Value: 'static + AsState + Send + Sync;
392
393    /// Alter a state in state machine.
394    /// * `tag` - the `Tag` of the handle which you want to alter.
395    async fn alter<T>(&self, tag: T, s: T::Value) -> Result<(), StateChangeError<T, Self::K>>
396    where
397        T: 'static + Clone + Debug + Into<Self::K> + KvAssoc + Send + Sync,
398        T::Value: 'static + AsState + Send + Sync;
399
400    /// Alter a state in state machine, and wait for all readers finishing responding actions.
401    /// * `tag` - the `Tag` of the handle which you want to alter.
402    async fn wait_alter<T>(&self, tag: T, s: T::Value) -> Result<(), StateChangeError<T, Self::K>>
403    where
404        T: 'static + Clone + Debug + Into<Self::K> + KvAssoc + Send + Sync,
405        T::Value: 'static + AsState + Send + Sync;
406
407    /// Amend a state in state machine, with a closure which take the current state value as parameter and return new state value.
408    /// * `tag` - the `Tag` of the handle which you want to amend.
409    async fn amend<T>(
410        &self,
411        tag: T,
412        f: impl FnOnce(T::Value) -> T::Value + Send,
413    ) -> Result<(), StateChangeError<T, Self::K>>
414    where
415        T: 'static + Clone + Debug + Into<Self::K> + KvAssoc + Send + Sync,
416        T::Value: 'static + AsState + Send + Sync;
417
418    /// 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.
419    /// * `tag` - the `Tag` of the handle which you want to amend.
420    async fn wait_amend<T>(
421        &self,
422        tag: T,
423        f: impl FnOnce(T::Value) -> T::Value + Send,
424    ) -> Result<(), StateChangeError<T, Self::K>>
425    where
426        T: 'static + Clone + Debug + Into<Self::K> + KvAssoc + Send + Sync,
427        T::Value: 'static + AsState + Send + Sync;
428
429    watch_decl!(1);
430    watch_decl!(2);
431    watch_decl!(3);
432    watch_decl!(4);
433    watch_decl!(5);
434    watch_decl!(6);
435    watch_decl!(7);
436    watch_decl!(8);
437    watch_decl!(9);
438    watch_decl!(10);
439
440    merge_reader_decl!(2);
441    merge_reader_decl!(3);
442    merge_reader_decl!(4);
443    merge_reader_decl!(5);
444    merge_reader_decl!(6);
445    merge_reader_decl!(7);
446    merge_reader_decl!(8);
447    merge_reader_decl!(9);
448    merge_reader_decl!(10);
449
450    split_reader_decl!(2);
451    split_reader_decl!(3);
452    split_reader_decl!(4);
453    split_reader_decl!(5);
454    split_reader_decl!(6);
455    split_reader_decl!(7);
456    split_reader_decl!(8);
457    split_reader_decl!(9);
458    split_reader_decl!(10);
459}
460
461#[async_trait]
462impl<M> UseStateMachine for M
463where
464    M: 'static + HasStateMachine + Send + Sync,
465{
466    async fn add_source<T>(
467        &self,
468        tag: T,
469        capacity: usize,
470        pass_checks: Option<Vec<Box<dyn AsPassCheck + Send + Sync>>>,
471    ) -> Result<(), AddHandleError<Self::K>>
472    where
473        T: 'static + Clone + Debug + Display + Into<Self::K> + KvAssoc + Send + Sync,
474        T::Value: 'static + AsState + Send + Sync,
475    {
476        self.state_machine()
477            .add_source(tag.clone(), capacity, pass_checks)
478            .await?;
479        Ok(())
480    }
481
482    fn del_handle<T>(&self, tag: &T) -> Result<bool, GetHandleError<Self::K>>
483    where
484        T: 'static + Clone + Debug + Into<Self::K> + KvAssoc,
485        T::Value: AsState + Send + Sync,
486    {
487        self.state_machine().del_handle(tag)
488    }
489
490    fn has_handle<T>(&self, tag: &T) -> bool
491    where
492        T: Clone + Into<Self::K>,
493    {
494        self.state_machine().has_handle(tag)
495    }
496
497    fn reader<T>(&self, tag: T) -> Result<Reader<T::Value>, GetHandleError<Self::K>>
498    where
499        T: Clone + Debug + Into<Self::K> + KvAssoc,
500        T::Value: AsState + Send + Sync,
501    {
502        self.state_machine().reader(tag)
503    }
504
505    async fn add_reader<T>(
506        &self,
507        tag: T,
508        reader: Reader<T::Value>,
509    ) -> Result<(), AddHandleError<Self::K>>
510    where
511        T: 'static + Clone + Debug + Display + Into<Self::K> + KvAssoc + Send + Sync,
512        T::Value: 'static + AsState + Send + Sync,
513    {
514        self.state_machine().add_reader(tag, reader).await?;
515        Ok(())
516    }
517
518    fn value<T>(&self, tag: T) -> Result<T::Value, GetHandleError<Self::K>>
519    where
520        T: 'static + Clone + Debug + Into<Self::K> + KvAssoc + Send + Sync,
521        T::Value: 'static + AsState + Send + Sync,
522    {
523        self.state_machine().value(tag)
524    }
525
526    fn state<T>(&self, tag: T) -> Result<State<T::Value>, GetHandleError<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().state(tag)
532    }
533
534    #[instrument(level = "trace", skip(self))]
535    async fn touch<T>(&self, tag: T) -> 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().touch(tag).await
541    }
542
543    #[instrument(level = "trace", skip(self))]
544    async fn wait_touch<T>(&self, tag: T) -> 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_touch(tag).await
550    }
551
552    #[instrument(level = "trace", skip(self))]
553    async fn alter<T>(&self, tag: T, s: T::Value) -> Result<(), StateChangeError<T, Self::K>>
554    where
555        T: 'static + Clone + Debug + Into<Self::K> + KvAssoc + Send + Sync,
556        T::Value: 'static + AsState + Send + Sync,
557    {
558        self.state_machine().alter(tag, s).await
559    }
560
561    #[instrument(level = "trace", skip(self))]
562    async fn wait_alter<T>(&self, tag: T, s: T::Value) -> Result<(), StateChangeError<T, Self::K>>
563    where
564        T: 'static + Clone + Debug + Into<Self::K> + KvAssoc + Send + Sync,
565        T::Value: 'static + AsState + Send + Sync,
566    {
567        self.state_machine().wait_alter(tag, s).await
568    }
569
570    #[instrument(level = "trace", skip(self, f))]
571    async fn amend<T>(
572        &self,
573        tag: T,
574        f: impl FnOnce(T::Value) -> T::Value + Send,
575    ) -> Result<(), StateChangeError<T, Self::K>>
576    where
577        T: 'static + Clone + Debug + Into<Self::K> + KvAssoc + Send + Sync,
578        T::Value: 'static + AsState + Send + Sync,
579    {
580        self.state_machine().amend(tag, f).await
581    }
582
583    #[instrument(level = "trace", skip(self, f))]
584    async fn wait_amend<T>(
585        &self,
586        tag: T,
587        f: impl FnOnce(T::Value) -> T::Value + Send,
588    ) -> Result<(), StateChangeError<T, Self::K>>
589    where
590        T: 'static + Clone + Debug + Into<Self::K> + KvAssoc + Send + Sync,
591        T::Value: 'static + AsState + Send + Sync,
592    {
593        self.state_machine().wait_amend(tag, f).await
594    }
595
596    watch_impl!(1);
597    watch_impl!(2);
598    watch_impl!(3);
599    watch_impl!(4);
600    watch_impl!(5);
601    watch_impl!(6);
602    watch_impl!(7);
603    watch_impl!(8);
604    watch_impl!(9);
605    watch_impl!(10);
606
607    merge_reader_impl!(2);
608    merge_reader_impl!(3);
609    merge_reader_impl!(4);
610    merge_reader_impl!(5);
611    merge_reader_impl!(6);
612    merge_reader_impl!(7);
613    merge_reader_impl!(8);
614    merge_reader_impl!(9);
615    merge_reader_impl!(10);
616
617    split_reader_impl!(2);
618    split_reader_impl!(3);
619    split_reader_impl!(4);
620    split_reader_impl!(5);
621    split_reader_impl!(6);
622    split_reader_impl!(7);
623    split_reader_impl!(8);
624    split_reader_impl!(9);
625    split_reader_impl!(10);
626}
627
628/// StateChangeError
629#[derive(Debug, Error)]
630pub enum StateChangeError<T, K>
631where
632    T: Debug + Into<K> + KvAssoc,
633    T::Value: Default,
634    K: AsTag,
635{
636    #[error(transparent)]
637    GetHandleError(#[from] GetHandleError<K>),
638    #[error(transparent)]
639    StateChangeError(#[from] HandleStateChangeError<T::Value>),
640}
641
642/// GetHandleError
643#[derive(Debug, Error)]
644pub enum GetHandleError<K>
645where
646    K: AsTag,
647{
648    #[error("State handle for tag [{0:?}] not exist.")]
649    HandleNotExist(K),
650    #[error("Type of state value does not match.")]
651    TypeNotMatch,
652}
653
654/// AddHandleError
655#[derive(Debug, Error)]
656pub enum AddHandleError<K>
657where
658    K: AsTag,
659{
660    #[error("State handle for tag [{0:?}] already exist.")]
661    AlreadyExist(K),
662    #[error("The state channel has been closed.")]
663    ChannelClosed,
664}
state_m/state_machine.rs

state_m/
state_machine.rs
