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::protocol::{
10    ChildEvent, ChildStopped, PeerEvent, PeerStopped, ShutdownEvent, ShutdownRequested, TimeEvent,
11    TimeReached, WorkerEvent, WorkerStopped,
12};
13use crate::{Exit, Step};
14
15/// The complete protocol of a behavior that supports graceful shutdown.
16#[derive(Debug, Clone, PartialEq, Eq)]
17pub enum ShutdownProtocol<E> {
18    Inner(E),
19    ShutdownRequested(ShutdownRequested),
20}
21
22impl<E> ShutdownEvent for ShutdownProtocol<E> {
23    fn shutdown_requested(event: ShutdownRequested) -> Option<Self> {
24        Some(Self::ShutdownRequested(event))
25    }
26}
27
28impl<E: UserEvent> UserEvent for ShutdownProtocol<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            shutdown @ Self::ShutdownRequested(_) => Err(shutdown),
40        }
41    }
42}
43
44impl<E: TimeEvent> TimeEvent for ShutdownProtocol<E> {
45    fn time_reached(event: TimeReached) -> Option<Self> {
46        E::time_reached(event).map(Self::Inner)
47    }
48}
49
50impl<E: PeerEvent<A>, A: Address> PeerEvent<A> for ShutdownProtocol<E> {
51    fn peer_stopped(event: PeerStopped<A>) -> Option<Self> {
52        E::peer_stopped(event).map(Self::Inner)
53    }
54}
55
56impl<E: ChildEvent<A>, A: Address> ChildEvent<A> for ShutdownProtocol<E> {
57    fn child_stopped(event: ChildStopped<A>) -> Option<Self> {
58        E::child_stopped(event).map(Self::Inner)
59    }
60}
61
62impl<E: WorkerEvent<A>, A: Address> WorkerEvent<A> for ShutdownProtocol<E> {
63    fn worker_stopped(event: WorkerStopped<A>) -> Option<Self> {
64        E::worker_stopped(event).map(Self::Inner)
65    }
66}
67
68/// Stop normally when the shutdown lane is received.
69pub struct StopOnShutdown<B> {
70    inner: B,
71}
72
73impl<B> StopOnShutdown<B> {
74    #[must_use]
75    pub fn new(inner: B) -> Self {
76        Self { inner }
77    }
78
79    #[must_use]
80    pub fn inner(&self) -> &B {
81        &self.inner
82    }
83}
84
85/// A final shutdown fold. Its sends and fresh creations are retained, while
86/// its become verdict is replaced with `Stop(Normal)`.
87pub type ShutdownReaction<B> = fn(
88    &mut B,
89    ShutdownRequested,
90) -> Result<
91    Actions<
92        <B as Behavior>::Addr,
93        <B as Behavior>::Ph,
94        <B as Behavior>::Sends,
95        <B as Behavior>::Birth,
96    >,
97    <B as Behavior>::Error,
98>;
99
100/// Run one explicit final fold and then stop normally.
101pub struct FinalizeOnShutdown<B: Behavior> {
102    inner: B,
103    finalize: ShutdownReaction<B>,
104}
105
106impl<B: Behavior> FinalizeOnShutdown<B> {
107    #[must_use]
108    pub fn new(inner: B, finalize: ShutdownReaction<B>) -> Self {
109        Self { inner, finalize }
110    }
111
112    #[must_use]
113    pub fn inner(&self) -> &B {
114        &self.inner
115    }
116}
117
118macro_rules! impl_shutdown_behavior {
119    ($wrapper:ident, $shutdown:expr) => {
120        impl<B, A, Ph, Sends, Br> Behavior for $wrapper<B>
121        where
122            A: Address + Send,
123            Sends: SendAlgebra,
124            Br: BirthMode,
125            B: Behavior<Addr = A, Ph = Ph, Sends = Sends, Birth = Br> + Send,
126            B::Event: Send,
127            B::Msg: Send,
128        {
129            type Addr = A;
130            type Msg = B::Msg;
131            type Event = ShutdownProtocol<B::Event>;
132            type Sends = Sends;
133            type Ph = Ph;
134            type Error = B::Error;
135            type Birth = Br;
136
137            async fn init(&mut self) -> Result<Actions<A, Ph, Sends, Br>, B::Error> {
138                self.inner.init().await
139            }
140
141            async fn step(
142                &mut self,
143                event: Self::Event,
144            ) -> Result<Actions<A, Ph, Sends, Br>, B::Error> {
145                match event {
146                    ShutdownProtocol::Inner(event) => self.inner.step(event).await,
147                    ShutdownProtocol::ShutdownRequested(request) => $shutdown(self, request),
148                }
149            }
150        }
151    };
152}
153
154impl_shutdown_behavior!(StopOnShutdown, |_this: &mut StopOnShutdown<B>, _request| {
155    Ok(Actions::stop(Exit::Normal))
156});
157
158impl_shutdown_behavior!(
159    FinalizeOnShutdown,
160    |this: &mut FinalizeOnShutdown<B>, request| {
161        let actions = (this.finalize)(&mut this.inner, request)?;
162        Ok(Actions {
163            sends: actions.sends,
164            creates: actions.creates,
165            become_: Step::Stop(Exit::Normal),
166        })
167    }
168);
behavior/shutdown.rs

behavior/
shutdown.rs
