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

behavior/
deadlined.rs
