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 {
53            disposition: Disposition::Forward,
54            effects: Vec::new(),
55        }
56    }
57
58    /// Drop the input, emit nothing. Allocates nothing.
59    pub fn drop() -> Self {
60        Self {
61            disposition: Disposition::Drop,
62            effects: Vec::new(),
63        }
64    }
65
66    /// Append an effect, leaving the disposition unchanged. Chainable.
67    pub fn emit(mut self, effect: E) -> Self {
68        self.effects.push(effect);
69        self
70    }
71}
72
73impl<E> Default for Decision<E> {
74    fn default() -> Self {
75        Self::forward()
76    }
77}
78
79/// The pure core of a stage.
80///
81/// Both methods are synchronous and total — no `.await`, no I/O — so they
82/// cannot be cancelled mid-step. All state mutation happens here; `perform`
83/// (in the runtime) is `&self` so a dropped future can't tear state.
84///
85/// `Effect` is the stage's own command vocabulary: plain data defined in core.
86/// The methods emit commands describing *what should happen*; the runtime's
87/// `perform` interprets them and does the I/O.
88///
89/// # Control calls
90///
91/// "No I/O" has one carve-out. `decide_*` may issue *control calls*:
92/// synchronous, non-blocking, idempotent, infallible operations on owned
93/// engines. The canonical example is `cancel()`, which flips an atomic flag an
94/// engine's worker observes; it cannot block, fail, allocate unboundedly, or
95/// tear state, so invoking it from the synchronous, `&mut self` decide step is
96/// sound. Anything that can block, fail, allocate unboundedly, or perform real
97/// I/O is *not* a control call — it remains an [`Effect`](Processor::Effect)
98/// for `perform` to carry out.
99///
100/// # Defaults
101///
102/// Both methods default to [`Decision::forward()`]: an un-overridden stage is a
103/// transparent pass-through. Override only the variants you care about.
104pub trait Processor {
105    /// The command vocabulary this stage can emit; plain data, no I/O.
106    type Effect;
107
108    /// Maps an incoming data frame to a disposition and zero or more effects.
109    ///
110    /// Called for every [`DataFrame`] arriving on the data lane. The default
111    /// forwards the frame unchanged.
112    fn decide_data(&mut self, _frame: &DataFrame) -> Decision<Self::Effect> {
113        Decision::forward()
114    }
115
116    /// Maps an incoming system frame to a disposition and zero or more effects.
117    ///
118    /// Called for every [`SystemFrame`] arriving on the system lane, along with
119    /// the [`Direction`] it is travelling. The default forwards the frame.
120    fn decide_system(&mut self, _dir: Direction, _frame: &SystemFrame) -> Decision<Self::Effect> {
121        Decision::forward()
122    }
123}
pipecrab_core/processor.rs

pipecrab_core/
processor.rs
