XML DSL (Aggregator)

One of the more common scenarios is the ability to combine different messages together based on certain relationships so they can then be processed as a composite. When we need to combine multiple messages together we will use the Aggregator pattern.

In the following examples, we will use a stateful filter (an Aggregator), to collect and persist individual messages until it receives a complete set of related messages. Once all messages have been received it will then publish the composite message.

XML DSL Simple Example

As an example, suppose we needed to aggregate responses from several different financial institutions to gather their quote for a specific loan request. Our strategy is to pick the institution with the best quote. We will need to implement some criteria to determine the cheapest loan and use our aggregation strategy to perform it.

<route>
        <from uri="jms:topic:loan:qoute"/>
        <aggregate strategyRef="bestQuote">
                <correlationExpression>
                        <header>loanId</header>
                </correlationExpression>
                <completionPredicate>
                        <simple>${header.CamelAggregatedSize} > 2</simple>
                </completionPredicate>
        </aggregate>
        <to uri="jsmith:bestQuote"/>
</route>

<bean id="bestQuote"
        class="com.mycompany.BestQuoteProcess"/>

NOTE: In this example, we are using simple language to declare the completion predicate. It is a basic language that only supports a primitive set of operators. The operation of ${header. CamelAggregatedSize} will get the header that contains the number of messages aggregated.

XML DSL Simple Example

However, if the completed predicate that is required is more complex, then we can invoke a method call ton a Bean so we can do a more complex evaluation using pure Java code:

<completionPredicate>
        <method bean="quoteService" method="isComplete"/>
</compledtionPrediacate>
public boolean isComplete(@Header(Exchange.AGGREGATED_SIZE)
        int count, String body) {
        return body.equals("STOP");
}

NOTE: Notice how, in this example, we are using a Bean Binding Parameter to get the aggregation size as a parameter. This frees us from having to search for it within the message.