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).

Variants

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).

Interfaces

Rust

#[passthrough_accumulator]
#[stream_accumulator(type = "...", topic = "...", group = "...")]
#[polling_accumulator(interval = "...")]
#[batch_accumulator(flush_interval = "...", max_buffer_size = 100)]
#[state_accumulator(capacity = 64)]

Python

@cloaca.passthrough_accumulator
@cloaca.stream_accumulator(type="...", topic="...", group="...")
@cloaca.polling_accumulator(interval="5s")
@cloaca.batch_accumulator(flush_interval="1s", max_buffer_size=100)
@cloaca.state_accumulator(capacity=64)

All five accumulator types have Python decorator parity.

Key facts

• process() returns an Option — None silently drops the event.
• Named source: an accumulator emits under a SourceName that must match the reactor’s accumulators list and the graph’s entry-node source name.
• Choosing one: see Choosing Accumulator Types.

See also

• Reactor · Boundary event · Computation Graph