behavior/
deadlined.rs

1//! Pure one-shot time composition. Scheduling is an ordinary send to the
2//! interpreter-provided clock actor.
3
4use tokio::time::Instant;
5
6use crate::behavior::{
7    Actions, Address, Become, Behavior, BirthMode, SendAlgebra, SendProduct, User, UserEvent,
8};
9use crate::supervising::{ChildEvent, ChildStopped};
10use crate::watching::{PeerEvent, PeerStopped};
11use crate::{Exit, Step};
12
13#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
14pub struct AtId(pub u64);
15
16#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
17pub struct AtGeneration(pub u64);
18
19#[derive(Debug, Clone, Copy, PartialEq, Eq)]
20pub struct ScheduleAt {
21    pub id: AtId,
22    pub generation: AtGeneration,
23    pub at: Instant,
24}
25
26#[derive(Debug, Clone, Copy, PartialEq, Eq)]
27pub struct TimeReached {
28    pub id: AtId,
29    pub generation: AtGeneration,
30    pub at: Instant,
31}
32
33#[derive(Debug, Clone, PartialEq, Eq)]
34pub enum AtEvent<E> {
35    Inner(E),
36    Reached(TimeReached),
37}
38
39pub trait TimeEvent: Sized {
40    fn time_reached(event: TimeReached) -> Option<Self>;
41}
42
43impl<E> TimeEvent for AtEvent<E> {
44    fn time_reached(event: TimeReached) -> Option<Self> {
45        Some(Self::Reached(event))
46    }
47}
48
49impl<E: UserEvent> UserEvent for AtEvent<E> {
50    type Addr = E::Addr;
51    type Message = E::Message;
52
53    fn user(from: Self::Addr, message: Self::Message) -> Self {
54        Self::Inner(E::user(from, message))
55    }
56
57    fn into_user(self) -> Result<User<Self::Addr, Self::Message>, Self> {
58        match self {
59            Self::Inner(event) => event.into_user().map_err(Self::Inner),
60            reached @ Self::Reached(_) => Err(reached),
61        }
62    }
63}
64
65impl<E: PeerEvent<A>, A: Address> PeerEvent<A> for AtEvent<E> {
66    fn peer_stopped(event: PeerStopped<A>) -> Option<Self> {
67        E::peer_stopped(event).map(Self::Inner)
68    }
69}
70
71impl<E: ChildEvent<A>, A: Address> ChildEvent<A> for AtEvent<E> {
72    fn child_stopped(event: ChildStopped<A>) -> Option<Self> {
73        E::child_stopped(event).map(Self::Inner)
74    }
75}
76
77pub type AtReaction<B> =
78    fn(&mut B) -> Result<Become<<B as Behavior>::Addr>, <B as Behavior>::Error>;
79
80pub type AtSends<B> = SendProduct<<B as Behavior>::Sends, Vec<ScheduleAt>>;
81
82pub type AtActions<B> =
83    Actions<<B as Behavior>::Addr, <B as Behavior>::Ph, AtSends<B>, <B as Behavior>::Birth>;
84
85pub struct At<B: Behavior> {
86    inner: B,
87    id: AtId,
88    scheduled: Option<(AtGeneration, Instant)>,
89    on_reached: AtReaction<B>,
90}
91
92impl<B: Behavior> At<B> {
93    #[must_use]
94    pub fn new(inner: B, id: AtId, at: Option<Instant>, on_reached: AtReaction<B>) -> Self {
95        Self {
96            inner,
97            id,
98            scheduled: at.map(|at| (AtGeneration(0), at)),
99            on_reached,
100        }
101    }
102
103    #[must_use]
104    pub fn inner(&self) -> &B {
105        &self.inner
106    }
107}
108
109impl<B, A, Ph, Sends, Br> Behavior for At<B>
110where
111    A: Address + Send,
112    Sends: SendAlgebra,
113    Br: BirthMode,
114    B: Behavior<
115            Addr = A,
116            Ph = Ph,
117            Sends = Sends,
118            Birth = Br,
119            Effect = Actions<A, Ph, Sends, Br>,
120            Done = Exit<A>,
121        > + Send,
122    B::Event: TimeEvent + Send,
123    B::Msg: Send,
124{
125    type Addr = A;
126    type Msg = B::Msg;
127    type Event = AtEvent<B::Event>;
128    type Sends = SendProduct<Sends, Vec<ScheduleAt>>;
129    type Ph = Ph;
130    type Error = B::Error;
131    type Birth = Br;
132    type Effect = Actions<A, Ph, Self::Sends, Br>;
133    type Done = Exit<A>;
134
135    async fn init(&mut self) -> Result<Self::Effect, B::Error> {
136        let actions = self.inner.init().await?;
137        let own = self.scheduled.map_or_else(Vec::new, |(generation, at)| {
138            vec![ScheduleAt {
139                id: self.id,
140                generation,
141                at,
142            }]
143        });
144        Ok(Self::wrap(actions, own))
145    }
146
147    async fn step(&mut self, event: Self::Event) -> Result<Self::Effect, B::Error> {
148        match event {
149            AtEvent::Reached(event)
150                if event.id == self.id && self.scheduled == Some((event.generation, event.at)) =>
151            {
152                self.scheduled = None;
153                let become_ = match (self.on_reached)(&mut self.inner)? {
154                    Step::Continue => Step::Continue,
155                    Step::Goto(never) => match never {},
156                    Step::Stop(exit) => Step::Stop(exit),
157                };
158                Ok(Actions {
159                    sends: SendProduct {
160                        inner: B::Sends::empty(),
161                        own: Vec::new(),
162                    },
163                    creates: Vec::new(),
164                    become_,
165                })
166            }
167            AtEvent::Reached(event) => match B::Event::time_reached(event) {
168                Some(inner) => self
169                    .inner
170                    .step(inner)
171                    .await
172                    .map(|actions| Self::wrap(actions, Vec::new())),
173                None => Ok(Actions::cont()),
174            },
175            AtEvent::Inner(event) => self
176                .inner
177                .step(event)
178                .await
179                .map(|actions| Self::wrap(actions, Vec::new())),
180        }
181    }
182}
183
184impl<B: Behavior> At<B> {
185    fn wrap(
186        actions: Actions<B::Addr, B::Ph, B::Sends, B::Birth>,
187        own: Vec<ScheduleAt>,
188    ) -> AtActions<B> {
189        Actions {
190            sends: SendProduct {
191                inner: actions.sends,
192                own,
193            },
194            creates: actions.creates,
195            become_: actions.become_,
196        }
197    }
198}
behavior/deadlined.rs

behavior/
deadlined.rs
