Reactor
A Reactor is the long-lived process that binds a set of named Accumulators to a Computation Graph, maintains an input cache from the Boundary events they emit, and fires the graph when its criteria are satisfied. The firing criterion is a top-level primitive: a graph subscribes to a reactor by name.
- A reactor declares which accumulators feed it and a criterion:
when_any(fire when any source has new data) orwhen_all(wait for all). - It keeps an input cache and an input strategy —
latest(overwrite with newest) orsequential. - When the criterion is met, it calls the compiled graph with the current cache.
#[reactor] declares the reactor as a top-level primitive on a struct:
#[cloacina_macros::reactor(
name = "pricing_reactor",
accumulators = [orderbook],
criteria = when_any(orderbook),
)]
pub struct PricingReactor;
(The lower-level runtime exposes Reactor::new(graph_fn, ReactionCriteria::WhenAny, InputStrategy::Latest, …) — see the embedded tutorials.)
@cloaca.reactor decorates a class; criteria is the mode argument:
import cloaca
@cloaca.reactor(
name="pricing_reactor",
accumulators=["orderbook"],
mode="when_any", # or "when_all"
)
class PricingReactor:
pass
- A reactor is a specialized trigger (CLOACI-S-0011): where a workflow trigger polls or follows a cron schedule, a reactor consumes accumulator boundary events and fires a computation graph.
- Criteria:
when_any/when_allover the named accumulator sources. - Input strategy:
latestorsequential(RustInputStrategy). There is noinput_strategyclause on the#[reactor]macro — set it programmatically or in the package manifest; see Sequential input strategy. - Naming: the
accumulatorsnames must match the accumulator source names and the graph’s entry-node source names. - Operator controls: manual fire is REST
(
POST /v1/health/reactors/{name}/fire,cloacinactl reactor fire|force-fire); pause/resume is WebSocket-only (/v1/ws/reactor/{name}) — there is no REST pause route, and reactor→workflow subscriptions have no HTTP API at all (runner API only).