Accumulator
An Accumulator sits between an external data source and a
Reactor. It receives raw
events, optionally aggregates or filters them, and emits typed
Boundary events under a named
source. Its core operation is process(event) -> Option<output> — returning
nothing drops the event (filtering/dedup).
| Variant | What it does |
|---|---|
| passthrough | No buffering; converts each event to an output (or drops it). |
| stream | Consumes from a stream backend (e.g. Kafka topic/group). |
| polling | Pulls from a source on a timer (interval). |
| batch | Buffers events, flushes on flush_interval / max_buffer_size. |
| state | A bounded, DAL-persisted history buffer (capacity). Rust-only. |
#[passthrough_accumulator]
#[stream_accumulator(type = "...", topic = "...", group = "...")]
#[polling_accumulator(interval = "...")]
#[batch_accumulator(flush_interval = "...", max_buffer_size = 100)]
#[state_accumulator(capacity = 64)] // Rust-only
@cloaca.passthrough_accumulator
@cloaca.stream_accumulator(type="...", topic="...", group="...")
@cloaca.polling_accumulator(interval="5s")
@cloaca.batch_accumulator(flush_interval="1s", max_buffer_size=100)
Parity gapPython exposes four accumulator types. The Rust-only#[state_accumulator](a bounded DAL-persisted history buffer) has no Python decorator yet — tracked in [CLOACI-T-0688]. Until it lands, author state accumulators in Rust.
process()returns anOption—Nonesilently drops the event.- Named source: an accumulator emits under a
SourceNamethat must match the reactor’saccumulatorslist and the graph’s entry-node source name. - Choosing one: see Choosing Accumulator Types.