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

behavior/
compose.rs
