behavior/
deadlined.rs

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

behavior/
deadlined.rs
