behavior/
shutdown.rs

1//! Typed graceful-shutdown composition.
2//!
3//! Shutdown is a Bombay policy expressed as an ordinary behavior transition,
4//! not an additional actor-model effect. An interpreter may construct the
5//! shutdown lane, but ingress closure and mailbox ordering remain interpreter
6//! concerns.
7
8use crate::behavior::{Actions, Address, Behavior, BirthMode, SendAlgebra, User, UserEvent};
9use crate::deadlined::{TimeEvent, TimeReached};
10use crate::supervising::{ChildEvent, ChildStopped, WorkerEvent, WorkerStopped};
11use crate::watching::{PeerEvent, PeerStopped};
12use crate::{Exit, Step};
13
14/// A request to finish through one serialized behavior transition.
15#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash)]
16pub struct ShutdownRequested;
17
18/// Construction of the shutdown lane through a composed event type.
19pub trait ShutdownEvent: Sized {
20    fn shutdown_requested(event: ShutdownRequested) -> Option<Self>;
21}
22
23/// The complete protocol of a behavior that supports graceful shutdown.
24#[derive(Debug, Clone, PartialEq, Eq)]
25pub enum ShutdownProtocol<E> {
26    Inner(E),
27    ShutdownRequested(ShutdownRequested),
28}
29
30impl<E> ShutdownEvent for ShutdownProtocol<E> {
31    fn shutdown_requested(event: ShutdownRequested) -> Option<Self> {
32        Some(Self::ShutdownRequested(event))
33    }
34}
35
36impl<E: UserEvent> UserEvent for ShutdownProtocol<E> {
37    type Addr = E::Addr;
38    type Message = E::Message;
39
40    fn user(from: Self::Addr, message: Self::Message) -> Self {
41        Self::Inner(E::user(from, message))
42    }
43
44    fn into_user(self) -> Result<User<Self::Addr, Self::Message>, Self> {
45        match self {
46            Self::Inner(event) => event.into_user().map_err(Self::Inner),
47            shutdown @ Self::ShutdownRequested(_) => Err(shutdown),
48        }
49    }
50}
51
52impl<E: TimeEvent> TimeEvent for ShutdownProtocol<E> {
53    fn time_reached(event: TimeReached) -> Option<Self> {
54        E::time_reached(event).map(Self::Inner)
55    }
56}
57
58impl<E: PeerEvent<A>, A: Address> PeerEvent<A> for ShutdownProtocol<E> {
59    fn peer_stopped(event: PeerStopped<A>) -> Option<Self> {
60        E::peer_stopped(event).map(Self::Inner)
61    }
62}
63
64impl<E: ChildEvent<A>, A: Address> ChildEvent<A> for ShutdownProtocol<E> {
65    fn child_stopped(event: ChildStopped<A>) -> Option<Self> {
66        E::child_stopped(event).map(Self::Inner)
67    }
68}
69
70impl<E: WorkerEvent<A>, A: Address> WorkerEvent<A> for ShutdownProtocol<E> {
71    fn worker_stopped(event: WorkerStopped<A>) -> Option<Self> {
72        E::worker_stopped(event).map(Self::Inner)
73    }
74}
75
76/// Stop normally when the shutdown lane is received.
77pub struct StopOnShutdown<B> {
78    inner: B,
79}
80
81impl<B> StopOnShutdown<B> {
82    #[must_use]
83    pub fn new(inner: B) -> Self {
84        Self { inner }
85    }
86
87    #[must_use]
88    pub fn inner(&self) -> &B {
89        &self.inner
90    }
91}
92
93/// A final shutdown fold. Its sends and fresh creations are retained, while
94/// its become verdict is replaced with `Stop(Normal)`.
95pub type ShutdownReaction<B> = fn(
96    &mut B,
97    ShutdownRequested,
98) -> Result<
99    Actions<
100        <B as Behavior>::Addr,
101        <B as Behavior>::Ph,
102        <B as Behavior>::Sends,
103        <B as Behavior>::Birth,
104    >,
105    <B as Behavior>::Error,
106>;
107
108/// Run one explicit final fold and then stop normally.
109pub struct FinalizeOnShutdown<B: Behavior> {
110    inner: B,
111    finalize: ShutdownReaction<B>,
112}
113
114impl<B: Behavior> FinalizeOnShutdown<B> {
115    #[must_use]
116    pub fn new(inner: B, finalize: ShutdownReaction<B>) -> Self {
117        Self { inner, finalize }
118    }
119
120    #[must_use]
121    pub fn inner(&self) -> &B {
122        &self.inner
123    }
124}
125
126macro_rules! impl_shutdown_behavior {
127    ($wrapper:ident, $shutdown:expr) => {
128        impl<B, A, Ph, Sends, Br> Behavior for $wrapper<B>
129        where
130            A: Address + Send,
131            Sends: SendAlgebra,
132            Br: BirthMode,
133            B: Behavior<
134                    Addr = A,
135                    Ph = Ph,
136                    Sends = Sends,
137                    Birth = Br,
138                    Effect = Actions<A, Ph, Sends, Br>,
139                    Done = Exit<A>,
140                > + Send,
141            B::Event: Send,
142            B::Msg: Send,
143        {
144            type Addr = A;
145            type Msg = B::Msg;
146            type Event = ShutdownProtocol<B::Event>;
147            type Sends = Sends;
148            type Ph = Ph;
149            type Error = B::Error;
150            type Birth = Br;
151            type Effect = Actions<A, Ph, Sends, Br>;
152            type Done = Exit<A>;
153
154            async fn init(&mut self) -> Result<Self::Effect, B::Error> {
155                self.inner.init().await
156            }
157
158            async fn step(&mut self, event: Self::Event) -> Result<Self::Effect, B::Error> {
159                match event {
160                    ShutdownProtocol::Inner(event) => self.inner.step(event).await,
161                    ShutdownProtocol::ShutdownRequested(request) => $shutdown(self, request),
162                }
163            }
164        }
165    };
166}
167
168impl_shutdown_behavior!(StopOnShutdown, |_this: &mut StopOnShutdown<B>, _request| {
169    Ok(Actions::stop(Exit::Normal))
170});
171
172impl_shutdown_behavior!(
173    FinalizeOnShutdown,
174    |this: &mut FinalizeOnShutdown<B>, request| {
175        let actions = (this.finalize)(&mut this.inner, request)?;
176        Ok(Actions {
177            sends: actions.sends,
178            creates: actions.creates,
179            become_: Step::Stop(Exit::Normal),
180        })
181    }
182);
behavior/shutdown.rs

behavior/
shutdown.rs
