Struct AggregatorService 

pub struct AggregatorService { /* private fields */ }

Implementations

impl AggregatorService

pub fn new( config: AggregatorConfig, late_tx: Sender<Exchange>, 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.

Trait Implementations

impl Clone for AggregatorService

fn clone(&self) -> AggregatorService

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,