Aggregations | Pricefx Knowledge Base

Overview

Aggregations implement the Aggregator EIP — they collect multiple exchanges and combine them into one. Use them to batch records before loading, collect results from a split, or group data by a key.

IM pre-registers a set of aggregation strategy beans at startup. Reference them by bean name in the strategyRef attribute of the <aggregate> element.

Pre-Registered Aggregation Beans

Bean Name	Class	Description
`aggregateToList`	`net.pricefx.integration.aggregation.AggregateToList`	Aggregates exchange bodies into a `List`
`enrichPricingParameterIdAggregationStrategy`	`...EnrichPricingParameterIdAggregationStrategy`	Copies a header from new exchange to old exchange to preserve pricing parameter ID
`propagateFailureAggregationToListStrategy`	`...PropagateFailureAggregationToListStrategy`	Aggregates into a list and propagates exceptions to the new exchange
`recordsCountAggregation`	`...RecordsCountAggregation`	Aggregates record counts
`groupedBodyAggregationStrategy`	`GroupedBodyAggregationStrategy` (Camel)	Collects all bodies into a `List<Object>`
`groupedExchangeAggregationStrategy`	`GroupedExchangeAggregationStrategy` (Camel)	Collects all exchanges into a `List<Exchange>`
`useLatestAggregationStrategy`	`UseLatestAggregationStrategy` (Camel)	Keeps only the latest exchange
`useOriginalAggregationStrategy`	`UseOriginalAggregationStrategy` (Camel)	Keeps the original exchange
`stringAggregationStrategy`	`StringAggregationStrategy` (Camel)	Concatenates bodies into a single String
`zipAggregationStrategy`	`ZipAggregationStrategy` (Camel)	Aggregates all incoming messages into a ZIP file

Disable auto-registration:

integration.aggregations.enabled=false

Examples

Collect all records into a batch before loading

The most common pattern — collect N records from a split and load them as a batch:

XML

<routes xmlns="http://camel.apache.org/schema/spring">
    <route id="batchImport">
        <from uri="file:{{inbound.path}}"/>
        <to uri="pfx-csv:unmarshal?delimiter=,"/>
        <!-- split into individual records -->
        <split>
            <simple>${body}</simple>
            <!-- aggregate into batches of 500 -->
            <aggregate strategyRef="aggregateToList" completionSize="500">
                <correlationExpression>
                    <constant>true</constant>
                </correlationExpression>
                <to uri="pfx-api:loaddata?objectType=P&amp;mapper=productMapper"/>
            </aggregate>
        </split>
        <onCompletion onCompleteOnly="true">
            <to uri="pfx-api:internalCopy?label=Product"/>
        </onCompletion>
    </route>
</routes>

Aggregate with time-based completion

Collect records for up to 5 seconds, then process whatever was gathered:

XML

<aggregate strategyRef="aggregateToList" completionTimeout="5000">
    <correlationExpression>
        <simple>${header.batchKey}</simple>
    </correlationExpression>
    <to uri="pfx-api:loaddata?objectType=DM&amp;dsUniqueName=MyDS&amp;mapper=myMapper"/>
</aggregate>
<onCompletion onCompleteOnly="true">
    <to uri="pfx-api:flush?objectType=DM&amp;dsUniqueName=MyDS"/>
</onCompletion>

Important: Always wrap pfx-api:flush in <onCompletion onCompleteOnly="true">. This ensures flush runs only after the entire route succeeds — if loading fails mid-way, data won't be flushed in a broken state.

Collect all bodies into a ZIP file

Use zipAggregationStrategy to combine multiple file exchanges into a single ZIP:

XML

<routes xmlns="http://camel.apache.org/schema/spring">
    <route id="zipExports">
        <from uri="timer://zip?repeatCount=1"/>
        <!-- fetch multiple object types -->
        <to uri="pfx-api:fetch?objectType=P&amp;filter=allProducts"/>
        <to uri="pfx-csv:marshal"/>
        <setHeader name="CamelFileName"><constant>products.csv</constant></setHeader>

        <aggregate strategyRef="zipAggregationStrategy" completionSize="1">
            <correlationExpression><constant>zip</constant></correlationExpression>
            <setHeader name="CamelFileName"><constant>export.zip</constant></setHeader>
            <to uri="file:{{outbound.path}}"/>
        </aggregate>
    </route>
</routes>

Keep only the latest result (de-duplicate)

When multiple exchanges arrive for the same key and you only care about the latest value:

XML

<aggregate strategyRef="useLatestAggregationStrategy"
           completionTimeout="1000"
           completionSize="100">
    <correlationExpression>
        <simple>${body[sku]}</simple>
    </correlationExpression>
    <to uri="pfx-api:integrate?objectType=P&amp;mapper=productMapper"/>
</aggregate>

Custom aggregation strategy (Groovy class)

For logic that pre-registered beans don't cover, write your own in classes/:

Groovy

// classes/SumByCategory.groovy
import org.apache.camel.Exchange
import org.apache.camel.AggregationStrategy

class SumByCategory implements AggregationStrategy {

    @Override
    Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
        def row = newExchange.in.body as Map
        if (oldExchange == null) {
            newExchange.in.body = [(row.category): (row.amount as BigDecimal)]
            return newExchange
        }
        def sums = oldExchange.in.body as Map
        sums[row.category] = (sums[row.category] ?: 0) + (row.amount as BigDecimal)
        oldExchange.in.body = sums
        return oldExchange
    }
}

XML

<aggregate strategyRef="sumByCategory" completionSize="1000">
    <correlationExpression><constant>true</constant></correlationExpression>
    <to uri="direct:loadCategorySums"/>
</aggregate>

Completion Conditions

An aggregate must have at least one completion condition. Common options:

Attribute	Description
`completionSize`	Complete when N exchanges collected
`completionTimeout`	Complete after N milliseconds of inactivity
`completionInterval`	Complete every N milliseconds regardless of count
`completionPredicate`	Complete when a Camel expression is true