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

behavior/
deadlined.rs
