Struct AsyncAggregatorAgent 

pub struct AsyncAggregatorAgent { /* private fields */ }

An agent that aggregates events by correlation ID.

This agent waits for all specified event types to arrive for a given correlation ID before processing them together. This is useful for workflows where multiple independent operations must complete before a final action can be taken.

Examples

use mojentic::agents::AsyncAggregatorAgent;
use std::any::TypeId;

let agent = AsyncAggregatorAgent::new(vec![
    TypeId::of::<Event1>(),
    TypeId::of::<Event2>(),
]);

Implementations

impl AsyncAggregatorAgent

pub fn new(event_types_needed: Vec<TypeId>) -> Self

Create a new AsyncAggregatorAgent.

Arguments

• event_types_needed - Vector of TypeIds representing the event types that must be collected before processing

pub async fn wait_for_events( &self, correlation_id: &str, timeout: Option<Duration>, ) -> Result<Vec<Box<dyn Event>>>

Wait for all needed events for a specific correlation ID.

This method blocks until all required event types have been received for the given correlation ID, or until the timeout expires.

Arguments

• correlation_id - The correlation ID to wait for
• timeout - Optional timeout duration

Returns

Vector of all events collected for this correlation ID

pub async fn process_events( &self, _events: Vec<Box<dyn Event>>, ) -> Result<Vec<Box<dyn Event>>>

Process collected events.

This method is called when all needed event types have been collected. Override this in subclasses to implement custom processing logic.

Arguments

• events - All collected events for a correlation ID

Returns

Vector of new events to emit

Trait Implementations

impl BaseAsyncAgent for AsyncAggregatorAgent

fn receive_event_async<'life0, 'async_trait>( &'life0 self, event: Box<dyn Event>, ) -> Pin<Box<dyn Future<Output = Result<Vec<Box<dyn Event>>>> + Send + 'async_trait>>

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

Process an event asynchronously and return resulting events.