behavior/
compose.rs

1//! Intent-facing typestate composition. Every method immediately builds a
2//! concrete pure behavior; there is no separate intent representation.
3
4use std::time::Duration;
5
6use tokio::time::Instant;
7
8use crate::behavior::{Address, Behavior, BirthMode, Births};
9use crate::next::Never;
10use crate::protocol::TimerId;
11use crate::shutdown::{FinalizeOnShutdown, ShutdownReaction, StopOnShutdown};
12use crate::stash::{Stash, StashRoute};
13use crate::supervision::{RestartPolicy, Strategy, SupervisionFailureReaction, Supervisor};
14use crate::timing::{Deadline, DeadlineReaction};
15use crate::timing::{ReceiveTimeout, ReceiveTimeoutReaction};
16use crate::watch::{LinkReaction, Watch};
17use crate::{Actions, BehaviorFn, Handler, Machine, Move, Pure, SendAlgebra, delegate_transition};
18
19const DEFAULT_STRATEGY: Strategy = Strategy::OneForOne;
20const DEFAULT_POLICY: RestartPolicy = RestartPolicy::Transient;
21const DEFAULT_BUDGET: (u32, Duration) = (1, Duration::from_secs(5));
22
23fn identity_nonce<N: From<u64>>(index: usize) -> N {
24    N::from(u64::try_from(index).expect("fleet index fits u64"))
25}
26
27pub struct Compose<B> {
28    behavior: B,
29    next_timer: u64,
30}
31
32impl<S: Handler<O, Br, E>, O, Br: BirthMode, E> Compose<Pure<S, O, Br, E>> {
33    #[must_use]
34    pub fn new(state: S) -> Self {
35        Self {
36            behavior: Pure::new(state),
37            next_timer: 0,
38        }
39    }
40}
41
42impl<A, S, M, P, E> Compose<Machine<A, S, M, P, E>>
43where
44    A: Address,
45    P: Copy + PartialEq,
46{
47    #[must_use]
48    pub fn machine(state: S, phase: P, on: fn(P, &mut S, &M) -> Result<Move<P>, E>) -> Self {
49        Self {
50            behavior: Machine::new(state, phase, on),
51            next_timer: 0,
52        }
53    }
54}
55
56impl<S, A, M, Sends, Br, E, I, F> Compose<BehaviorFn<S, A, M, Sends, Br, E, I, F>>
57where
58    A: Address,
59    Sends: SendAlgebra,
60    Br: BirthMode,
61    I: FnMut(&mut S) -> crate::Acted<A, Never, Sends, Br, E>,
62    F: FnMut(&mut S, A, M) -> crate::Acted<A, Never, Sends, Br, E>,
63{
64    /// Define a concrete behavior from explicit initialization and user-event
65    /// folds, ready for typed wrapper composition.
66    #[must_use]
67    pub fn from_fns(state: S, initialize: I, transition: F) -> Self {
68        Self {
69            behavior: BehaviorFn::new(state, initialize, transition),
70            next_timer: 0,
71        }
72    }
73}
74
75impl<B: Behavior> Compose<B> {
76    fn map_behavior<Mapped>(self, map: impl FnOnce(B) -> Mapped) -> Compose<Mapped> {
77        Compose {
78            behavior: map(self.behavior),
79            next_timer: self.next_timer,
80        }
81    }
82
83    fn map_timer<Mapped>(self, map: impl FnOnce(B, TimerId) -> Mapped) -> Compose<Mapped> {
84        let timer = TimerId(self.next_timer);
85        Compose {
86            behavior: map(self.behavior, timer),
87            next_timer: self
88                .next_timer
89                .checked_add(1)
90                .expect("timer identity exhausted"),
91        }
92    }
93
94    #[must_use]
95    pub fn from_behavior(behavior: B) -> Self {
96        Self {
97            behavior,
98            next_timer: 0,
99        }
100    }
101
102    #[must_use]
103    pub fn build(self) -> B {
104        self.behavior
105    }
106
107    #[must_use]
108    pub fn behavior(&self) -> &B {
109        &self.behavior
110    }
111
112    /// Stop normally when a typed shutdown request is folded.
113    #[must_use]
114    pub fn stop_on_shutdown(self) -> Compose<StopOnShutdown<B>> {
115        self.map_behavior(StopOnShutdown::new)
116    }
117
118    /// Apply one final pure fold, retain its sends and creations, and stop
119    /// normally regardless of the fold's become verdict.
120    #[must_use]
121    pub fn finalize_on_shutdown(
122        self,
123        finalize: ShutdownReaction<B>,
124    ) -> Compose<FinalizeOnShutdown<B>> {
125        self.map_behavior(|behavior| FinalizeOnShutdown::new(behavior, finalize))
126    }
127
128    /// Observe a peer and apply a pure reaction when it stops.
129    #[must_use]
130    pub fn watch(self, peer: B::Addr, on_stopped: LinkReaction<B>) -> Compose<Watch<B>> {
131        self.map_behavior(|behavior| Watch::new(behavior, peer, on_stopped))
132    }
133
134    /// Apply a pure reaction when the given absolute time is reached.
135    ///
136    /// # Panics
137    ///
138    /// Panics if one specification composes more than `u64::MAX` timer
139    /// capabilities.
140    #[must_use]
141    pub fn deadline(
142        self,
143        when: Option<Instant>,
144        on_reached: DeadlineReaction<B>,
145    ) -> Compose<Deadline<B>> {
146        self.map_timer(|behavior, timer| Deadline::new(behavior, timer, when, on_reached))
147    }
148
149    /// Notify the behavior once after an idle period containing no successful
150    /// user communication.
151    ///
152    /// Initialization and each successful continuing user fold emit a relative
153    /// schedule. Service events never reset inactivity. A matching delivery is
154    /// consumed before `on_elapsed` runs, and a continuing reaction remains
155    /// unarmed until another successful continuing user communication.
156    ///
157    /// # Panics
158    ///
159    /// Panics if one specification composes more than `u64::MAX` timer
160    /// capabilities.
161    #[must_use]
162    pub fn receive_timeout(
163        self,
164        after: Duration,
165        on_elapsed: ReceiveTimeoutReaction<B>,
166    ) -> Compose<ReceiveTimeout<B>> {
167        self.map_timer(|behavior, timer| ReceiveTimeout::new(behavior, timer, after, on_elapsed))
168    }
169
170    /// Hold messages selected by `route` and replay them on `Release`.
171    #[must_use]
172    pub fn stash(self, route: fn(&B::Msg) -> StashRoute) -> Compose<Stash<B>>
173    where
174        B: Behavior<Ph = Never>,
175    {
176        self.map_behavior(|behavior| Stash::new(behavior, route))
177    }
178
179    /// Create a supervised child topology. Concrete proxy and monitor types
180    /// remain hidden in the returned typestate.
181    #[must_use]
182    pub fn children<C>(self, fleet: (usize, fn(usize) -> C)) -> Compose<Supervisor<B, C>>
183    where
184        B: Behavior<Birth = Births<C>>,
185        C: Behavior<Ph = Never, Addr = B::Addr>,
186        <B::Addr as Address>::Nonce: From<u64>,
187    {
188        self.children_with_nonces(identity_nonce, fleet.0, fleet.1)
189    }
190
191    #[must_use]
192    pub fn children_with_nonces<C>(
193        self,
194        nonces: fn(usize) -> <B::Addr as Address>::Nonce,
195        count: usize,
196        build: fn(usize) -> C,
197    ) -> Compose<Supervisor<B, C>>
198    where
199        B: Behavior<Birth = Births<C>>,
200        C: Behavior<Ph = Never, Addr = B::Addr>,
201    {
202        self.map_behavior(|behavior| {
203            Supervisor::new(
204                behavior,
205                nonces,
206                count,
207                build,
208                DEFAULT_STRATEGY,
209                DEFAULT_POLICY,
210                DEFAULT_BUDGET.0,
211                DEFAULT_BUDGET.1,
212            )
213        })
214    }
215}
216
217impl<B, C> Compose<Supervisor<B, C>>
218where
219    B: Behavior<Birth = Births<C>>,
220    C: Behavior<Ph = Never, Addr = B::Addr>,
221{
222    #[must_use]
223    pub fn restart(self, strategy: Strategy) -> Self {
224        Self {
225            behavior: self.behavior.with_strategy(strategy),
226            next_timer: self.next_timer,
227        }
228    }
229
230    #[must_use]
231    pub fn when(self, policy: RestartPolicy) -> Self {
232        Self {
233            behavior: self.behavior.with_policy(policy),
234            next_timer: self.next_timer,
235        }
236    }
237
238    #[must_use]
239    pub fn within(self, maximum: u32, window: Duration) -> Self {
240        Self {
241            behavior: self.behavior.with_budget(maximum, window),
242            next_timer: self.next_timer,
243        }
244    }
245
246    /// Apply a pure reaction when supervision can no longer preserve its
247    /// child topology.
248    #[must_use]
249    pub fn on_supervision_failure(self, reaction: SupervisionFailureReaction<B>) -> Self {
250        Self {
251            behavior: self.behavior.with_failure_reaction(reaction),
252            next_timer: self.next_timer,
253        }
254    }
255}
256
257impl<B, A, Ph, Sends, Br> Behavior for Compose<B>
258where
259    A: Address,
260    Sends: SendAlgebra,
261    Br: BirthMode,
262    B: Behavior<Addr = A, Ph = Ph, Sends = Sends, Birth = Br>,
263{
264    type Addr = A;
265    type Msg = B::Msg;
266    type Event = B::Event;
267    type Sends = Sends;
268    type Ph = Ph;
269    type Error = B::Error;
270    type Birth = Br;
271
272    fn init(&mut self) -> Result<Actions<A, Ph, Sends, Br>, B::Error> {
273        self.behavior.init()
274    }
275
276    fn transition(&mut self, event: B::Event) -> Result<Actions<A, Ph, Sends, Br>, B::Error> {
277        delegate_transition(&mut self.behavior, event)
278    }
279}
behavior/compose.rs

behavior/
compose.rs
