pipecrab_core/
processor.rs

1use crate::frame::{DataFrame, Direction, SystemFrame};
2
3/// What happens to the frame that [`Processor::decide_data`] or
4/// [`Processor::decide_system`] just received.
5#[derive(Debug, Clone, Copy, PartialEq, Eq)]
6pub enum Disposition {
7    /// Pass the input onward — data downstream, system in its travel direction.
8    Forward,
9    /// Consume the input; it does not propagate.
10    Drop,
11}
12
13/// Outcome of [`Processor::decide_data`] / [`Processor::decide_system`]:
14/// a disposition for the input frame plus zero or more effects to emit.
15///
16/// # Four forms
17///
18/// ```
19/// use pipecrab_core::{Decision, Disposition};
20///
21/// // 1. Pass-through: forward the input, emit nothing.
22/// let d: Decision<u32> = Decision::forward();
23/// assert_eq!(d.disposition, Disposition::Forward);
24/// assert!(d.effects.is_empty());
25///
26/// // 2. Transform (drop-on-emit): consume the input, emit a replacement.
27/// let d: Decision<u32> = Decision::drop().emit(42);
28/// assert_eq!(d.disposition, Disposition::Drop);
29/// assert_eq!(d.effects, [42]);
30///
31/// // 3. Observe-and-pass: forward the input *and* emit a derived value.
32/// let d: Decision<u32> = Decision::forward().emit(42);
33/// assert_eq!(d.disposition, Disposition::Forward);
34/// assert_eq!(d.effects, [42]);
35///
36/// // 4. Silent drop: consume the input, emit nothing.
37/// let d: Decision<u32> = Decision::drop();
38/// assert_eq!(d.disposition, Disposition::Drop);
39/// assert!(d.effects.is_empty());
40/// ```
41#[derive(Debug, Clone, PartialEq, Eq)]
42pub struct Decision<E> {
43    /// What happens to the input frame.
44    pub disposition: Disposition,
45    /// Effects to perform, in order, after the disposition is acted on.
46    pub effects: Vec<E>,
47}
48
49impl<E> Decision<E> {
50    /// Forward the input, emit nothing. Allocates nothing.
51    pub fn forward() -> Self {
52        Self { disposition: Disposition::Forward, effects: Vec::new() }
53    }
54
55    /// Drop the input, emit nothing. Allocates nothing.
56    pub fn drop() -> Self {
57        Self { disposition: Disposition::Drop, effects: Vec::new() }
58    }
59
60    /// Append an effect, leaving the disposition unchanged. Chainable.
61    pub fn emit(mut self, effect: E) -> Self {
62        self.effects.push(effect);
63        self
64    }
65}
66
67impl<E> Default for Decision<E> {
68    fn default() -> Self {
69        Self::forward()
70    }
71}
72
73/// The pure core of a stage.
74///
75/// Both methods are synchronous and total — no `.await`, no I/O — so they
76/// cannot be cancelled mid-step. All state mutation happens here; `perform`
77/// (in the runtime) is `&self` so a dropped future can't tear state.
78///
79/// `Effect` is the stage's own command vocabulary: plain data defined in core.
80/// The methods emit commands describing *what should happen*; the runtime's
81/// `perform` interprets them and does the I/O.
82///
83/// # Control calls
84///
85/// "No I/O" has one carve-out. `decide_*` may issue *control calls*:
86/// synchronous, non-blocking, idempotent, infallible operations on owned
87/// engines. The canonical example is `cancel()`, which flips an atomic flag an
88/// engine's worker observes; it cannot block, fail, allocate unboundedly, or
89/// tear state, so invoking it from the synchronous, `&mut self` decide step is
90/// sound. Anything that can block, fail, allocate unboundedly, or perform real
91/// I/O is *not* a control call — it remains an [`Effect`](Processor::Effect)
92/// for `perform` to carry out.
93///
94/// # Defaults
95///
96/// Both methods default to [`Decision::forward()`]: an un-overridden stage is a
97/// transparent pass-through. Override only the variants you care about.
98pub trait Processor {
99    /// The command vocabulary this stage can emit; plain data, no I/O.
100    type Effect;
101
102    /// Maps an incoming data frame to a disposition and zero or more effects.
103    ///
104    /// Called for every [`DataFrame`] arriving on the data lane. The default
105    /// forwards the frame unchanged.
106    fn decide_data(&mut self, _frame: &DataFrame) -> Decision<Self::Effect> {
107        Decision::forward()
108    }
109
110    /// Maps an incoming system frame to a disposition and zero or more effects.
111    ///
112    /// Called for every [`SystemFrame`] arriving on the system lane, along with
113    /// the [`Direction`] it is travelling. The default forwards the frame.
114    fn decide_system(&mut self, _dir: Direction, _frame: &SystemFrame) -> Decision<Self::Effect> {
115        Decision::forward()
116    }
117}
pipecrab_core/processor.rs

pipecrab_core/
processor.rs
