Struct AggregatorService 

pub struct AggregatorService { /* private fields */ }

Implementations

impl AggregatorService

pub fn new( config: AggregatorConfig, late_tx: Sender<AggregateEmission>, language_registry: SharedLanguageRegistry, route_cancel: CancellationToken, ) -> Self

Lifecycle invariant: sweep_cancel is initially seeded from the route’s cancellation token and wrapped in a swappable Arc<Mutex<...>> cell. StepLifecycle::start replaces it with a fresh token so the sweep respawns after a restart. Construction is runtime-free (no tokio::spawn here). The maintenance sweep task (TTL eviction when bucket_ttl is set, queue-depth sampling when queue_metrics is set) is spawned LAZILY on the first poll_ready and bound to the current sweep_cancel token via select!. The route owner MUST call shutdown to cancel it; Drop also aborts it as defense-in-depth.

pub fn with_queue_metrics( self, metrics: Arc<dyn MetricsCollector>, label: impl Into<String>, ) -> Self

Inject queue-depth reporting: the TTL-sweep maintenance pass publishes the buffered group count (open correlation buckets) as camel_queue_depth{queue = label}. label is the route-scoped closed-set identifier (aggregator:<route>).

pub fn config(&self) -> &AggregatorConfig

pub fn has_timeout(&self) -> bool

pub fn force_complete_all(&self)

pub fn release_unarmed_buckets(&self)

Consumer-exit release (bd rc-iioeq): discard buckets with NO armed timeout task, leave armed buckets untouched.

After a natural consumer exit no further exchanges arrive, so a bucket without an armed timeout task can never complete on its own — its buffered exchanges are released eagerly (the same discard semantics force_complete_all applies with force_completion_on_stop=false). A bucket WITH an armed timeout task is left alone: that task owns its completion and emits through the late channel when the timeout fires.

Arming is decided per bucket in call: when the max_timeout_tasks cap is reached, a timeout-configured bucket stays unarmed. Without this release such a bucket would be orphaned forever — the bucket_ttl sweep only runs inside call on the next exchange, which never arrives after consumer exit.

impl AggregatorService

pub async fn submit_with_claim( &self, exchange: Exchange, claim: Option<InFlightClaim>, ) -> AggregationReceipt

Submit one exchange with its drainclaim sidecar (drainclaim claim propagation) — the route loop’s aggregate entry point.

Behaviorally identical to the tower Service::call this is extracted from, except the claim travels with the exchange: a pending stash moves it into the bucket, a sync completion returns it (with the rest of the bucket’s claims) to the caller to hold across the aggregated exchange’s continuation, and every rejection path (bad correlation key, bucket caps) drops it — rejected means released.

Trait Implementations

impl Clone for AggregatorService

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for AggregatorService

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Drop for AggregatorService

fn drop(&mut self)

Executes the destructor for this type.

fn pin_drop(self: Pin<&mut Self>)

🔬This is a nightly-only experimental API. (pin_ergonomics)

Execute the destructor for this type, but different to Drop::drop, it requires self to be pinned.

impl Service<Exchange> for AggregatorService

type Response = Exchange

Responses given by the service.

type Error = CamelError

Errors produced by the service.

type Future = Pin<Box<dyn Future<Output = Result<Exchange, CamelError>> + Send>>

The future response value.

fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), CamelError>>

Returns Poll::Ready(Ok(())) when the service is able to process requests.

fn call(&mut self, exchange: Exchange) -> Self::Future

Process the request and return the response asynchronously.

impl StepLifecycle for AggregatorService

fn start<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<(), CamelError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Resets the sweep for a fresh lifecycle: replaces the cancellation token with a new one and aborts any existing sweep handle so the next poll_ready respawns the sweep bound to the new token.

fn name(&self) -> &'static str

Stable name for logging/diagnostics.

fn shutdown<'life0, 'async_trait>( &'life0 self, reason: StepShutdownReason, ) -> Pin<Box<dyn Future<Output = Result<(), CamelError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,